Vorrania/backend/tests/test_oidc.py

"""Single Sign-on: Konfiguration lesen und Konten zuordnen.

Beides laeuft ohne HTTP – die Anbieter-Konfiguration ist eine reine Funktion
ueber einem Umgebungs-Woerterbuch, und die Zuordnung bekommt ein fertiges
Claims-Woerterbuch, wie es nach der Token-Pruefung vorliegt.
"""

from datetime import datetime, timedelta, timezone
from types import SimpleNamespace

import pytest
from fastapi import HTTPException

from app.models import Role, User, UserIdentity
from app.oidc_config import OidcProvider, provider_aus_umgebung
from app.routers.auth import login
from app.services import oidc_state
from app.security import hash_password, verify_password
from app.services.oidc import (
    OidcFehler,
    benutzer_aus_claims,
    identitaet_loesen,
    identitaet_verknuepfen,
)

BASIS = "https://vorrania.example.com"

VOLLE_UMGEBUNG = {
    "OIDC_PROVIDERS": "authentik",
    "OIDC_AUTHENTIK_DISPLAY_NAME": "Authentik",
    "OIDC_AUTHENTIK_DISCOVERY_URL": "https://idp.example.com/application/o/v/.well-known/openid-configuration",
    "OIDC_AUTHENTIK_CLIENT_ID": "web-client",
    "OIDC_AUTHENTIK_CLIENT_SECRET": "geheim",
    "OIDC_AUTHENTIK_NATIVE_CLIENT_IDS": "ios-client, ipad-client",
}


@pytest.fixture
def provider() -> OidcProvider:
    return OidcProvider(
        slug="authentik",
        display_name="Authentik",
        discovery_url="https://idp.example.com/.well-known/openid-configuration",
        client_id="web-client",
        client_secret="geheim",
        native_client_ids=("ios-client",),
        redirect_uri=f"{BASIS}/api/auth/oidc/authentik/callback",
    )


def claims(**felder) -> dict:
    grund = {"sub": "abc-123", "email": "eva@example.com", "email_verified": True}
    grund.update(felder)
    return grund


# ---- Konfiguration --------------------------------------------------------


def test_ohne_konfiguration_kein_anbieter():
    """Ohne OIDC_PROVIDERS bleibt alles beim Passwort-Login.

    Wichtig, weil update.sh bestehende .env-Dateien nicht nachtraegt – ein alter
    Bestand darf durch das Update nicht ploetzlich beim Start scheitern.
    """
    assert provider_aus_umgebung({}, BASIS) == []


def test_anbieter_wird_vollstaendig_gelesen():
    (p,) = provider_aus_umgebung(VOLLE_UMGEBUNG, BASIS)
    assert p.slug == "authentik"
    assert p.display_name == "Authentik"
    assert p.client_id == "web-client"
    assert p.client_secret == "geheim"
    assert p.native_client_ids == ("ios-client", "ipad-client")
    # Das /api-Praefix entsteht erst in nginx und muss deshalb hier drinstehen.
    assert p.redirect_uri == f"{BASIS}/api/auth/oidc/authentik/callback"
    # Der Browser-Weg braucht kein Refresh-Token – offline_access dort anzufragen
    # loeste nur einen ueberfluessigen Zustimmungsdialog aus.
    assert p.scopes == "openid profile email"
    # Die App dagegen erneuert selbst und braucht es zwingend: ohne diesen Scope
    # liefert Authentik seit 2024.2 gar kein Refresh-Token.
    assert "offline_access" in p.native_scopes
    assert p.erlaubte_audiences == ("web-client", "ios-client", "ipad-client")


def test_zweiter_anbieter_ist_reine_konfiguration():
    """Ein weiterer Anbieter darf keine Codeaenderung brauchen."""
    umgebung = dict(VOLLE_UMGEBUNG)
    umgebung["OIDC_PROVIDERS"] = "authentik,keycloak"
    umgebung["OIDC_KEYCLOAK_DISCOVERY_URL"] = "https://kc.example.com/.well-known/openid-configuration"
    umgebung["OIDC_KEYCLOAK_CLIENT_ID"] = "kc-web"

    anbieter = provider_aus_umgebung(umgebung, BASIS)
    assert [p.slug for p in anbieter] == ["authentik", "keycloak"]
    assert anbieter[1].display_name == "Keycloak"  # Vorgabe aus dem Kuerzel


def test_unvollstaendiger_anbieter_wird_uebersprungen():
    """Ein Tippfehler in der .env darf den Serverstart nicht verhindern."""
    umgebung = {"OIDC_PROVIDERS": "authentik"}  # weder Discovery noch Client-ID
    assert provider_aus_umgebung(umgebung, BASIS) == []


