05/06 Phase 2: updated and added new files

app/admin/system_users/routes.py

@@ -1,7 +1,143 @@
 """
 app/admin/system_users/routes.py
-Phase 2 implementation.
+System user management: create, edit, deactivate, force password reset.
+All actions logged to AuditLog. Superadmin only.
 """
-from flask import Blueprint
+
+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("/<int:user_id>/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("/<int:user_id>/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("/<int:user_id>/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"))