certain/backend/tests/test_oidc.py

"""Anmeldung ueber einen fremden Dienst.

Getestet wird gegen einen ECHTEN, aber gefaelschten Anbieter: eigenes
RSA-Schluesselpaar, eigenes Discovery-Dokument, eigenes JWKS. Ein Doppelgaenger,
der nur "ja" sagt, wuerde genau das nicht pruefen, worauf es hier ankommt –
naemlich dass ein falsch signiertes, abgelaufenes oder fuer jemand anderen
ausgestelltes Token abgewiesen wird.
"""

import json
import secrets
import time

import pytest
from authlib.jose import JsonWebKey, JsonWebToken

from conftest import ADMIN_PASSWORD, login_ready, new_user

ISSUER = "https://authentik.test"
CLIENT_ID = "certain-test"
DISCOVERY = f"{ISSUER}/application/o/certain/.well-known/openid-configuration"


@pytest.fixture()
def schluessel():
    """Ein RSA-Schluesselpaar, mit dem der gefaelschte Anbieter signiert."""
    return JsonWebKey.generate_key("RSA", 2048, is_private=True)


@pytest.fixture()
def anbieter(monkeypatch, schluessel):
    """Richtet einen Anbieter ein und faengt seine Netzaufrufe ab.

    Gibt eine Funktion zurueck, mit der ein Test bestimmt, welche Angaben im
    naechsten ID-Token stehen.
    """
    from app import oidc_config
    from app.services import oidc as oidc_service

    monkeypatch.setenv("APP_OIDC_PROVIDERS", "authentik")
    monkeypatch.setenv("APP_OIDC_AUTHENTIK_DISCOVERY_URL", DISCOVERY)
    monkeypatch.setenv("APP_OIDC_AUTHENTIK_CLIENT_ID", CLIENT_ID)
    monkeypatch.setenv("APP_OIDC_AUTHENTIK_CLIENT_SECRET", "geheim")
    monkeypatch.setenv("APP_OIDC_AUTHENTIK_LABEL", "Authentik")
    oidc_config.zuruecksetzen()
    oidc_service.cache_leeren()

    zustand = {
        "claims": {},
        "signieren_mit": schluessel,
        "token_status": 200,
    }

    oeffentlich = json.loads(schluessel.as_json(is_private=False))
    oeffentlich["kid"] = oeffentlich.get("kid") or "test"
    jwks = {"keys": [oeffentlich]}
    dokument = {
        "issuer": ISSUER,
        "authorization_endpoint": f"{ISSUER}/application/o/authorize/",
        "token_endpoint": f"{ISSUER}/application/o/token/",
        "jwks_uri": f"{ISSUER}/jwks",
        "userinfo_endpoint": f"{ISSUER}/application/o/userinfo/",
    }

    def _id_token(nonce: str) -> str:
        jetzt = int(time.time())
        inhalt = {
            "iss": ISSUER, "aud": CLIENT_ID, "sub": "sub-standard",
            "iat": jetzt, "exp": jetzt + 300, "nonce": nonce,
        }
        inhalt.update(zustand["claims"])
        jwt = JsonWebToken(["RS256"])
        return jwt.encode({"alg": "RS256"}, inhalt,
                          zustand["signieren_mit"]).decode("ascii")

    class FakeAntwort:
        def __init__(self, daten, status=200):
            self._daten = daten
            self.status_code = status

        def json(self):
            return self._daten

        def raise_for_status(self):
            if self.status_code >= 400:
                import httpx
                raise httpx.HTTPError(f"status {self.status_code}")

    class FakeClient:
        def __init__(self, *a, **kw):
            pass

        def __enter__(self):
            return self

        def __exit__(self, *a):
            return False

        def get(self, url, **kw):
            if url == DISCOVERY:
                return FakeAntwort(dokument)
            if url == dokument["jwks_uri"]:
                return FakeAntwort(jwks)
            if url == dokument["userinfo_endpoint"]:
                return FakeAntwort({"sub": "sub-standard"})
            return FakeAntwort({}, 404)

        def post(self, url, data=None, headers=None, auth=None, **kw):
            if url != dokument["token_endpoint"]:
                return FakeAntwort({}, 404)
            if zustand["token_status"] >= 400:
                return FakeAntwort({"error": "invalid_grant"}, zustand["token_status"])
            zustand["gesehen"] = {"data": dict(data or {}), "auth": auth}
            nonce = zustand.get("nonce", "")
            return FakeAntwort({
                "access_token": "at", "token_type": "Bearer",
                "id_token": _id_token(nonce),
            })

    monkeypatch.setattr("app.services.oidc.httpx.Client", FakeClient)
    yield zustand
    oidc_config.zuruecksetzen()
    oidc_service.cache_leeren()


