""" app/admin/auth/routes.py Superadmin portal authentication: login, logout, password reset. Rate-limited. Brute-force lockout enforced. """ import logging import secrets from datetime import datetime, timezone, timedelta from flask import ( Blueprint, render_template, redirect, url_for, flash, request, current_app, ) from flask_login import login_user, logout_user, login_required, current_user from app.extensions import db, bcrypt, limiter, mail logger = logging.getLogger(__name__) admin_auth_bp = Blueprint("admin_auth", __name__, url_prefix="") @admin_auth_bp.route("/") def admin_index(): if current_user.is_authenticated: return redirect(url_for("admin_auth.dashboard_redirect")) return redirect(url_for("admin_auth.login")) @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": from app.models.platform import SystemUser, AuditLog email = request.form.get("email", "").strip().lower() password = request.form.get("password", "") user = SystemUser.query.filter_by(email=email).first() if not user or not user.is_active: logger.warning("Admin login failed: unknown/inactive email=%s", email) error = "Invalid credentials." elif user.is_locked(): logger.warning("Admin login: account locked email=%s", email) error = "Account is temporarily locked. Please try again later." elif not bcrypt.check_password_hash(user.password_hash, password): max_attempts = current_app.config.get("MAX_LOGIN_ATTEMPTS", 5) lockout_minutes = current_app.config.get("LOGIN_LOCKOUT_MINUTES", 15) user.record_failed_login(max_attempts, lockout_minutes) db.session.commit() error = "Invalid credentials." else: user.record_login() db.session.commit() AuditLog.log( actor_id=user.id, actor_type="system_user", action="login", ip_address=request.remote_addr, ) db.session.commit() login_user(user, remember=False) logger.info("Admin login success: id=%s email=%s", user.id, email) next_page = request.args.get("next") return redirect(next_page or url_for("admin_auth.dashboard_redirect")) return render_template("admin/auth/login.html", error=error) @admin_auth_bp.route("/dashboard") @login_required def dashboard_redirect(): return redirect(url_for("analytics.index")) @admin_auth_bp.route("/logout") @login_required def logout(): from app.models.platform import AuditLog user_id = current_user.id email = current_user.email AuditLog.log( actor_id=user_id, actor_type="system_user", action="logout", ip_address=request.remote_addr, ) db.session.commit() logout_user() logger.info("Admin logout: id=%s email=%s", user_id, email) flash("You have been logged out.", "info") return redirect(url_for("admin_auth.login")) @admin_auth_bp.route("/password-reset", methods=["GET", "POST"]) @limiter.limit("5 per hour") def password_reset_request(): if request.method == "POST": from app.models.platform import SystemUser email = request.form.get("email", "").strip().lower() user = SystemUser.query.filter_by(email=email, is_active=True).first() if user: token = secrets.token_urlsafe(32) user.password_reset_token = bcrypt.generate_password_hash(token).decode("utf-8") user.password_reset_expires_at = datetime.now(timezone.utc) + timedelta(hours=1) db.session.commit() reset_url = url_for("admin_auth.password_reset_confirm", token=token, _external=True) try: from flask_mail import Message msg = Message( subject="Admin Portal — Password Reset", recipients=[user.email], body=f"Reset link (valid 1 hour):\n\n{reset_url}", ) mail.send(msg) logger.info("Password reset email sent to admin: %s", email) except Exception as exc: logger.error("Failed to send password reset email: %s", exc) flash("If that email is registered, a reset link has been sent.", "info") return redirect(url_for("admin_auth.login")) return render_template("admin/auth/password_reset_request.html") @admin_auth_bp.route("/password-reset/", methods=["GET", "POST"]) @limiter.limit("10 per hour") def password_reset_confirm(token: str): from app.models.platform import SystemUser now = datetime.now(timezone.utc) candidates = SystemUser.query.filter( SystemUser.password_reset_token.isnot(None), SystemUser.password_reset_expires_at > now, ).all() user = None for candidate in candidates: if bcrypt.check_password_hash(candidate.password_reset_token, token): user = candidate break if not user: flash("This reset link is invalid or has expired.", "danger") return redirect(url_for("admin_auth.password_reset_request")) if request.method == "POST": new_password = request.form.get("password", "") confirm = request.form.get("confirm_password", "") from app.forms import validate_password_strength error = validate_password_strength(new_password, confirm) if error: return render_template("admin/auth/password_reset_confirm.html", error=error, token=token) user.password_hash = bcrypt.generate_password_hash(new_password).decode("utf-8") user.password_reset_token = None user.password_reset_expires_at = None user.failed_login_attempts = 0 user.locked_until = None db.session.commit() logger.info("Admin password reset completed: id=%s", user.id) flash("Password updated. Please log in.", "success") return redirect(url_for("admin_auth.login")) return render_template("admin/auth/password_reset_confirm.html", token=token)