def test_ohne_basisadresse_kein_anbieter():
    """Ohne PUBLIC_BASE_URL laesst sich die Rueckleit-Adresse nicht bilden."""
    assert provider_aus_umgebung(VOLLE_UMGEBUNG, "") == []


def test_ausdrueckliche_rueckleitadresse_ersetzt_die_basisadresse():
    umgebung = dict(VOLLE_UMGEBUNG)
    umgebung["OIDC_AUTHENTIK_REDIRECT_URI"] = "https://anders.example.com/api/auth/oidc/authentik/callback"
    (p,) = provider_aus_umgebung(umgebung, "")
    assert p.redirect_uri == "https://anders.example.com/api/auth/oidc/authentik/callback"


# ---- Zuordnung ------------------------------------------------------------


def test_bekannte_identitaet_meldet_denselben_nutzer_an(db, provider):
    eva = User(username="eva", password_hash=hash_password("geheim123"), role=Role.admin)
    db.add(eva)
    db.flush()
    db.add(UserIdentity(user_id=eva.id, provider="authentik", subject="abc-123"))
    db.commit()

    gefunden = benutzer_aus_claims(db, provider, claims())

    assert gefunden.id == eva.id
    assert gefunden.role == Role.admin  # Rolle bleibt, was der Admin gesetzt hat
    assert db.query(User).count() == 1


def test_erstverknuepfung_ueber_bestaetigte_email(db, provider):
    """Der Regelfall beim Umstieg: Admin traegt seine Adresse ein, meldet sich an."""
    eva = User(
        username="eva",
        password_hash=hash_password("geheim123"),
        email="eva@example.com",
        role=Role.admin,
    )
    db.add(eva)
    db.commit()

    gefunden = benutzer_aus_claims(db, provider, claims())

    assert gefunden.id == eva.id
    assert gefunden.role == Role.admin
    assert db.query(User).count() == 1
    # Ab jetzt zaehlt die sub, nicht mehr die Adresse.
    identitaet = db.query(UserIdentity).one()
    assert (identitaet.provider, identitaet.subject) == ("authentik", "abc-123")


def test_unbestaetigte_email_verknuepft_nicht(db, provider):
    """Sonst genuegt eine frei gewaehlte Adresse im IdP zur Kontouebernahme."""
    db.add(
        User(
            username="eva",
            password_hash=hash_password("geheim123"),
            email="eva@example.com",
            role=Role.admin,
        )
    )
    db.commit()

    gefunden = benutzer_aus_claims(db, provider, claims(email_verified=False))

    assert gefunden.username != "eva"
    assert gefunden.role == Role.user
    assert db.query(User).count() == 2
    # Die Adresse gehoert weiterhin allein dem Bestandskonto.
    assert gefunden.email is None


def test_ohne_auto_anlage_wird_abgewiesen(db, provider):
    from dataclasses import replace

    ohne = replace(provider, auto_create=False)

    with pytest.raises(OidcFehler) as fehler:
        benutzer_aus_claims(db, ohne, claims())

    assert fehler.value.schluessel == "kein_konto"
    assert db.query(User).count() == 0


def test_auto_anlage_erzeugt_konto_ohne_passwort(db, provider):
    neu = benutzer_aus_claims(db, provider, claims(preferred_username="eva.mueller"))

    assert neu.username == "eva.mueller"
    assert neu.email == "eva@example.com"
    # Rolle kommt NIE aus dem Anbieter.
    assert neu.role == Role.user
    # Reines SSO-Konto: kein Passwort, also auch kein Passwort-Login moeglich.
    assert neu.password_hash is None


def test_namenskollision_uebernimmt_kein_fremdes_konto(db, provider):
    """Ein passend gewaehlter preferred_username darf kein Konto kapern."""
    fremd = User(username="admin", password_hash=hash_password("geheim123"), role=Role.admin)
    db.add(fremd)
    db.commit()

    neu = benutzer_aus_claims(db, provider, claims(preferred_username="admin"))

    assert neu.id != fremd.id
    assert neu.username == "admin-2"
    assert neu.role == Role.user
    # Das bestehende Admin-Konto bleibt unangetastet.
    db.refresh(fremd)
    assert fremd.role == Role.admin
    assert fremd.password_hash is not None