def _beginne(client, pfad="/api/auth/oidc/authentik/start", **params):
    """Anmeldung/Verknuepfung starten und den state aus der Adresse holen."""
    from urllib.parse import parse_qs, urlparse

    r = client.get(pfad, params=params, follow_redirects=False)
    assert r.status_code == 303, r.text
    frage = parse_qs(urlparse(r.headers["location"]).query)
    return frage["state"][0], frage["nonce"][0], frage


def _zurueck(client, zustand, state, nonce, code="guter-code"):
    zustand["nonce"] = nonce
    return client.get("/api/auth/oidc/authentik/callback",
                      params={"code": code, "state": state},
                      follow_redirects=False)


# --- Einrichtung -------------------------------------------------------------


def test_ohne_konfiguration_gibt_es_keinen_anbieter(client):
    """Nichts eingerichtet heisst: Die Login-Seite zeigt keinen Knopf."""
    from app import oidc_config

    oidc_config.zuruecksetzen()
    r = client.get("/api/auth/oidc/providers")
    assert r.status_code == 200
    assert r.json()["providers"] == []


def test_anbieter_wird_ohne_geheimnis_gemeldet(client, anbieter):
    r = client.get("/api/auth/oidc/providers")
    assert r.json()["providers"] == [{"name": "authentik", "label": "Authentik"}]
    assert "geheim" not in r.text, "Client-Secret nach aussen gegeben"


def test_unbekannter_anbieter_ist_ein_vierhundertvier(client, anbieter):
    r = client.get("/api/auth/oidc/gibtsnicht/start", follow_redirects=False)
    assert r.status_code == 404


# --- Der Weg hin -------------------------------------------------------------


def test_start_schickt_mit_pkce_und_nonce_zum_anbieter(client, anbieter):
    state, nonce, frage = _beginne(client)
    assert frage["client_id"] == [CLIENT_ID]
    assert frage["response_type"] == ["code"]
    assert frage["code_challenge_method"] == ["S256"], "kein PKCE"
    assert frage["code_challenge"][0]
    assert state and nonce
    assert "openid" in frage["scope"][0]
    # Der verifier darf NICHT mitgeschickt werden – sonst waere PKCE sinnlos.
    assert "code_verifier" not in frage


def test_zwei_anlaeufe_bekommen_verschiedene_kennungen(client, anbieter):
    s1, n1, _ = _beginne(client)
    s2, n2, _ = _beginne(client)
    assert s1 != s2 and n1 != n2


# --- Der Weg zurueck ---------------------------------------------------------


def test_erste_anmeldung_ohne_verknuepfung_wird_abgewiesen(client, anbieter):
    """Vorgabe: kein Konto von selbst. Wer schon eines hat, soll verknuepfen –
    sonst bekaeme er ein zweites, leeres."""
    state, nonce, _ = _beginne(client)
    r = _zurueck(client, anbieter, state, nonce)
    assert r.status_code == 303
    assert "oidc_error" in r.headers["location"]
    assert client.get("/api/auth/me").status_code == 401, "trotzdem angemeldet"


def test_mit_auto_create_entsteht_ein_konto(client, anbieter, monkeypatch):
    from app import oidc_config

    monkeypatch.setenv("APP_OIDC_AUTHENTIK_AUTO_CREATE", "true")
    oidc_config.zuruecksetzen()
    anbieter["claims"] = {"sub": "neu-" + secrets.token_hex(4),
                          "email": "neu@example.com", "email_verified": True,
                          "preferred_username": "neuer" + secrets.token_hex(3)}
    state, nonce, _ = _beginne(client)
    r = _zurueck(client, anbieter, state, nonce)
    assert r.status_code == 303, r.text
    assert "oidc_error" not in r.headers["location"], r.headers["location"]

    ich = client.get("/api/auth/me")
    assert ich.status_code == 200, "nicht angemeldet"
    assert ich.json()["role"] == "user", "neues Konto darf kein Admin sein"
    client.post("/api/auth/logout")


