Thomas Stallinger
576613ed0f
Ersetzt die handgestrickte bcrypt+Session-Auth durch fastapi-users (CookieTransport+JWTStrategy statt Starlette-Session, custom psycopg3-async-Adapter statt SQLAlchemy). Bestehende bcrypt-Hashes bleiben gültig und werden beim nächsten Login automatisch auf Argon2 angehoben (pwdlib-Default-PasswordHelper). ist_admin wird zu is_superuser, damit fastapi-users' eingebaute Semantik direkt nutzbar ist - require_admin/get_current_user bleiben als Wrapper unter gleichem Namen, alle Router-Dateien bis auf geraete.py (dict- zu attribut-Zugriff auf user.organization_id) unverändert. Lokal end-to-end gegen eine Wegwerf-Postgres-Instanz (podman) verifiziert: Migration, Bestandskonto-Login mit altem Passwort samt Argon2-Upgrade, Admin-Kontoanlage über beide Aufrufstellen (create_user.py und admin_kunden.py), 403 für Nicht-Admins, Logout. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
93 lines
3.3 KiB
Python
93 lines
3.3 KiB
Python
from fastapi import APIRouter, Depends, Form, Request
|
|
from fastapi.responses import RedirectResponse
|
|
|
|
from anode_client import anode_request
|
|
from auth import UserCreate, UserManager, get_user_manager, require_admin
|
|
from db import get_database_connection
|
|
from templating import templates
|
|
|
|
|
|
router = APIRouter(prefix="/admin/kunden")
|
|
|
|
|
|
def fetch_benutzer_fuer_organisation(organization_id: str) -> list[dict]:
|
|
with get_database_connection() as conn:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"SELECT email, is_superuser, created_at FROM benutzer WHERE organization_id = %s ORDER BY email",
|
|
(organization_id,),
|
|
)
|
|
return [
|
|
{"email": email, "is_superuser": is_superuser, "created_at": created_at}
|
|
for email, is_superuser, created_at in cur.fetchall()
|
|
]
|
|
|
|
|
|
@router.get("")
|
|
def index(request: Request, user: dict = Depends(require_admin)):
|
|
result = anode_request("GET", "/api/v1/organizations")
|
|
organizations = result.get("organizations", []) if result.get("success") else []
|
|
|
|
return templates.TemplateResponse(
|
|
request, "admin/kunden_index.html", {"user": user, "organizations": organizations}
|
|
)
|
|
|
|
|
|
@router.post("")
|
|
def organisation_anlegen(name: str = Form(...), user: dict = Depends(require_admin)):
|
|
anode_request("POST", "/api/v1/organizations", json={"name": name})
|
|
|
|
return RedirectResponse("/admin/kunden", status_code=303)
|
|
|
|
|
|
@router.get("/{organization_id}")
|
|
def detail(request: Request, organization_id: str, user: dict = Depends(require_admin)):
|
|
orgs_result = anode_request("GET", "/api/v1/organizations")
|
|
organizations = orgs_result.get("organizations", []) if orgs_result.get("success") else []
|
|
organization = next((o for o in organizations if o["id"] == organization_id), None)
|
|
|
|
codes_result = anode_request("GET", f"/api/v1/organizations/{organization_id}/activation-codes")
|
|
activation_codes = codes_result.get("activation_codes", []) if codes_result.get("success") else []
|
|
|
|
benutzer = fetch_benutzer_fuer_organisation(organization_id)
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/kunden_detail.html",
|
|
{
|
|
"user": user,
|
|
"organization": organization,
|
|
"organization_id": organization_id,
|
|
"activation_codes": activation_codes,
|
|
"benutzer": benutzer,
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/{organization_id}/activation-codes")
|
|
def aktivierungscode_anlegen(organization_id: str, user: dict = Depends(require_admin)):
|
|
anode_request("POST", f"/api/v1/organizations/{organization_id}/activation-codes")
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}", status_code=303)
|
|
|
|
|
|
@router.post("/{organization_id}/konten")
|
|
async def konto_anlegen(
|
|
organization_id: str,
|
|
email: str = Form(...),
|
|
password: str = Form(...),
|
|
is_superuser: str = Form(default=""),
|
|
user: dict = Depends(require_admin),
|
|
user_manager: UserManager = Depends(get_user_manager),
|
|
):
|
|
await user_manager.create(
|
|
UserCreate(
|
|
email=email,
|
|
password=password,
|
|
organization_id=organization_id,
|
|
is_superuser=is_superuser == "on",
|
|
)
|
|
)
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}", status_code=303)
|