def test_zweite_kennung_auf_verknuepftes_konto_wird_abgelehnt(db, provider):
    eva = User(username="eva", password_hash=None, email="eva@example.com", role=Role.user)
    db.add(eva)
    db.flush()
    db.add(UserIdentity(user_id=eva.id, provider="authentik", subject="alt-999"))
    db.commit()

    with pytest.raises(OidcFehler) as fehler:
        benutzer_aus_claims(db, provider, claims(sub="neu-111"))

    assert fehler.value.schluessel == "konto_bereits_verknuepft"


def test_sso_konto_ist_per_passwort_nicht_anmeldbar():
    """Ohne Hash muss die Pruefung schlicht False sagen – kein AttributeError.

    Sonst antwortet POST /auth/login mit 500 statt mit einem ehrlichen 401.
    """
    assert verify_password("irgendwas", None) is False
    assert verify_password("irgendwas", "") is False


# ---- Merkzettel fuer den Browser-Weg ---------------------------------------


def test_zettel_gilt_genau_einmal(db):
    """Ein zweiter Rueckweg mit demselben state muss ins Leere laufen."""
    zettel = oidc_state.anlegen(db, "authentik")

    erster = oidc_state.einloesen(db, zettel.state)
    assert erster is not None
    assert erster.provider == "authentik"
    assert len(erster.code_verifier) >= 43  # RFC 7636

    assert oidc_state.einloesen(db, zettel.state) is None


def test_abgelaufener_zettel_wird_abgelehnt(db):
    zettel = oidc_state.anlegen(db, "authentik")
    spaeter = datetime.now(timezone.utc) + oidc_state.GUELTIG + timedelta(seconds=1)

    assert oidc_state.einloesen(db, zettel.state, jetzt=spaeter) is None


def test_unbekannter_zettel_wird_abgelehnt(db):
    assert oidc_state.einloesen(db, "nie-vergeben") is None


def test_aufraeumen_entfernt_nur_alte_zettel(db):
    alt = oidc_state.anlegen(db, "authentik")
    neu = oidc_state.anlegen(db, "authentik")
    # Den einen kuenstlich altern lassen.
    alt.created_at = datetime.now(timezone.utc) - oidc_state.GUELTIG - timedelta(minutes=1)
    db.commit()

    assert oidc_state.aufraeumen(db) == 1
    assert oidc_state.einloesen(db, neu.state) is not None


def test_zettel_tragen_unterschiedliche_werte(db):
    a = oidc_state.anlegen(db, "authentik")
    b = oidc_state.anlegen(db, "authentik")
    assert a.state != b.state
    assert a.nonce != b.nonce
    assert a.code_verifier != b.code_verifier


# ---- Passwort-Login gegen ein reines SSO-Konto -----------------------------


def test_passwort_login_gegen_sso_konto_gibt_401(db):
    """Kein 500: das Konto hat keinen Hash, das ist ein Anmeldefehler."""
    db.add(User(username="eva", password_hash=None, role=Role.user))
    db.commit()

    with pytest.raises(HTTPException) as fehler:
        login(form=SimpleNamespace(username="eva", password="beliebig"), db=db)

    assert fehler.value.status_code == 401


def test_sehr_langer_name_passt_in_die_spalte(db, provider):
    """username ist String(64) – ein langer preferred_username darf nicht sprengen."""
    neu = benutzer_aus_claims(db, provider, claims(preferred_username="e" * 200))
    assert len(neu.username) <= 64


# ---- Verknuepfung eines bestehenden Kontos ---------------------------------


def test_verknuepfung_haengt_kennung_an_das_angemeldete_konto(db, provider):
    """Der Weg fuer Konten ohne (passende) E-Mail: Berechtigung aus der Sitzung."""
    eva = User(username="eva", password_hash=hash_password("geheim123"), role=Role.admin)
    db.add(eva)
    db.commit()

    identitaet = identitaet_verknuepfen(db, eva, provider, claims(email=None))

    assert identitaet.user_id == eva.id
    assert identitaet.subject == "abc-123"
    # Ab jetzt meldet dieselbe Kennung genau dieses Konto an – mit seiner Rolle.
    gefunden = benutzer_aus_claims(db, provider, claims())
    assert gefunden.id == eva.id
    assert gefunden.role == Role.admin


def test_verknuepfung_braucht_keine_bestaetigte_email(db, provider):
    """Anders als die Erstverknuepfung: hier buergt die laufende Sitzung."""
    eva = User(username="eva", password_hash=hash_password("geheim123"), role=Role.admin)
    db.add(eva)
    db.commit()

    identitaet_verknuepfen(db, eva, provider, claims(email_verified=False))

    assert db.query(UserIdentity).count() == 1