def test_falsche_signatur_kommt_nicht_herein(client, anbieter, monkeypatch):
    """Der Kern: Ein Token, das jemand anders signiert hat, ist wertlos."""
    from app import oidc_config

    monkeypatch.setenv("APP_OIDC_AUTHENTIK_AUTO_CREATE", "true")
    oidc_config.zuruecksetzen()
    anbieter["signieren_mit"] = JsonWebKey.generate_key("RSA", 2048, is_private=True)
    anbieter["claims"] = {"sub": "faelscher"}
    state, nonce, _ = _beginne(client)
    r = _zurueck(client, anbieter, state, nonce)
    assert "oidc_error" in r.headers["location"]
    assert client.get("/api/auth/me").status_code == 401


def test_fremder_empfaenger_kommt_nicht_herein(client, anbieter, monkeypatch):
    """Ein Token, das fuer eine ANDERE Anwendung ausgestellt wurde."""
    from app import oidc_config

    monkeypatch.setenv("APP_OIDC_AUTHENTIK_AUTO_CREATE", "true")
    oidc_config.zuruecksetzen()
    anbieter["claims"] = {"sub": "fremd", "aud": "eine-andere-app"}
    state, nonce, _ = _beginne(client)
    r = _zurueck(client, anbieter, state, nonce)
    assert "oidc_error" in r.headers["location"]
    assert client.get("/api/auth/me").status_code == 401


def test_abgelaufenes_token_kommt_nicht_herein(client, anbieter, monkeypatch):
    from app import oidc_config

    monkeypatch.setenv("APP_OIDC_AUTHENTIK_AUTO_CREATE", "true")
    oidc_config.zuruecksetzen()
    jetzt = int(time.time())
    anbieter["claims"] = {"sub": "alt", "iat": jetzt - 4000, "exp": jetzt - 3600}
    state, nonce, _ = _beginne(client)
    r = _zurueck(client, anbieter, state, nonce)
    assert "oidc_error" in r.headers["location"]
    assert client.get("/api/auth/me").status_code == 401


def test_falscher_nonce_kommt_nicht_herein(client, anbieter, monkeypatch):
    """Sonst liesse sich ein anderswo erbeutetes, gueltiges Token einreichen."""
    from app import oidc_config

    monkeypatch.setenv("APP_OIDC_AUTHENTIK_AUTO_CREATE", "true")
    oidc_config.zuruecksetzen()
    anbieter["claims"] = {"sub": "nonce-test"}
    state, nonce, _ = _beginne(client)
    r = _zurueck(client, anbieter, state, "ein-ganz-anderer-nonce")
    assert "oidc_error" in r.headers["location"]
    assert client.get("/api/auth/me").status_code == 401


def test_unbekannter_state_wird_abgewiesen(client, anbieter):
    """Ohne diese Pruefung waere jede Rueckkehr gueltig."""
    r = client.get("/api/auth/oidc/authentik/callback",
                   params={"code": "x", "state": "nie-vergeben"},
                   follow_redirects=False)
    assert r.status_code == 303
    assert "oidc_error" in r.headers["location"]


def test_ein_state_gilt_nur_einmal(client, anbieter, monkeypatch):
    """Ein Code darf genau einmal eingeloest werden."""
    from app import oidc_config

    monkeypatch.setenv("APP_OIDC_AUTHENTIK_AUTO_CREATE", "true")
    oidc_config.zuruecksetzen()
    anbieter["claims"] = {"sub": "einmal-" + secrets.token_hex(4),
                          "preferred_username": "einmal" + secrets.token_hex(3)}
    state, nonce, _ = _beginne(client)
    assert "oidc_error" not in _zurueck(client, anbieter, state, nonce).headers["location"]
    client.post("/api/auth/logout")

    zweiter = _zurueck(client, anbieter, state, nonce)
    assert "oidc_error" in zweiter.headers["location"], "state ein zweites Mal gueltig"
    assert client.get("/api/auth/me").status_code == 401


