JiriUhlir
460 lines
20 KiB
Python
460 lines
20 KiB
Python
import html
|
||
import re
|
||
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,
|
||
delete_user,
|
||
get_user,
|
||
list_users,
|
||
set_gitea_username,
|
||
set_user_enabled,
|
||
update_user_role,
|
||
)
|
||
from app.gitea_provisioning import apply_gitea_credentials, sync_gitea_user
|
||
from app.logging_config import get_logger
|
||
from app.templates.layout import page
|
||
|
||
router = APIRouter()
|
||
logger = get_logger(__name__)
|
||
|
||
# Gitea handle: začíná alfanumerickým znakem, dál povolen . _ -; max 40 znaků.
|
||
GITEA_USERNAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,39}$")
|
||
MIN_GITEA_PASSWORD_LEN = 8
|
||
|
||
|
||
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'<option value="{role}"{selected_attr}>{role}</option>')
|
||
return "".join(options)
|
||
|
||
|
||
def render_enabled_pill(enabled) -> str:
|
||
if enabled:
|
||
return '<span class="pill pill-success">enabled</span>'
|
||
return '<span class="pill pill-muted">disabled</span>'
|
||
|
||
|
||
def render_sync_pill(status: str) -> str:
|
||
status = (status or "not_required").lower()
|
||
if status == "synced":
|
||
return '<span class="pill pill-success">synced</span>'
|
||
if status == "error":
|
||
return '<span class="pill pill-danger">error</span>'
|
||
if status == "syncing":
|
||
return '<span class="pill pill-muted">syncing</span>'
|
||
return '<span class="pill pill-muted">not required</span>'
|
||
|
||
|
||
def render_role_overview() -> str:
|
||
role_rows = (
|
||
("admin", "Plný přístup: vše co developer + mazání služeb, správa záloh a jejich obnova, správa uživatelů a rolí, admin-only operace."),
|
||
("developer", "Vidí běžící služby a dokumentaci jako viewer + provozní workflow služeb: Git/klonování, nasazení, proměnné, prostředky, metadata. Nemůže mazat služby ani do Administrace / Uživatelů."),
|
||
("viewer", "Pouze pro čtení: vidí jaké služby běží a jejich dokumentaci (Swagger). Bez Gitu, úprav, nasazení a mazání. Bez přístupu do Administrace / Uživatelů."),
|
||
("guest", "Výchozí role nového účtu po prvním přihlášení. Nemá přístup nikam – vidí jen výzvu, aby požádal administrátora o přidělení práv."),
|
||
)
|
||
rows = ""
|
||
for role, description in role_rows:
|
||
rows += f"""
|
||
<tr>
|
||
<td><span class="pill pill-muted">{html.escape(role)}</span></td>
|
||
<td>{html.escape(description)}</td>
|
||
</tr>
|
||
"""
|
||
return f"""
|
||
<div class="card">
|
||
<h2>Role Permissions</h2>
|
||
<table>
|
||
<tr>
|
||
<th>Role</th>
|
||
<th>Access</th>
|
||
</tr>
|
||
{rows}
|
||
</table>
|
||
</div>
|
||
"""
|
||
|
||
|
||
@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'<p class="alert">{html.escape(message)}</p>'
|
||
if error:
|
||
notice = f'<p class="alert alert-danger">{html.escape(error)}</p>'
|
||
|
||
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'<img class="user-avatar" src="{avatar_url}" alt="">' 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 = (managed_user.get("gitea_sync_error", "") or "").strip()
|
||
gitea_synced_at = (managed_user.get("gitea_synced_at", "") or "").strip()
|
||
gitea_username_value = (managed_user.get("gitea_username") or "").strip()
|
||
current_gitea_username = gitea_username_value or (f"portal-{user_id}" if gitea_user_id else "")
|
||
toggle_label = "Vypnout" if enabled else "Zapnout"
|
||
toggle_action = "disable" if enabled else "enable"
|
||
|
||
# Gitea: pill (synced/…) + (i) s detaily v tooltipu (ID, jméno, čas synchronizace, chyba).
|
||
detail_parts = []
|
||
if gitea_user_id:
|
||
detail_parts.append(f"ID {gitea_user_id}")
|
||
if current_gitea_username:
|
||
detail_parts.append(f"@{current_gitea_username}")
|
||
if gitea_synced_at:
|
||
detail_parts.append(f"synced {gitea_synced_at}")
|
||
if gitea_sync_error:
|
||
detail_parts.append(f"chyba: {gitea_sync_error}")
|
||
tooltip = html.escape(" · ".join(detail_parts) or "Bez Gitea účtu", quote=True)
|
||
gitea_info = (
|
||
f'<span class="info-dot" tabindex="0" role="note" aria-label="{tooltip}" data-tooltip="{tooltip}">'
|
||
'<i class="fa-solid fa-circle-info" aria-hidden="true"></i></span>'
|
||
)
|
||
|
||
avatar_block = avatar or '<span class="user-avatar user-avatar-empty"><i class="fa-solid fa-user" aria-hidden="true"></i></span>'
|
||
provider_label = auth_provider or "—"
|
||
email_line = f'<br><span class="muted">{email}</span>' if email else ""
|
||
|
||
# Nastavení Gitea jména a hesla – v modálu (nezarovnává řádek a neořízne se scrollem tabulky).
|
||
gitea_modal_id = f"gitea-modal-{user_id_html}"
|
||
gitea_modal = f"""
|
||
<dialog class="resource-modal" id="{gitea_modal_id}">
|
||
<form method="post" action="/portal/admin/users/{user_id_html}/gitea-credentials">
|
||
<div class="modal-header">
|
||
<h3>Gitea účet – {display_name or username}</h3>
|
||
<button type="button" class="icon-button" onclick="closeResourceModal('{gitea_modal_id}')" aria-label="Zavřít">×</button>
|
||
</div>
|
||
<p class="muted">Nastaví uživatelské jméno a/nebo heslo přímo v Gitea. Heslo se nikam neukládá.</p>
|
||
<label>Gitea uživatelské jméno</label>
|
||
<input name="gitea_username" value="{html.escape(current_gitea_username, quote=True)}" placeholder="portal-{user_id_html}" autocomplete="off">
|
||
<label>Nové heslo</label>
|
||
<input type="password" name="gitea_password" placeholder="prázdné = beze změny" autocomplete="new-password">
|
||
<div class="modal-actions">
|
||
<button type="button" class="btn btn-secondary" onclick="closeResourceModal('{gitea_modal_id}')"><i class="fa-solid fa-xmark" aria-hidden="true"></i> Zrušit</button>
|
||
<button type="submit"><i class="fa-solid fa-key" aria-hidden="true"></i> Uložit</button>
|
||
</div>
|
||
</form>
|
||
</dialog>
|
||
"""
|
||
|
||
rows += f"""
|
||
<tr class="user-row">
|
||
<td class="service-name-cell">
|
||
<div class="service-title">
|
||
{avatar_block}
|
||
<div>
|
||
<strong>{display_name or username}</strong><br>
|
||
<span class="muted">@{username}</span>{email_line}
|
||
</div>
|
||
</div>
|
||
</td>
|
||
<td>
|
||
<form method="post" action="/portal/admin/users/{user_id_html}/role" class="inline-form">
|
||
<select name="role">{render_role_options(role)}</select>
|
||
<button type="submit" class="icon-action" title="Uložit roli" aria-label="Uložit roli"><i class="fa-solid fa-floppy-disk" aria-hidden="true"></i></button>
|
||
</form>
|
||
</td>
|
||
<td>{render_enabled_pill(enabled)}<br><span class="muted">{provider_label}</span></td>
|
||
<td>{render_sync_pill(gitea_sync_status)} {gitea_info}</td>
|
||
<td class="muted-cell">
|
||
<span class="muted">vytvořeno</span><br>{created_at or "—"}<br>
|
||
<span class="muted">poslední přihlášení</span><br>{last_login_at or "—"}
|
||
</td>
|
||
<td class="actions-cell service-actions">
|
||
<form method="post" action="/portal/admin/users/{user_id_html}/{toggle_action}" class="inline-form">
|
||
<button type="submit" class="icon-action" title="{toggle_label}" aria-label="{toggle_label}"><i class="fa-solid fa-power-off" aria-hidden="true"></i></button>
|
||
</form>
|
||
<form method="post" action="/portal/admin/users/{user_id_html}/sync-gitea" class="inline-form">
|
||
<button type="submit" class="icon-action" title="Synchronizovat Gitea" aria-label="Synchronizovat Gitea"><i class="fa-solid fa-rotate" aria-hidden="true"></i></button>
|
||
</form>
|
||
<button type="button" class="icon-action" onclick="return openResourceModal('{gitea_modal_id}')" title="Gitea jméno a heslo" aria-label="Gitea jméno a heslo"><i class="fa-solid fa-code-branch" aria-hidden="true"></i></button>
|
||
<form method="post" action="/portal/admin/users/{user_id_html}/delete" class="inline-form" onsubmit="return confirm('Smazat uživatele {username}? Odstraní se portálový účet.');">
|
||
<button type="submit" class="icon-action icon-action-danger" title="Smazat" aria-label="Smazat"><i class="fa-solid fa-trash" aria-hidden="true"></i></button>
|
||
</form>
|
||
{gitea_modal}
|
||
</td>
|
||
</tr>
|
||
"""
|
||
|
||
if not rows:
|
||
rows = '<tr><td colspan="6">Žádní uživatelé.</td></tr>'
|
||
|
||
return page(
|
||
"Users",
|
||
f"""
|
||
<div class="card">
|
||
<h2><i class="fa-solid fa-users" aria-hidden="true"></i> Users</h2>
|
||
<p class="muted">Administration of portal users and roles.</p>
|
||
{notice}
|
||
</div>
|
||
|
||
<div class="card">
|
||
<h2>Seznam uživatelů</h2>
|
||
<table>
|
||
<tr>
|
||
<th>Uživatel</th>
|
||
<th>Role</th>
|
||
<th>Stav</th>
|
||
<th>Gitea</th>
|
||
<th>Aktivita</th>
|
||
<th>Akce</th>
|
||
</tr>
|
||
{rows}
|
||
</table>
|
||
</div>
|
||
{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)
|
||
|
||
|
||
@router.post("/admin/users/{user_id}/gitea-credentials")
|
||
def update_gitea_credentials_action(
|
||
user_id: int,
|
||
gitea_username: str = Form(""),
|
||
gitea_password: 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")
|
||
|
||
new_username = (gitea_username or "").strip()
|
||
new_password = gitea_password or ""
|
||
|
||
if not new_username and not new_password:
|
||
return RedirectResponse(
|
||
url="/portal/admin/users?error=" + quote("Zadejte Gitea jméno nebo nové heslo."),
|
||
status_code=303,
|
||
)
|
||
if new_username and not GITEA_USERNAME_RE.match(new_username):
|
||
return RedirectResponse(
|
||
url="/portal/admin/users?error=" + quote("Neplatné Gitea jméno (povoleno A-Z, 0-9, . _ -, max 40 znaků)."),
|
||
status_code=303,
|
||
)
|
||
if new_password and len(new_password) < MIN_GITEA_PASSWORD_LEN:
|
||
return RedirectResponse(
|
||
url="/portal/admin/users?error=" + quote(f"Heslo musí mít alespoň {MIN_GITEA_PASSWORD_LEN} znaků."),
|
||
status_code=303,
|
||
)
|
||
|
||
result = apply_gitea_credentials(target, new_username, new_password)
|
||
applied = result.get("applied") or []
|
||
errors = result.get("errors") or []
|
||
|
||
if result.get("renamed"):
|
||
set_gitea_username(user_id, result.get("gitea_username") or new_username)
|
||
|
||
log_audit_event(
|
||
user,
|
||
action="user.gitea.credentials_updated",
|
||
target_type="user",
|
||
target_id=user_id,
|
||
metadata={
|
||
"username": target.get("username"),
|
||
"gitea_username": result.get("gitea_username"),
|
||
"applied": applied,
|
||
"errors": errors, # heslo se nikdy nezaznamenává
|
||
},
|
||
)
|
||
|
||
status = result.get("status")
|
||
if status == "error":
|
||
detail = result.get("error") or "; ".join(errors) or "neznámá chyba"
|
||
logger.error("Změna Gitea údajů selhala pro uživatele id=%s: %s", user_id, detail)
|
||
return RedirectResponse(url="/portal/admin/users?error=" + quote(f"Gitea: {detail}"), status_code=303)
|
||
|
||
if errors:
|
||
msg = f"Gitea částečně aktualizováno ({', '.join(applied)}). Nepovedlo se: {'; '.join(errors)}"
|
||
logger.warning("Gitea údaje částečně pro uživatele id=%s: applied=%s errors=%s", user_id, applied, errors)
|
||
return RedirectResponse(url="/portal/admin/users?error=" + quote(msg), status_code=303)
|
||
|
||
logger.info("Gitea údaje aktualizovány pro uživatele id=%s: %s", user_id, applied)
|
||
return RedirectResponse(
|
||
url="/portal/admin/users?message=" + quote(f"Gitea aktualizováno: {', '.join(applied) or 'beze změny'}."),
|
||
status_code=303,
|
||
)
|
||
|
||
|
||
@router.post("/admin/users/{user_id}/delete")
|
||
def delete_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 int(user.get("id")) == user_id:
|
||
return RedirectResponse(
|
||
url="/portal/admin/users?error=" + quote("Cannot delete your own user account."),
|
||
status_code=303,
|
||
)
|
||
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 delete 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)
|
||
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,
|
||
)
|
||
|
||
delete_user(user_id)
|
||
log_audit_event(
|
||
user,
|
||
action="user.deleted",
|
||
target_type="user",
|
||
target_id=user_id,
|
||
metadata={"username": target.get("username"), "role": target.get("role"), "gitea_sync_status": sync_result.get("status")},
|
||
)
|
||
return RedirectResponse(url="/portal/admin/users?message=" + quote("User deleted."), status_code=303)
|