Files
certain/backend/tests/test_oidc.py
T
ScarriffleandClaude Opus 5 62cb6a017c fix: Ein unerreichbarer Anmeldedienst zeigte eine Seite mit rohem JSON
Auf "Mit Authentik anmelden" folgt der BROWSER einem Link – da steht kein
fetch(), das eine Fehlerantwort auswerten koennte. Scheiterte der Aufruf des
Anmeldedienstes (Tippfehler in der Discovery-Adresse, Name nicht aufloesbar),
landete der Nutzer auf einer weissen Seite mit {"detail": ...} und musste
zurueckdruecken.

Jetzt geht es zurueck in die Oberflaeche, mit dem Grund in der Adresse – genau
wie bei der Rueckkehr vom Anbieter. Beim Anmelden auf die Login-Seite, beim
Verknuepfen auf "Mein Konto".

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-09 19:54:20 +02:00

486 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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