def test_offene_weiterleitung_ist_nicht_moeglich(client, anbieter):
    """`weiter` darf nur auf diese Seite zeigen – sonst waere der Endpunkt eine
    Schleuder auf fremde Seiten, die wie Certain aussehen."""
    state, nonce, _ = _beginne(client, weiter="https://boese.example/phish")
    r = _zurueck(client, anbieter, state, nonce)
    assert not r.headers["location"].startswith("http"), r.headers["location"]


# --- Verknuepfen -------------------------------------------------------------


def test_bestehendes_konto_verknuepfen_und_danach_damit_anmelden(admin, client, anbieter):
    """Der eigentliche Weg: einmal normal anmelden, verknuepfen, fertig."""
    sub = "guido-" + secrets.token_hex(4)
    anbieter["claims"] = {"sub": sub, "email": "guido@example.com",
                          "email_verified": True}

    state, nonce, _ = _beginne(admin, "/api/auth/oidc/authentik/link")
    r = _zurueck(admin, anbieter, state, nonce)
    assert "oidc_error" not in r.headers["location"], r.headers["location"]
    assert "oidc_linked" in r.headers["location"]

    ident = admin.get("/api/auth/oidc/identities").json()
    assert [i["provider"] for i in ident["identities"]] == ["authentik"]
    assert ident["hat_passwort"] is True

    # Abmelden und jetzt ueber den Anbieter hereinkommen – dasselbe Konto.
    admin.post("/api/auth/logout")
    state, nonce, _ = _beginne(admin)
    r = _zurueck(admin, anbieter, state, nonce)
    assert "oidc_error" not in r.headers["location"], r.headers["location"]
    ich = admin.get("/api/auth/me")
    assert ich.status_code == 200
    assert ich.json()["username"] == "admin", "in einem fremden Konto gelandet"

    # Aufraeumen, damit die uebrigen Tests den Admin unveraendert vorfinden.
    ident = admin.get("/api/auth/oidc/identities").json()["identities"][0]
    assert admin.delete(f"/api/auth/oidc/identities/{ident['id']}").status_code == 204
    admin.post("/api/auth/login",
               data={"username": "admin", "password": ADMIN_PASSWORD})


def test_eine_kennung_gehoert_nur_einem_konto(admin, anbieter):
    """Sonst landete derselbe Anmeldende mal hier und mal dort."""
    sub = "geteilt-" + secrets.token_hex(4)
    anbieter["claims"] = {"sub": sub}

    state, nonce, _ = _beginne(admin, "/api/auth/oidc/authentik/link")
    assert "oidc_linked" in _zurueck(admin, anbieter, state, nonce).headers["location"]
    eigene = admin.get("/api/auth/oidc/identities").json()["identities"][0]

    username, passwort = new_user(admin)
    anderer, _ = login_ready(username, passwort)
    state, nonce, _ = _beginne(anderer, "/api/auth/oidc/authentik/link")
    r = _zurueck(anderer, anbieter, state, nonce)
    assert "oidc_error" in r.headers["location"], "dieselbe Kennung zweimal verknuepft"
    assert anderer.get("/api/auth/oidc/identities").json()["identities"] == []

    admin.delete(f"/api/auth/oidc/identities/{eigene['id']}")


def test_verknuepfen_verlangt_eine_anmeldung(client, anbieter):
    r = client.get("/api/auth/oidc/authentik/link", follow_redirects=False)
    assert r.status_code == 401


