Phase 3: Anmeldung, Rate-Limits und Sicherheits-Header
Lesen darf jeder, schreiben nur der Betreiber - mit der bewussten Ausnahme des Reservierens. Dazu Dockerfile und Compose-Datei, damit sich das lokal ausprobieren lässt. Beim Bauen sind zwei Fehler aufgefallen, die ohne Test nicht aufgefallen wären: 1. slowapi zählt pro URL-Pfad. Weil jedes Kleidungsstück eine eigene URL hat, bekam jedes seinen eigenen Zähler - ein Skript hätte also den gesamten Bestand reservieren können, ohne je an ein Limit zu stossen. Genau der Missbrauch, gegen den das Limit gedacht ist. Behoben mit shared_limit und festem scope. 2. Der erste Anlauf des Tests machte fünf Anfragen gegen ein Limit von fünf und konnte damit gar nichts zeigen. Geprüft wird jetzt der tatsächlich ausgelieferte Standardwert, mit mehr Anfragen als erlaubt. Weiter umgesetzt: - Passwort als bcrypt-Hash aus der Umgebung, einmal beim Start gebildet und gemerkt. Bei jeder Anfrage neu gehasht liesse sich die Anwendung sonst ohne Anmeldung lahmlegen - bcrypt ist absichtlich langsam. - Ohne hinterlegtes Passwort bleibt der Erfassungsbereich gesperrt (503) statt offen zu stehen. Kein mitgeliefertes Standardpasswort. - Sitzung als signiertes Cookie, HttpOnly, SameSite=Lax (blockt seitenfremde POSTs), Secure abschaltbar nur fürs lokale Testen, Abmeldung nach zwei Stunden Ruhe. - CSP mit script-src 'self', nosniff, frame-ancestors none, dazu noindex und robots.txt: die Galerie ist frei zugänglich, soll aber nicht dauerhaft im Suchindex stehen. - Der Betreiber darf Reservierungen ohne Token aufheben, damit sich eine Missbrauchswelle aufräumen lässt. - uvicorn mit --proxy-headers: hinter einem Reverse-Proxy zählte sonst alles auf dessen IP, und ein einzelner Besucher sperrte alle aus. - Container läuft nicht als root; Code gehört root, nur Daten und Bilder dem Dienstbenutzer. Umgebungsvariablen heissen jetzt ausdrücklich englisch (ADMIN_PASSWORD, SECRET_KEY, ...), passend zur Anleitung und zur Kantone-App. 59 Tests, alle grün. Zusätzlich gegen den laufenden Container geprüft. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01GR4bNaj9GtRu57J4Niii8o
This commit is contained in:
Claude Opus 5
+151
@@ -0,0 +1,151 @@
|
||||
"""Anmeldung, Sitzung, CSRF und Rate-Limits.
|
||||
|
||||
Grundsatz: **lesen darf jeder, schreiben nur der Betreiber** - mit der einen
|
||||
Ausnahme des Reservierens, das bewusst ohne Anmeldung möglich ist.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import secrets
|
||||
|
||||
import bcrypt
|
||||
from fastapi import Depends, HTTPException, Request, status
|
||||
from slowapi import Limiter
|
||||
from slowapi.util import get_remote_address
|
||||
|
||||
from app.config import einstellungen
|
||||
|
||||
# Der Schlüssel für die Rate-Limits ist die IP. Hinter einem Reverse-Proxy
|
||||
# liefert get_remote_address nur dann die echte Adresse, wenn uvicorn mit
|
||||
# --proxy-headers läuft (siehe Dockerfile) - sonst zählt alles auf die IP
|
||||
# des Proxys, und ein einzelner Besucher sperrt alle anderen aus.
|
||||
limiter = Limiter(key_func=get_remote_address)
|
||||
|
||||
|
||||
# ------------------------------------------------------------- Passwort ---
|
||||
|
||||
def passwort_hash(klartext: str) -> str:
|
||||
return bcrypt.hashpw(klartext.encode(), bcrypt.gensalt()).decode()
|
||||
|
||||
|
||||
def passwort_stimmt(klartext: str, hash_wert: str) -> bool:
|
||||
"""Prüft das Passwort. bcrypt vergleicht in konstanter Zeit."""
|
||||
try:
|
||||
# bcrypt schneidet nach 72 Byte ab und wirft bei längeren Eingaben
|
||||
# je nach Version einen Fehler - abfangen statt 500 liefern.
|
||||
return bcrypt.checkpw(klartext.encode()[:72], hash_wert.encode())
|
||||
except (ValueError, TypeError):
|
||||
return False
|
||||
|
||||
|
||||
def _hinterlegter_hash() -> str | None:
|
||||
"""Hash aus der Umgebung. Ergebnis wird gemerkt.
|
||||
|
||||
Wichtig: der Hash wird **einmal** gebildet und nicht bei jeder Anfrage.
|
||||
bcrypt ist absichtlich langsam; würde man ihn pro Aufruf berechnen,
|
||||
liesse sich die Anwendung mit ein paar gleichzeitigen Aufrufen der
|
||||
Anmeldeseite lahmlegen - noch dazu ohne Anmeldung.
|
||||
"""
|
||||
cfg = einstellungen()
|
||||
if cfg.admin_passwort_hash:
|
||||
return cfg.admin_passwort_hash
|
||||
if cfg.admin_passwort:
|
||||
return passwort_hash(cfg.admin_passwort)
|
||||
return None
|
||||
|
||||
|
||||
_hash_zwischenspeicher: dict[str, str | None] = {}
|
||||
|
||||
|
||||
def hinterlegter_hash() -> str | None:
|
||||
if "wert" not in _hash_zwischenspeicher:
|
||||
_hash_zwischenspeicher["wert"] = _hinterlegter_hash()
|
||||
return _hash_zwischenspeicher["wert"]
|
||||
|
||||
|
||||
def hash_zwischenspeicher_leeren() -> None:
|
||||
"""Nur für Tests - im Betrieb ändert sich das Passwort nicht zur Laufzeit."""
|
||||
_hash_zwischenspeicher.clear()
|
||||
|
||||
|
||||
# -------------------------------------------------------------- Sitzung ---
|
||||
|
||||
SITZUNG_ANGEMELDET = "betreiber"
|
||||
SITZUNG_LETZTE_AKTIVITAET = "letzte_aktivitaet"
|
||||
SITZUNG_CSRF = "csrf"
|
||||
|
||||
|
||||
def anmelden(request: Request) -> None:
|
||||
import time
|
||||
|
||||
# Sitzung neu aufbauen statt ergänzen: verhindert, dass eine vorher
|
||||
# untergeschobene Sitzungskennung nach der Anmeldung weitergilt.
|
||||
request.session.clear()
|
||||
request.session[SITZUNG_ANGEMELDET] = True
|
||||
request.session[SITZUNG_LETZTE_AKTIVITAET] = int(time.time())
|
||||
|
||||
|
||||
def abmelden(request: Request) -> None:
|
||||
request.session.clear()
|
||||
|
||||
|
||||
def ist_betreiber(request: Request) -> bool:
|
||||
import time
|
||||
|
||||
if not request.session.get(SITZUNG_ANGEMELDET):
|
||||
return False
|
||||
|
||||
cfg = einstellungen()
|
||||
letzte = request.session.get(SITZUNG_LETZTE_AKTIVITAET, 0)
|
||||
if letzte and (time.time() - letzte) > cfg.sitzung_max_ruhe_sekunden:
|
||||
request.session.clear()
|
||||
return False
|
||||
|
||||
request.session[SITZUNG_LETZTE_AKTIVITAET] = int(time.time())
|
||||
return True
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ CSRF ---
|
||||
|
||||
def csrf_token(request: Request) -> str:
|
||||
if not request.session.get(SITZUNG_CSRF):
|
||||
request.session[SITZUNG_CSRF] = secrets.token_urlsafe(32)
|
||||
return request.session[SITZUNG_CSRF]
|
||||
|
||||
|
||||
def csrf_pruefen(request: Request, gesendet: str | None) -> bool:
|
||||
hinterlegt = request.session.get(SITZUNG_CSRF)
|
||||
if not hinterlegt or not gesendet:
|
||||
return False
|
||||
return secrets.compare_digest(hinterlegt, gesendet)
|
||||
|
||||
|
||||
# ----------------------------------------------------------- Abhängigkeit ---
|
||||
|
||||
def betreiber_noetig(request: Request) -> None:
|
||||
"""Schützt alles, was den Bestand verändert.
|
||||
|
||||
Nicht darunter: Lesen und Reservieren - beides ist bewusst offen.
|
||||
"""
|
||||
if not einstellungen().admin_aktiv:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||||
detail=(
|
||||
"Der Erfassungsbereich ist nicht eingerichtet: "
|
||||
"ADMIN_PASSWORD oder ADMIN_PASSWORD_HASH setzen."
|
||||
),
|
||||
)
|
||||
if not ist_betreiber(request):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Bitte zuerst anmelden.",
|
||||
)
|
||||
|
||||
|
||||
BetreiberNoetig = Depends(betreiber_noetig)
|
||||
|
||||
|
||||
def betreiber_optional(request: Request) -> bool:
|
||||
"""Für Stellen, die je nach Anmeldung unterschiedlich reagieren -
|
||||
etwa das Aufheben einer Reservierung (Token oder Betreiber)."""
|
||||
return ist_betreiber(request)
|
||||
Reference in New Issue
Block a user