"""OpenID Connect: Discovery, Token-Pruefung und Zuordnung zu Vorrania-Konten. Der Protokoll-Teil laeuft ueber Authlib – Signaturpruefung, PKCE und Code-Tausch sind nichts, was man selbst schreiben sollte. Bewusst die SYNCHRONE Variante von Authlib: alle Routen dieses Projekts sind normale ``def``-Funktionen im Threadpool, die Datenbank-Sitzung ist synchron, und ausgehendes HTTP laeuft schon in ``off.py`` ueber ein synchrones ``httpx``. :func:`benutzer_aus_claims` ist absichtlich frei von HTTP: die gesamte Zuordnungs- und Anlage-Logik laesst sich damit mit einer Datenbank-Sitzung und einem selbstgebauten Claims-Wörterbuch pruefen. """ from __future__ import annotations import logging import re import secrets import time from datetime import datetime, timedelta, timezone from typing import Any import httpx from authlib.integrations.httpx_client import OAuth2Client from authlib.jose import JsonWebKey, jwt from authlib.jose.errors import JoseError from authlib.oidc.core import CodeIDToken from sqlalchemy import func from sqlalchemy.orm import Session from ..models import OidcPendingLink, Role, User, UserIdentity from ..oidc_config import OidcProvider from ..security import hash_api_token, verify_password logger = logging.getLogger(__name__) TIMEOUT_SECONDS = 10.0 # Discovery-Dokument und Schluessel aendern sich selten; eine Stunde Vorrat spart # bei jeder Anmeldung zwei Abfragen beim Anbieter. CACHE_SEKUNDEN = 3600.0 # Etwas Spielraum fuer auseinanderlaufende Uhren zwischen Server und Anbieter. UHREN_SPIELRAUM = 60 # So lange darf die Entscheidung „neues Konto oder verknuepfen" offenbleiben. OFFEN_GUELTIG = timedelta(minutes=10) class OidcFehler(Exception): """Anmeldung ueber den Anbieter ist gescheitert. ``schluessel`` landet als Kennung in der Weiterleitung zur Web-Oberflaeche, ``args[0]`` ist der Klartext fuer Protokoll und API-Antwort. """ def __init__(self, schluessel: str, meldung: str) -> None: super().__init__(meldung) self.schluessel = schluessel _metadaten_cache: dict[str, tuple[float, dict[str, Any]]] = {} _jwks_cache: dict[str, tuple[float, Any]] = {} def cache_leeren() -> None: """Nur fuer Tests und zum erzwungenen Neuladen.""" _metadaten_cache.clear() _jwks_cache.clear() def _holen(url: str) -> dict[str, Any]: try: antwort = httpx.get(url, timeout=TIMEOUT_SECONDS, follow_redirects=True) antwort.raise_for_status() return antwort.json() except (httpx.HTTPError, ValueError) as fehler: # Anders als beim Produkt-Lookup darf das hier NICHT still zu None werden – # ein nicht erreichbarer Anbieter muss als Fehler sichtbar sein. raise OidcFehler( "anbieter_nicht_erreichbar", f"Anmeldeanbieter nicht erreichbar ({url}): {fehler}", ) from fehler def metadaten(discovery_url: str, erzwingen: bool = False) -> dict[str, Any]: """Discovery-Dokument (gepuffert). Gepuffert wird nach der Adresse, nicht nach dem Anbieter: Browser- und App-Client koennen bei Authentik zwei getrennte Anwendungen mit je eigenem Issuer und eigenem Schluesselsatz sein. """ eintrag = _metadaten_cache.get(discovery_url) if not erzwingen and eintrag and eintrag[0] > time.monotonic(): return eintrag[1] daten = _holen(discovery_url) for pflicht in ("issuer", "authorization_endpoint", "token_endpoint", "jwks_uri"): if not daten.get(pflicht): raise OidcFehler( "discovery_unvollstaendig", f"Discovery-Dokument unter {discovery_url} ohne '{pflicht}'.", ) _metadaten_cache[discovery_url] = (time.monotonic() + CACHE_SEKUNDEN, daten) return daten def _jwks(discovery_url: str, erzwingen: bool = False) -> Any: eintrag = _jwks_cache.get(discovery_url) if not erzwingen and eintrag and eintrag[0] > time.monotonic(): return eintrag[1] schluessel = JsonWebKey.import_key_set(_holen(metadaten(discovery_url)["jwks_uri"])) _jwks_cache[discovery_url] = (time.monotonic() + CACHE_SEKUNDEN, schluessel) return schluessel def _client(provider: OidcProvider, redirect_uri: str) -> OAuth2Client: return OAuth2Client( client_id=provider.client_id, # Die App ist ein oeffentlicher Client und schickt nie ein Secret; nur der # vertrauliche Web-Client hat eins. client_secret=provider.client_secret or None, scope=provider.scopes, redirect_uri=redirect_uri, code_challenge_method="S256", timeout=TIMEOUT_SECONDS, ) def anmelde_url(provider: OidcProvider, state: str, nonce: str, code_verifier: str) -> str: """Adresse, zu der der Browser fuer die Anmeldung geschickt wird.""" meta = metadaten(provider.discovery_url) with _client(provider, provider.redirect_uri) as client: url, _ = client.create_authorization_url( meta["authorization_endpoint"], state=state, nonce=nonce, code_verifier=code_verifier, ) return url def code_einloesen(provider: OidcProvider, code: str, code_verifier: str) -> dict[str, Any]: """Tauscht den Autorisierungs-Code gegen Tokens (vertraulicher Web-Client).""" meta = metadaten(provider.discovery_url) try: with _client(provider, provider.redirect_uri) as client: return client.fetch_token( meta["token_endpoint"], grant_type="authorization_code", code=code, code_verifier=code_verifier, redirect_uri=provider.redirect_uri, ) except OidcFehler: raise except Exception as fehler: # Authlib wirft je nach Ursache sehr Verschiedenes. raise OidcFehler( "code_tausch_fehlgeschlagen", f"Code-Tausch gescheitert: {fehler}" ) from fehler def pruefe_id_token( provider: OidcProvider, id_token: str, nonce: str | None = None, erlaubte_audiences: tuple[str, ...] | None = None, discovery_url: str | None = None, ) -> dict[str, Any]: """Prueft Signatur, ``iss``, ``aud``, ``exp`` (und ``nonce``, falls bekannt). ``erlaubte_audiences`` schraenkt die zulaessigen Empfaenger ein: der Web-Rueckweg akzeptiert nur die vertrauliche Client-ID, der App-Weg nur die der registrierten oeffentlichen Clients. ``discovery_url`` bestimmt, gegen welchen Issuer und welchen Schluesselsatz geprueft wird. """ quelle = discovery_url or provider.discovery_url meta = metadaten(quelle) audiences = erlaubte_audiences or provider.erlaubte_audiences if not audiences: raise OidcFehler( "keine_audience", f"Fuer '{provider.slug}' ist keine Client-ID hinterlegt." ) optionen = { "iss": {"essential": True, "value": meta["issuer"]}, "aud": {"essential": True, "values": list(audiences)}, "exp": {"essential": True}, "sub": {"essential": True}, } params = {"nonce": nonce} if nonce else None def _entschluesseln(schluessel: Any) -> Any: return jwt.decode( id_token, key=schluessel, claims_cls=CodeIDToken, claims_options=optionen, claims_params=params, ) try: try: claims = _entschluesseln(_jwks(quelle)) except JoseError: # Kann ein frisch gedrehter Signaturschluessel sein – einmal neu laden. claims = _entschluesseln(_jwks(quelle, erzwingen=True)) claims.validate(leeway=UHREN_SPIELRAUM) except JoseError as fehler: raise OidcFehler("id_token_ungueltig", f"ID-Token abgelehnt: {fehler}") from fehler # Doppelt gehalten: aud steht schon in den claims_options, aber ein falsch # adressiertes Token darf unter keinen Umstaenden durchrutschen. roh_aud = claims.get("aud") token_aud = set(roh_aud) if isinstance(roh_aud, list) else {roh_aud} if not token_aud & set(audiences): raise OidcFehler("id_token_ungueltig", "ID-Token ist fuer einen anderen Client bestimmt.") return dict(claims) # --------------------------------------------------------------------------- # Zuordnung zu Vorrania-Konten (ohne HTTP – direkt testbar) # --------------------------------------------------------------------------- _ERLAUBT_IM_NAMEN = re.compile(r"[^A-Za-z0-9._-]") def _jetzt() -> datetime: return datetime.now(timezone.utc) def _namensvorschlag(claims: dict[str, Any]) -> str: roh = ( claims.get("preferred_username") or claims.get("nickname") or (claims.get("email") or "").split("@")[0] or claims.get("name") or "" ) sauber = _ERLAUBT_IM_NAMEN.sub("", str(roh).strip().replace(" ", "."))[:64].strip("._-") return sauber or "sso-nutzer" def _freier_name(db: Session, vorschlag: str) -> str: """Macht den Namen eindeutig, ohne je ein bestehendes Konto zu uebernehmen.""" if db.query(User).filter(User.username == vorschlag).first() is None: return vorschlag stamm = vorschlag[:58].rstrip("._-") or "sso-nutzer" for n in range(2, 1000): kandidat = f"{stamm}-{n}" if db.query(User).filter(User.username == kandidat).first() is None: return kandidat raise OidcFehler("kein_freier_name", "Kein freier Benutzername zu finden.") def _email_aus_claims(claims: dict[str, Any]) -> str | None: roh = claims.get("email") if not roh or not isinstance(roh, str): return None return roh.strip().lower() or None def identitaet_verknuepfen( db: Session, nutzer: User, provider: OidcProvider, claims: dict[str, Any] ) -> UserIdentity: """Haengt eine Anmeldedienst-Kennung an ein bestehendes, angemeldetes Konto. Der Weg fuer alle, deren Konto keine (oder eine andere) E-Mail-Adresse hat: Die Berechtigung kommt hier aus der laufenden Vorrania-Sitzung, nicht aus einem Claim – deshalb braucht es weder eine bestaetigte Adresse noch einen Abgleich ueber den Benutzernamen. """ subject = str(claims.get("sub") or "").strip() if not subject: raise OidcFehler("kein_subject", "ID-Token ohne 'sub'.") vorhanden = ( db.query(UserIdentity) .filter(UserIdentity.provider == provider.slug, UserIdentity.subject == subject) .first() ) if vorhanden is not None: if vorhanden.user_id == nutzer.id: return vorhanden # schon verknuepft – nichts zu tun # Diese Kennung gehoert einem ANDEREN Vorrania-Konto. Umhaengen waere # eine Uebernahme, auch wenn hier gerade jemand angemeldet ist. raise OidcFehler( "kennung_fremd", "Diese Anmeldung ist bereits mit einem anderen Vorrania-Konto verknuepft.", ) schon_dabei = ( db.query(UserIdentity) .filter(UserIdentity.provider == provider.slug, UserIdentity.user_id == nutzer.id) .first() ) if schon_dabei is not None: raise OidcFehler( "schon_verknuepft", f"Dieses Konto ist bereits mit '{provider.display_name}' verknuepft.", ) identitaet = UserIdentity( user_id=nutzer.id, provider=provider.slug, subject=subject, last_login_at=_jetzt(), ) db.add(identitaet) # Adresse nachtragen, solange sie frei ist – dann findet auch die # Erstverknuepfung anderer Konten spaeter das Richtige. email = _email_aus_claims(claims) if email and not nutzer.email: if db.query(User).filter(func.lower(User.email) == email).first() is None: nutzer.email = email db.commit() db.refresh(identitaet) logger.info("SSO '%s': mit Konto '%s' verknuepft.", provider.slug, nutzer.username) return identitaet def identitaet_loesen(db: Session, nutzer: User, provider_slug: str) -> None: """Entfernt eine Verknuepfung – aber nie die letzte Anmeldemoeglichkeit.""" identitaet = ( db.query(UserIdentity) .filter(UserIdentity.provider == provider_slug, UserIdentity.user_id == nutzer.id) .first() ) if identitaet is None: raise OidcFehler("nicht_verknuepft", "Diese Verknuepfung besteht nicht.") uebrige = ( db.query(UserIdentity) .filter(UserIdentity.user_id == nutzer.id, UserIdentity.id != identitaet.id) .count() ) if not nutzer.password_hash and uebrige == 0: # Ohne Passwort und ohne weitere Kennung waere das Konto danach von # niemandem mehr erreichbar – auch nicht vom Besitzer. raise OidcFehler( "letzte_anmeldung", "Das ist die einzige Anmeldemoeglichkeit dieses Kontos. Vergib zuerst " "ein Passwort oder verknuepfe einen weiteren Anmeldedienst.", ) db.delete(identitaet) db.commit() logger.info("SSO '%s': Verknuepfung von '%s' geloest.", provider_slug, nutzer.username) def konto_suchen( db: Session, provider: OidcProvider, claims: dict[str, Any] ) -> User | None: """Sucht das Konto zu einem geprueften ID-Token – ohne je eines anzulegen. 1. Bekannte Identitaet (Anbieter + ``sub``) – der Normalfall ab dem zweiten Mal. 2. Erstverknuepfung ueber die **bestaetigte** E-Mail-Adresse. ``None`` heisst: die Kennung ist echt, gehoert aber noch niemandem. Der Browser-Weg fragt dann nach (neues Konto oder bestehendes verknuepfen), statt still ein zweites Konto anzulegen. """ subject = str(claims.get("sub") or "").strip() if not subject: raise OidcFehler("kein_subject", "ID-Token ohne 'sub'.") email = _email_aus_claims(claims) # Authentik liefert das als echtes bool; andere Anbieter schicken schon mal # den String "true". roh_bestaetigt = claims.get("email_verified") email_bestaetigt = roh_bestaetigt is True or str(roh_bestaetigt).lower() == "true" # 1) Bereits verknuepft. identitaet = ( db.query(UserIdentity) .filter(UserIdentity.provider == provider.slug, UserIdentity.subject == subject) .first() ) if identitaet is not None: nutzer = db.get(User, identitaet.user_id) if nutzer is None: # Verwaiste Verknuepfung (Konto geloescht) – aufraeumen und neu entscheiden. db.delete(identitaet) db.commit() else: identitaet.last_login_at = _jetzt() # Adresse nachtragen, falls das Konto noch keine hat – aber nur, wenn # sie nicht schon einem anderen Konto gehoert (eindeutiger Index). if email and not nutzer.email: if db.query(User).filter(func.lower(User.email) == email).first() is None: nutzer.email = email db.commit() db.refresh(nutzer) return nutzer # 2) Erstverknuepfung ueber die E-Mail. if email and (email_bestaetigt or not provider.require_verified_email): treffer = db.query(User).filter(func.lower(User.email) == email).first() if treffer is not None: belegt = ( db.query(UserIdentity) .filter( UserIdentity.provider == provider.slug, UserIdentity.user_id == treffer.id, ) .first() ) if belegt is not None: # Dasselbe Konto haengt schon an einer ANDEREN Kennung dieses # Anbieters. Stillschweigend umhaengen waere eine Uebernahme. raise OidcFehler( "konto_bereits_verknuepft", f"Das Konto '{treffer.username}' ist bereits mit einer anderen " f"Kennung bei '{provider.slug}' verknuepft.", ) db.add( UserIdentity( user_id=treffer.id, provider=provider.slug, subject=subject, last_login_at=_jetzt(), ) ) db.commit() db.refresh(treffer) return treffer elif email and not email_bestaetigt: logger.info( "SSO '%s': E-Mail '%s' ist beim Anbieter nicht bestaetigt – keine " "Verknuepfung mit einem bestehenden Konto.", provider.slug, email, ) return None def konto_anlegen(db: Session, provider: OidcProvider, claims: dict[str, Any]) -> User: """Legt ein neues Konto zu einer geprueften, noch freien Kennung an.""" subject = str(claims.get("sub") or "").strip() if not subject: raise OidcFehler("kein_subject", "ID-Token ohne 'sub'.") email = _email_aus_claims(claims) # Die Adresse ist eindeutig indiziert. Sie kann hier trotzdem schon vergeben # sein – naemlich wenn Schritt 2 sie wegen fehlender Bestaetigung NICHT zum # Verknuepfen benutzen durfte. Dann bleibt das neue Konto ohne Adresse, statt # den Login an einem Datenbankfehler scheitern zu lassen. email_frei = email is not None and ( db.query(User).filter(func.lower(User.email) == email).first() is None ) if email and not email_frei: logger.info( "SSO '%s': Adresse '%s' ist schon vergeben – neues Konto bleibt ohne Adresse.", provider.slug, email, ) # Rolle kommt NIE aus dem Anbieter: sonst genuegte ein passend gesetzter Claim, # um sich Administratorrechte zu geben. nutzer = User( username=_freier_name(db, _namensvorschlag(claims)), password_hash=None, email=email if email_frei else None, role=Role.user, ) db.add(nutzer) db.flush() db.add( UserIdentity( user_id=nutzer.id, provider=provider.slug, subject=subject, last_login_at=_jetzt(), ) ) db.commit() db.refresh(nutzer) logger.info("SSO '%s': neues Konto '%s' angelegt.", provider.slug, nutzer.username) return nutzer def benutzer_aus_claims( db: Session, provider: OidcProvider, claims: dict[str, Any] ) -> User: """Sucht das Konto und legt es notfalls an – der Weg ohne Rueckfrage. Genutzt vom App-Anmeldeweg, der keinen Zwischenschritt anzeigen kann. """ nutzer = konto_suchen(db, provider, claims) if nutzer is not None: return nutzer if not provider.auto_create: raise OidcFehler( "kein_konto", "Zu dieser Anmeldung gibt es kein Vorrania-Konto. Melde dich einmal im " "Browser an, um es zu verknuepfen, oder lass dir eines anlegen.", ) return konto_anlegen(db, provider, claims) # --------------------------------------------------------------------------- # Offene Erstanmeldung: Kennung ist echt, gehoert aber noch keinem Konto # --------------------------------------------------------------------------- def _mit_zeitzone(wert: datetime) -> datetime: # SQLite gibt Zeitstempel ohne Zonenangabe zurueck. return wert if wert.tzinfo is not None else wert.replace(tzinfo=timezone.utc) def offen_anlegen(db: Session, provider: OidcProvider, claims: dict[str, Any]) -> str: """Merkt die geprueften Angaben vor und gibt den Einmalcode im Klartext zurueck. Gespeichert wird nur, was aus dem bereits geprueften ID-Token stammt – nie das Token selbst. Vom Code liegt hier nur der Hash, wie bei den API-Tokens. """ subject = str(claims.get("sub") or "").strip() if not subject: raise OidcFehler("kein_subject", "ID-Token ohne 'sub'.") roh_bestaetigt = claims.get("email_verified") code = secrets.token_urlsafe(32) db.add( OidcPendingLink( code_hash=hash_api_token(code), provider=provider.slug, subject=subject, email=_email_aus_claims(claims), email_verified=roh_bestaetigt is True or str(roh_bestaetigt).lower() == "true", username_vorschlag=_namensvorschlag(claims), created_at=_jetzt(), ) ) db.commit() return code def offen_lesen( db: Session, code: str, jetzt: datetime | None = None ) -> OidcPendingLink | None: """Liest den Merkzettel, ohne ihn zu verbrauchen (fuer die Anzeige).""" eintrag = db.get(OidcPendingLink, hash_api_token(code)) if eintrag is None: return None if _mit_zeitzone(eintrag.created_at) + OFFEN_GUELTIG < (jetzt or _jetzt()): db.delete(eintrag) db.commit() return None return eintrag def offen_aufraeumen(db: Session, jetzt: datetime | None = None) -> int: grenze = (jetzt or _jetzt()) - OFFEN_GUELTIG alte = [ e for e in db.query(OidcPendingLink).all() if _mit_zeitzone(e.created_at) < grenze ] for eintrag in alte: db.delete(eintrag) if alte: db.commit() return len(alte) def _claims_aus_offen(eintrag: OidcPendingLink) -> dict[str, Any]: return { "sub": eintrag.subject, "email": eintrag.email, "email_verified": eintrag.email_verified, "preferred_username": eintrag.username_vorschlag, } def offen_neues_konto( db: Session, provider: OidcProvider, eintrag: OidcPendingLink ) -> User: """Legt das neue Konto an und verbraucht den Merkzettel.""" if not provider.auto_create: raise OidcFehler( "anlegen_nicht_erlaubt", "Auf diesem Server duerfen sich keine neuen Konten selbst anlegen.", ) claims = _claims_aus_offen(eintrag) db.delete(eintrag) db.flush() return konto_anlegen(db, provider, claims) def offen_verknuepfen( db: Session, provider: OidcProvider, eintrag: OidcPendingLink, username: str, passwort: str, ) -> User: """Haengt die Kennung an ein bestehendes Konto – nach Passwort-Nachweis. Das Passwort ist hier der Eigentumsnachweis. Ohne ihn koennte sich jeder mit einem beliebigen Konto beim Anmeldedienst an ein fremdes Vorrania-Konto haengen – genau deshalb fragt der Anmeldeweg selbst nie „welches Konto?". """ nutzer = db.query(User).filter(User.username == username).first() if nutzer is None or not verify_password(passwort, nutzer.password_hash): # Bewusst dieselbe Meldung fuer „gibt es nicht" und „Passwort falsch": # sonst verraet die Antwort, welche Benutzernamen existieren. raise OidcFehler("anmeldung_falsch", "Benutzername oder Passwort falsch.") claims = _claims_aus_offen(eintrag) identitaet_verknuepfen(db, nutzer, provider, claims) # Erst nach erfolgreicher Verknuepfung verbrauchen – ein Tippfehler im # Passwort soll nicht den ganzen Anmeldevorgang wegwerfen. db.delete(eintrag) db.commit() db.refresh(nutzer) return nutzer