112 lines
4.0 KiB
Python
112 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from flask import jsonify, g
|
|
|
|
from ..db import connect, configured
|
|
from ..keycloak import require_auth, require_portal_admin
|
|
from ..provisioning import provision_access_request
|
|
|
|
|
|
def register(app) -> None:
|
|
@app.route("/api/admin/access/requests", methods=["GET"])
|
|
@require_auth
|
|
def admin_list_requests() -> Any:
|
|
ok, resp = require_portal_admin()
|
|
if not ok:
|
|
return resp
|
|
if not configured():
|
|
return jsonify({"error": "server not configured"}), 503
|
|
|
|
try:
|
|
with connect() as conn:
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT request_code, username, contact_email, note, status, created_at
|
|
FROM access_requests
|
|
WHERE status = 'pending'
|
|
ORDER BY created_at ASC
|
|
LIMIT 200
|
|
"""
|
|
).fetchall()
|
|
except Exception:
|
|
return jsonify({"error": "failed to load requests"}), 502
|
|
|
|
output: list[dict[str, Any]] = []
|
|
for row in rows:
|
|
output.append(
|
|
{
|
|
"id": row["request_code"],
|
|
"username": row["username"],
|
|
"email": row.get("contact_email") or "",
|
|
"request_code": row["request_code"],
|
|
"created_at": (row.get("created_at").isoformat() if row.get("created_at") else ""),
|
|
"note": row.get("note") or "",
|
|
}
|
|
)
|
|
return jsonify({"requests": output})
|
|
|
|
@app.route("/api/admin/access/requests/<username>/approve", methods=["POST"])
|
|
@require_auth
|
|
def admin_approve_request(username: str) -> Any:
|
|
ok, resp = require_portal_admin()
|
|
if not ok:
|
|
return resp
|
|
if not configured():
|
|
return jsonify({"error": "server not configured"}), 503
|
|
|
|
decided_by = getattr(g, "keycloak_username", "") or ""
|
|
try:
|
|
with connect() as conn:
|
|
row = conn.execute(
|
|
"""
|
|
UPDATE access_requests
|
|
SET status = 'accounts_building', decided_at = NOW(), decided_by = %s
|
|
WHERE username = %s AND status = 'pending'
|
|
RETURNING request_code
|
|
""",
|
|
(decided_by or None, username),
|
|
).fetchone()
|
|
except Exception:
|
|
return jsonify({"error": "failed to approve request"}), 502
|
|
|
|
if not row:
|
|
return jsonify({"ok": True, "request_code": ""})
|
|
|
|
# Provision the account best-effort (Keycloak user + Mailu password + sync).
|
|
try:
|
|
provision_access_request(row["request_code"])
|
|
except Exception:
|
|
# Keep the request in accounts_building; status checks will surface it.
|
|
pass
|
|
return jsonify({"ok": True, "request_code": row["request_code"]})
|
|
|
|
@app.route("/api/admin/access/requests/<username>/deny", methods=["POST"])
|
|
@require_auth
|
|
def admin_deny_request(username: str) -> Any:
|
|
ok, resp = require_portal_admin()
|
|
if not ok:
|
|
return resp
|
|
if not configured():
|
|
return jsonify({"error": "server not configured"}), 503
|
|
|
|
decided_by = getattr(g, "keycloak_username", "") or ""
|
|
try:
|
|
with connect() as conn:
|
|
row = conn.execute(
|
|
"""
|
|
UPDATE access_requests
|
|
SET status = 'denied', decided_at = NOW(), decided_by = %s
|
|
WHERE username = %s AND status = 'pending'
|
|
RETURNING request_code
|
|
""",
|
|
(decided_by or None, username),
|
|
).fetchone()
|
|
except Exception:
|
|
return jsonify({"error": "failed to deny request"}), 502
|
|
|
|
if not row:
|
|
return jsonify({"ok": True, "request_code": ""})
|
|
return jsonify({"ok": True, "request_code": row["request_code"]})
|