certain/backend/app/services/sessions.py

"""Server-seitige Sitzungen: anlegen, prüfen, widerrufen.

Im Cookie liegt nur das Klartext-Token; in der Datenbank nur dessen SHA-256-Hash.
So lässt sich eine Sitzung serverseitig beenden (Logout, „überall abmelden"),
und ein Blick in die DB verrät kein gültiges Cookie.
"""

from datetime import datetime, timedelta, timezone

from sqlalchemy.orm import Session as DBSession

from ..config import get_settings
from ..models import Session as UserSession, User
from ..security import generate_session_token, hash_session_token


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _aware(dt: datetime) -> datetime:
    """SQLite gibt datetimes teils ohne tzinfo zurück – als UTC interpretieren."""
    return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc)


# Wie lange der Zwischenzustand „Passwort ok, Code fehlt" gilt. Kurz, weil in
# dieser Zeit nur noch der zweite Faktor eingegeben werden soll.
PENDING_TTL_MINUTES = 10
# So viele Fehlversuche verträgt eine Pending-Sitzung, dann ist sie verbraucht.
MAX_2FA_ATTEMPTS = 5


def create_session(
    db: DBSession, user: User, user_agent: str | None = None, pending_2fa: bool = False
) -> str:
    """Legt eine Sitzung an und gibt das Klartext-Token zurück (nur hier sichtbar)."""
    settings = get_settings()
    gueltig_bis = (
        _now() + timedelta(minutes=PENDING_TTL_MINUTES)
        if pending_2fa
        else _now() + timedelta(days=settings.session_ttl_days)
    )
    raw = generate_session_token()
    entry = UserSession(
        user_id=user.id,
        token_hash=hash_session_token(raw),
        expires_at=gueltig_bis,
        last_seen_at=_now(),
        user_agent=(user_agent or "")[:255] or None,
        pending_2fa=pending_2fa,
    )
    db.add(entry)
    db.commit()
    return raw


def _lookup(db: DBSession, raw_token: str) -> UserSession | None:
    """Sucht die Sitzung und räumt sie weg, falls sie abgelaufen ist."""
    if not raw_token:
        return None
    entry = (
        db.query(UserSession)
        .filter(UserSession.token_hash == hash_session_token(raw_token))
        .first()
    )
    if entry is None:
        return None
    if _aware(entry.expires_at) < _now():
        db.delete(entry)
        db.commit()
        return None
    return entry


def validate(db: DBSession, raw_token: str) -> User | None:
    """Gibt den Benutzer zurück, wenn das Token gültig und nicht abgelaufen ist.

    Sitzungen, bei denen der zweite Faktor noch aussteht, gelten hier bewusst
    als *nicht* angemeldet – sonst wäre das Zwischen-Cookie ein vollwertiger
    Zugang."""
    entry = _lookup(db, raw_token)
    if entry is None or entry.pending_2fa:
        return None
    user = db.get(User, entry.user_id)
    if user is None or not user.is_active:
        return None
    entry.last_seen_at = _now()
    db.commit()
    return user


def validate_pending(db: DBSession, raw_token: str) -> tuple[UserSession, User] | None:
    """Gegenstück zu ``validate`` für den Zwei-Faktor-Schritt."""
    entry = _lookup(db, raw_token)
    if entry is None or not entry.pending_2fa:
        return None
    user = db.get(User, entry.user_id)
    if user is None or not user.is_active:
        return None
    return entry, user


def note_failed_2fa(db: DBSession, entry: UserSession) -> bool:
    """Zählt einen Fehlversuch. Gibt ``True`` zurück, wenn die Sitzung verfällt."""
    entry.failed_2fa += 1
    verbraucht = entry.failed_2fa >= MAX_2FA_ATTEMPTS
    if verbraucht:
        db.delete(entry)
    db.commit()
    return verbraucht


def revoke(db: DBSession, raw_token: str) -> None:
    entry = (
        db.query(UserSession)
        .filter(UserSession.token_hash == hash_session_token(raw_token))
        .first()
    )
    if entry is not None:
        db.delete(entry)
        db.commit()


def revoke_all(db: DBSession, user: User) -> None:
    db.query(UserSession).filter(UserSession.user_id == user.id).delete()
    db.commit()


def revoke_all_everywhere(db: DBSession) -> None:
    """Beendet sämtliche Sitzungen – nach einer systemweiten Wiederherstellung,
    bei der die Benutzertabelle komplett ausgetauscht wurde."""
    db.query(UserSession).delete()
    db.commit()