"""Routes: admin users""" from datetime import datetime from flask import abort, flash, redirect, render_template, request, session, url_for from werkzeug.security import generate_password_hash from nexora.core import app from nexora.database.connection import execute, query from nexora.utils.validators import sanitize_field, validate_username_format from nexora.security.permissions import ( PERMISSION_KEYS, PERMISSION_LABELS, PERMISSION_SECTIONS, admin_required, forbidden_response, has_permission, log_role_audit, permission_required, ) @app.route("/admin/users", methods=["GET", "POST"]) @admin_required @permission_required("users_view") def admin_users(): roles = query("SELECT * FROM roles ORDER BY name") if request.method == "POST": if not has_permission("users_create"): return forbidden_response() username = sanitize_field(request.form.get("username", ""), 60) password = request.form.get("password", "") role_id = request.form.get("role_id", type=int) full_name = sanitize_field(request.form.get("full_name", ""), 120) if not validate_username_format(username): flash("Username must be 3-60 characters: letters, numbers, dots, hyphens, or underscores.", "danger") elif len(password) < 8: flash("Password must be at least 8 characters.", "danger") elif not role_id: flash("A role must be selected.", "danger") elif query("SELECT id FROM admin_users WHERE username = ?", (username,), one=True): flash("That username already exists.", "danger") else: execute("INSERT INTO admin_users (username, password_hash, full_name, role_id, active, created_at) VALUES (?, ?, ?, ?, 1, ?)", (username, generate_password_hash(password), full_name, role_id, datetime.now().isoformat(timespec="seconds"))) log_role_audit("user_created", username, f"role_id={role_id}") flash("Staff user created.", "success") return redirect(url_for("admin_users")) users = query("SELECT admin_users.*, roles.name AS role_name FROM admin_users LEFT JOIN roles ON roles.id = admin_users.role_id ORDER BY admin_users.id DESC") user_permissions = {} for u in users: user_permissions[u["id"]] = [row["permission_key"] for row in query("SELECT permission_key FROM user_permissions WHERE user_id = ?", (u["id"],))] return render_template("admin_users.html", users=users, roles=roles, user_permissions=user_permissions, permission_sections=PERMISSION_SECTIONS, permission_labels=PERMISSION_LABELS) @app.route("/admin/users//toggle", methods=["POST"]) @admin_required @permission_required("users_update") def toggle_admin_user(user_id): user = query("SELECT username FROM admin_users WHERE id = ?", (user_id,), one=True) if not user: abort(404) execute("UPDATE admin_users SET active = 1 - active WHERE id = ?", (user_id,)) log_role_audit("user_status_changed", user["username"]) return redirect(url_for("admin_users")) @app.route("/admin/users//edit", methods=["POST"]) @admin_required @permission_required("users_update") def edit_admin_user(user_id): user = query("SELECT username FROM admin_users WHERE id = ?", (user_id,), one=True) if not user: abort(404) username = request.form.get("username", "").strip() full_name = request.form.get("full_name", "").strip() role_id = request.form.get("role_id", type=int) active = 1 if request.form.get("active") else 0 if not username or not role_id: flash("Username and role are required.", "danger") return redirect(url_for("admin_users")) if query("SELECT id FROM admin_users WHERE username = ? AND id != ?", (username, user_id), one=True): flash("That username already exists.", "danger") return redirect(url_for("admin_users")) execute( "UPDATE admin_users SET username = ?, full_name = ?, role_id = ?, active = ? WHERE id = ?", (username, full_name, role_id, active, user_id), ) log_role_audit("user_updated", username, f"role_id={role_id}") flash("Staff user updated.", "success") return redirect(url_for("admin_users")) @app.route("/admin/users//delete", methods=["POST"]) @admin_required @permission_required("users_delete") def delete_admin_user(user_id): user = query("SELECT username FROM admin_users WHERE id = ?", (user_id,), one=True) if not user: abort(404) if session.get("admin_username") == user["username"]: flash("You cannot delete the account you are currently logged in with.", "danger") return redirect(url_for("admin_users")) execute("DELETE FROM admin_users WHERE id = ?", (user_id,)) log_role_audit("user_deleted", user["username"]) flash(f'Staff user "{user["username"]}" deleted.', "success") return redirect(url_for("admin_users")) @app.route("/admin/users//reset-password", methods=["POST"]) @admin_required @permission_required("users_update") def reset_user_password(user_id): user = query("SELECT username FROM admin_users WHERE id = ?", (user_id,), one=True) if not user: flash("User not found.", "danger") return redirect(url_for("admin_users")) new_password = request.form.get("new_password", "") if len(new_password) < 8: flash("Password must be at least 8 characters.", "danger") return redirect(url_for("admin_users")) execute("UPDATE admin_users SET password_hash = ? WHERE id = ?", (generate_password_hash(new_password), user_id)) flash(f'Password reset for "{user["username"]}" done.', "success") return redirect(url_for("admin_users")) @app.route("/admin/users//permissions", methods=["POST"]) @admin_required @permission_required("users_update") def save_user_permissions(user_id): user = query("SELECT username FROM admin_users WHERE id = ?", (user_id,), one=True) if not user: flash("User not found.", "danger") return redirect(url_for("admin_users")) selected = [key for key in request.form.getlist("permissions") if key in PERMISSION_KEYS] execute("DELETE FROM user_permissions WHERE user_id = ?", (user_id,)) for key in selected: execute("INSERT INTO user_permissions (user_id, permission_key) VALUES (?, ?)", (user_id, key)) log_role_audit("user_permissions_updated", user["username"], f"direct_perms={','.join(selected)}") flash(f'Permissions updated for "{user["username"]}".', "success") return redirect(url_for("admin_users"))