"""BFF authentication routes. The production path is Google OIDC authorization code + PKCE. Until the DB session table is wired, the issued browser sessions are server-side in-proc sessions backed by an opaque HttpOnly cookie. Local development also has a dev-only server login endpoint so Playwright can exercise auth without trusting browser localStorage. """ from __future__ import annotations import base64 import hashlib import hmac import json import logging import secrets import time from dataclasses import dataclass from typing import Annotated, Literal, Optional from urllib.parse import urlencode, urlsplit import httpx from fastapi import APIRouter, Cookie, HTTPException, Query, Request, Response, status from fastapi.responses import RedirectResponse from pydantic import BaseModel from ..auth_types import AccountStatus, RoleName from ..auth_sessions import ( ManagedUser, get_managed_user, get_managed_user_by_email, has_admin_access, InactiveUserError, is_super_admin_email, SessionUser, create_session, record_user_consent, revoke_session, withdraw_user_consent, ) from ..config import settings from ..deps import CurrentPrincipal, Principal, Role from ..saml import ( SamlIdentity, acs_url_for_entity_id, build_authn_request, parse_fixture_response, redirect_binding_url, ) router = APIRouter(prefix="/auth", tags=["auth"]) logger = logging.getLogger(__name__) GOOGLE_AUTHORIZE_URL = "https://accounts.google.com/o/oauth2/v2/auth" GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token" GOOGLE_TOKENINFO_URL = "https://oauth2.googleapis.com/tokeninfo" OAUTH_STATE_TTL_SECONDS = 10 * 60 OAUTH_STATE_COOKIE_NAME = "__Host-vignette_oauth_state" DEV_OAUTH_STATE_COOKIE_NAME = "vignette_oauth_state" GENERIC_ADMIN_ENTRY_PATHS = {"/", "/learn", "/teach", "/login", "/onboarding"} @dataclass(slots=True) class OAuthState: code_verifier: str next_path: str created_at: float @dataclass(slots=True) class SamlState: request_id: str next_path: str created_at: float _oauth_states: dict[str, OAuthState] = {} _saml_states: dict[str, SamlState] = {} class MeResponse(BaseModel): user_id: str email: str display_name: str role: RoleName admin_access: bool = False super_admin: bool = False account_status: AccountStatus = "approved" approval_required: bool = False cohort_ids: list[str] consent_at: float | None = None onboarding_completed_at: float | None = None nickname: str = "" self_introduction: str = "" avatar_url: str = "" class ConsentRequest(BaseModel): accepted: bool = True class ConsentResponse(BaseModel): consent_at: float | None = None class AuthProviderStatus(BaseModel): provider: Literal["google", "saml"] configured: bool enabled: bool login_path: str class AuthConfigResponse(BaseModel): google_oauth_configured: bool saml_configured: bool providers: list[AuthProviderStatus] allowed_email_domains: list[str] redirect_uri: str dev_login_enabled: bool class DevLoginRequest(BaseModel): email: str role: RoleName = "learner" display_name: str | None = None def _normalize_domain(domain: str | None) -> str: return (domain or "").strip().lower().lstrip("@") def _normalize_email(email: str | None) -> str: return (email or "").strip().lower() def _email_domain(email: str | None) -> str: value = _normalize_email(email) if "@" not in value: return "" return value.rsplit("@", 1)[1] def _normalize_email_set(values: list[str]) -> set[str]: return {email for value in values if (email := _normalize_email(value))} def _google_configured() -> bool: return bool(settings.oauth_google_client_id and settings.oauth_google_client_secret) def _saml_configured() -> bool: return bool( settings.auth_saml_enabled and settings.saml_sp_entity_id.strip() and settings.saml_sso_url.strip() ) def _auth_provider_statuses() -> list[AuthProviderStatus]: google_ready = _google_configured() saml_ready = _saml_configured() return [ AuthProviderStatus( provider="google", configured=google_ready, enabled=google_ready, login_path="/auth/login?provider=google", ), AuthProviderStatus( provider="saml", configured=saml_ready, enabled=saml_ready, login_path="/auth/login?provider=saml", ), ] def allowed_email_domains() -> set[str]: """Configured login email domains, normalized for claim checks.""" return { normalized for domain in settings.auth_allowed_email_domains if (normalized := _normalize_domain(domain)) } def validate_google_identity_domain( *, email: str | None, email_verified: bool, hosted_domain: str | None = None, ) -> str: """Reject Google identities outside the allowed email domain list. Google Console authorized domains protect app/redirect domains, not user email domains. After id_token signature/audience/issuer validation, call this check with the `email`, `email_verified`, and optional `hd` claims. """ normalized_email = _normalize_email(email) domain = _email_domain(normalized_email) allowed = allowed_email_domains() if not allowed: raise HTTPException( status.HTTP_500_INTERNAL_SERVER_ERROR, detail="allowed email domains are not configured", ) if not normalized_email or not domain: raise HTTPException(status.HTTP_403_FORBIDDEN, detail="email claim is required") if not email_verified: raise HTTPException(status.HTTP_403_FORBIDDEN, detail="email is not verified") if domain not in allowed: raise HTTPException(status.HTTP_403_FORBIDDEN, detail="email domain is not allowed") hd = _normalize_domain(hosted_domain) if hd and hd not in allowed: raise HTTPException(status.HTTP_403_FORBIDDEN, detail="hosted domain is not allowed") return normalized_email async def validate_login_identity_email( *, email: str | None, email_verified: bool, hosted_domain: str | None = None, ) -> tuple[str, ManagedUser | None]: """Validate provider email, allowing exact admin-created managed accounts.""" async def managed_user_for_email(normalized_email: str) -> ManagedUser | None: try: return await get_managed_user_by_email(normalized_email) except HTTPException: return None try: normalized_email = validate_google_identity_domain( email=email, email_verified=email_verified, hosted_domain=hosted_domain, ) except HTTPException as exc: normalized_email = _normalize_email(email) if ( exc.status_code == status.HTTP_403_FORBIDDEN and normalized_email and _email_domain(normalized_email) and email_verified ): managed_user = await managed_user_for_email(normalized_email) if managed_user is not None: return normalized_email, managed_user raise return normalized_email, await managed_user_for_email(normalized_email) def _role_for_email(email: str) -> Role: normalized = _normalize_email(email) if normalized in _normalize_email_set(settings.auth_super_admin_emails): return Role.ADMIN if normalized in _normalize_email_set(settings.auth_admin_emails): return Role.ADMIN if normalized in _normalize_email_set(settings.auth_teacher_emails): return Role.TEACHER return Role.LEARNER def _role_for_managed_user(managed_user: ManagedUser | None, fallback: Role) -> Role: if managed_user is None: return fallback if managed_user.role == "admin": return Role.ADMIN if managed_user.role == "teacher": return Role.TEACHER return Role.LEARNER def _cohort_ids_for_managed_user(managed_user: ManagedUser | None, fallback: list[str]) -> list[str]: if managed_user is not None and managed_user.cohort_ids: return list(managed_user.cohort_ids) return fallback def _role_for_saml_identity(identity: SamlIdentity) -> Role: hinted = (identity.role_hint or "").strip().lower() if hinted in {"admin", "administrator"}: return Role.ADMIN if hinted in {"teacher", "instructor", "faculty"}: return Role.TEACHER return _role_for_email(identity.email) def _split_cohort_values(value: str | None) -> list[str]: if not value: return [] return [item.strip() for item in value.split(",") if item.strip()] def _append_unique(items: list[str], values: list[str]) -> None: seen = {item.lower() for item in items} for value in values: key = value.lower() if key and key not in seen: items.append(value) seen.add(key) def _configured_cohort_ids( *, email: str, hosted_domain: str | None = None, claim_hint: str | None = None, ) -> list[str]: normalized_email = _normalize_email(email) domain = _email_domain(normalized_email) hd = _normalize_domain(hosted_domain) email_map = { _normalize_email(key): value for key, value in settings.auth_email_cohort_map.items() if _normalize_email(key) } domain_map = { _normalize_domain(key): value for key, value in settings.auth_domain_cohort_map.items() if _normalize_domain(key) } cohorts: list[str] = [] _append_unique(cohorts, _split_cohort_values(email_map.get(normalized_email))) _append_unique(cohorts, _split_cohort_values(domain_map.get(domain))) if hd and hd != domain: _append_unique(cohorts, _split_cohort_values(domain_map.get(hd))) _append_unique(cohorts, _split_cohort_values(claim_hint)) return cohorts def _provider_external_id(provider: str, subject: str | None, email: str) -> str: value = (subject or "").strip() or _normalize_email(email) return f"{provider}:{value.lower()}" def _safe_next_path(next_path: str | None) -> str: if not next_path or not next_path.startswith("/") or next_path.startswith("//"): return "/" return next_path def _is_generic_admin_entry_path(next_path: str) -> bool: path = urlsplit(_safe_next_path(next_path)).path.rstrip("/") or "/" return path in GENERIC_ADMIN_ENTRY_PATHS def _post_login_next_path( next_path: str, *, email: str, role: Role, managed_user: ManagedUser | None, ) -> str: safe_next = _safe_next_path(next_path) stored_admin_access = managed_user.admin_access if managed_user is not None else False if _is_generic_admin_entry_path(safe_next) and has_admin_access( email, role.value, stored_admin_access, ): return "/admin" return safe_next def _url_origin(value: str | None) -> str | None: if not value: return None parsed = urlsplit(value) if parsed.scheme not in {"http", "https"} or not parsed.netloc: return None return f"{parsed.scheme}://{parsed.netloc}".rstrip("/") def _is_local_origin(origin: str) -> bool: host = urlsplit(origin).hostname or "" return host in {"localhost", "127.0.0.1", "::1"} def _frontend_origin_map() -> dict[str, str]: mapped: dict[str, str] = {} for api_host, frontend_origin in settings.frontend_origin_map.items(): host = (api_host or "").strip().lower().rstrip(".") origin = _url_origin(frontend_origin) if host and origin: mapped[host] = origin return mapped def _dev_login_extra_origins() -> set[str]: return { origin for value in settings.auth_dev_login_extra_origins if (origin := _url_origin(value)) } def _dev_login_extra_hosts() -> set[str]: return { host for origin in _dev_login_extra_origins() if (host := (urlsplit(origin).hostname or "").lower()) } def _is_dev_login_allowed_origin(origin: str) -> bool: host = (urlsplit(origin).hostname or "").lower() return _is_local_origin(origin) or origin in _dev_login_extra_origins() or host in _dev_login_extra_hosts() def _configured_frontend_origins() -> list[str]: origins: list[str] = [] for value in [ settings.frontend_base_url, *settings.frontend_origin_map.values(), *settings.cors_origins, *settings.auth_dev_login_extra_origins, ]: origin = _url_origin(value) if origin and origin not in origins: origins.append(origin) return origins def _frontend_origin_for_request(request: Request | None = None) -> str: origins = _configured_frontend_origins() fallback = (_url_origin(settings.frontend_base_url) or "http://localhost:5173").rstrip("/") if request is not None: for header_name in ("origin", "referer"): candidate = _url_origin(request.headers.get(header_name)) if candidate in origins: return candidate forwarded_host = request.headers.get("x-forwarded-host") host = (forwarded_host or request.headers.get("host") or "").split(",", 1)[0].strip() hostname = host.rsplit(":", 1)[0].lower() if host else "" if mapped_origin := _frontend_origin_map().get(hostname): return mapped_origin forwarded_proto = ( request.headers.get("x-forwarded-proto", "").split(",", 1)[0].strip().lower() ) if forwarded_host and forwarded_proto in {"http", "https"}: forwarded_origin = _url_origin(f"{forwarded_proto}://{host}") if ( settings.environment == "dev" and forwarded_origin and _is_dev_login_allowed_origin(forwarded_origin) ): return forwarded_origin if hostname in {"localhost", "127.0.0.1", "::1"}: return fallback if not _is_local_origin(fallback): return fallback for origin in origins: hostname = (urlsplit(origin).hostname or "").lower() if not _is_local_origin(origin) and hostname != "api-vignette.chanpaca.net": return origin return fallback def _frontend_url(path: str, request: Request | None = None) -> str: base = _frontend_origin_for_request(request) return f"{base}{_safe_next_path(path)}" def _pkce_challenge(verifier: str) -> str: digest = hashlib.sha256(verifier.encode("ascii")).digest() return base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=") def _b64url_encode(value: bytes) -> str: return base64.urlsafe_b64encode(value).decode("ascii").rstrip("=") def _b64url_decode(value: str) -> bytes: padded = value + ("=" * (-len(value) % 4)) return base64.urlsafe_b64decode(padded.encode("ascii")) def _oauth_state_signature(payload: str) -> str: digest = hmac.new( settings.session_secret.encode("utf-8"), f"oauth-state:{payload}".encode("utf-8"), hashlib.sha256, ).digest() return _b64url_encode(digest) def _oauth_code_verifier_for_state(state: str) -> str: digest = hmac.new( settings.session_secret.encode("utf-8"), f"oauth-pkce:{state}".encode("utf-8"), hashlib.sha256, ).digest() return _b64url_encode(digest) def _build_oauth_state(next_path: str | None) -> tuple[str, OAuthState]: created_at = time.time() safe_next = _safe_next_path(next_path) payload = { "iat": created_at, "next": safe_next, "nonce": secrets.token_urlsafe(24), } payload_blob = _b64url_encode( json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8") ) state = f"{payload_blob}.{_oauth_state_signature(payload_blob)}" return state, OAuthState( code_verifier=_oauth_code_verifier_for_state(state), next_path=safe_next, created_at=created_at, ) def _oauth_state_from_signed_token(state: str | None) -> OAuthState | None: if not state or "." not in state: return None payload_blob, signature = state.rsplit(".", 1) if not payload_blob or not signature: return None if not hmac.compare_digest(signature, _oauth_state_signature(payload_blob)): return None try: payload = json.loads(_b64url_decode(payload_blob).decode("utf-8")) created_at = float(payload.get("iat", 0)) except (ValueError, TypeError, json.JSONDecodeError): return None if created_at <= 0 or created_at < time.time() - OAUTH_STATE_TTL_SECONDS: return None if not isinstance(payload.get("nonce"), str): return None next_path = payload.get("next") if not isinstance(next_path, str): return None return OAuthState( code_verifier=_oauth_code_verifier_for_state(state), next_path=_safe_next_path(next_path), created_at=created_at, ) def _oauth_state_for_callback(state: str | None, cookie_state: str | None) -> OAuthState | None: if not state: return None if cookie_state != state: return None stored = _oauth_states.pop(state, None) if stored is not None: return stored return _oauth_state_from_signed_token(state) def _prune_oauth_states() -> None: cutoff = time.time() - OAUTH_STATE_TTL_SECONDS stale = [key for key, value in _oauth_states.items() if value.created_at < cutoff] for key in stale: _oauth_states.pop(key, None) def _prune_saml_states() -> None: cutoff = time.time() - OAUTH_STATE_TTL_SECONDS stale = [key for key, value in _saml_states.items() if value.created_at < cutoff] for key in stale: _saml_states.pop(key, None) def _cookie_secure() -> bool: # The __Host- prefix requires Secure, Path=/, and no Domain. Modern Chrome # accepts Secure cookies on localhost, which keeps dev and prod semantics # aligned. return settings.is_prod or settings.cookie_name.startswith("__Host-") def _set_session_cookie(response: Response, sid: str) -> None: response.set_cookie( key=settings.cookie_name, value=sid, max_age=settings.session_ttl_seconds, httponly=True, secure=_cookie_secure(), samesite="lax", path="/", ) if settings.environment == "dev": response.set_cookie( key="vignette_sid", value=sid, max_age=settings.session_ttl_seconds, httponly=True, secure=False, samesite="lax", path="/", ) def _set_oauth_state_cookie(response: Response, state: str) -> None: response.set_cookie( key=OAUTH_STATE_COOKIE_NAME, value=state, max_age=OAUTH_STATE_TTL_SECONDS, httponly=True, secure=True, samesite="lax", path="/", ) if settings.environment == "dev": response.set_cookie( key=DEV_OAUTH_STATE_COOKIE_NAME, value=state, max_age=OAUTH_STATE_TTL_SECONDS, httponly=True, secure=False, samesite="lax", path="/", ) def _delete_oauth_state_cookie(response: Response) -> None: response.delete_cookie( OAUTH_STATE_COOKIE_NAME, httponly=True, secure=True, samesite="lax", path="/", ) if settings.environment == "dev": response.delete_cookie( DEV_OAUTH_STATE_COOKIE_NAME, httponly=True, secure=False, samesite="lax", path="/", ) def _delete_session_cookie(response: Response) -> None: response.delete_cookie( settings.cookie_name, httponly=True, secure=_cookie_secure(), samesite="lax", path="/", ) if settings.environment == "dev": response.delete_cookie( "vignette_sid", httponly=True, secure=False, samesite="lax", path="/", ) async def _me_response(user: SessionUser | Principal) -> MeResponse: managed = await get_managed_user(user.user_id) onboarding_completed_at = getattr(user, "profile_completed_at", None) if managed: onboarding_completed_at = ( managed.profile_completed_at if ( managed.profile_completed_at is not None and managed.terms_agreed_at is not None and managed.privacy_agreed_at is not None and managed.nickname.strip() and managed.self_introduction.strip() ) else None ) account_status = ( managed.account_status if managed else getattr(user, "account_status", "approved") ) email = getattr(user, "email", "") role = user.role.value if isinstance(user.role, Role) else user.role stored_admin_access = managed.admin_access if managed else getattr(user, "admin_access", False) return MeResponse( user_id=user.user_id, email=email, display_name=( (managed.display_name if managed else "") or getattr(user, "display_name", "") or email ), role=role, admin_access=has_admin_access(email, role, stored_admin_access), super_admin=is_super_admin_email(email), account_status=account_status, approval_required=account_status != "approved", cohort_ids=user.cohort_ids, consent_at=(managed.consent_at if managed else getattr(user, "consent_at", None)), onboarding_completed_at=onboarding_completed_at, nickname=(managed.nickname if managed else ""), self_introduction=(managed.self_introduction if managed else ""), avatar_url=(managed.avatar_url if managed else ""), ) def _frontend_login_redirect(reason: str, request: Request) -> RedirectResponse: base_url = _frontend_origin_for_request(request) return RedirectResponse(f"{base_url}/login?{urlencode({'oauth': reason})}", status_code=302) def _oauth_callback_error(reason: str, request: Request) -> RedirectResponse: response = _frontend_login_redirect(reason, request) _delete_oauth_state_cookie(response) return response def _log_oauth_callback_failure(request: Request, reason: str, **fields: object) -> None: """Log OAuth callback failures without authorization codes, tokens, or raw user IDs.""" host = request.headers.get("x-forwarded-host") or request.headers.get("host") logger.warning( "google_oauth_callback_failed reason=%s host=%s forwarded_proto=%s details=%s", reason, host, request.headers.get("x-forwarded-proto"), fields, ) def _dev_login_available(request: Request) -> bool: if settings.environment != "dev" or not settings.auth_dev_login_enabled: return False saw_browser_origin = False for header_name in ("origin", "referer"): origin = _url_origin(request.headers.get(header_name)) if origin: saw_browser_origin = True if not _is_dev_login_allowed_origin(origin): return False if saw_browser_origin: return True forwarded_host = request.headers.get("x-forwarded-host") host = (forwarded_host or request.headers.get("host") or "").split(",", 1)[0].strip() forwarded_proto = ( request.headers.get("x-forwarded-proto", "").split(",", 1)[0].strip().lower() ) scheme = forwarded_proto if forwarded_host and forwarded_proto in {"http", "https"} else "http" origin = _url_origin(f"{scheme}://{host}") if host else None return bool(origin and _is_dev_login_allowed_origin(origin)) def _dev_oauth_redirect_unavailable(request: Request) -> bool: redirect_origin = _url_origin(settings.oauth_redirect_uri) return bool( _dev_login_available(request) and redirect_origin and not _is_local_origin(redirect_origin) ) @router.get("/config", response_model=AuthConfigResponse) async def auth_config(request: Request) -> AuthConfigResponse: """Return non-secret login configuration for the browser login screen.""" google_ready = _google_configured() saml_ready = _saml_configured() return AuthConfigResponse( google_oauth_configured=google_ready, saml_configured=saml_ready, providers=_auth_provider_statuses(), allowed_email_domains=sorted(allowed_email_domains()), redirect_uri=settings.oauth_redirect_uri, dev_login_enabled=_dev_login_available(request), ) @router.get("/login") async def login( request: Request, provider: Annotated[str, Query()] = "google", next: Annotated[str | None, Query()] = None, ) -> RedirectResponse: """Start Google OIDC authorization code + PKCE login.""" if provider == "saml": if not _saml_configured(): return _frontend_login_redirect("saml_not_configured", request) _prune_saml_states() relay_state = secrets.token_urlsafe(32) acs_url = acs_url_for_entity_id(settings.saml_sp_entity_id) request_id, authn_request_xml = build_authn_request( sp_entity_id=settings.saml_sp_entity_id, sso_url=settings.saml_sso_url, acs_url=acs_url, ) _saml_states[relay_state] = SamlState( request_id=request_id, next_path=_safe_next_path(next), created_at=time.time(), ) return RedirectResponse( redirect_binding_url( sso_url=settings.saml_sso_url, authn_request_xml=authn_request_xml, relay_state=relay_state, ), status_code=302, ) if provider != "google": return _frontend_login_redirect("unsupported_provider", request) if not _google_configured(): return _frontend_login_redirect("not_configured", request) if _dev_oauth_redirect_unavailable(request): return _frontend_login_redirect("local_oauth_unavailable", request) _prune_oauth_states() state, stored_state = _build_oauth_state(next) _oauth_states[state] = stored_state params = { "client_id": settings.oauth_google_client_id, "redirect_uri": settings.oauth_redirect_uri, "response_type": "code", "scope": "openid email profile", "state": state, "code_challenge": _pkce_challenge(stored_state.code_verifier), "code_challenge_method": "S256", "prompt": "select_account", } response = RedirectResponse(f"{GOOGLE_AUTHORIZE_URL}?{urlencode(params)}", status_code=302) _set_oauth_state_cookie(response, state) return response @router.get("/callback") async def callback( request: Request, code: Annotated[Optional[str], Query()] = None, state: Annotated[Optional[str], Query()] = None, error: Annotated[Optional[str], Query()] = None, error_description: Annotated[Optional[str], Query()] = None, oauth_state_cookie: Annotated[Optional[str], Cookie(alias=OAUTH_STATE_COOKIE_NAME)] = None, dev_oauth_state_cookie: Annotated[Optional[str], Cookie(alias=DEV_OAUTH_STATE_COOKIE_NAME)] = None, ) -> RedirectResponse: """Exchange Google auth code, validate identity, and issue a BFF cookie.""" if error: reason = "access_denied" if error == "access_denied" else "provider_error" _log_oauth_callback_failure( request, reason, provider_error=error, has_state=bool(state), has_error_description=bool(error_description), ) return _oauth_callback_error(reason, request) if not code or not state: _log_oauth_callback_failure( request, "missing_callback", has_code=bool(code), has_state=bool(state), ) return _oauth_callback_error("missing_callback", request) _prune_oauth_states() cookie_state = oauth_state_cookie or ( dev_oauth_state_cookie if settings.environment == "dev" else None ) stored = _oauth_state_for_callback(state, cookie_state) if stored is None: _log_oauth_callback_failure( request, "invalid_state", has_cookie=bool(cookie_state), state_in_memory=state in _oauth_states if state else False, ) return _oauth_callback_error("invalid_state", request) async with httpx.AsyncClient(timeout=10.0) as client: token_res = await client.post( GOOGLE_TOKEN_URL, data={ "client_id": settings.oauth_google_client_id, "client_secret": settings.oauth_google_client_secret, "code": code, "grant_type": "authorization_code", "redirect_uri": settings.oauth_redirect_uri, "code_verifier": stored.code_verifier, }, headers={"Accept": "application/json"}, ) if token_res.status_code >= 400: token_error: object try: token_body = token_res.json() token_error = { "error": token_body.get("error"), "error_description": token_body.get("error_description"), } except Exception: token_error = "non_json_error" _log_oauth_callback_failure( request, "token_exchange_failed", status_code=token_res.status_code, token_error=token_error, ) return _oauth_callback_error("token_exchange_failed", request) token_payload = token_res.json() id_token = token_payload.get("id_token") if not isinstance(id_token, str) or not id_token: _log_oauth_callback_failure(request, "id_token_missing") return _oauth_callback_error("id_token_missing", request) info_res = await client.get(GOOGLE_TOKENINFO_URL, params={"id_token": id_token}) if info_res.status_code >= 400: _log_oauth_callback_failure( request, "id_token_invalid", status_code=info_res.status_code, ) return _oauth_callback_error("id_token_invalid", request) claims = info_res.json() if claims.get("aud") != settings.oauth_google_client_id: _log_oauth_callback_failure(request, "audience_mismatch") return _oauth_callback_error("audience_mismatch", request) issuer = claims.get("iss") if issuer not in {"accounts.google.com", "https://accounts.google.com"}: _log_oauth_callback_failure(request, "issuer_mismatch", issuer=issuer) return _oauth_callback_error("issuer_mismatch", request) try: email, managed_user = await validate_login_identity_email( email=claims.get("email"), email_verified=claims.get("email_verified") in {True, "true", "True", "1", 1}, hosted_domain=claims.get("hd"), ) except HTTPException: _log_oauth_callback_failure( request, "domain_not_allowed", email_domain=_email_domain(str(claims.get("email") or "")), hosted_domain=_normalize_domain(str(claims.get("hd") or "")), ) return _oauth_callback_error("domain_not_allowed", request) role = _role_for_managed_user(managed_user, _role_for_email(email)) display_name = str(claims.get("name") or email) cohort_ids = _cohort_ids_for_managed_user( managed_user, _configured_cohort_ids( email=email, hosted_domain=str(claims.get("hd") or ""), ), ) external_id = _provider_external_id("google", str(claims.get("sub") or ""), email) try: sid, _ = await create_session( email=email, display_name=display_name, role=role.value, cohort_ids=cohort_ids, external_id=external_id, ) except InactiveUserError as exc: _log_oauth_callback_failure( request, "inactive_user", email_domain=_email_domain(email), ) return _oauth_callback_error("inactive_user", request) next_path = _post_login_next_path( stored.next_path, email=email, role=role, managed_user=managed_user, ) response = RedirectResponse(_frontend_url(next_path, request), status_code=302) _set_session_cookie(response, sid) _delete_oauth_state_cookie(response) return response @router.post("/saml/acs") async def saml_acs(request: Request) -> RedirectResponse: """Accept a minimal unsigned SAMLResponse for local fixture SAML proof. Signed SAML verification is intentionally not implemented. When SAML_X509_CERT_FINGERPRINT is configured, this endpoint refuses to trust the response so production does not silently run unsigned SAML. """ if not _saml_configured(): return _frontend_login_redirect("saml_not_configured", request) if settings.saml_x509_cert_fingerprint.strip(): return _frontend_login_redirect("saml_signature_verification_required", request) if settings.environment != "dev": return _frontend_login_redirect("saml_fixture_acs_dev_only", request) form = await request.form() relay_state = str(form.get("RelayState") or "") encoded_response = str(form.get("SAMLResponse") or "") if not relay_state or not encoded_response: return _frontend_login_redirect("saml_missing_callback", request) _prune_saml_states() stored = _saml_states.pop(relay_state, None) if stored is None: return _frontend_login_redirect("saml_invalid_state", request) try: identity = parse_fixture_response(encoded_response) email, managed_user = await validate_login_identity_email( email=identity.email, email_verified=True, hosted_domain=_email_domain(identity.email), ) except (HTTPException, ValueError): return _frontend_login_redirect("saml_assertion_invalid", request) role = _role_for_managed_user(managed_user, _role_for_saml_identity(identity)) cohort_ids = _cohort_ids_for_managed_user( managed_user, _configured_cohort_ids(email=email, claim_hint=identity.cohort_hint), ) external_id = _provider_external_id("saml", identity.subject, email) try: sid, _ = await create_session( email=email, display_name=identity.display_name or email, role=role.value, cohort_ids=cohort_ids, external_id=external_id, ) except InactiveUserError: return _frontend_login_redirect("inactive_user", request) next_path = _post_login_next_path( stored.next_path, email=email, role=role, managed_user=managed_user, ) response = RedirectResponse(_frontend_url(next_path, request), status_code=302) _set_session_cookie(response, sid) return response @router.post("/dev-login", response_model=MeResponse) async def dev_login(request: Request, body: DevLoginRequest, response: Response) -> MeResponse: """Dev-only server login for local E2E and manual testing. This is not a browser-side auth shortcut: the role is stored server-side and the browser only gets the same opaque HttpOnly cookie used by OAuth. """ if not _dev_login_available(request): raise HTTPException(status.HTTP_404_NOT_FOUND, detail="dev login is disabled") email, managed_user = await validate_login_identity_email( email=str(body.email), email_verified=True, hosted_domain=_email_domain(str(body.email)), ) try: sid, user = await create_session( email=email, display_name=body.display_name or email, role=_role_for_managed_user(managed_user, Role(body.role)).value, cohort_ids=_cohort_ids_for_managed_user( managed_user, _configured_cohort_ids(email=email), ), external_id=_provider_external_id("dev", email, email), ) except InactiveUserError as exc: raise HTTPException(status.HTTP_403_FORBIDDEN, detail="user is inactive") from exc _set_session_cookie(response, sid) return await _me_response(user) @router.post("/logout") async def logout( response: Response, session_cookie: Annotated[Optional[str], Cookie(alias="__Host-vignette_sid")] = None, dev_session_cookie: Annotated[Optional[str], Cookie(alias="vignette_sid")] = None, ) -> dict[str, bool]: """Revoke the current server session and expire the browser cookie.""" await revoke_session(session_cookie or (dev_session_cookie if settings.environment == "dev" else None)) _delete_session_cookie(response) return {"ok": True} @router.post("/consent", response_model=ConsentResponse) async def accept_consent( body: ConsentRequest, principal: CurrentPrincipal, ) -> ConsentResponse: """Record the current learner's practice-session consent receipt.""" if principal.role != Role.LEARNER and principal.can_access_role(Role.LEARNER): principal = principal.with_role(Role.LEARNER) if principal.role != Role.LEARNER: raise HTTPException(status.HTTP_403_FORBIDDEN, detail="learner consent only") if not body.accepted: raise HTTPException(status.HTTP_400_BAD_REQUEST, detail="consent_not_accepted") consent_at = await record_user_consent(principal.user_id) if consent_at is None: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="user not found") principal.consent_at = consent_at return ConsentResponse(consent_at=consent_at) @router.delete("/consent", response_model=ConsentResponse) async def withdraw_consent(principal: CurrentPrincipal) -> ConsentResponse: """Withdraw practice-session consent until the learner accepts again.""" if principal.role != Role.LEARNER and principal.can_access_role(Role.LEARNER): principal = principal.with_role(Role.LEARNER) if principal.role != Role.LEARNER: raise HTTPException(status.HTTP_403_FORBIDDEN, detail="learner consent only") changed = await withdraw_user_consent(principal.user_id) if not changed: raise HTTPException(status.HTTP_404_NOT_FOUND, detail="user not found") principal.consent_at = None return ConsentResponse(consent_at=None) @router.get("/me", response_model=MeResponse) async def me(principal: CurrentPrincipal) -> MeResponse: """Return the current authenticated user. Unauthenticated requests are 401.""" return await _me_response(principal)