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()