paskia/paskia/fastapi/api.py

import logging
from contextlib import suppress
from datetime import datetime, timedelta, timezone

from fastapi import (
    Depends,
    FastAPI,
    HTTPException,
    Query,
    Request,
    Response,
)
from fastapi.responses import JSONResponse
from fastapi.security import HTTPBearer

from paskia import db
from paskia.authsession import (
    EXPIRES,
    get_reset,
    get_session,
    refresh_session_token,
)
from paskia.fastapi import authz, session, user
from paskia.fastapi.session import AUTH_COOKIE, AUTH_COOKIE_NAME
from paskia.globals import passkey as global_passkey
from paskia.util import frontend, hostutil, htmlutil, passphrase, userinfo

bearer_auth = HTTPBearer(auto_error=True)

app = FastAPI()

app.mount("/user", user.app)


@app.exception_handler(HTTPException)
async def http_exception_handler(_request: Request, exc: HTTPException):
    """Ensure auth cookie is cleared on 401 responses (JSON responses only)."""
    if exc.status_code == 401:
        resp = JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
        session.clear_session_cookie(resp)
        return resp
    return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})


# Refresh only if at least this much of the session lifetime has been *consumed*.
# Consumption is derived from (now + EXPIRES) - current_expires.
# This guarantees a minimum spacing between DB writes even with frequent /validate calls.
_REFRESH_INTERVAL = timedelta(minutes=5)


@app.exception_handler(ValueError)
async def value_error_handler(_request: Request, exc: ValueError):
    return JSONResponse(status_code=400, content={"detail": str(exc)})


@app.exception_handler(authz.AuthException)
async def auth_exception_handler(_request: Request, exc: authz.AuthException):
    """Handle AuthException with auth info for UI."""
    return JSONResponse(
        status_code=exc.status_code,
        content=await authz.auth_error_content(exc),
    )


@app.exception_handler(Exception)
async def general_exception_handler(
    _request: Request, exc: Exception
):  # pragma: no cover
    logging.exception("Unhandled exception in API app")
    return JSONResponse(status_code=500, content={"detail": "Internal server error"})


@app.post("/validate")
async def validate_token(
    request: Request,
    response: Response,
    perm: list[str] = Query([]),
    max_age: str | None = Query(None),
    auth=AUTH_COOKIE,
):
    """Validate the current session and extend its expiry.

    Always refreshes the session (sliding expiration) and re-sets the cookie with a
    renewed max-age. This keeps active users logged in without needing a separate
    refresh endpoint.
    """
    try:
        ctx = await authz.verify(
            auth,
            perm,
            host=request.headers.get("host"),
            max_age=max_age,
        )
    except HTTPException:
        # Global handler will clear cookie if 401
        raise
    renewed = False
    if auth:
        consumed = EXPIRES - (ctx.session.expiry - datetime.now(timezone.utc))
        if not timedelta(0) < consumed < _REFRESH_INTERVAL:
            try:
                await refresh_session_token(
                    auth,
                    ip=request.client.host if request.client else "",
                    user_agent=request.headers.get("user-agent") or "",
                )
                session.set_session_cookie(response, auth)
                renewed = True
            except ValueError:
                # Session disappeared, e.g. due to concurrent logout; global handler will clear
                raise authz.AuthException(
                    status_code=401, detail="Session expired", mode="login"
                )
    return {
        "valid": True,
        "user_uuid": str(ctx.session.user_uuid),
        "renewed": renewed,
    }


