"""
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, Depends, 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 core.rate_limit import rate_limit
from db.database import session_scope
router = APIRouter(prefix="/auth", tags=["auth"])
# These are the only routes that must answer before an identity exists, so they
# are the only ones worth limiting — everything else is already owner-gated.
# `/callback` and `/refresh-callback` each trigger an outbound token exchange
# with Casdoor, and `/me` opens a DB session per call; all three are
# unauthenticated work an attacker controls. See core/rate_limit.py.
_limit_callback = [Depends(rate_limit("auth:callback", limit=10))]
_limit_me = [Depends(rate_limit("auth:me"))]
_limit_redirect = [Depends(rate_limit("auth:redirect"))]
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", dependencies=_limit_redirect)
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", dependencies=_limit_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", dependencies=_limit_redirect)
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", dependencies=_limit_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", dependencies=_limit_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)