Anmeldung wahlweise ueber OpenID Connect (Authentik)

Neben dem Passwort-Login gibt es jetzt einen zweiten, gleichberechtigten Weg:
OpenID Connect gegen einen selbst gehosteten Anmeldedienst. Ohne Konfiguration
aendert sich nichts - keine SSO-Knoepfe, Passwort-Login wie bisher.

Der Anmeldedienst ist NUR fuer die Anmeldung zustaendig. Danach stellt Vorrania
sein eigenes Token aus wie beim Passwort-Login, deshalb bleiben deps.py, die
Rollenpruefung und die langlebigen API-Tokens unveraendert.

Backend
- Anbieter kommen aus der Umgebung (OIDC_PROVIDERS + OIDC_<KUERZEL>_*). Ein
  zweiter Anbieter ist damit reine Konfiguration, ohne Codeaenderung.
- Browser-Weg (vertraulicher Client) und App-Weg (oeffentlicher Client, PKCE)
  enden beide im selben Token-Schema.
- state/nonce/PKCE-Verifier liegen in einer kurzlebigen Tabelle statt in einem
  Cookie: Starlettes CORS-Schicht spiegelt bei allow_origins=["*"] zusammen mit
  allow_credentials die Herkunft zurueck, sobald ein Request ein Cookie traegt.
  Ohne Cookie entsteht diese Wechselwirkung erst gar nicht - und der Ablauf
  bleibt mit der db-Fixture pruefbar.
- Zuordnung ueber die sub des Anbieters; die Erstverknuepfung laeuft ueber die
  BESTAETIGTE E-Mail-Adresse. Rollen kommen nie aus dem Anbieter, und ein
  gleichlautender preferred_username uebernimmt nie ein bestehendes Konto.
- users.password_hash darf jetzt leer sein (reine SSO-Konten). verify_password
  faengt das ab, sonst antwortete der Login mit 500 statt 401.

Verknuepfen bestehender Konten
- Neue Seite "Mein Konto" (fuer alle Benutzer, nicht nur Admins): verknuepfen
  und wieder loesen. Die Berechtigung stammt aus der laufenden Sitzung, es
  braucht also weder eine hinterlegte Adresse noch Admin-Rechte.
- Der Start laeuft ueber POST, weil eine Seitennavigation das Bearer-Token
  nicht mitschicken kann.
- Die letzte Anmeldemoeglichkeit eines Kontos laesst sich nicht loesen.

iOS
- AppAuth (erste Paketabhaengigkeit) mit PKCE im System-Browser, nie in einem
  eingebetteten WebView, und ohne Client-Secret.
- offline_access nur im App-Weg: der Browser-Weg braucht kein Refresh-Token und
  bekaeme sonst einen ueberfluessigen Zustimmungsdialog.
- Anmeldezustand je Serverprofil im Keychain; abgelaufene Token erneuert die App
  still im Hintergrund. Gleichzeitige 401er teilen sich eine Erneuerung.

Deployment
- docker-compose reicht die ganze .env durch. Einzeln aufzuzaehlen ginge nur mit
  fest verdrahteten Kuerzeln - genau das soll ein weiterer Anbieter nicht
  erfordern.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
2026-09-09 21:55:38 +02:00
co-authored by Claude Opus 5
parent 8d4954610b
commit 3d0ff54bf5
35 changed files with 2772 additions and 24 deletions
+412
View File
@@ -0,0 +1,412 @@
"""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