"""Anmeldung ueber einen OpenID-Connect-Anbieter (z.B. Authentik). Zwei Wege, ein Ergebnis: In beiden Faellen endet die Anmeldung mit einem ganz normalen Vorrania-Token aus :func:`create_access_token`. Dadurch bleiben ``deps.py``, die Rollenpruefung und die langlebigen API-Tokens unveraendert – der Anbieter ist ausschliesslich fuer den Anmeldevorgang zustaendig. * **Browser** (vertraulicher Client): ``/login`` leitet zum Anbieter, ``/callback`` loest den Code mit dem Client-Secret ein. * **App** (oeffentlicher Client, PKCE): die App fuehrt den Flow mit AppAuth im System-Browser selbst durch und reicht nur ihr ID-Token bei ``/native`` ein. """ from __future__ import annotations import logging import secrets from urllib.parse import quote from fastapi import APIRouter, Depends, HTTPException, status from fastapi.responses import RedirectResponse from sqlalchemy.orm import Session from ..database import get_db from ..deps import get_current_user from ..models import OidcPendingLink, User, UserIdentity from ..oidc_config import OidcProvider, provider_finden, registrierte_provider from ..schemas import ( OidcIdentityOut, OidcLinkStart, OidcNativeLogin, OidcPendingOut, OidcPendingVerknuepfen, OidcProviderOut, Token, ) from ..security import create_access_token from .settings import get_session_minutes from ..services import oidc as dienst from ..services import oidc_state logger = logging.getLogger(__name__) router = APIRouter(prefix="/auth/oidc", tags=["auth"]) # Wohin die Weboberflaeche nach dem Rueckweg geschickt wird. Absichtlich relativ: # der Browser loest das gegen die oeffentliche Adresse auf, mit der er ohnehin # gerade spricht – eine korrekt gesetzte PUBLIC_BASE_URL braucht es dafuer nicht. WEB_ERFOLG = "/auth/callback" WEB_FEHLER = "/login" # Eine Verknuepfung endet dort, wo sie angestossen wurde – nicht auf der # Anmeldemaske: Der Nutzer ist ja die ganze Zeit angemeldet geblieben. WEB_KONTO = "/konto" # Wohin es geht, wenn die Kennung echt ist, aber noch keinem Konto gehoert. WEB_ENTSCHEIDUNG = "/sso-neu" def _provider_oder_404(slug: str) -> OidcProvider: provider = provider_finden(slug) if provider is None: raise HTTPException( status.HTTP_404_NOT_FOUND, f"Kein Anmeldeanbieter '{slug}' eingerichtet" ) return provider def _token_fuer(user: User, db: Session) -> Token: # Die Sitzungsdauer steht in den Einstellungen und gilt fuer JEDEN Weg # herein - nach Single Sign-on genauso wie nach dem Passwort-Login. token = create_access_token( user_id=user.id, role=user.role.value, gueltig_minuten=get_session_minutes(db) ) return Token(access_token=token, role=user.role, username=user.username) @router.get("/providers", response_model=list[OidcProviderOut]) def anbieter_auflisten() -> list[OidcProviderOut]: """Welche Anmeldeanbieter eingerichtet sind – ohne Anmeldung abrufbar. Die Weboberflaeche kennt keine Build-Variablen und die App muss die Angaben je Serverprofil erfragen; beide erfahren erst hier, ob es ueberhaupt einen SSO-Knopf zu zeigen gibt. Client-Secrets stehen hier nie drin. """ return [ OidcProviderOut( slug=p.slug, display_name=p.display_name, # Die App prueft gegen ihre eigene Anwendung beim Anbieter. discovery_url=p.app_discovery_url, native_client_id=p.native_client_ids[0] if p.native_client_ids else None, native_redirect_uri=p.native_redirect_uri, scopes=p.native_scopes, ) for p in registrierte_provider() ] @router.get("/{slug}/login") def anmeldung_starten(slug: str, db: Session = Depends(get_db)) -> RedirectResponse: """Schickt den Browser zum Anbieter.""" provider = _provider_oder_404(slug) # Abgebrochene Anmeldungen hinterlassen Zettel – hier ist der guenstigste # Moment, sie loszuwerden. oidc_state.aufraeumen(db) zettel = oidc_state.anlegen(db, provider.slug) try: ziel = dienst.anmelde_url( provider, state=zettel.state, nonce=zettel.nonce, code_verifier=zettel.code_verifier, ) except dienst.OidcFehler as fehler: logger.warning("SSO '%s' konnte nicht gestartet werden: %s", slug, fehler) return _fehler_weiterleitung(fehler.schluessel) return RedirectResponse(ziel, status_code=status.HTTP_303_SEE_OTHER) def _fehler_weiterleitung(schluessel: str, verknuepfung: bool = False) -> RedirectResponse: # Nur die Kennung, kein Klartext: die Oberflaeche uebersetzt sie, und interne # Details bleiben im Serverprotokoll. ziel = WEB_KONTO if verknuepfung else WEB_FEHLER return RedirectResponse( f"{ziel}#sso_error={quote(schluessel)}", status_code=status.HTTP_303_SEE_OTHER ) @router.get("/{slug}/callback") def anmeldung_abschliessen( slug: str, db: Session = Depends(get_db), code: str | None = None, state: str | None = None, error: str | None = None, ) -> RedirectResponse: """Rueckweg vom Anbieter. Zwei Ausgaenge, je nachdem womit der Vorgang gestartet wurde: eine Anmeldung endet mit einem Vorrania-Token, eine Verknuepfung haengt die Kennung an das bereits angemeldete Konto und laesst die Sitzung unberuehrt. """ provider = _provider_oder_404(slug) if error: # Nutzer hat abgebrochen oder der Anbieter verweigert – kein Serverfehler. # Zu diesem Zeitpunkt ist noch unbekannt, ob es eine Verknuepfung war; # die Anmeldemaske ist der unschaedlichere Landeplatz. logger.info("SSO '%s' vom Anbieter abgelehnt: %s", slug, error) return _fehler_weiterleitung("abgebrochen") if not code or not state: return _fehler_weiterleitung("unvollstaendige_antwort") # Schutz vor untergeschobenen Anmeldungen: der Zettel muss von uns stammen, # darf nicht abgelaufen und noch nicht verbraucht sein. Das Einloesen # entfernt ihn, ein zweiter Rueckweg laeuft also ins Leere. gemerkt = oidc_state.einloesen(db, state) if gemerkt is None: logger.warning("SSO '%s': unbekannter oder abgelaufener state.", slug) return _fehler_weiterleitung("sitzung_abgelaufen") if not secrets.compare_digest(gemerkt.provider, slug): logger.warning("SSO '%s': state gehoert zu Anbieter '%s'.", slug, gemerkt.provider) return _fehler_weiterleitung("state_falsch") verknuepfen_fuer = gemerkt.link_user_id try: tokens = dienst.code_einloesen(provider, code=code, code_verifier=gemerkt.code_verifier) roh_id_token = tokens.get("id_token") if not roh_id_token: raise dienst.OidcFehler( "kein_id_token", "Der Anbieter hat kein ID-Token geliefert." ) claims = dienst.pruefe_id_token( provider, roh_id_token, nonce=gemerkt.nonce, # Der Browser-Weg laeuft ueber den vertraulichen Client – ein Token # fuer die App-Client-ID darf hier nicht gelten. erlaubte_audiences=(provider.client_id,), ) if verknuepfen_fuer is not None: # Verknuepfung: Es wird niemand angemeldet. Wer das Konto ist, stand # schon beim Start fest (aus der damaligen Sitzung) – die Claims # liefern nur die Kennung, der wir es zuordnen. konto = db.get(User, verknuepfen_fuer) if konto is None: raise dienst.OidcFehler( "konto_verschwunden", "Das zu verknuepfende Konto gibt es nicht mehr." ) dienst.identitaet_verknuepfen(db, konto, provider, claims) logger.info("SSO '%s': Verknuepfung fuer '%s' abgeschlossen.", slug, konto.username) return RedirectResponse( f"{WEB_KONTO}#sso_linked={quote(provider.slug)}", status_code=status.HTTP_303_SEE_OTHER, ) nutzer = dienst.konto_suchen(db, provider, claims) if nutzer is None: # Die Kennung ist echt, gehoert aber noch niemandem. Statt still ein # zweites Konto anzulegen, nachfragen – wer schon ein Konto hat, will # meistens genau das verknuepfen. dienst.offen_aufraeumen(db) merkzettel = dienst.offen_anlegen(db, provider, claims) logger.info("SSO '%s': unbekannte Kennung, frage nach.", slug) return RedirectResponse( f"{WEB_ENTSCHEIDUNG}#code={quote(merkzettel)}", status_code=status.HTTP_303_SEE_OTHER, ) except dienst.OidcFehler as fehler: logger.warning("SSO '%s' gescheitert: %s", slug, fehler) return _fehler_weiterleitung(fehler.schluessel, verknuepfung=verknuepfen_fuer is not None) token = _token_fuer(nutzer, db) logger.info("SSO '%s': '%s' angemeldet.", slug, nutzer.username) # Token im Fragment, nicht in der Abfragezeichenkette: Fragmente werden nicht # an den Server geschickt und landen damit auch nicht in den nginx-Logs. return RedirectResponse( f"{WEB_ERFOLG}#token={quote(token.access_token)}", status_code=status.HTTP_303_SEE_OTHER, ) @router.post("/{slug}/native", response_model=Token) def anmeldung_app(slug: str, payload: OidcNativeLogin, db: Session = Depends(get_db)) -> Token: """Tauscht das ID-Token aus der App gegen ein Vorrania-Token. Die App hat den Flow mit PKCE im System-Browser selbst gefuehrt; der Server prueft hier nur noch, dass das Token echt, unverfallen und **fuer einen der registrierten App-Clients ausgestellt** ist. """ provider = _provider_oder_404(slug) if not provider.native_client_ids: raise HTTPException( status.HTTP_400_BAD_REQUEST, f"Fuer '{slug}' ist kein App-Client eingerichtet", ) try: claims = dienst.pruefe_id_token( provider, payload.id_token, # AppAuth prueft die nonce bereits auf dem Geraet. Schickt die App sie # mit, prueft der Server sie ein zweites Mal – erst das schliesst aus, # dass ein anderswo abgegriffenes ID-Token desselben Clients hier # eingereicht wird. nonce=payload.nonce, erlaubte_audiences=provider.native_client_ids, # Eigener Issuer und Schluesselsatz, falls der App-Client beim # Anbieter eine getrennte Anwendung ist (bei Authentik der Normalfall). discovery_url=provider.app_discovery_url, ) nutzer = dienst.benutzer_aus_claims(db, provider, claims) except dienst.OidcFehler as fehler: logger.warning("SSO '%s' (App) gescheitert: %s", slug, fehler) raise HTTPException(status.HTTP_401_UNAUTHORIZED, str(fehler)) from fehler logger.info("SSO '%s' (App): '%s' angemeldet.", slug, nutzer.username) return _token_fuer(nutzer, db) # --------------------------------------------------------------------------- # Verknuepfung eines bestehenden Kontos (fuer bereits angemeldete Nutzer) # --------------------------------------------------------------------------- @router.get("/identities", response_model=list[OidcIdentityOut]) def verknuepfungen_auflisten( current_user: User = Depends(get_current_user), db: Session = Depends(get_db) ) -> list[OidcIdentityOut]: """Mit welchen Anmeldediensten das eigene Konto verknuepft ist.""" eintraege = ( db.query(UserIdentity) .filter(UserIdentity.user_id == current_user.id) .order_by(UserIdentity.provider) .all() ) return [ OidcIdentityOut( provider=e.provider, # Ein inzwischen abgeschalteter Anbieter soll trotzdem sichtbar # bleiben – sonst haengt eine Verknuepfung unsichtbar am Konto. display_name=(p.display_name if (p := provider_finden(e.provider)) else e.provider), created_at=e.created_at, last_login_at=e.last_login_at, ) for e in eintraege ] @router.post("/{slug}/link", response_model=OidcLinkStart) def verknuepfung_starten( slug: str, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> OidcLinkStart: """Startet die Verknuepfung des eigenen Kontos mit einem Anmeldedienst. Bewusst POST und nicht einfach ein Link: Die Berechtigung steckt im Bearer-Token der laufenden Sitzung, und das laesst sich bei einer normalen Seitennavigation nicht mitschicken. Die Oberflaeche holt hier also erst die Adresse und leitet dann selbst dorthin weiter. Damit braucht die Verknuepfung weder eine bestaetigte E-Mail-Adresse noch einen Abgleich ueber den Benutzernamen – beides waere eine Uebernahmeflaeche. """ provider = _provider_oder_404(slug) schon_da = ( db.query(UserIdentity) .filter(UserIdentity.provider == provider.slug, UserIdentity.user_id == current_user.id) .first() ) if schon_da is not None: raise HTTPException( status.HTTP_409_CONFLICT, f"Dein Konto ist bereits mit '{provider.display_name}' verknüpft", ) oidc_state.aufraeumen(db) zettel = oidc_state.anlegen(db, provider.slug, link_user_id=current_user.id) try: ziel = dienst.anmelde_url( provider, state=zettel.state, nonce=zettel.nonce, code_verifier=zettel.code_verifier, ) except dienst.OidcFehler as fehler: logger.warning("SSO '%s': Verknuepfung nicht startbar: %s", slug, fehler) raise HTTPException(status.HTTP_502_BAD_GATEWAY, str(fehler)) from fehler return OidcLinkStart(authorization_url=ziel) # response_model=None ist noetig, weil "from __future__ import annotations" # aus "-> None" den Typ NoneType macht - den FastAPI sonst als Antwortkoerper # deutet und mit 204 nicht vereinbar ist. @router.delete("/{slug}/link", status_code=status.HTTP_204_NO_CONTENT, response_model=None) def verknuepfung_loesen( slug: str, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> None: """Loest die Verknuepfung wieder. Der Anbieter muss dafuer NICHT mehr eingerichtet sein – sonst liesse sich eine Altlast nach dem Abschalten von SSO nicht mehr entfernen. """ try: dienst.identitaet_loesen(db, current_user, slug) except dienst.OidcFehler as fehler: # "letzte Anmeldemoeglichkeit" ist kein Serverfehler, sondern eine # bewusste Weigerung – der Text erklaert den Ausweg. raise HTTPException(status.HTTP_409_CONFLICT, str(fehler)) from fehler # --------------------------------------------------------------------------- # Erstanmeldung: Kennung ist echt, gehoert aber noch keinem Konto # --------------------------------------------------------------------------- def _offen_oder_404(db: Session, code: str) -> OidcPendingLink: eintrag = dienst.offen_lesen(db, code) if eintrag is None: raise HTTPException( status.HTTP_404_NOT_FOUND, "Der Anmeldevorgang ist abgelaufen. Bitte noch einmal anmelden.", ) return eintrag @router.get("/pending/{code}", response_model=OidcPendingOut) def offene_anmeldung(code: str, db: Session = Depends(get_db)) -> OidcPendingOut: """Was die Entscheidungsseite anzeigen soll. Ohne Anmeldung abrufbar – der Einmalcode aus der Adresse ist der Nachweis. """ eintrag = _offen_oder_404(db, code) provider = _provider_oder_404(eintrag.provider) return OidcPendingOut( provider=eintrag.provider, display_name=provider.display_name, email=eintrag.email, username_vorschlag=eintrag.username_vorschlag, anlegen_erlaubt=provider.auto_create, ) @router.post("/pending/{code}/anlegen", response_model=Token) def offene_anmeldung_neues_konto(code: str, db: Session = Depends(get_db)) -> Token: """Legt ein neues Vorrania-Konto zu dieser Kennung an.""" eintrag = _offen_oder_404(db, code) provider = _provider_oder_404(eintrag.provider) try: nutzer = dienst.offen_neues_konto(db, provider, eintrag) except dienst.OidcFehler as fehler: raise HTTPException(status.HTTP_403_FORBIDDEN, str(fehler)) from fehler logger.info("SSO '%s': neues Konto '%s' auf Wunsch angelegt.", provider.slug, nutzer.username) return _token_fuer(nutzer, db) @router.post("/pending/{code}/verknuepfen", response_model=Token) def offene_anmeldung_verknuepfen( code: str, payload: OidcPendingVerknuepfen, db: Session = Depends(get_db) ) -> Token: """Haengt die Kennung an ein bestehendes Konto – nach Passwort-Nachweis.""" eintrag = _offen_oder_404(db, code) provider = _provider_oder_404(eintrag.provider) try: nutzer = dienst.offen_verknuepfen( db, provider, eintrag, payload.username, payload.password ) except dienst.OidcFehler as fehler: # Falsches Passwort ist 401, eine belegte Kennung ein Konflikt. code_http = ( status.HTTP_401_UNAUTHORIZED if fehler.schluessel == "anmeldung_falsch" else status.HTTP_409_CONFLICT ) raise HTTPException(code_http, str(fehler)) from fehler logger.info("SSO '%s': mit bestehendem Konto '%s' verknuepft.", provider.slug, nutzer.username) return _token_fuer(nutzer, db)