certain/backend/app/services/twofactor.py

"""Zwei-Faktor per TOTP (Authenticator-App) plus Einmal-Wiederherstellungscodes.

Der QR-Code wird hier serverseitig als PNG-Data-URI erzeugt. Das spart eine
JavaScript-Abhängigkeit im Frontend und erspart es, fremdes SVG ins DOM zu
schreiben – das Bild landet einfach in einem ``<img src=…>``.
"""

from __future__ import annotations

import base64
import io
import secrets
import time
from datetime import datetime, timezone

import pyotp
import qrcode
from sqlalchemy.orm import Session

from ..branding import APP_NAME
from ..models import RecoveryCode, User
from ..security import hash_session_token

STEP_SECONDS = 30
# Ein Schritt Toleranz nach vorn und hinten – deckt übliche Uhrendrift ab,
# ohne das Zeitfenster unnötig aufzuweiten.
VALID_WINDOW = 1

RECOVERY_CODE_COUNT = 10
RECOVERY_CODE_LENGTH = 10
# Ohne 0/O/1/I/L – abgeschriebene Codes sollen nicht an Verwechslungen scheitern.
_ALPHABET = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"


def new_secret() -> str:
    return pyotp.random_base32()


def otpauth_uri(user: User, secret: str) -> str:
    return pyotp.TOTP(secret).provisioning_uri(name=user.username, issuer_name=APP_NAME)


def qr_data_uri(uri: str) -> str:
    """QR-Code als ``data:image/png;base64,…``."""
    img = qrcode.make(uri)
    puffer = io.BytesIO()
    img.save(puffer, format="PNG")
    return "data:image/png;base64," + base64.b64encode(puffer.getvalue()).decode("ascii")


def matching_step(secret: str, code: str) -> int | None:
    """Zu welchem 30-Sekunden-Fenster gehört dieser Code? ``None`` = zu keinem.

    Nicht ``TOTP.verify`` benutzt, weil das nur ja/nein sagt. Für den
    Wiederverwendungsschutz muss aber feststehen, *welches* Fenster verbraucht
    wurde – sonst liesse sich ein Code aus dem nächsten Fenster im nächsten
    Moment erneut einlösen."""
    code = (code or "").strip().replace(" ", "")
    if not code.isdigit():
        return None
    totp = pyotp.TOTP(secret)
    jetzt = int(time.time()) // STEP_SECONDS
    for versatz in range(-VALID_WINDOW, VALID_WINDOW + 1):
        schritt = jetzt + versatz
        if secrets.compare_digest(totp.at(schritt * STEP_SECONDS), code):
            return schritt
    return None


def consume_code(db: Session, user: User, code: str) -> bool:
    """Prüft den Code und verbraucht sein Zeitfenster.

    Ein einmal benutzter Code bleibt zwar bis zu 30 Sekunden rechnerisch gültig,
    darf aber kein zweites Mal zählen – sonst könnte ein abgefangener Code sofort
    nachgespielt werden."""
    if not user.totp_secret:
        return False
    schritt = matching_step(user.totp_secret, code)
    if schritt is None:
        return False
    if user.totp_last_step is not None and schritt <= user.totp_last_step:
        return False
    user.totp_last_step = schritt
    db.commit()
    return True


def _format_code(raw: str) -> str:
    """Fünfergruppen mit Bindestrich – deutlich leichter abzuschreiben."""
    return f"{raw[:5]}-{raw[5:]}"


def generate_recovery_codes(db: Session, user: User) -> list[str]:
    """Erzeugt einen frischen Satz Codes; alte werden dabei ungültig.

    Die Klartext-Codes gibt es nur dieses eine Mal – in der Datenbank steht
    ausschliesslich ihr Hash."""
    db.query(RecoveryCode).filter(RecoveryCode.user_id == user.id).delete()
    codes: list[str] = []
    for _ in range(RECOVERY_CODE_COUNT):
        roh = "".join(secrets.choice(_ALPHABET) for _ in range(RECOVERY_CODE_LENGTH))
        codes.append(_format_code(roh))
        db.add(RecoveryCode(user_id=user.id, code_hash=hash_session_token(roh)))
    db.commit()
    return codes


def consume_recovery_code(db: Session, user: User, code: str) -> bool:
    """Setzt einen Wiederherstellungscode ein. Jeder Code zählt genau einmal."""
    roh = (code or "").strip().upper().replace("-", "").replace(" ", "")
    if not roh:
        return False
    treffer = (
        db.query(RecoveryCode)
        .filter(
            RecoveryCode.user_id == user.id,
            RecoveryCode.code_hash == hash_session_token(roh),
            RecoveryCode.used_at.is_(None),
        )
        .first()
    )
    if treffer is None:
        return False
    treffer.used_at = datetime.now(timezone.utc)
    db.commit()
    return True


def codes_left(db: Session, user: User) -> int:
    return (
        db.query(RecoveryCode)
        .filter(RecoveryCode.user_id == user.id, RecoveryCode.used_at.is_(None))
        .count()
    )


def disable(db: Session, user: User) -> None:
    """Schaltet 2FA ab und räumt Secret sowie Wiederherstellungscodes weg."""
    db.query(RecoveryCode).filter(RecoveryCode.user_id == user.id).delete()
    user.totp_secret = None
    user.totp_enabled = False
    user.totp_confirmed_at = None
    user.totp_last_step = None
    db.commit()