def test_letzte_verknuepfung_bleibt_wenn_es_kein_passwort_gibt(admin, anbieter, monkeypatch):
    """Sonst schliesst sich der Nutzer aus seinem eigenen Konto aus."""
    from app import oidc_config
    from app.database import SessionLocal
    from app.models import User, UserIdentity

    monkeypatch.setenv("APP_OIDC_AUTHENTIK_AUTO_CREATE", "true")
    oidc_config.zuruecksetzen()
    sub = "ohnepw-" + secrets.token_hex(4)
    anbieter["claims"] = {"sub": sub, "preferred_username": "ohnepw" + secrets.token_hex(3)}
    state, nonce, _ = _beginne(admin)
    admin.post("/api/auth/logout")
    # Frischer Client, damit die Sitzung des neuen Kontos gilt.
    from fastapi.testclient import TestClient
    from app.main import app
    neu = TestClient(app)
    state, nonce, _ = _beginne(neu)
    assert "oidc_error" not in _zurueck(neu, anbieter, state, nonce).headers["location"]

    daten = neu.get("/api/auth/oidc/identities").json()
    assert daten["hat_passwort"] is False, "SSO-Konto hat wider Erwarten ein Passwort"
    ident_id = daten["identities"][0]["id"]
    r = neu.delete(f"/api/auth/oidc/identities/{ident_id}")
    assert r.status_code == 400, "letzte Anmeldung liess sich entfernen"
    assert "Passwort" in r.json()["detail"]

    with SessionLocal() as s:
        eintrag = s.query(UserIdentity).filter(UserIdentity.subject == sub).first()
        assert eintrag is not None, "Verknuepfung wurde doch geloescht"
        s.delete(s.get(User, eintrag.user_id))
        s.commit()


# --- Zusammenspiel mit dem Passwort-Login ------------------------------------


def test_passwort_login_bleibt_unveraendert(client):
    """Der neue Weg kommt DAZU. Der alte muss weiter funktionieren."""
    r = client.post("/api/auth/login",
                    data={"username": "admin", "password": ADMIN_PASSWORD})
    assert r.status_code == 200, r.text
    client.post("/api/auth/logout")


def test_konto_ohne_passwort_faellt_beim_passwort_login_nicht_durch(client, anbieter,
                                                                   monkeypatch):
    """argon2 bekaeme sonst None und wuerfe einen Fehler, aus dem sich ablesen
    liesse, dass es diesen Benutzernamen gibt."""
    from app import oidc_config
    from app.database import SessionLocal
    from app.models import User, UserIdentity

    monkeypatch.setenv("APP_OIDC_AUTHENTIK_AUTO_CREATE", "true")
    oidc_config.zuruecksetzen()
    name = "sso" + secrets.token_hex(4)
    sub = "sso-" + secrets.token_hex(4)
    anbieter["claims"] = {"sub": sub, "preferred_username": name}
    state, nonce, _ = _beginne(client)
    assert "oidc_error" not in _zurueck(client, anbieter, state, nonce).headers["location"]
    client.post("/api/auth/logout")

    r = client.post("/api/auth/login", data={"username": name, "password": "irgendwas"})
    assert r.status_code == 401, r.text
    assert "Passwort falsch" in r.json()["detail"]

    with SessionLocal() as s:
        eintrag = s.query(UserIdentity).filter(UserIdentity.subject == sub).first()
        s.delete(s.get(User, eintrag.user_id))
        s.commit()


# --- Rueckleit-Adresse hinter einem Proxy ------------------------------------


def test_oeffentliche_adresse_schlaegt_die_der_anfrage(client, anbieter, monkeypatch):
    """Hinter einem Reverse Proxy kaeme aus der Anfrage die INTERNE Adresse –
    der Anmeldedienst weist sie ab, und die Fehlermeldung sagt nicht, warum."""
    from urllib.parse import parse_qs, urlparse

    from app.config import get_settings

    get_settings.cache_clear()
    monkeypatch.setenv("APP_PUBLIC_BASE_URL", "https://certain.example.com")
    try:
        r = client.get("/api/auth/oidc/authentik/start", follow_redirects=False)
        frage = parse_qs(urlparse(r.headers["location"]).query)
        assert frage["redirect_uri"] == [
            "https://certain.example.com/api/auth/oidc/authentik/callback"
        ], frage["redirect_uri"]
    finally:
        get_settings.cache_clear()


def test_ohne_einstellung_kommt_die_adresse_aus_der_anfrage(client, anbieter):
    """Ohne Proxy und in der Entwicklung soll nichts eingestellt werden muessen."""
    from urllib.parse import parse_qs, urlparse

    from app.config import get_settings

    get_settings.cache_clear()
    r = client.get("/api/auth/oidc/authentik/start", follow_redirects=False)
    frage = parse_qs(urlparse(r.headers["location"]).query)
    assert frage["redirect_uri"][0].endswith("/api/auth/oidc/authentik/callback")


