""" app/admin/system_users/routes.py System user management: create, edit, deactivate, force password reset. All actions logged to AuditLog. Superadmin only. """ import logging import secrets from datetime import datetime, timezone, timedelta from flask import ( Blueprint, render_template, redirect, url_for, flash, request, ) from flask_login import login_required, current_user from app.extensions import db, bcrypt, mail from app.models.platform import SystemUser from app.admin.utils import superadmin_required, log_admin_action, model_to_dict logger = logging.getLogger(__name__) system_users_bp = Blueprint("system_users", __name__, url_prefix="/system-users") _AUDIT_FIELDS = ["email", "name", "role", "is_active"] @system_users_bp.route("/") @login_required @superadmin_required def index(): users = SystemUser.query.order_by(SystemUser.created_at.desc()).all() return render_template("admin/system_users/index.html", users=users) @system_users_bp.route("/new", methods=["GET", "POST"]) @login_required @superadmin_required def create(): if request.method == "POST": email = request.form.get("email", "").strip().lower() name = request.form.get("name", "").strip() role = request.form.get("role", "superadmin").strip() password = request.form.get("password", "") confirm = request.form.get("confirm_password", "") from app.forms import validate_password_strength error = validate_password_strength(password, confirm) if error: return render_template("admin/system_users/form.html", error=error, mode="create") if SystemUser.query.filter_by(email=email).first(): return render_template( "admin/system_users/form.html", error="A user with that email already exists.", mode="create", ) user = SystemUser( email=email, name=name, role=role, password_hash=bcrypt.generate_password_hash(password).decode("utf-8"), is_active=True, ) db.session.add(user) db.session.flush() log_admin_action("system_user.create", "system_user", user.id, after=model_to_dict(user, _AUDIT_FIELDS)) db.session.commit() logger.info("System user created: id=%s email=%s", user.id, email) flash(f"System user \'{email}\' created.", "success") return redirect(url_for("system_users.index")) return render_template("admin/system_users/form.html", mode="create") @system_users_bp.route("//edit", methods=["GET", "POST"]) @login_required @superadmin_required def edit(user_id): user = SystemUser.query.get_or_404(user_id) before = model_to_dict(user, _AUDIT_FIELDS) if request.method == "POST": user.name = request.form.get("name", user.name).strip() user.role = request.form.get("role", user.role).strip() user.is_active = request.form.get("is_active") == "1" log_admin_action("system_user.edit", "system_user", user.id, before=before, after=model_to_dict(user, _AUDIT_FIELDS)) db.session.commit() logger.info("System user edited: id=%s", user.id) flash(f"User \'{user.email}\' updated.", "success") return redirect(url_for("system_users.index")) return render_template("admin/system_users/form.html", mode="edit", user=user) @system_users_bp.route("//toggle-active", methods=["POST"]) @login_required @superadmin_required def toggle_active(user_id): user = SystemUser.query.get_or_404(user_id) if user.id == current_user.id: flash("You cannot deactivate your own account.", "danger") return redirect(url_for("system_users.index")) before = model_to_dict(user, _AUDIT_FIELDS) user.is_active = not user.is_active action = "system_user.activate" if user.is_active else "system_user.deactivate" log_admin_action(action, "system_user", user.id, before=before, after=model_to_dict(user, _AUDIT_FIELDS)) db.session.commit() status = "activated" if user.is_active else "deactivated" logger.info("System user %s: id=%s", status, user.id) flash(f"User \'{user.email}\' {status}.", "success") return redirect(url_for("system_users.index")) @system_users_bp.route("//force-password-reset", methods=["POST"]) @login_required @superadmin_required def force_password_reset(user_id): user = SystemUser.query.get_or_404(user_id) 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=24) user.failed_login_attempts = 0 user.locked_until = None log_admin_action("system_user.force_password_reset", "system_user", user.id) 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 by Administrator", recipients=[user.email], body=f"A password reset has been initiated.\n\nReset link (24 hours):\n\n{reset_url}", ) mail.send(msg) flash(f"Password reset email sent to \'{user.email}\'.", "success") except Exception as exc: logger.error("Force reset email failed: %s", exc) flash(f"Token generated but email failed. Reset URL: {reset_url}", "warning") return redirect(url_for("system_users.index"))