"""OpenID Connect: Anmeldung ueber einen fremden Anbieter (Authorization Code Flow). Geprueft wird mit Authlib, nicht von Hand. Ein ID-Token selbst zu zerlegen und seine Signatur nachzurechnen ist die Art Code, die in neunzig Prozent der Faelle funktioniert und in den uebrigen zehn jemanden hereinlaesst, der nicht hereingehoert. Was hier passiert und warum: ``state`` Verknuepft die Rueckkehr mit dem Aufbruch. Ohne ihn koennte jemand eine fremde Rueckleitung in den Browser des Nutzers schieben. ``nonce`` Steht im ID-Token und wird gegen den gemerkten Wert geprueft. Ohne ihn liesse sich ein anderswo erbeutetes, gueltiges Token einreichen. ``PKCE`` Bindet den Code an den, der ihn angefordert hat. Bei einem vertraulichen Client (Secret) nicht zwingend, aber kostenlos – und der Code bleibt derselbe, falls einmal ein oeffentlicher Client dazukommt. Discovery und Schluessel werden kurz zwischengespeichert. Nicht laenger: Ein Anbieter wechselt seine Signaturschluessel, und ein zu lange gehaltener Schluesselsatz weist danach jeden ab. """ from __future__ import annotations import base64 import hashlib import logging import secrets import time from dataclasses import dataclass import httpx from authlib.jose import JsonWebKey, JsonWebToken from authlib.jose.errors import JoseError from ..oidc_config import Provider log = logging.getLogger(__name__) ZEITLIMIT = 15.0 # So lange gelten Discovery-Dokument und Schluesselsatz als frisch. CACHE_SEKUNDEN = 600 # Wie viel Uhrzeit-Unterschied zwischen uns und dem Anbieter geduldet wird. TOLERANZ_SEKUNDEN = 60 # Nur asymmetrische Verfahren. "none" waere keine Signatur, und HS256 pruefte # gegen das Client-Secret – beides gehoert hier nicht hin. ERLAUBTE_ALGORITHMEN = ["RS256", "RS384", "RS512", "ES256", "ES384", "ES512", "PS256"] class OidcFehler(RuntimeError): """Fehler mit einem Grund, den man dem Nutzer zeigen kann.""" @dataclass class Anmeldung: """Was nach erfolgreicher Pruefung ueber den Anmeldenden feststeht.""" subject: str email: str | None email_verified: bool name: str | None preferred_username: str | None _cache: dict[str, tuple[float, dict]] = {} _schluessel_cache: dict[str, tuple[float, object]] = {} def _hole(url: str) -> dict: try: with httpx.Client(timeout=ZEITLIMIT, follow_redirects=True) as c: antwort = c.get(url) antwort.raise_for_status() return antwort.json() except httpx.HTTPError as e: raise OidcFehler(f"Der Anmeldedienst ist nicht erreichbar: {e}") from e except ValueError as e: raise OidcFehler("Der Anmeldedienst antwortet unverstaendlich.") from e def discovery(p: Provider) -> dict: """Das Discovery-Dokument – gecacht.""" eintrag = _cache.get(p.name) if eintrag and eintrag[0] > time.time(): return eintrag[1] daten = _hole(p.discovery_url) for pflicht in ("issuer", "authorization_endpoint", "token_endpoint", "jwks_uri"): if not daten.get(pflicht): raise OidcFehler( f"Das Discovery-Dokument von '{p.name}' hat kein '{pflicht}'. " f"Zeigt die Adresse wirklich auf eine OpenID-Konfiguration?" ) _cache[p.name] = (time.time() + CACHE_SEKUNDEN, daten) return daten def _schluesselsatz(p: Provider, jwks_uri: str, erzwinge_neu: bool = False): eintrag = _schluessel_cache.get(p.name) if eintrag and eintrag[0] > time.time() and not erzwinge_neu: return eintrag[1] satz = JsonWebKey.import_key_set(_hole(jwks_uri)) _schluessel_cache[p.name] = (time.time() + CACHE_SEKUNDEN, satz) return satz def cache_leeren() -> None: _cache.clear() _schluessel_cache.clear() # --- Schritt 1: hinschicken -------------------------------------------------- def _pkce() -> tuple[str, str]: """(verifier, challenge) nach S256.""" verifier = secrets.token_urlsafe(64) rumpf = hashlib.sha256(verifier.encode("ascii")).digest() return verifier, base64.urlsafe_b64encode(rumpf).rstrip(b"=").decode("ascii") def start(p: Provider, redirect_uri: str) -> dict: """Adresse, zu der der Nutzer geschickt wird – plus was wir uns merken muessen.""" from urllib.parse import urlencode daten = discovery(p) verifier, challenge = _pkce() state = secrets.token_urlsafe(32) nonce = secrets.token_urlsafe(32) frage = { "response_type": "code", "client_id": p.client_id, "redirect_uri": redirect_uri, "scope": p.scopes, "state": state, "nonce": nonce, "code_challenge": challenge, "code_challenge_method": "S256", } return { "url": f"{daten['authorization_endpoint']}?{urlencode(frage)}", "state": state, "nonce": nonce, "code_verifier": verifier, } # --- Schritt 2: zurueckkommen ------------------------------------------------ def _tausche_code(p: Provider, daten: dict, code: str, redirect_uri: str, code_verifier: str) -> dict: formular = { "grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri, "client_id": p.client_id, "code_verifier": code_verifier, } kopfzeilen = {"Accept": "application/json"} # Ein oeffentlicher Client hat kein Secret und darf auch keines schicken. auth = (p.client_id, p.client_secret) if p.client_secret else None if auth is None: formular.pop("client_id", None) formular["client_id"] = p.client_id try: with httpx.Client(timeout=ZEITLIMIT) as c: antwort = c.post(daten["token_endpoint"], data=formular, headers=kopfzeilen, auth=auth) except httpx.HTTPError as e: raise OidcFehler(f"Der Anmeldedienst ist nicht erreichbar: {e}") from e if antwort.status_code >= 400: grund = "" try: inhalt = antwort.json() grund = inhalt.get("error_description") or inhalt.get("error") or "" except ValueError: pass raise OidcFehler( f"Der Anmeldedienst hat den Code abgelehnt{': ' + grund if grund else '.'}" ) try: return antwort.json() except ValueError as e: raise OidcFehler("Der Anmeldedienst antwortet unverstaendlich.") from e def _pruefe_id_token(p: Provider, daten: dict, id_token: str, nonce: str) -> dict: """Signatur, Aussteller, Empfaenger, Laufzeit und nonce. Bei einem unbekannten Schluessel wird der Satz EINMAL neu geholt: Anbieter tauschen ihre Signaturschluessel aus, und ein gecachter Satz wiese danach jeden ab, bis der Zwischenspeicher von selbst ablaeuft. """ anspruch = { "iss": {"essential": True, "value": daten["issuer"]}, "aud": {"essential": True, "value": p.client_id}, "exp": {"essential": True}, } def _lauf(erzwinge_neu: bool) -> dict: satz = _schluesselsatz(p, daten["jwks_uri"], erzwinge_neu) jwt = JsonWebToken(ERLAUBTE_ALGORITHMEN) gelesen = jwt.decode(id_token, satz, claims_options=anspruch) gelesen.validate(leeway=TOLERANZ_SEKUNDEN) return gelesen try: gelesen = _lauf(False) except (JoseError, ValueError): try: gelesen = _lauf(True) except JoseError as e: raise OidcFehler(f"Das Token des Anmeldedienstes ist ungueltig: {e}") from e except ValueError as e: raise OidcFehler("Das Token des Anmeldedienstes ist unlesbar.") from e # Der nonce wird von Authlib nicht mitgeprueft – hier von Hand, sonst # liesse sich ein anderswo erbeutetes, gueltiges Token einreichen. if nonce and gelesen.get("nonce") != nonce: raise OidcFehler("Die Antwort gehoert nicht zu dieser Anmeldung (nonce).") if not gelesen.get("sub"): raise OidcFehler("Das Token nennt keinen Nutzer (sub fehlt).") return dict(gelesen) def schliesse_ab(p: Provider, code: str, redirect_uri: str, code_verifier: str, nonce: str) -> Anmeldung: """Code einloesen, ID-Token pruefen, und sagen wer da hereinkommt.""" daten = discovery(p) tokens = _tausche_code(p, daten, code, redirect_uri, code_verifier) id_token = tokens.get("id_token") if not id_token: raise OidcFehler( "Der Anmeldedienst hat kein ID-Token geschickt. Fehlt der Scope " "'openid' in der Konfiguration?" ) anspruch = _pruefe_id_token(p, daten, id_token, nonce) email = anspruch.get("email") verifiziert = bool(anspruch.get("email_verified")) # Manche Anbieter liefern die E-Mail erst am userinfo-Endpunkt. if not email and daten.get("userinfo_endpoint") and tokens.get("access_token"): try: with httpx.Client(timeout=ZEITLIMIT) as c: antwort = c.get( daten["userinfo_endpoint"], headers={"Authorization": f"Bearer {tokens['access_token']}"}, ) if antwort.status_code < 400: weiteres = antwort.json() # Nur vom selben Nutzer – sonst mischten wir zwei Identitaeten. if weiteres.get("sub") == anspruch["sub"]: email = weiteres.get("email") or email verifiziert = bool(weiteres.get("email_verified")) or verifiziert anspruch.setdefault("name", weiteres.get("name")) anspruch.setdefault("preferred_username", weiteres.get("preferred_username")) except (httpx.HTTPError, ValueError): log.info("userinfo von '%s' nicht abrufbar – ID-Token genuegt", p.name) return Anmeldung( subject=str(anspruch["sub"]), email=(email or "").strip().lower() or None, email_verified=verifiziert, name=anspruch.get("name"), preferred_username=anspruch.get("preferred_username"), )