108 lines
3.5 KiB
Python
108 lines
3.5 KiB
Python
"""
|
|
app/admin/auth/routes.py — Admin portal authentication.
|
|
Routes: /admin/login, /admin/logout
|
|
Brute-force lockout: 5 failures → locked for LOGIN_LOCKOUT_MINUTES.
|
|
"""
|
|
|
|
import logging
|
|
from datetime import datetime, timezone, timedelta
|
|
from flask import (
|
|
Blueprint, render_template, redirect, url_for,
|
|
flash, request, current_app, session,
|
|
)
|
|
from flask_login import login_user, logout_user, login_required, current_user
|
|
from flask_limiter import Limiter
|
|
from flask_limiter.util import get_remote_address
|
|
import bcrypt
|
|
|
|
from app.extensions import db, limiter
|
|
from app.models.platform import SystemUser, AuditLog
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
admin_auth_bp = Blueprint("admin_auth", __name__, url_prefix="/admin")
|
|
|
|
|
|
@admin_auth_bp.route("/login", methods=["GET", "POST"])
|
|
@limiter.limit("10 per minute")
|
|
def login():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for("admin_auth.dashboard_redirect"))
|
|
|
|
error = None
|
|
|
|
if request.method == "POST":
|
|
email = request.form.get("email", "").strip().lower()
|
|
password = request.form.get("password", "")
|
|
|
|
user = SystemUser.query.filter_by(email=email, is_active=True).first()
|
|
|
|
max_attempts = current_app.config.get("MAX_LOGIN_ATTEMPTS", 5)
|
|
lockout_minutes = current_app.config.get("LOGIN_LOCKOUT_MINUTES", 15)
|
|
|
|
if user and user.is_locked():
|
|
logger.warning("Admin login blocked — account locked: %s", email)
|
|
error = f"Account locked. Try again in {lockout_minutes} minutes."
|
|
elif user and bcrypt.checkpw(password.encode(), user.password_hash.encode()):
|
|
# Success — reset lockout counters
|
|
user.failed_login_attempts = 0
|
|
user.locked_until = None
|
|
user.last_login_at = datetime.now(timezone.utc)
|
|
db.session.commit()
|
|
|
|
login_user(user, remember=False)
|
|
session.permanent = True
|
|
|
|
AuditLog.log(
|
|
actor_id=user.id,
|
|
actor_type="system_user",
|
|
action="auth.login",
|
|
ip_address=request.remote_addr,
|
|
)
|
|
db.session.commit()
|
|
|
|
logger.info("Admin login success: %s", email)
|
|
next_url = request.args.get("next") or url_for("admin_auth.dashboard_redirect")
|
|
return redirect(next_url)
|
|
else:
|
|
# Failed attempt
|
|
if user:
|
|
user.failed_login_attempts += 1
|
|
if user.failed_login_attempts >= max_attempts:
|
|
user.locked_until = datetime.now(timezone.utc) + timedelta(
|
|
minutes=lockout_minutes
|
|
)
|
|
logger.warning(
|
|
"Admin account locked after %d failures: %s",
|
|
max_attempts, email,
|
|
)
|
|
db.session.commit()
|
|
|
|
logger.warning("Admin login failed: %s", email)
|
|
error = "Invalid email or password."
|
|
|
|
return render_template("admin/auth/login.html", error=error)
|
|
|
|
|
|
@admin_auth_bp.route("/logout")
|
|
@login_required
|
|
def logout():
|
|
AuditLog.log(
|
|
actor_id=current_user.id,
|
|
actor_type="system_user",
|
|
action="auth.logout",
|
|
ip_address=request.remote_addr,
|
|
)
|
|
db.session.commit()
|
|
logout_user()
|
|
session.clear()
|
|
logger.info("Admin logout")
|
|
return redirect(url_for("admin_auth.login"))
|
|
|
|
|
|
@admin_auth_bp.route("/")
|
|
@login_required
|
|
def dashboard_redirect():
|
|
# Will route to the analytics dashboard in Phase 2
|
|
return redirect(url_for("admin_auth.login"))
|