diff --git a/app.py b/app.py index d59679a..6ebec97 100644 --- a/app.py +++ b/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) diff --git a/auth.py b/auth.py index e313371..cb988dc 100644 --- a/auth.py +++ b/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) diff --git a/routers/admin_kunden.py b/routers/admin_kunden.py index d47706a..50fb59b 100644 --- a/routers/admin_kunden.py +++ b/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( - UserCreate( - email=email, - password=password, - organization_id=organization_id, - is_superuser=is_superuser == "on", - ist_organisationsadmin=ist_organisationsadmin == "on", + 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) diff --git a/routers/organisation.py b/routers/organisation.py index 3da2a6c..0ced976 100644 --- a/routers/organisation.py +++ b/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( - UserCreate( - email=email, - password=password, - organization_id=user.organization_id, - ist_organisationsadmin=ist_organisationsadmin == "on", + try: + await user_manager.create( + UserCreate( + email=email, + password=password, + 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) diff --git a/scripts/reset_password.py b/scripts/reset_password.py new file mode 100644 index 0000000..44a71d3 --- /dev/null +++ b/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()