Security-Hardening vor öffentlicher Freigabe: Rate-Limiting, Passwort-Policy, Docs-Endpunkte, Security-Header

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>

app.py

@@ -1,4 +1,4 @@
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 from fastapi.staticfiles import StaticFiles
 
 import auth
@@ -15,9 +15,36 @@ from routers import (
 )
 
 
-app = FastAPI(title="Tuxflotte Kundenplattform")
+# Sicherheits-Review 25.08.2026 (vor der geplanten oeffentlichen Freigabe
+# ueber Pangolin): /docs, /redoc, /openapi.json waren zuvor unauthentifiziert
+# erreichbar und legten die komplette interne Routenstruktur (inkl. aller
+# /admin/*-Pfade) offen - fuer eine Plattform in aktiver Entwicklung
+# unnoetige Aufklaerungshilfe fuer Angreifer.
+app = FastAPI(title="Tuxflotte Kundenplattform", docs_url=None, redoc_url=None, openapi_url=None)
 app.mount("/static", StaticFiles(directory="static"), name="static")
 
+
+@app.middleware("http")
+async def security_headers(request: Request, call_next):
+    """
+    Basis-Security-Header (Sicherheits-Review 25.08.2026). Pangolin
+    terminiert TLS und kann HSTS ergaenzen, das hier ist die App-seitige
+    Schicht unabhaengig davon. CSP erlaubt 'unsafe-inline' nur fuer Styles
+    (einzelne style="display:inline"-Attribute in ein paar Admin-Templates),
+    nicht fuer Skripte - die Templates enthalten ohnehin kein Inline-JS.
+    """
+
+    response = await call_next(request)
+    response.headers.setdefault("X-Content-Type-Options", "nosniff")
+    response.headers.setdefault("X-Frame-Options", "DENY")
+    response.headers.setdefault("Referrer-Policy", "same-origin")
+    response.headers.setdefault(
+        "Content-Security-Policy",
+        "default-src 'self'; script-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' data:",
+    )
+
+    return response
+
 app.include_router(auth.router)
 app.include_router(geraete.router)
 app.include_router(organisation.router)

auth.py

@@ -1,12 +1,14 @@
 import os
+import time
 import uuid
+from collections import defaultdict, deque
 from dataclasses import dataclass
 from types import SimpleNamespace
 from typing import Optional
 
 from fastapi import APIRouter, Depends, Form, HTTPException, Request
 from fastapi.responses import RedirectResponse
-from fastapi_users import BaseUserManager, FastAPIUsers, UUIDIDMixin, schemas
+from fastapi_users import BaseUserManager, FastAPIUsers, UUIDIDMixin, exceptions, schemas
 from fastapi_users.authentication import AuthenticationBackend, CookieTransport, JWTStrategy
 from fastapi_users.db import BaseUserDatabase
 
@@ -19,6 +21,58 @@ SESSION_SECRET = os.environ.get("KUNDENPLATTFORM_SESSION_SECRET", "")
 COOKIE_NAME = "kundenplattform_session"
 COOKIE_MAX_AGE = 14 * 24 * 60 * 60  # 14 Tage, wie zuvor Starlettes SessionMiddleware-Default
 
