145 lines
5.0 KiB
Python
145 lines
5.0 KiB
Python
"""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.exit import router as exit_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(exit_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)
|
|
api = create_api(application, host=host)
|
|
server = uvicorn.Server(uvicorn.Config(api, **options))
|
|
api.state.uvicorn_server = server
|
|
server.run()
|
|
|
|
|
|
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
|