"""Anmelden über einen fremden Dienst – und ein bestehendes Konto damit verknüpfen. Zwei Wege durch dieselbe Strecke: **Verknüpfen** (angemeldet). Der eigentliche Weg für alle, die schon ein Konto haben: einmal wie gewohnt anmelden, in den Einstellungen „Mit … verknüpfen", fertig. Danach geht beides – Passwort und SSO. **Anmelden** (nicht angemeldet). Ist die Kennung verknüpft, wird die Sitzung angelegt. Ist sie es nicht, entsteht je nach Einstellung ein NEUES Konto – oder die Anmeldung wird abgewiesen (Vorgabe). Was es bewusst nicht gibt: automatisches Zusammenführen über die E-Mail. Es wäre der bequeme Weg und der gefährliche – wer beim Anbieter eine fremde Adresse einträgt, übernähme damit ein fremdes Konto. Der zweite Faktor von Certain wird beim SSO-Login übersprungen: Der ist beim Anbieter schon passiert, und ein zweites Mal fragen hiesse, ihm nicht zu glauben, obwohl man ihn gerade zur Anmeldung ermächtigt hat. """ from datetime import datetime, timedelta, timezone from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response, status from fastapi.responses import RedirectResponse from sqlalchemy.orm import Session from .. import oidc_config from ..config import get_settings from ..database import get_db from ..deps import get_current_user from ..models import OidcState, Role, User, UserIdentity from ..services import oidc as oidc_service from ..services import sessions router = APIRouter(prefix="/auth/oidc", tags=["auth"]) # Wie lange ein begonnener Anmeldeversuch gilt. Kurz: In dieser Zeit soll der # Nutzer nur beim Anbieter auf "Anmelden" druecken. STATE_TTL_MINUTEN = 15 def _now() -> datetime: return datetime.now(timezone.utc) def _redirect_uri(request: Request, provider: str) -> str: """Die Adresse, an die der Anbieter zurueckschickt. Sie muss beim Anbieter EXAKT so eingetragen sein – beim Einloesen des Codes wird sie noch einmal mitgeschickt und dort verglichen. Ein Zeichen Unterschied, und die Anmeldung scheitert. Steht ``APP_PUBLIC_BASE_URL``, gilt die. Sonst wird sie aus der Anfrage abgeleitet – das genuegt in der Entwicklung, geht hinter einem Reverse Proxy aber schief: Dort kaeme die INTERNE Adresse heraus (http://10.0.0.5:8071/...), weil uvicorn die Weiterleitungs-Kopfzeilen nicht auswertet. Genau deshalb gibt es die Einstellung. """ pfad = request.url_for("oidc_callback", provider=provider).path basis = get_settings().public_base_url.strip().rstrip("/") return f"{basis}{pfad}" if basis else str(request.url_for( "oidc_callback", provider=provider)) def _fehlerseite(weiter: str, meldung: str) -> RedirectResponse: """Zurueck in die Oberflaeche, mit dem Grund in der Adresse. Kein JSON: Hier landet ein BROWSER nach einer Weiterleitung, kein Programm. Eine nackte Fehlerantwort waere eine weisse Seite. """ from urllib.parse import quote return RedirectResponse(f"{weiter}?oidc_error={quote(meldung[:300])}", status_code=303) def _sicheres_ziel(weiter: str | None) -> str: """Nur Ziele auf dieser Seite. Ohne diese Pruefung waere der Endpunkt eine offene Weiterleitung: Ein Link auf ``…/start?weiter=https://woanders`` schickte den Nutzer nach der Anmeldung auf eine fremde Seite, die wie ein Teil von Certain aussieht. """ if not weiter or not weiter.startswith("/") or weiter.startswith("//"): return "/" return weiter[:255] @router.get("/providers") def list_providers() -> dict: """Welche Anmeldedienste eingerichtet sind. Ohne Anmeldung erreichbar: Die Login-Seite muss vor der Anmeldung wissen, ob sie einen Knopf zeigen soll. Client-Secrets stehen hier nie. """ return { "providers": [ {"name": p.name, "label": p.label} for p in oidc_config.providers().values() ] } def _hole_provider(provider: str): p = oidc_config.get(provider) if p is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Kein Anmeldedienst '{provider}' eingerichtet.", ) return p def _beginne(db: Session, request: Request, provider: str, weiter: str | None, link_user: User | None) -> RedirectResponse: p = _hole_provider(provider) ziel = _redirect_uri(request, p.name) try: begonnen = oidc_service.start(p, ziel) except oidc_service.OidcFehler as e: raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(e)) from e # Abgelaufene Versuche wegraeumen – sonst waechst die Tabelle still. db.query(OidcState).filter(OidcState.expires_at < _now()).delete() db.add(OidcState( state=begonnen["state"], provider=p.name, nonce=begonnen["nonce"], code_verifier=begonnen["code_verifier"], redirect_uri=ziel, link_user_id=link_user.id if link_user is not None else None, weiter=_sicheres_ziel(weiter), expires_at=_now() + timedelta(minutes=STATE_TTL_MINUTEN), )) db.commit() return RedirectResponse(begonnen["url"], status_code=303) @router.get("/{provider}/start") def oidc_start(provider: str, request: Request, weiter: str | None = Query(default=None), db: Session = Depends(get_db)): """Anmeldung beginnen. Ohne Anmeldung erreichbar – das ist der Sinn.""" return _beginne(db, request, provider, weiter, None) @router.get("/{provider}/link") def oidc_link(provider: str, request: Request, weiter: str | None = Query(default=None), db: Session = Depends(get_db), user: User = Depends(get_current_user)): """Das eigene, bestehende Konto mit dem Anbieter verknuepfen.""" return _beginne(db, request, provider, weiter or "/konto", user) def _freier_benutzername(db: Session, vorschlag: str) -> str: """Ein Benutzername, den es noch nicht gibt.""" basis = "".join(c for c in (vorschlag or "").strip().lower() if c.isalnum() or c in "._-")[:48] or "nutzer" name = basis n = 2 while db.query(User.id).filter(User.username == name).first() is not None: name = f"{basis}{n}"[:64] n += 1 return name def _darf_anlegen(p, anmeldung: oidc_service.Anmeldung) -> bool: if not p.auto_create: return False if not p.allowed_domains: return True if not anmeldung.email: return False return anmeldung.email.rsplit("@", 1)[-1] in p.allowed_domains def _finde_oder_lege_an(db: Session, p, anmeldung: oidc_service.Anmeldung) -> User: """Zu wem gehoert dieser Anmeldende? Zwei Faelle, und dazwischen ausdruecklich NICHTS: 1. **Bekannte Kennung** – die Verknuepfung gibt es schon. Eindeutig. 2. **Neu anlegen**, wenn der Anbieter das darf. Was hier bewusst FEHLT, ist der bequeme dritte Fall: ueber die E-Mail ein vorhandenes Konto suchen und stillschweigend verknuepfen. Wer beim Anbieter eine fremde Adresse eintragen kann, uebernaehme damit ein fremdes Konto – und die E-Mail ist die Angabe, die sich dort am leichtesten aendern laesst. Ein bestehendes Konto wird ausdruecklich verknuepft, im angemeldeten Zustand, ueber ``/link``. """ vorhanden = ( db.query(UserIdentity) .filter(UserIdentity.provider == p.name, UserIdentity.subject == anmeldung.subject) .first() ) if vorhanden is not None: vorhanden.email = anmeldung.email vorhanden.last_login_at = _now() nutzer = db.get(User, vorhanden.user_id) if nutzer is None: raise oidc_service.OidcFehler("Das verknuepfte Konto gibt es nicht mehr.") return nutzer if not _darf_anlegen(p, anmeldung): raise oidc_service.OidcFehler( "Zu dieser Anmeldung gibt es kein Konto in Certain. Melde dich mit " "Benutzername und Passwort an und verknuepfe dein Konto in den " "Einstellungen." ) nutzer = User( username=_freier_benutzername( db, anmeldung.preferred_username or (anmeldung.email.split("@")[0] if anmeldung.email else "")), # Kein Passwort: Dieses Konto meldet sich ueber den Anbieter an. password_hash=None, display_name=(anmeldung.name or "")[:120] or None, role=Role.user, is_active=True, ) db.add(nutzer) db.flush() db.add(UserIdentity(user_id=nutzer.id, provider=p.name, subject=anmeldung.subject, email=anmeldung.email, last_login_at=_now())) return nutzer @router.get("/{provider}/callback", name="oidc_callback") def oidc_callback(provider: str, request: Request, response: Response, code: str | None = Query(default=None), state: str | None = Query(default=None), error: str | None = Query(default=None), error_description: str | None = Query(default=None), db: Session = Depends(get_db)): """Rueckkehr vom Anbieter.""" eintrag = ( db.query(OidcState).filter(OidcState.state == (state or "")).first() if state else None ) # Das Ziel steht im gemerkten Versuch; ohne ihn bleibt nur die Startseite. weiter = _sicheres_ziel(eintrag.weiter if eintrag is not None else None) if error: return _fehlerseite(weiter, error_description or error) if eintrag is None: # Auch bei falschem state: Ohne Zuordnung waere jede Rueckkehr gueltig. return _fehlerseite(weiter, "Diese Anmeldung ist abgelaufen oder gehoert " "nicht hierher. Bitte noch einmal versuchen.") verbraucht = eintrag # ab hier gilt er als benutzt, egal wie es ausgeht abgelaufen = eintrag.expires_at.replace(tzinfo=timezone.utc) < _now() \ if eintrag.expires_at.tzinfo is None else eintrag.expires_at < _now() nonce, verifier, ziel_uri = eintrag.nonce, eintrag.code_verifier, eintrag.redirect_uri link_user_id = eintrag.link_user_id db.delete(verbraucht) db.commit() if abgelaufen: return _fehlerseite(weiter, "Die Anmeldung hat zu lange gedauert. " "Bitte noch einmal versuchen.") if eintrag.provider != provider: return _fehlerseite(weiter, "Die Antwort gehoert zu einem anderen Anmeldedienst.") if not code: return _fehlerseite(weiter, "Der Anmeldedienst hat keinen Code geschickt.") p = oidc_config.get(provider) if p is None: return _fehlerseite(weiter, f"Kein Anmeldedienst '{provider}' eingerichtet.") try: anmeldung = oidc_service.schliesse_ab(p, code, ziel_uri, verifier, nonce) except oidc_service.OidcFehler as e: return _fehlerseite(weiter, str(e)) # --- Verknuepfen: der Nutzer ist schon angemeldet ----------------------- if link_user_id is not None: nutzer = db.get(User, link_user_id) if nutzer is None: return _fehlerseite(weiter, "Das Konto gibt es nicht mehr.") schon = ( db.query(UserIdentity) .filter(UserIdentity.provider == p.name, UserIdentity.subject == anmeldung.subject) .first() ) if schon is not None and schon.user_id != nutzer.id: return _fehlerseite( weiter, "Dieses Konto beim Anmeldedienst gehoert bereits zu einem " "anderen Benutzer in Certain.") if schon is None: db.add(UserIdentity(user_id=nutzer.id, provider=p.name, subject=anmeldung.subject, email=anmeldung.email, last_login_at=_now())) db.commit() from urllib.parse import quote return RedirectResponse(f"{weiter}?oidc_linked={quote(p.name)}", status_code=303) # --- Anmelden ----------------------------------------------------------- try: nutzer = _finde_oder_lege_an(db, p, anmeldung) except oidc_service.OidcFehler as e: db.rollback() return _fehlerseite(weiter, str(e)) if not nutzer.is_active: db.rollback() return _fehlerseite(weiter, "Dieses Konto ist deaktiviert.") db.commit() # Kein zweiter Faktor und kein Passwortwechsel: Beides gehoert zum # Passwort-Login. Der Anbieter hat den Nutzer geprueft, und ein Passwort, # das gewechselt werden muesste, gibt es hier womoeglich gar nicht. raw = sessions.create_session(db, nutzer, request.headers.get("user-agent")) antwort = RedirectResponse(weiter, status_code=303) settings = get_settings() antwort.set_cookie( key=settings.cookie_name, value=raw, max_age=settings.session_ttl_days * 86400, httponly=True, secure=settings.cookie_secure, samesite=settings.cookie_samesite, path="/", ) return antwort @router.get("/identities") def my_identities(db: Session = Depends(get_db), user: User = Depends(get_current_user)) -> dict: """Womit dieses Konto verknuepft ist – fuer die Kontoseite.""" bekannt = oidc_config.providers() return { "identities": [ { "id": i.id, "provider": i.provider, "label": bekannt[i.provider].label if i.provider in bekannt else i.provider, "email": i.email, "last_login_at": i.last_login_at.isoformat() if i.last_login_at else None, } for i in db.query(UserIdentity) .filter(UserIdentity.user_id == user.id) .order_by(UserIdentity.id) ], "hat_passwort": user.password_hash is not None, } @router.delete("/identities/{identity_id}", status_code=status.HTTP_204_NO_CONTENT) def unlink(identity_id: int, db: Session = Depends(get_db), user: User = Depends(get_current_user)) -> None: """Eine Verknuepfung loesen. Nicht die letzte, wenn es kein Passwort gibt: Sonst schliesst sich der Nutzer aus seinem eigenen Konto aus, und niemand koennte ihn wieder hereinlassen. """ eintrag = ( db.query(UserIdentity) .filter(UserIdentity.id == identity_id, UserIdentity.user_id == user.id) .first() ) if eintrag is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Verknuepfung nicht gefunden") uebrig = ( db.query(UserIdentity) .filter(UserIdentity.user_id == user.id, UserIdentity.id != identity_id) .count() ) if uebrig == 0 and user.password_hash is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=("Das ist die einzige Anmeldung fuer dieses Konto. Setz erst " "ein Passwort, sonst kommst du nicht mehr herein."), ) db.delete(eintrag) db.commit()