import html
from urllib.parse import quote
from fastapi import APIRouter, Depends, Form, HTTPException, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from app.auth import require_user
from app.db.audit import log_audit_event
from app.db.users import ROLES, count_enabled_admins, get_user, list_users, set_user_enabled, update_user_role
from app.gitea_provisioning import sync_gitea_user
from app.templates.layout import page
router = APIRouter()
def require_admin(user: dict) -> None:
if (user.get("role") or "").lower() != "admin":
raise HTTPException(status_code=403, detail="Only admins can manage users")
def render_role_options(selected: str) -> str:
options = []
for role in ROLES:
selected_attr = " selected" if selected == role else ""
options.append(f'{role} ')
return "".join(options)
def render_enabled_pill(enabled) -> str:
if enabled:
return 'enabled '
return 'disabled '
def render_sync_pill(status: str) -> str:
status = (status or "not_required").lower()
if status == "synced":
return 'synced '
if status == "error":
return 'error '
if status == "syncing":
return 'syncing '
return 'not required '
def render_role_overview() -> str:
role_rows = (
("admin", "Full administration, user management, role changes, disabling/enabling users, and admin-only operations."),
("developer", "Operational portal access for application and service workflows. No access to Administration / Users."),
("viewer", "Default Google role. Read-oriented portal access where routes allow it. No access to Administration / Users."),
)
rows = ""
for role, description in role_rows:
rows += f"""
{html.escape(role)}
{html.escape(description)}
"""
return f"""
"""
@router.get("/admin/users", response_class=HTMLResponse)
def users_page(request: Request, message: str = "", error: str = "", user=Depends(require_user)):
require_admin(user)
notice = ""
if message:
notice = f'{html.escape(message)}
'
if error:
notice = f'{html.escape(error)}
'
rows = ""
for managed_user in list_users():
user_id = int(managed_user.get("id"))
user_id_html = html.escape(str(user_id))
username = html.escape(managed_user.get("username", "") or "")
display_name = html.escape(managed_user.get("display_name", "") or "")
email = html.escape(managed_user.get("email", "") or "")
role = html.escape(managed_user.get("role", "") or "")
auth_provider = html.escape(managed_user.get("auth_provider", "") or "")
avatar_url = html.escape(managed_user.get("avatar_url", "") or "", quote=True)
avatar = f' ' if avatar_url else ""
created_at = html.escape(managed_user.get("created_at", "") or "")
last_login_at = html.escape(managed_user.get("last_login_at", "") or "")
enabled = bool(managed_user.get("is_enabled", True))
gitea_user_id = managed_user.get("gitea_user_id")
gitea_sync_status = managed_user.get("gitea_sync_status", "not_required") or "not_required"
gitea_sync_error = html.escape(managed_user.get("gitea_sync_error", "") or "")
gitea_synced_at = html.escape(managed_user.get("gitea_synced_at", "") or "")
gitea_details = f"ID {html.escape(str(gitea_user_id))}" if gitea_user_id else "No account"
if gitea_synced_at:
gitea_details += f"{gitea_synced_at} "
if gitea_sync_error:
gitea_details += f'{gitea_sync_error} '
toggle_label = "Disable" if enabled else "Enable"
toggle_action = "disable" if enabled else "enable"
rows += f"""
{user_id_html}
{username}
{display_name}
{avatar}
{email}
{auth_provider}
{render_enabled_pill(enabled)}
{render_sync_pill(gitea_sync_status)}{gitea_details}
{created_at}
{last_login_at}
"""
if not rows:
rows = 'No users found. '
return page(
"Users",
f"""
Users
Administration of portal users and roles.
{notice}
User List
ID
Username
Display name
Avatar
Email
Provider
Role
Status
Gitea sync
Created
Last login
Actions
{rows}
{render_role_overview()}
""",
user=user,
)
@router.post("/admin/users/{user_id}/role")
def update_role_action(user_id: int, role: str = Form(...), user=Depends(require_user)):
require_admin(user)
target = get_user(user_id)
if not target:
raise HTTPException(status_code=404, detail="User not found")
if role not in ROLES:
raise HTTPException(status_code=400, detail="Invalid role")
old_role = (target.get("role") or "").lower()
if old_role == "admin" and role != "admin" and count_enabled_admins(excluding_user_id=user_id) == 0:
return RedirectResponse(
url="/portal/admin/users?error=" + quote("Cannot remove admin role from the last enabled admin."),
status_code=303,
)
update_user_role(user_id, role)
target = get_user(user_id) or target
sync_result = sync_gitea_user(target)
if old_role != role:
log_audit_event(
user,
action="user.role.updated",
target_type="user",
target_id=user_id,
metadata={
"username": target.get("username"),
"old_role": old_role,
"new_role": role,
"gitea_sync_status": sync_result.get("status"),
},
)
if sync_result.get("status") == "error":
return RedirectResponse(
url="/portal/admin/users?error=" + quote(f"User role updated, but Gitea sync failed: {sync_result.get('error')}"),
status_code=303,
)
return RedirectResponse(url="/portal/admin/users?message=" + quote("User role updated and Gitea synchronized."), status_code=303)
@router.post("/admin/users/{user_id}/disable")
def disable_user_action(user_id: int, user=Depends(require_user)):
require_admin(user)
target = get_user(user_id)
if not target:
raise HTTPException(status_code=404, detail="User not found")
if (target.get("role") or "").lower() == "admin" and count_enabled_admins(excluding_user_id=user_id) == 0:
return RedirectResponse(
url="/portal/admin/users?error=" + quote("Cannot disable the last enabled admin."),
status_code=303,
)
set_user_enabled(user_id, False)
target = get_user(user_id) or target
sync_result = sync_gitea_user(target)
log_audit_event(
user,
action="user.disabled",
target_type="user",
target_id=user_id,
metadata={"username": target.get("username"), "gitea_sync_status": sync_result.get("status")},
)
if sync_result.get("status") == "error":
return RedirectResponse(
url="/portal/admin/users?error=" + quote(f"User disabled, but Gitea sync failed: {sync_result.get('error')}"),
status_code=303,
)
return RedirectResponse(url="/portal/admin/users?message=" + quote("User disabled and Gitea synchronized."), status_code=303)
@router.post("/admin/users/{user_id}/enable")
def enable_user_action(user_id: int, user=Depends(require_user)):
require_admin(user)
target = get_user(user_id)
if not target:
raise HTTPException(status_code=404, detail="User not found")
set_user_enabled(user_id, True)
target = get_user(user_id) or target
sync_result = sync_gitea_user(target)
log_audit_event(
user,
action="user.enabled",
target_type="user",
target_id=user_id,
metadata={"username": target.get("username"), "gitea_sync_status": sync_result.get("status")},
)
if sync_result.get("status") == "error":
return RedirectResponse(
url="/portal/admin/users?error=" + quote(f"User enabled, but Gitea sync failed: {sync_result.get('error')}"),
status_code=303,
)
return RedirectResponse(url="/portal/admin/users?message=" + quote("User enabled and Gitea synchronized."), status_code=303)
@router.post("/admin/users/{user_id}/sync-gitea")
def sync_gitea_user_action(user_id: int, user=Depends(require_user)):
require_admin(user)
target = get_user(user_id)
if not target:
raise HTTPException(status_code=404, detail="User not found")
sync_result = sync_gitea_user(target)
log_audit_event(
user,
action="user.gitea.synced",
target_type="user",
target_id=user_id,
metadata={"username": target.get("username"), **sync_result},
)
if sync_result.get("status") == "error":
return RedirectResponse(
url="/portal/admin/users?error=" + quote(f"Gitea sync failed: {sync_result.get('error')}"),
status_code=303,
)
return RedirectResponse(url="/portal/admin/users?message=" + quote("Gitea synchronized."), status_code=303)