05/06 admin login issues
This commit is contained in:
+118
-54
@@ -1,26 +1,31 @@
|
||||
"""
|
||||
app/admin/auth/routes.py — Admin portal authentication.
|
||||
Routes: /admin/login, /admin/logout
|
||||
Brute-force lockout: 5 failures → locked for LOGIN_LOCKOUT_MINUTES.
|
||||
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, session,
|
||||
flash, request, current_app,
|
||||
)
|
||||
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
|
||||
from app.extensions import db, bcrypt, limiter, mail
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
admin_auth_bp = Blueprint("admin_auth", __name__, url_prefix="/admin")
|
||||
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"])
|
||||
@@ -30,78 +35,137 @@ def login():
|
||||
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, is_active=True).first()
|
||||
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)
|
||||
|
||||
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)
|
||||
user.record_failed_login(max_attempts, lockout_minutes)
|
||||
db.session.commit()
|
||||
error = "Invalid credentials."
|
||||
else:
|
||||
user.record_login()
|
||||
db.session.commit()
|
||||
|
||||
login_user(user, remember=False)
|
||||
session.permanent = True
|
||||
|
||||
AuditLog.log(
|
||||
actor_id=user.id,
|
||||
actor_type="system_user",
|
||||
action="auth.login",
|
||||
action="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."
|
||||
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():
|
||||
# Phase 2: redirect to analytics/tenants dashboard
|
||||
flash("Welcome to the Admin Portal. Feature modules coming in Phase 2.", "info")
|
||||
return render_template("admin/auth/login.html", error=None)
|
||||
|
||||
|
||||
@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=current_user.id,
|
||||
actor_id=user_id,
|
||||
actor_type="system_user",
|
||||
action="auth.logout",
|
||||
action="logout",
|
||||
ip_address=request.remote_addr,
|
||||
)
|
||||
db.session.commit()
|
||||
logout_user()
|
||||
session.clear()
|
||||
logger.info("Admin logout")
|
||||
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("/")
|
||||
@login_required
|
||||
def dashboard_redirect():
|
||||
# Will route to the analytics dashboard in Phase 2
|
||||
@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/<token>", 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)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
{% extends "layouts/base.html" %}
|
||||
{% extends "admin/layouts/base.html" %}
|
||||
{% block title %}Admin Login{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
{% extends "layouts/base.html" %}
|
||||
{% extends "admin/layouts/base.html" %}
|
||||
{% block title %}Set New Password{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
{% extends "layouts/base.html" %}
|
||||
{% extends "admin/layouts/base.html" %}
|
||||
{% block title %}Reset Password{% endblock %}
|
||||
|
||||
{% block content %}
|
||||
|
||||
@@ -13,7 +13,7 @@
|
||||
{% if current_user.is_authenticated %}
|
||||
<nav class="navbar navbar-expand-lg navbar-dark bg-dark">
|
||||
<div class="container-fluid">
|
||||
<a class="navbar-brand fw-bold" href="{{ url_for('tenants.index') }}">
|
||||
<a class="navbar-brand fw-bold" href="#">
|
||||
<i class="bi bi-shield-lock-fill me-2"></i>Admin Portal
|
||||
</a>
|
||||
<div class="navbar-nav ms-auto">
|
||||
@@ -32,37 +32,37 @@
|
||||
<ul class="nav flex-column">
|
||||
<li class="nav-item">
|
||||
<a class="nav-link {% if request.endpoint and request.endpoint.startswith('tenants') %}active fw-bold{% endif %}"
|
||||
href="{{ url_for('tenants.index') }}">
|
||||
href="#">
|
||||
<i class="bi bi-building me-2"></i>Tenants
|
||||
</a>
|
||||
</li>
|
||||
<li class="nav-item">
|
||||
<a class="nav-link {% if request.endpoint and request.endpoint.startswith('system_users') %}active fw-bold{% endif %}"
|
||||
href="{{ url_for('system_users.index') }}">
|
||||
href="#">
|
||||
<i class="bi bi-people me-2"></i>System Users
|
||||
</a>
|
||||
</li>
|
||||
<li class="nav-item">
|
||||
<a class="nav-link {% if request.endpoint and request.endpoint.startswith('plans') %}active fw-bold{% endif %}"
|
||||
href="{{ url_for('plans.index') }}">
|
||||
href="#">
|
||||
<i class="bi bi-card-list me-2"></i>Plans
|
||||
</a>
|
||||
</li>
|
||||
<li class="nav-item">
|
||||
<a class="nav-link {% if request.endpoint and request.endpoint.startswith('billing') %}active fw-bold{% endif %}"
|
||||
href="{{ url_for('billing.index') }}">
|
||||
href="#">
|
||||
<i class="bi bi-receipt me-2"></i>Billing
|
||||
</a>
|
||||
</li>
|
||||
<li class="nav-item">
|
||||
<a class="nav-link {% if request.endpoint and request.endpoint.startswith('audit_log') %}active fw-bold{% endif %}"
|
||||
href="{{ url_for('audit_log.index') }}">
|
||||
href="#">
|
||||
<i class="bi bi-journal-text me-2"></i>Audit Log
|
||||
</a>
|
||||
</li>
|
||||
<li class="nav-item">
|
||||
<a class="nav-link {% if request.endpoint and request.endpoint.startswith('analytics') %}active fw-bold{% endif %}"
|
||||
href="{{ url_for('analytics.index') }}">
|
||||
href="#">
|
||||
<i class="bi bi-bar-chart me-2"></i>Analytics
|
||||
</a>
|
||||
</li>
|
||||
|
||||
Reference in New Issue
Block a user