nest/uvm/server.py

"""FastAPI application factory and Uvicorn server launcher for uvm."""

from __future__ import annotations

from ipaddress import ip_address
from pathlib import Path
import ssl
from typing import Any

from .app import Application, build_application
from .errors import ConfigurationError, ConflictError, NotFoundError, UvmError, ValidationError


def create_api(application: Application | None = None, *, host: str | None = None) -> Any:
    """Build the HTTP API after validating its intended bind address."""

    if host is None:
        raise UvmError("create_api requires an explicit host. Start the API with uvm --serve.")

    resolved_application = application or build_application(emit=None)
    _validate_server_settings(
        resolved_application.settings.api_token,
        resolved_application.settings.api_tls_cert,
        resolved_application.settings.api_tls_key,
        host,
    )

    try:
        from fastapi import FastAPI
        from fastapi.responses import JSONResponse, RedirectResponse
        from fastapi.staticfiles import StaticFiles
    except ModuleNotFoundError as error:
        raise UvmError(
            "FastAPI server support is not installed. Install this project with its dependencies."
        ) from error

    from .routers.health import router as health_router
    from .routers.installation import router as installation_router
    from .routers.vms import router as vms_router
    from .routers.diagnostics import router as diagnostics_router
    from .routers.dashboard import router as dashboard_router

    api = FastAPI(
        title="uvm",
        version="0.1.0",
        description="Local Firecracker microVM management API.",
    )
    api.state.uvm_application = resolved_application

    @api.exception_handler(UvmError)
    async def handle_uvm_error(_request: Any, error: UvmError) -> Any:
        return JSONResponse(
            status_code=_http_status_for(error),
            content={"error": {"message": str(error)}},
        )

    api.include_router(health_router)
    api.include_router(installation_router)
    api.include_router(vms_router)
    api.include_router(diagnostics_router)

    static_dir = Path(__file__).parent / "dashboard" / "static"
    api.mount("/dashboard/static", StaticFiles(directory=static_dir), name="dashboard_static")
    api.include_router(dashboard_router, prefix="/dashboard")

    @api.get("/")
    async def root() -> Any:
        return RedirectResponse(url="/dashboard")

    return api


def run_server(application: Application, *, host: str, port: int) -> None:
    """Run Uvicorn after enforcing the local-management security boundary."""

    settings = application.settings
    _validate_server_settings(
        settings.api_token,
        settings.api_tls_cert,
        settings.api_tls_key,
        host,
    )
    try:
        import uvicorn
    except ModuleNotFoundError as error:
        raise UvmError(
            "Uvicorn server support is not installed. Install this project with its dependencies."
        ) from error

    options: dict[str, Any] = {"host": host, "port": port}
    if settings.api_tls_cert is not None:
        options["ssl_certfile"] = str(settings.api_tls_cert)
        options["ssl_keyfile"] = str(settings.api_tls_key)
    uvicorn.run(create_api(application, host=host), **options)


def _is_loopback_host(host: str) -> bool:
    try:
        return ip_address(host).is_loopback
    except ValueError:
        return False


def _validate_server_settings(
    api_token: str | None,
    cert: Path | None,
    key: Path | None,
    host: str,
) -> None:
    if not api_token:
        raise UvmError("UVM_API_TOKEN is required before starting the management API")
    if (cert is None) != (key is None):
        raise UvmError("UVM_API_TLS_CERT and UVM_API_TLS_KEY must be configured together")
    if cert is not None:
        assert key is not None
        if not cert.is_file() or not key.is_file():
            raise UvmError("configured API TLS certificate or key does not exist")
        try:
            context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
            context.load_cert_chain(certfile=str(cert), keyfile=str(key))
        except (OSError, ssl.SSLError) as error:
            raise UvmError(f"configured API TLS certificate or key is invalid: {error}") from error
    if not _is_loopback_host(host) and cert is None:
        raise UvmError(
            "refusing to bind the API to a non-loopback host without TLS. "
            "Set UVM_API_TLS_CERT and UVM_API_TLS_KEY or bind behind a TLS reverse proxy."
        )


def _http_status_for(error: UvmError) -> int:
    if isinstance(error, NotFoundError):
        return 404
    if isinstance(error, ConflictError):
        return 409
    if isinstance(error, (ConfigurationError, ValidationError)):
        return 422
    if "needs root" in str(error):
        return 403
    return 500