+MIN_PASSWORD_LENGTH = 12
+
+# Login-Rate-Limiting (Sicherheits-Review 25.08.2026): einfacher In-Memory-
+# Zaehler statt einer zusaetzlichen Abhaengigkeit (slowapi o.ae.) - der
+# Dienst laeuft laut kundenplattform.service ohne --workers, also als
+# einzelner uvicorn-Prozess, ein In-Process-Dict ist damit ausreichend (kein
+# Multi-Worker-Sync-Problem). Ueberlebt keinen Neustart - das ist akzeptabel,
+# CrowdSec am Pangolin-Proxy ist die persistente/IP-uebergreifende Schicht,
+# das hier ist die zusaetzliche Anwendungsschicht gegen gezieltes Erraten
+# eines bekannten Accounts, auch von wechselnden Quell-IPs aus.
+LOGIN_RATE_LIMIT = 8
+LOGIN_RATE_WINDOW_SECONDS = 10 * 60
+
+_login_attempts_je_email: dict[str, deque] = defaultdict(deque)
+_login_attempts_je_ip: dict[str, deque] = defaultdict(deque)
+
+
+def _client_ip(request: Request) -> str:
+    # Pangolin (Reverse Proxy) setzt X-Forwarded-For; ohne Proxy (z.B. beim
+    # lokalen QEMU-Test) faellt das auf die direkte Verbindung zurueck.
+    weitergeleitet = request.headers.get("x-forwarded-for")
+    if weitergeleitet:
+        return weitergeleitet.split(",")[0].strip()
+
+    return request.client.host if request.client else "unbekannt"
+
+
+def _rate_limited(zaehler: dict[str, deque], schluessel: str) -> bool:
+    jetzt = time.monotonic()
+    versuche = zaehler[schluessel]
+
+    while versuche and jetzt - versuche[0] > LOGIN_RATE_WINDOW_SECONDS:
+        versuche.popleft()
+
+    return len(versuche) >= LOGIN_RATE_LIMIT
+
+
+def _login_rate_limited(email: str, ip: str) -> bool:
+    return _rate_limited(_login_attempts_je_email, email) or _rate_limited(_login_attempts_je_ip, ip)
+
+
+def _record_failed_login(email: str, ip: str) -> None:
+    jetzt = time.monotonic()
+    _login_attempts_je_email[email].append(jetzt)
+    _login_attempts_je_ip[ip].append(jetzt)
+
+
+def _clear_failed_logins(email: str, ip: str) -> None:
+    _login_attempts_je_email.pop(email, None)
+    _login_attempts_je_ip.pop(ip, None)
+
+
 BENUTZER_SPALTEN = (
     "id, organization_id, email, hashed_password, is_active, is_superuser, "
     "ist_organisationsadmin, is_verified"
@@ -136,6 +190,21 @@ class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
     reset_password_token_secret = SESSION_SECRET
     verification_token_secret = SESSION_SECRET
 
+    async def validate_password(self, password: str, user) -> None:
+        # Sicherheits-Review 25.08.2026: fastapi_users' Default ist ein
+        # No-Op, das lies bislang jedes Passwort (auch leer) zu - live als
+        # `root@default-lab.test` / `test123` (Superuser!) gefunden.
+        if len(password) < MIN_PASSWORD_LENGTH:
+            raise exceptions.InvalidPasswordException(
+                reason=f"Das Passwort muss mindestens {MIN_PASSWORD_LENGTH} Zeichen lang sein."
+            )
+
+        email = getattr(user, "email", None)
+        if email and password.lower() == email.lower():
+            raise exceptions.InvalidPasswordException(
+                reason="Das Passwort darf nicht mit der E-Mail-Adresse identisch sein."
+            )
+
 
 async def get_user_manager(user_db: UserDatabase = Depends(get_user_db)):
     yield UserManager(user_db)
@@ -250,13 +319,27 @@ async def login_submit(
     password: str = Form(...),
     user_manager: UserManager = Depends(get_user_manager),
 ):
+    email_normalisiert = email.strip().lower()
+    ip = _client_ip(request)
+
+    if _login_rate_limited(email_normalisiert, ip):
+        return templates.TemplateResponse(
+            request,
+            "login.html",
+            {"error": "Zu viele fehlgeschlagene Loginversuche. Bitte in einigen Minuten erneut versuchen."},
+            status_code=429,
+        )
+
     user = await user_manager.authenticate(SimpleNamespace(username=email, password=password))
 
     if user is None or not user.is_active:
+        _record_failed_login(email_normalisiert, ip)
         return templates.TemplateResponse(
             request, "login.html", {"error": "E-Mail oder Passwort ist falsch."}, status_code=401
         )
 
+    _clear_failed_logins(email_normalisiert, ip)
+
     strategy = get_jwt_strategy()
     token = await strategy.write_token(user)
 

routers/admin_kunden.py

@@ -1,5 +1,6 @@
 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
@@ -232,15 +233,20 @@ async def konto_anlegen(
     user: dict = Depends(require_admin),
     user_manager: UserManager = Depends(get_user_manager),
 ):
-    await user_manager.create(
+    try:
-        UserCreate(
+        await user_manager.create(
-            email=email,
+            UserCreate(
-            password=password,
+                email=email,
-            organization_id=organization_id,
+                password=password,
-            is_superuser=is_superuser == "on",
+                organization_id=organization_id,
-            ist_organisationsadmin=ist_organisationsadmin == "on",
+                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)
 

routers/organisation.py

@@ -1,5 +1,6 @@
 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 (
@@ -72,14 +73,19 @@ async def konto_anlegen(
     # is_superuser wird beim Self-Service-Aufruf nie gesetzt (Default False)
     # - das bleibt exklusiv dem internen Admin-Bereich vorbehalten (siehe
     # ADR zu is_superuser vs. ist_organisationsadmin).
-    await user_manager.create(
+    try:
-        UserCreate(
+        await user_manager.create(
-            email=email,
+            UserCreate(
-            password=password,
+                email=email,
-            organization_id=user.organization_id,
+                password=password,
-            ist_organisationsadmin=ist_organisationsadmin == "on",
+                organization_id=user.organization_id,
+                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("/organisation", status_code=303)
 

scripts/reset_password.py

@@ -0,0 +1,57 @@
+"""
+Passwort eines bestehenden Kontos zuruecksetzen (Gegenstueck zu
+create_user.py). Beispiel:
+
+    KUNDENPLATTFORM_DATABASE_URL=... .venv/bin/python scripts/reset_password.py \\
+        --email admin@schule-beispiel.de
+"""
+
+import argparse
+import asyncio
+import getpass
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from fastapi_users import exceptions  # noqa: E402
+
+from auth import UserDatabase, UserManager  # noqa: E402
+from db import get_async_database_connection  # noqa: E402
+
+
+async def passwort_zuruecksetzen(email: str, password: str):
+    async with await get_async_database_connection() as connection:
+        user_manager = UserManager(UserDatabase(connection))
+        user = await user_manager.get_by_email(email)
+
+        await user_manager.validate_password(password, user)
+        await user_manager.user_db.update(user, {"hashed_password": user_manager.password_helper.hash(password)})
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--email", required=True)
+    args = parser.parse_args()
+
+    if not os.environ.get("KUNDENPLATTFORM_DATABASE_URL"):
+        sys.exit("KUNDENPLATTFORM_DATABASE_URL ist nicht gesetzt.")
+
+    password = getpass.getpass("Neues Passwort: ")
+    password_confirm = getpass.getpass("Neues Passwort (Wiederholung): ")
+
+    if password != password_confirm:
+        sys.exit("Passwörter stimmen nicht überein.")
+
+    try:
+        asyncio.run(passwort_zuruecksetzen(args.email, password))
+    except exceptions.UserNotExists:
+        sys.exit(f"Kein Konto mit der E-Mail {args.email} gefunden.")
+    except exceptions.InvalidPasswordException as fehler:
+        sys.exit(f"Passwort abgelehnt: {fehler.reason}")
+
+    print(f"Passwort für {args.email} zurückgesetzt.")
+
+
+if __name__ == "__main__":
+    main()