@app.get("/forward")
async def forward_authentication(
    request: Request,
    response: Response,
    perm: list[str] = Query([]),
    max_age: str | None = Query(None),
    auth=AUTH_COOKIE,
):
    """Forward auth validation for Caddy/Nginx.

    Query Params:
    - perm: repeated permission IDs the authenticated user must possess (ALL required).
    - max_age: maximum age of authentication (e.g., "5m", "1h", "30s"). If the session
               is older than this, user must re-authenticate.

    Success: 204 No Content with Remote-* headers describing the authenticated user.
    Failure (unauthenticated / unauthorized): 4xx response.
        - If Accept header contains "text/html": HTML page for authentication
          with data attributes for mode and other metadata.
        - Otherwise: JSON response with error details and an `iframe` field
          pointing to /auth/restricted/?mode=... for iframe-based authentication.
    """
    try:
        ctx = await authz.verify(
            auth, perm, host=request.headers.get("host"), max_age=max_age
        )
        role_permissions = set(ctx.role.permissions or [])
        if ctx.permissions:
            role_permissions.update(permission.scope for permission in ctx.permissions)

        remote_headers: dict[str, str] = {
            "Remote-User": str(ctx.user.uuid),
            "Remote-Name": ctx.user.display_name,
            "Remote-Groups": ",".join(sorted(role_permissions)),
            "Remote-Org": str(ctx.org.uuid),
            "Remote-Org-Name": ctx.org.display_name,
            "Remote-Role": str(ctx.role.uuid),
            "Remote-Role-Name": ctx.role.display_name,
            "Remote-Session-Expires": (
                ctx.session.expiry.astimezone(timezone.utc)
                .isoformat()
                .replace("+00:00", "Z")
                if ctx.session.expiry.tzinfo
                else ctx.session.expiry.replace(tzinfo=timezone.utc)
                .isoformat()
                .replace("+00:00", "Z")
            ),
            "Remote-Credential": str(ctx.session.credential_uuid),
        }
        return Response(status_code=204, headers=remote_headers)
    except authz.AuthException as e:
        # Clear cookie only if session is invalid (not for reauth)
        if e.clear_session:
            session.clear_session_cookie(response)

        # Check Accept header to decide response format
        accept = request.headers.get("accept", "")
        wants_html = "text/html" in accept

        if wants_html:
            # Browser request - return full-page HTML with metadata
            data_attrs = {"mode": e.mode, **e.metadata}
            html = (await frontend.read("/int/forward/index.html"))[0]
            html = htmlutil.patch_html_data_attrs(html, **data_attrs)
            return Response(
                html, status_code=e.status_code, media_type="text/html; charset=UTF-8"
            )
        else:
            # API request - return JSON with iframe srcdoc HTML
            return JSONResponse(
                status_code=e.status_code,
                content=await authz.auth_error_content(e),
            )


@app.get("/settings")
async def get_settings():
    pk = global_passkey.instance
    base_path = hostutil.ui_base_path()
    return {
        "rp_id": pk.rp_id,
        "rp_name": pk.rp_name,
        "ui_base_path": base_path,
        "auth_host": hostutil.dedicated_auth_host(),
        "auth_site_url": hostutil.auth_site_url(),
        "session_cookie": AUTH_COOKIE_NAME,
    }


@app.get("/token-info")
async def api_token_info(token: str):
    """Get information about a reset token.

    Returns:
        - type: "reset"
        - user_name: display name of the user
        - token_type: type of reset token
    """
    if not passphrase.is_well_formed(token):
        raise HTTPException(status_code=404, detail="Invalid token")

    # Check if this is a reset token
    try:
        reset_token = await get_reset(token)
        user = db.get_user_by_uuid(reset_token.user_uuid)
        return {
            "type": "reset",
            "user_name": user.display_name,
            "token_type": reset_token.token_type,
        }
    except (ValueError, Exception):
        raise HTTPException(status_code=404, detail="Token not found or expired")


@app.post("/user-info")
async def api_user_info(
    request: Request,
    response: Response,
    reset: str | None = None,
    auth=AUTH_COOKIE,
):
    """Get user information including credentials, sessions, and permissions.

    Can be called with either:
    - A session cookie (auth) for authenticated users
    - A reset token for users in password reset flow
    """
    authenticated = False
    session_record = None
    reset_token = None
    try:
        if reset:
            if not passphrase.is_well_formed(reset):
                raise ValueError("Invalid reset token")
            reset_token = await get_reset(reset)
            target_user_uuid = reset_token.user_uuid
        else:
            if auth is None:
                raise authz.AuthException(
                    status_code=401,
                    detail="Authentication required",
                    mode="login",
                )
            session_record = await get_session(auth, host=request.headers.get("host"))
            authenticated = True
            target_user_uuid = session_record.user_uuid
    except ValueError as e:
        raise HTTPException(401, str(e))

    # Return minimal response for reset tokens
    if not authenticated and reset_token:
        return await userinfo.format_reset_user_info(target_user_uuid, reset_token)

    # Return full user info for authenticated users
    assert auth is not None
    assert session_record is not None

    return await userinfo.format_user_info(
        user_uuid=target_user_uuid,
        auth=auth,
        session_record=session_record,
        request_host=request.headers.get("host"),
    )


@app.post("/logout")
async def api_logout(request: Request, response: Response, auth=AUTH_COOKIE):
    if not auth:
        return {"message": "Already logged out"}
    try:
        s = await get_session(auth, host=request.headers.get("host"))
    except ValueError:
        return {"message": "Already logged out"}
    with suppress(Exception):
        db.delete_session(auth, actor=str(s.user_uuid))
    session.clear_session_cookie(response)
    return {"message": "Logged out successfully"}


@app.post("/set-session")
async def api_set_session(
    request: Request, response: Response, auth=Depends(bearer_auth)
):
    user = await get_session(auth.credentials, host=request.headers.get("host"))
    session.set_session_cookie(response, auth.credentials)
    return {
        "message": "Session cookie set successfully",
        "user_uuid": str(user.user_uuid),
    }