hold-slayer/api/auth.py

"""
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(
            '<script>window.parent.postMessage('
            '{type:"hold-slayer-refresh",error:true},"*");</script>'
        )

    sdk = get_sdk()
    try:
        token = await sdk.get_oauth_token(code=code)
        access_token = token.get("access_token", "")
    except Exception:
        return HTMLResponse(
            '<script>window.parent.postMessage('
            '{type:"hold-slayer-refresh",error:true},"*");</script>'
        )

    return HTMLResponse(
        f'<script>window.parent.postMessage('
        f'{{type:"hold-slayer-refresh",token:"{access_token}"}},"*");</script>'
    )


@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)