paskia/paskia/fastapi/mainapp.py

import logging
import os
from contextlib import asynccontextmanager
from pathlib import Path

import msgspec
from fastapi import FastAPI, HTTPException, Request, Response
from fastapi.responses import FileResponse, RedirectResponse

from paskia import authcode, db, globals
from paskia.__main__ import DEVMODE
from paskia.bootstrap import bootstrap_if_needed
from paskia.db import start_background, stop_background
from paskia.db.background import flush
from paskia.db.logging import configure_db_logging
from paskia.fastapi import admin, api, auth_host, oid, ws
from paskia.fastapi.admin.adminapp import adminapp

# Import frontend instance
from paskia.fastapi.front import frontend
from paskia.fastapi.logging import AccessLogMiddleware, configure_access_logging
from paskia.fastapi.session import AUTH_COOKIE
from paskia.util import hostutil, passphrase, vitedev
from paskia.util.runtime import RuntimeConfig

# Configure custom logging
configure_access_logging()
configure_db_logging()

_access_logger = logging.getLogger("paskia.access")

# Path to examples/index.html when running from source tree
_EXAMPLES_DIR = Path(__file__).parent.parent.parent / "examples"


@asynccontextmanager
async def lifespan(app: FastAPI):  # pragma: no cover - startup path
    """Application lifespan to ensure globals (DB, passkey) are initialized in each process.

    Configuration is passed via PASKIA_CONFIG JSON env variable (set by the CLI entrypoint)
    so that uvicorn reload / multiprocess workers inherit the settings.
    All keys are guaranteed to exist; values are already normalized by __main__.py.
    """
    runtime = msgspec.json.decode(os.environ["PASKIA_CONFIG"], type=RuntimeConfig)

    try:
        await globals.init(
            rp_id=runtime.config.rp_id,
            rp_name=runtime.config.rp_name,
            origins=runtime.config.origins,
            bootstrap=False,
        )
    except ValueError as e:
        logging.error(f"⚠️ {e}")
        # Re-raise to fail fast
        raise

    # Bootstrap and persist config now that the full DB is loaded
    await bootstrap_if_needed(config=runtime.config)
    if runtime.save:
        db.update_config(runtime.config)
    await flush()

    # Restore uvicorn info logging (suppressed during startup in dev mode)
    # Keep uvicorn.error at WARNING to suppress WebSocket "connection open/closed" messages
    if app.debug:
        logging.getLogger("uvicorn").setLevel(logging.INFO)
    logging.getLogger("uvicorn.error").setLevel(logging.WARNING)
    await frontend.load()
    await start_background()
    yield
    await stop_background()
    await authcode.stop()


app = FastAPI(
    lifespan=lifespan,
    redirect_slashes=False,
    docs_url=None,
    redoc_url=None,
    openapi_url=None,
    debug=DEVMODE,
)

# Custom access logging (uvicorn's access_log is disabled)
app.add_middleware(AccessLogMiddleware)

# Apply redirections to auth-host if configured (deny access to restricted endpoints, remove /auth/)
app.middleware("http")(auth_host.redirect_middleware)

app.mount("/auth/api/admin/", admin.app)
app.mount("/auth/api/", api.app)
app.mount("/auth/ws/", ws.app)
app.mount("/auth/oidc/", oid.app)


# OIDC Well-Known endpoints (must be at site root)
@app.get("/.well-known/openid-configuration")
async def openid_configuration(request: Request):
    """OpenID Connect Discovery document."""
    # Build issuer URL from request
    scheme = request.headers.get("x-forwarded-proto", request.url.scheme)
    host = request.headers.get("host", request.url.netloc)
    issuer = f"{scheme}://{host}"

    return {
        "issuer": issuer,
        "authorization_endpoint": f"{issuer}/auth/restricted/oidc",
        "token_endpoint": f"{issuer}/auth/oidc/token",
        "userinfo_endpoint": f"{issuer}/auth/oidc/userinfo",
        "jwks_uri": f"{issuer}/auth/oidc/keys",
        "backchannel_logout_supported": True,
        "backchannel_logout_session_supported": True,
        "response_types_supported": ["code"],
        "grant_types_supported": ["authorization_code", "refresh_token"],
        "subject_types_supported": ["public"],
        "id_token_signing_alg_values_supported": ["EdDSA"],
        "scopes_supported": ["openid", "profile", "email"],
        "token_endpoint_auth_methods_supported": [
            "client_secret_post",
            "client_secret_basic",
        ],
        "code_challenge_methods_supported": ["S256"],
        "claims_supported": [
            "sub",
            "name",
            "preferred_username",
            "email",
            "groups",
            "sid",
        ],
    }


@app.get("/auth/restricted/iframe")
@app.get("/auth/restricted/oidc")
async def restricted_view(request: Request):
    """Serve the restricted/authentication UI for iframe or OpenID Connect."""
    return await vitedev.handle(request, frontend, "/auth/restricted/")


# Navigable URLs are defined here. We support both / and /auth/ as the base path
# / is used on a dedicated auth site, /auth/ on app domains with auth


@app.get("/")
@app.get("/auth/")
async def frontapp(request: Request, response: Response, auth=AUTH_COOKIE):
    """Serve the user profile app.

    The frontend handles mode detection (host mode vs full profile) based on settings.
    Access control is handled via APIs.
    """
    return await vitedev.handle(request, frontend, "/auth/")


@app.get("/admin", include_in_schema=False)
@app.get("/auth/admin", include_in_schema=False)
async def admin_root_redirect():
    return RedirectResponse(f"{hostutil.ui_base_path()}admin/", status_code=307)


@app.get("/admin/", include_in_schema=False)
@app.get("/auth/admin/", include_in_schema=False)
async def admin_root(request: Request, auth=AUTH_COOKIE):
    return await adminapp(request, auth)  # Delegated to admin app


@app.get("/auth/examples/", include_in_schema=False)
async def examples_page():
    """Serve examples/index.html when running from source tree.

    This provides a simple test page for API mode authentication flows
    without depending on the Vue frontend build.
    """
    index_file = _EXAMPLES_DIR / "index.html"
    if not index_file.is_file():
        raise HTTPException(
            status_code=404,
            detail="Examples not available (not running from source tree)",
        )
    return FileResponse(index_file, media_type="text/html")


# Frontend static files - must be before /{token} catch-all routes
frontend.route(app, "/")


# Note: this catch-all handler must be the last route defined
@app.get("/{token}")
@app.get("/auth/{token}")
async def token_link(request: Request, token: str):
    """Serve the reset app for reset tokens (password reset / device addition).

    The frontend will validate the token via /auth/api/token-info.
    """
    if not passphrase.is_well_formed(token):
        raise HTTPException(status_code=404)

    return await vitedev.handle(request, frontend, "/int/reset/")