def test_unerreichbarer_anbieter_zeigt_keine_json_seite(client, anbieter, monkeypatch):
    """Hier steht ein Browser, der einem Link gefolgt ist. Eine Fehlerantwort
    waere eine Seite mit rohem JSON – der Grund gehoert in die Oberflaeche."""
    from app.services import oidc as oidc_service

    def kaputt(*a, **kw):
        raise oidc_service.OidcFehler("Der Anmeldedienst ist nicht erreichbar: "
                                      "[Errno -2] Name or service not known")

    monkeypatch.setattr("app.services.oidc.start", kaputt)
    r = client.get("/api/auth/oidc/authentik/start", follow_redirects=False)
    assert r.status_code == 303, r.text
    ziel = r.headers["location"]
    assert ziel.startswith("/login"), ziel
    assert "oidc_error" in ziel
    assert "Name+or+service" in ziel or "Name%20or%20service" in ziel


# --- Der Benutzername ist ein Etikett, kein Schluessel ------------------------


def test_umbenennen_laesst_alle_daten_stehen(admin):
    """Der Kern: Buchungen, Konten und Vertraege haengen an der NUMMER des
    Nutzers. Aendert sich sein Name – etwa weil der Anmeldedienst einen neuen
    schickt –, darf davon nichts mitwandern."""
    username, temp = new_user(admin)
    # login_ready erledigt den Pflicht-Wechsel und gibt das NEUE Passwort
    # zurueck – das temporaere gilt danach nicht mehr.
    nutzer, passwort = login_ready(username, temp)

    konto = nutzer.post("/api/accounts", json={
        "name": "Giro " + secrets.token_hex(3), "type": "girokonto",
        "currency": "EUR", "opening_balance_cents": 12345}).json()
    nutzer.post("/api/transactions", json={
        "type": "expense", "date": "2026-05-05", "account_id": konto["id"],
        "amount_cents": 999})

    ich = nutzer.get("/api/auth/me").json()
    neuer_name = "umbenannt" + secrets.token_hex(4)
    r = admin.patch(f"/api/users/{ich['id']}", json={"username": neuer_name})
    assert r.status_code == 200, r.text
    assert r.json()["username"] == neuer_name
    assert r.json()["id"] == ich["id"], "Umbenennen hat einen NEUEN Nutzer erzeugt"

    # Dieselbe Sitzung, dieselben Daten – nichts ist umgezogen.
    assert nutzer.get("/api/auth/me").json()["username"] == neuer_name
    konten = nutzer.get("/api/accounts").json()
    assert [k["id"] for k in konten] == [konto["id"]]
    assert konten[0]["balance_cents"] == 12345 - 999

    # Und der neue Name funktioniert beim Anmelden.
    nutzer.post("/api/auth/logout")
    r = nutzer.post("/api/auth/login",
                    data={"username": neuer_name, "password": passwort})
    assert r.status_code == 200, r.text


def test_umbenennen_auf_einen_vergebenen_namen_wird_abgewiesen(admin):
    """Eindeutig muss er bleiben – er ist das, was eingetippt wird."""
    username, passwort = new_user(admin)
    nutzer, _ = login_ready(username, passwort)
    ich = nutzer.get("/api/auth/me").json()

    r = admin.patch(f"/api/users/{ich['id']}", json={"username": "admin"})
    assert r.status_code == 409, r.text
    assert nutzer.get("/api/auth/me").json()["username"] == username