def test_verknuepfung_zweimal_ist_folgenlos(db, provider):
    eva = User(username="eva", password_hash=hash_password("geheim123"), role=Role.user)
    db.add(eva)
    db.commit()

    erste = identitaet_verknuepfen(db, eva, provider, claims())
    zweite = identitaet_verknuepfen(db, eva, provider, claims())

    assert erste.id == zweite.id
    assert db.query(UserIdentity).count() == 1


def test_fremde_kennung_wird_nicht_umgehaengt(db, provider):
    """Auch eine gueltige Sitzung darf keine fremde Kennung an sich ziehen."""
    otto = User(username="otto", password_hash=hash_password("geheim123"), role=Role.user)
    eva = User(username="eva", password_hash=hash_password("geheim123"), role=Role.admin)
    db.add_all([otto, eva])
    db.flush()
    db.add(UserIdentity(user_id=otto.id, provider="authentik", subject="abc-123"))
    db.commit()

    with pytest.raises(OidcFehler) as fehler:
        identitaet_verknuepfen(db, eva, provider, claims())

    assert fehler.value.schluessel == "kennung_fremd"
    assert db.query(UserIdentity).one().user_id == otto.id


def test_zweite_kennung_desselben_dienstes_wird_abgelehnt(db, provider):
    eva = User(username="eva", password_hash=hash_password("geheim123"), role=Role.user)
    db.add(eva)
    db.flush()
    db.add(UserIdentity(user_id=eva.id, provider="authentik", subject="alt-999"))
    db.commit()

    with pytest.raises(OidcFehler) as fehler:
        identitaet_verknuepfen(db, eva, provider, claims(sub="neu-111"))

    assert fehler.value.schluessel == "schon_verknuepft"


def test_loesen_entfernt_die_verknuepfung(db, provider):
    eva = User(username="eva", password_hash=hash_password("geheim123"), role=Role.user)
    db.add(eva)
    db.flush()
    db.add(UserIdentity(user_id=eva.id, provider="authentik", subject="abc-123"))
    db.commit()

    identitaet_loesen(db, eva, "authentik")

    assert db.query(UserIdentity).count() == 0


def test_letzte_anmeldemoeglichkeit_bleibt_bestehen(db, provider):
    """Ein reines SSO-Konto darf sich nicht selbst aussperren."""
    eva = User(username="eva", password_hash=None, role=Role.user)
    db.add(eva)
    db.flush()
    db.add(UserIdentity(user_id=eva.id, provider="authentik", subject="abc-123"))
    db.commit()

    with pytest.raises(OidcFehler) as fehler:
        identitaet_loesen(db, eva, "authentik")

    assert fehler.value.schluessel == "letzte_anmeldung"
    assert db.query(UserIdentity).count() == 1


def test_zettel_merkt_sich_das_zu_verknuepfende_konto(db):
    """Nur so weiss der Rueckweg, dass er verknuepfen und nicht anmelden soll."""
    eva = User(username="eva", password_hash=hash_password("geheim123"), role=Role.user)
    db.add(eva)
    db.commit()

    zettel = oidc_state.anlegen(db, "authentik", link_user_id=eva.id)
    eingeloest = oidc_state.einloesen(db, zettel.state)

    assert eingeloest.link_user_id == eva.id
    # Eine normale Anmeldung traegt das Feld nicht.
    assert oidc_state.einloesen(db, oidc_state.anlegen(db, "authentik").state).link_user_id is None


def test_app_client_darf_eigene_discovery_adresse_haben():
    """In Authentik hat jede Anwendung standardmaessig einen eigenen Issuer.

    Web- und App-Client brauchen dort zwei getrennte Anwendungen – ohne eigene
    Discovery-Adresse wuerde die iss-Pruefung des App-Tokens fehlschlagen.
    """
    umgebung = dict(VOLLE_UMGEBUNG)
    umgebung["OIDC_AUTHENTIK_NATIVE_DISCOVERY_URL"] = (
        "https://idp.example.com/application/o/vorrania-ios/.well-known/openid-configuration"
    )
    (p,) = provider_aus_umgebung(umgebung, BASIS)

    assert p.app_discovery_url == umgebung["OIDC_AUTHENTIK_NATIVE_DISCOVERY_URL"]
    assert p.discovery_url != p.app_discovery_url


def test_ohne_eigene_adresse_teilen_sich_beide_wege_die_discovery():
    (p,) = provider_aus_umgebung(VOLLE_UMGEBUNG, BASIS)
    assert p.app_discovery_url == p.discovery_url