ScarriffleandClaude Opus 5
31b3624cc7
In Authentik hat jede Anwendung standardmaessig ihren eigenen Issuer, abgeleitet aus dem Application-Slug. Web- und App-Client brauchen dort aber zwei getrennte Anwendungen, weil ein Provider entweder vertraulich ODER oeffentlich ist. Bisher wurde das App-Token gegen den Issuer und den Schluesselsatz der Web-Anwendung geprueft - die App-Anmeldung waere damit an der iss-Pruefung gescheitert. Discovery und JWKS werden jetzt je Adresse gepuffert statt je Anbieter, und OIDC_<KUERZEL>_NATIVE_DISCOVERY_URL benennt die Anwendung des App-Clients. Ohne die Variable bleibt es bei einer gemeinsamen Adresse. Ausserdem im README: der Signing Key muss gesetzt sein, sonst signiert Authentik mit HS256 und das Token wird abgelehnt. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
475 lines
17 KiB
Python
475 lines
17 KiB
Python
"""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 time
|
||
from datetime import datetime, 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 Role, User, UserIdentity
|
||
from ..oidc_config import OidcProvider
|
||
|
||
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
|
||
|
||
|
||
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 benutzer_aus_claims(
|
||
db: Session, provider: OidcProvider, claims: dict[str, Any]
|
||
) -> User:
|
||
"""Findet oder erzeugt das Vorrania-Konto zu einem geprueften ID-Token.
|
||
|
||
Reihenfolge:
|
||
|
||
1. Bekannte Identitaet (Anbieter + ``sub``) – der Normalfall ab dem zweiten Mal.
|
||
2. Erstverknuepfung ueber die **bestaetigte** E-Mail-Adresse.
|
||
3. Neues Konto, falls fuer den Anbieter erlaubt.
|
||
"""
|
||
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,
|
||
)
|
||
|
||
# 3) Neu anlegen, falls erlaubt.
|
||
if not provider.auto_create:
|
||
raise OidcFehler(
|
||
"kein_konto",
|
||
"Zu dieser Anmeldung gibt es kein Vorrania-Konto. Ein Administrator muss "
|
||
"es zuerst anlegen und die E-Mail-Adresse hinterlegen.",
|
||
)
|
||
|
||
# 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
|