985 lines
33 KiB
Python
985 lines
33 KiB
Python
"""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_sessions import (
|
|
InactiveUserError,
|
|
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"
|
|
|
|
|
|
@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: str
|
|
cohort_ids: list[str]
|
|
consent_at: float | None = None
|
|
|
|
|
|
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: Literal["learner", "teacher", "admin"] = "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
|
|
|
|
|
|
def _role_for_email(email: str) -> Role:
|
|
normalized = _normalize_email(email)
|
|
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_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 _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="/",
|
|
)
|
|
|
|
|
|
def _me_response(user: SessionUser | Principal) -> MeResponse:
|
|
return MeResponse(
|
|
user_id=user.user_id,
|
|
email=getattr(user, "email", ""),
|
|
display_name=getattr(user, "display_name", "") or getattr(user, "email", ""),
|
|
role=user.role.value if isinstance(user.role, Role) else user.role,
|
|
cohort_ids=user.cohort_ids,
|
|
consent_at=getattr(user, "consent_at", None),
|
|
)
|
|
|
|
|
|
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 = validate_google_identity_domain(
|
|
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_email(email)
|
|
display_name = str(claims.get("name") or email)
|
|
cohort_ids = _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)
|
|
|
|
response = RedirectResponse(_frontend_url(stored.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 = validate_google_identity_domain(
|
|
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_saml_identity(identity)
|
|
cohort_ids = _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)
|
|
|
|
response = RedirectResponse(_frontend_url(stored.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 = validate_google_identity_domain(
|
|
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=body.role,
|
|
cohort_ids=_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 _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:
|
|
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:
|
|
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 _me_response(principal)
|