kleiderboerse/app/security.py

"""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:
    runden = einstellungen().bcrypt_runden
    return bcrypt.hashpw(klartext.encode(), bcrypt.gensalt(runden)).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)