""" OIDC authentication endpoints (Casdoor SSO). GET /auth/login → redirect to Casdoor authorization URL GET /auth/callback → exchange code for tokens, redirect to UI with token GET /auth/me → return current user info (requires Bearer token) GET /auth/silent-refresh → hidden-iframe refresh (re-auth with existing session) GET /auth/refresh-callback → post the refreshed token to the parent window GET /auth/logout → redirect to Casdoor logout URL The dashboard is owner-only; ``/auth/me`` returns ``is_owner`` so a signed-in non-owner sees an "access denied" screen instead of a bare 401. """ import secrets from urllib.parse import urlencode from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from auth import get_sdk, is_owner, resolve_from_header_or_query from config import get_settings from db.database import session_scope router = APIRouter(prefix="/auth", tags=["auth"]) def _build_casdoor_auth_url( callback: str, *, scope: str = "openid profile email", state: str | None = None, prompt: str | None = None, ) -> str: """Build the Casdoor authorization URL directly. The SDK's get_auth_link() doesn't support the ``prompt`` parameter that silent refresh needs, so build the URL manually. """ c = get_settings().casdoor params = { "client_id": c.client_id, "response_type": "code", "redirect_uri": callback, "scope": scope, "state": state or secrets.token_urlsafe(16), } if prompt: params["prompt"] = prompt return f"{c.endpoint.rstrip('/')}/login/oauth/authorize?{urlencode(params)}" @router.get("/login") async def login(request: Request, redirect_uri: str = Query(None)): """Redirect the browser to the Casdoor authorization page. No ``prompt=login`` — an existing Casdoor session auto-redirects back with a code without showing the login form (silent SSO across *.helu.ca). """ if not get_settings().casdoor.enabled: raise HTTPException(400, "Casdoor SSO is not enabled") callback = redirect_uri or f"{request.base_url}auth/callback" return RedirectResponse(url=_build_casdoor_auth_url(callback)) @router.get("/callback") async def callback( code: str = Query(...), state: str = Query(None), redirect_uri: str = Query(None), ): """Exchange the authorization code for tokens. Redirects to the dashboard with the access token in the URL *fragment* (``/#token=...``) so the token stays client-side and is stored in localStorage. """ if not get_settings().casdoor.enabled: raise HTTPException(400, "Casdoor SSO is not enabled") sdk = get_sdk() try: token = await sdk.get_oauth_token(code=code) except Exception as exc: raise HTTPException(400, f"Token exchange failed: {exc}") from exc access_token = token.get("access_token", "") return RedirectResponse(url=f"/#token={access_token}") @router.get("/silent-refresh") async def silent_refresh(request: Request): """Start a silent token refresh via hidden iframe (``prompt=none``). If the Casdoor session is still active, Casdoor redirects back to ``/auth/refresh-callback`` with a fresh code — no login form. Otherwise it returns an error and the iframe tells the parent to show the login overlay. """ if not get_settings().casdoor.enabled: raise HTTPException(400, "Casdoor SSO is not enabled") callback = f"{request.base_url}auth/refresh-callback" return RedirectResponse(url=_build_casdoor_auth_url(callback, prompt="none")) @router.get("/refresh-callback") async def refresh_callback( code: str = Query(None), error: str = Query(None), state: str = Query(None), ): """Handle the silent-refresh callback inside the hidden iframe. On success posts the new token to the parent window; on failure posts an error so the parent shows the login overlay. """ if not get_settings().casdoor.enabled: raise HTTPException(400, "Casdoor SSO is not enabled") if error or not code: return HTMLResponse( '' ) sdk = get_sdk() try: token = await sdk.get_oauth_token(code=code) access_token = token.get("access_token", "") except Exception: return HTMLResponse( '' ) return HTMLResponse( f'' ) @router.get("/me") async def me(request: Request): """Return the current authenticated user's profile + ``is_owner``. Resolved manually (not via the ``OwnerUser`` gate) so a signed-in non-owner gets a 200 with ``is_owner:false`` — the dashboard uses that to show the "not authorized" screen rather than treating it as a hard 401. """ auth_header = request.headers.get("authorization") q_token = request.query_params.get("token") async with session_scope() as session: user = await resolve_from_header_or_query(session, auth_header, q_token) if user is None: raise HTTPException( status_code=401, detail="Not authenticated", headers={"WWW-Authenticate": "Bearer"}, ) return JSONResponse( { "id": user.id, "name": user.name, "display_name": user.display_name, "email": user.email, "is_owner": is_owner(user), } ) @router.get("/logout") async def logout(request: Request): """Clear the Casdoor session and redirect back to the app. ``post_logout_redirect_uri`` must be absolute — Casdoor won't follow a relative ``/`` — so it's derived from ``request.base_url`` (works behind HAProxy/nginx with X-Forwarded-Proto/Host). """ c = get_settings().casdoor if not c.enabled: return RedirectResponse(url="/") app_url = str(request.base_url).rstrip("/") logout_url = ( f"{c.endpoint.rstrip('/')}/login/oauth/logout" f"?client_id={c.client_id}" f"&post_logout_redirect_uri={app_url}/auth/login" ) return RedirectResponse(url=logout_url)