def test_anbieter_darf_ein_reines_sso_konto_umbenennen(client, anbieter, monkeypatch):
    """Dort gibt es nichts einzutippen – der Name ist ein Etikett, und der
    Anbieter bestimmt es."""
    from app import oidc_config
    from app.database import SessionLocal
    from app.models import User, UserIdentity

    monkeypatch.setenv("APP_OIDC_AUTHENTIK_AUTO_CREATE", "true")
    oidc_config.zuruecksetzen()
    sub = "rename-" + secrets.token_hex(4)
    erst = "vorher" + secrets.token_hex(3)
    anbieter["claims"] = {"sub": sub, "preferred_username": erst, "name": "Vorher Nachname"}
    state, nonce, _ = _beginne(client)
    assert "oidc_error" not in _zurueck(client, anbieter, state, nonce).headers["location"]
    ich = client.get("/api/auth/me").json()
    assert ich["username"] == erst
    assert ich["display_name"] == "Vorher Nachname"
    client.post("/api/auth/logout")

    # Beim Anbieter umbenannt – naechste Anmeldung zieht nach.
    danach = "nachher" + secrets.token_hex(3)
    anbieter["claims"] = {"sub": sub, "preferred_username": danach,
                          "name": "Nachher Nachname"}
    state, nonce, _ = _beginne(client)
    assert "oidc_error" not in _zurueck(client, anbieter, state, nonce).headers["location"]
    jetzt = client.get("/api/auth/me").json()
    assert jetzt["id"] == ich["id"], "es ist ein ZWEITES Konto entstanden"
    assert jetzt["username"] == danach, "Namensaenderung nicht uebernommen"
    assert jetzt["display_name"] == "Nachher Nachname"
    client.post("/api/auth/logout")

    with SessionLocal() as s:
        eintrag = s.query(UserIdentity).filter(UserIdentity.subject == sub).first()
        s.delete(s.get(User, eintrag.user_id))
        s.commit()


def test_anbieter_ruehrt_den_namen_eines_passwort_kontos_nicht_an(admin, anbieter):
    """Dort gehoert der Name dem Nutzer. Ihn hinter dessen Ruecken zu aendern
    hiesse, ihm seine Anmeldung wegzunehmen."""
    sub = "eigen-" + secrets.token_hex(4)
    anbieter["claims"] = {"sub": sub, "preferred_username": "ganzandersername",
                          "name": "Anzeigename vom Anbieter"}

    state, nonce, _ = _beginne(admin, "/api/auth/oidc/authentik/link")
    assert "oidc_linked" in _zurueck(admin, anbieter, state, nonce).headers["location"]
    admin.post("/api/auth/logout")

    state, nonce, _ = _beginne(admin)
    assert "oidc_error" not in _zurueck(admin, anbieter, state, nonce).headers["location"]
    ich = admin.get("/api/auth/me").json()
    assert ich["username"] == "admin", "der Anbieter hat den Namen ueberschrieben"
    # Der Anzeigename darf er dagegen setzen – der ist reine Anzeige.
    assert ich["display_name"] == "Anzeigename vom Anbieter"

    ident = admin.get("/api/auth/oidc/identities").json()["identities"][0]
    admin.delete(f"/api/auth/oidc/identities/{ident['id']}")
    admin.patch(f"/api/users/{ich['id']}", json={"display_name": None})
    admin.post("/api/auth/login",
               data={"username": "admin", "password": ADMIN_PASSWORD})


def test_vergebener_name_vom_anbieter_laesst_den_alten_stehen(client, anbieter,
                                                              monkeypatch):
    """Eine Anmeldung daran scheitern zu lassen waere die schlechtere Antwort."""
    from app import oidc_config
    from app.database import SessionLocal
    from app.models import User, UserIdentity

    monkeypatch.setenv("APP_OIDC_AUTHENTIK_AUTO_CREATE", "true")
    oidc_config.zuruecksetzen()
    sub = "kollision-" + secrets.token_hex(4)
    eigen = "eigener" + secrets.token_hex(3)
    anbieter["claims"] = {"sub": sub, "preferred_username": eigen}
    state, nonce, _ = _beginne(client)
    _zurueck(client, anbieter, state, nonce)
    client.post("/api/auth/logout")

    # Der Anbieter schickt jetzt einen Namen, den es hier schon gibt.
    anbieter["claims"] = {"sub": sub, "preferred_username": "admin"}
    state, nonce, _ = _beginne(client)
    assert "oidc_error" not in _zurueck(client, anbieter, state, nonce).headers["location"]
    assert client.get("/api/auth/me").json()["username"] == eigen, \
        "fremder Name uebernommen"
    client.post("/api/auth/logout")

    with SessionLocal() as s:
        eintrag = s.query(UserIdentity).filter(UserIdentity.subject == sub).first()
        s.delete(s.get(User, eintrag.user_id))
        s.commit()