Thomas Stallinger
cb820b9790
Sicherheits-Review 25.08.2026 vor geplanter öffentlicher Erreichbarkeit über Pangolin (bisher nur Pangolin-interne Auth vorgeschaltet): - auth.py: In-Memory-Rate-Limiting auf /login (je Email UND je Quell-IP, 8 Fehlversuche / 10 Minuten), UserManager.validate_password() setzt jetzt eine Mindestlänge von 12 Zeichen durch (fastapi_users' Default war ein No-Op, ließ auch leere Passwörter zu). - routers/admin_kunden.py, routers/organisation.py: fangen InvalidPasswordException/UserAlreadyExists jetzt sauber ab statt 500. - app.py: /docs, /redoc, /openapi.json deaktiviert (legten zuvor die komplette Routenstruktur inkl. /admin/* offen), Security-Header- Middleware (X-Content-Type-Options, X-Frame-Options, Referrer-Policy, CSP). - scripts/reset_password.py: neues Skript, Gegenstück zu create_user.py, für die im selben Review gefundenen aktiven Testaccounts mit dem Passwort 'test123' (einer davon Superuser) gebraucht. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
297 lines
13 KiB
Python
297 lines
13 KiB
Python
from fastapi import APIRouter, Depends, Form, HTTPException, Request
|
|
from fastapi.responses import RedirectResponse
|
|
from fastapi_users import exceptions
|
|
|
|
from anode_client import anode_request
|
|
from auth import UserCreate, UserManager, fetch_benutzer_fuer_organisation, get_user_manager, require_admin
|
|
from templating import templates
|
|
|
|
|
|
router = APIRouter(prefix="/admin/kunden")
|
|
|
|
|
|
@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)
|
|
|
|
vorlagen_result = anode_request("GET", f"/api/v1/organizations/{organization_id}/bereitstellungsvorlagen")
|
|
bereitstellungsvorlagen = vorlagen_result.get("bereitstellungsvorlagen", []) if vorlagen_result.get("success") else []
|
|
|
|
workspaces_result = anode_request("GET", "/api/v1/workspaces")
|
|
workspaces = workspaces_result.get("workspaces", []) if workspaces_result.get("success") else []
|
|
|
|
backends_result = anode_request("GET", "/api/v1/backends")
|
|
backends = backends_result.get("backends", []) if backends_result.get("success") else []
|
|
|
|
gruppen_result = anode_request("GET", f"/api/v1/organizations/{organization_id}/gruppen")
|
|
gruppen = gruppen_result.get("gruppen", []) if gruppen_result.get("success") else []
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/kunden_detail.html",
|
|
{
|
|
"user": user,
|
|
"organization": organization,
|
|
"organization_id": organization_id,
|
|
"activation_codes": activation_codes,
|
|
"benutzer": benutzer,
|
|
"bereitstellungsvorlagen": bereitstellungsvorlagen,
|
|
"workspaces": workspaces,
|
|
"backends": backends,
|
|
"gruppen": gruppen,
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/{organization_id}/organisationsdaten")
|
|
def organisationsdaten_speichern(
|
|
organization_id: str,
|
|
name: str = Form(...),
|
|
kontakt_name: str = Form(default=""),
|
|
adresse: str = Form(default=""),
|
|
telefon: str = Form(default=""),
|
|
kontakt_email: str = Form(default=""),
|
|
notizen: str = Form(default=""),
|
|
user: dict = Depends(require_admin),
|
|
):
|
|
anode_request(
|
|
"PATCH",
|
|
f"/api/v1/organizations/{organization_id}",
|
|
json={
|
|
"name": name,
|
|
"kontakt_name": kontakt_name or None,
|
|
"adresse": adresse or None,
|
|
"telefon": telefon or None,
|
|
"kontakt_email": kontakt_email or None,
|
|
"notizen": notizen or None,
|
|
},
|
|
)
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}", status_code=303)
|
|
|
|
|
|
@router.post("/{organization_id}/gruppen")
|
|
def gruppe_anlegen(organization_id: str, name: str = Form(...), user: dict = Depends(require_admin)):
|
|
anode_request("POST", f"/api/v1/organizations/{organization_id}/gruppen", json={"name": name})
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}", status_code=303)
|
|
|
|
|
|
@router.post("/{organization_id}/gruppen/{gruppe_id}/loeschen")
|
|
def gruppe_loeschen(organization_id: str, gruppe_id: str, user: dict = Depends(require_admin)):
|
|
anode_request("DELETE", f"/api/v1/gruppen/{gruppe_id}")
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}", status_code=303)
|
|
|
|
|
|
@router.get("/{organization_id}/gruppen/{gruppe_id}")
|
|
def gruppe_detail(request: Request, organization_id: str, gruppe_id: str, user: dict = Depends(require_admin)):
|
|
gruppen_result = anode_request("GET", f"/api/v1/organizations/{organization_id}/gruppen")
|
|
gruppen = gruppen_result.get("gruppen", []) if gruppen_result.get("success") else []
|
|
gruppe = next((g for g in gruppen if g["id"] == gruppe_id), None)
|
|
|
|
if gruppe is None:
|
|
raise HTTPException(status_code=404, detail="Gruppe wurde nicht gefunden.")
|
|
|
|
merkmale_result = anode_request("GET", f"/api/v1/gruppen/{gruppe_id}/merkmale")
|
|
merkmale = merkmale_result.get("merkmale", []) if merkmale_result.get("success") else []
|
|
|
|
kategorien: dict[str, list[dict]] = {}
|
|
for eintrag in merkmale:
|
|
kategorie_name = eintrag["kategorie"]["name"] if eintrag["kategorie"] else "Ohne Kategorie"
|
|
kategorien.setdefault(kategorie_name, []).append(eintrag)
|
|
|
|
devices_result = anode_request("GET", f"/api/v1/organizations/{organization_id}/devices")
|
|
devices = devices_result.get("devices", []) if devices_result.get("success") else []
|
|
|
|
# anode bietet keinen direkten "Geräte einer Gruppe"-Endpunkt an, nur
|
|
# "Gruppen eines Geräts" (GET /devices/{id}/gruppen) - daher hier für
|
|
# jedes Gerät der Organisation dessen Gruppen abfragen und filtern.
|
|
mitglied_ids = set()
|
|
for device in devices:
|
|
dg_result = anode_request("GET", f"/api/v1/devices/{device['id']}/gruppen")
|
|
device_gruppen = dg_result.get("gruppen", []) if dg_result.get("success") else []
|
|
if any(g["id"] == gruppe_id for g in device_gruppen):
|
|
mitglied_ids.add(device["id"])
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/gruppe_detail.html",
|
|
{
|
|
"user": user,
|
|
"organization_id": organization_id,
|
|
"gruppe": gruppe,
|
|
"kategorien": kategorien,
|
|
"devices": devices,
|
|
"mitglied_ids": mitglied_ids,
|
|
},
|
|
)
|
|
|
|
|
|
@router.post("/{organization_id}/gruppen/{gruppe_id}/merkmale/{merkmal_key}/select")
|
|
def gruppe_merkmal_select(organization_id: str, gruppe_id: str, merkmal_key: str, user: dict = Depends(require_admin)):
|
|
anode_request("POST", f"/api/v1/gruppen/{gruppe_id}/merkmale/{merkmal_key}/select")
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}/gruppen/{gruppe_id}", status_code=303)
|
|
|
|
|
|
@router.post("/{organization_id}/gruppen/{gruppe_id}/merkmale/{merkmal_key}/deselect")
|
|
def gruppe_merkmal_deselect(organization_id: str, gruppe_id: str, merkmal_key: str, user: dict = Depends(require_admin)):
|
|
anode_request("POST", f"/api/v1/gruppen/{gruppe_id}/merkmale/{merkmal_key}/deselect")
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}/gruppen/{gruppe_id}", status_code=303)
|
|
|
|
|
|
@router.post("/{organization_id}/gruppen/{gruppe_id}/merkmale/{merkmal_key}/unset")
|
|
def gruppe_merkmal_unset(organization_id: str, gruppe_id: str, merkmal_key: str, user: dict = Depends(require_admin)):
|
|
anode_request("POST", f"/api/v1/gruppen/{gruppe_id}/merkmale/{merkmal_key}/unset")
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}/gruppen/{gruppe_id}", status_code=303)
|
|
|
|
|
|
@router.post("/{organization_id}/gruppen/{gruppe_id}/geraete/{device_id}/hinzufuegen")
|
|
def gruppe_geraet_hinzufuegen(organization_id: str, gruppe_id: str, device_id: str, user: dict = Depends(require_admin)):
|
|
anode_request("POST", f"/api/v1/devices/{device_id}/gruppen/{gruppe_id}")
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}/gruppen/{gruppe_id}", status_code=303)
|
|
|
|
|
|
@router.post("/{organization_id}/gruppen/{gruppe_id}/geraete/{device_id}/entfernen")
|
|
def gruppe_geraet_entfernen(organization_id: str, gruppe_id: str, device_id: str, user: dict = Depends(require_admin)):
|
|
anode_request("DELETE", f"/api/v1/devices/{device_id}/gruppen/{gruppe_id}")
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}/gruppen/{gruppe_id}", status_code=303)
|
|
|
|
|
|
@router.post("/{organization_id}/bereitstellungsvorlagen")
|
|
def bereitstellungsvorlage_anlegen(
|
|
organization_id: str,
|
|
workspace_id: str = Form(...),
|
|
backend_id: str = Form(...),
|
|
label: str = Form(...),
|
|
is_default: str = Form(default=""),
|
|
disk_encryption: str = Form(default=""),
|
|
root_filesystem: str = Form(default="ext4"),
|
|
secure_boot_required: str = Form(default=""),
|
|
user: dict = Depends(require_admin),
|
|
):
|
|
anode_request(
|
|
"POST",
|
|
f"/api/v1/organizations/{organization_id}/bereitstellungsvorlagen",
|
|
json={
|
|
"workspace_id": workspace_id,
|
|
"backend_id": backend_id,
|
|
"label": label,
|
|
"is_default": is_default == "on",
|
|
"disk_encryption": disk_encryption == "on",
|
|
"root_filesystem": root_filesystem,
|
|
"secure_boot_required": secure_boot_required == "on",
|
|
},
|
|
)
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}", status_code=303)
|
|
|
|
|
|
@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=""),
|
|
ist_organisationsadmin: str = Form(default=""),
|
|
user: dict = Depends(require_admin),
|
|
user_manager: UserManager = Depends(get_user_manager),
|
|
):
|
|
try:
|
|
await user_manager.create(
|
|
UserCreate(
|
|
email=email,
|
|
password=password,
|
|
organization_id=organization_id,
|
|
is_superuser=is_superuser == "on",
|
|
ist_organisationsadmin=ist_organisationsadmin == "on",
|
|
)
|
|
)
|
|
except exceptions.InvalidPasswordException as fehler:
|
|
raise HTTPException(status_code=400, detail=fehler.reason)
|
|
except exceptions.UserAlreadyExists:
|
|
raise HTTPException(status_code=400, detail="Ein Konto mit dieser E-Mail-Adresse existiert bereits.")
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}", status_code=303)
|
|
|
|
|
|
@router.get("/{organization_id}/organisationseinheiten/{oe_id}/merkmale")
|
|
def oe_merkmale_ansicht(request: Request, organization_id: str, oe_id: str, user: dict = Depends(require_admin)):
|
|
"""
|
|
Minimale Admin-Ansicht einer OE - anders als bei Gruppen kein volles
|
|
CRUD hier (OE-Struktur bleibt laut ADR-0012 Kunden-Self-Service unter
|
|
/struktur), nur der Merkmale-Editor als Sprungziel von der
|
|
Auftragskatalog-Herkunftsanzeige aus.
|
|
"""
|
|
|
|
merkmale_result = anode_request("GET", f"/api/v1/organisationseinheiten/{oe_id}/merkmale")
|
|
merkmale = merkmale_result.get("merkmale", []) if merkmale_result.get("success") else []
|
|
|
|
kategorien: dict[str, list[dict]] = {}
|
|
for eintrag in merkmale:
|
|
kategorie_name = eintrag["kategorie"]["name"] if eintrag["kategorie"] else "Ohne Kategorie"
|
|
kategorien.setdefault(kategorie_name, []).append(eintrag)
|
|
|
|
return templates.TemplateResponse(
|
|
request,
|
|
"admin/oe_merkmale.html",
|
|
{"user": user, "organization_id": organization_id, "oe_id": oe_id, "kategorien": kategorien},
|
|
)
|
|
|
|
|
|
@router.post("/{organization_id}/organisationseinheiten/{oe_id}/merkmale/{merkmal_key}/select")
|
|
def admin_oe_merkmal_select(organization_id: str, oe_id: str, merkmal_key: str, user: dict = Depends(require_admin)):
|
|
anode_request("POST", f"/api/v1/organisationseinheiten/{oe_id}/merkmale/{merkmal_key}/select")
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}/organisationseinheiten/{oe_id}/merkmale", status_code=303)
|
|
|
|
|
|
@router.post("/{organization_id}/organisationseinheiten/{oe_id}/merkmale/{merkmal_key}/deselect")
|
|
def admin_oe_merkmal_deselect(organization_id: str, oe_id: str, merkmal_key: str, user: dict = Depends(require_admin)):
|
|
anode_request("POST", f"/api/v1/organisationseinheiten/{oe_id}/merkmale/{merkmal_key}/deselect")
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}/organisationseinheiten/{oe_id}/merkmale", status_code=303)
|
|
|
|
|
|
@router.post("/{organization_id}/organisationseinheiten/{oe_id}/merkmale/{merkmal_key}/unset")
|
|
def admin_oe_merkmal_unset(organization_id: str, oe_id: str, merkmal_key: str, user: dict = Depends(require_admin)):
|
|
anode_request("POST", f"/api/v1/organisationseinheiten/{oe_id}/merkmale/{merkmal_key}/unset")
|
|
|
|
return RedirectResponse(f"/admin/kunden/{organization_id}/organisationseinheiten/{oe_id}/merkmale", status_code=303)
|