nest/uvm/config.py

"""Configuration and filesystem layout for the uvm command-line application."""

from __future__ import annotations

import math
import os
from dataclasses import dataclass, field
from ipaddress import IPv4Address, IPv4Network
from pathlib import Path

from .errors import ConfigurationError


# Pin published kernels with built-in virtio-rng support. The artifact series
# is independent of the VMM release; the old ARM quickstart kernel lacks this driver.
DEFAULT_GUEST_ASSETS_URL = "https://s3.amazonaws.com/spec.ccfc.min/firecracker-ci/v1.15"
MIN_STORAGE_MIB = 30 * 1024
MAX_STORAGE_MIB = (2**63 - 1) // (1024 * 1024)


@dataclass(frozen=True, slots=True)
class Settings:
    """All host-specific values consumed by the application."""

    app_name: str = "uvm"
    base: Path = Path("/var/lib/uvm")
    network: IPv4Network = IPv4Network("10.42.0.0/24")
    gateway: IPv4Address = IPv4Address("10.42.0.1")
    bridge: str = "uvm0"
    architecture: str = field(default_factory=lambda: os.uname().machine)
    firecracker_version: str = "v1.16.1"
    default_ram_mib: int = 512
    default_storage_mib: int = MIN_STORAGE_MIB
    default_vcpu: float = 1
    default_ssh_user: str = "root"
    kernel_url: str = ""
    rootfs_url: str = ""  # Empty selects the locally built, package-managed Ubuntu image.
    firecracker_sha256: str | None = None
    firecracker_binary_sha256: str | None = None
    kernel_sha256: str | None = None
    rootfs_sha256: str | None = None
    allow_unverified_downloads: bool = False
    api_token: str | None = None
    api_tls_cert: Path | None = None
    api_tls_key: Path | None = None
    known_hosts_files: tuple[Path, ...] | None = None
    api_timeout_s: float = 2.0
    api_socket_timeout_s: float = 5.0
    terminate_timeout_s: float = 5.0
    tool_workers: int = 2
    tool_ready_timeout_s: float = 120.0
    tool_install_timeout_s: float = 1200.0

    def __post_init__(self) -> None:
        architecture = self.architecture.lower()
        architecture = {"amd64": "x86_64", "arm64": "aarch64"}.get(architecture, architecture)
        if architecture not in ("x86_64", "aarch64"):
            raise ConfigurationError(f"unsupported host architecture: {self.architecture}")
        object.__setattr__(self, "architecture", architecture)

        if type(self.tool_workers) is not int or not 1 <= self.tool_workers <= 8:
            raise ConfigurationError("tool_workers must be an integer between 1 and 8")
        for name in ("tool_ready_timeout_s", "tool_install_timeout_s"):
            value = getattr(self, name)
            if type(value) not in (int, float) or not 0 < value <= 3600 or not math.isfinite(value):
                raise ConfigurationError(f"{name} must be finite, greater than zero, and at most 3600 seconds")

        # Resolve defaults at construction time, while preserving individual URL overrides.
        asset_base = f"{DEFAULT_GUEST_ASSETS_URL}/{architecture}"
        if not self.kernel_url:
            object.__setattr__(self, "kernel_url", f"{asset_base}/vmlinux-6.1.155")

        if self.gateway not in self.network:
            raise ConfigurationError(
                f"gateway {self.gateway} is outside configured network {self.network}"
            )
        if self.gateway in (self.network.network_address, self.network.broadcast_address):
            raise ConfigurationError("gateway must be a usable host address")
        if len(self.bridge) > 15:
            raise ConfigurationError("bridge name must be 15 characters or fewer")
        if self.known_hosts_files is not None and any(not path.is_absolute() for path in self.known_hosts_files):
            raise ConfigurationError("UVM_KNOWN_HOSTS_FILES must contain absolute file paths")
        if self.api_token is not None:
            if not self.api_token or not all(33 <= ord(character) <= 126 for character in self.api_token):
                raise ConfigurationError(
                    "UVM_API_TOKEN must contain only visible ASCII characters without whitespace"
                )

    @property
    def bin_dir(self) -> Path:
        return self.base / "bin"

    @property
    def images_dir(self) -> Path:
        return self.base / "images"

    @property
    def vms_dir(self) -> Path:
        return self.base / "vms"

    @property
    def state_path(self) -> Path:
        return self.base / "state.json"

    @property
    def state_lock_path(self) -> Path:
        return self.base / "state.lock"

    @property
    def operation_lock_path(self) -> Path:
        return self.base / "operations.lock"

    @property
    def firecracker_binary(self) -> Path:
        return self.bin_dir / "firecracker"

    @property
    def jailer_binary(self) -> Path:
        return self.bin_dir / "jailer"

    @property
    def kernel_image(self) -> Path:
        return self.images_dir / "vmlinux"

    @property
    def rootfs_image(self) -> Path:
        return self.images_dir / "ubuntu.ext4"

    @property
    def integrity_manifest_path(self) -> Path:
        return self.base / "integrity.json"

    def vm_dir(self, vm_id: str) -> Path:
        return self.vms_dir / vm_id

    def guest_socket_path(self, vm_id: str) -> Path:
        return self.vm_dir(vm_id) / "vsock.sock"

    @classmethod
    def from_environment(cls) -> "Settings":
        """Load optional deployment overrides while retaining script defaults."""

        try:
            network = IPv4Network(os.environ.get("UVM_NETWORK", "10.42.0.0/24"))
            gateway = IPv4Address(os.environ.get("UVM_GATEWAY", "10.42.0.1"))
        except ValueError as error:
            raise ConfigurationError(f"invalid network configuration: {error}") from error

        return cls(
            base=Path(os.environ.get("UVM_BASE", "/var/lib/uvm")).expanduser(),
            network=network,
            gateway=gateway,
            bridge=os.environ.get("UVM_BRIDGE", "uvm0"),
            kernel_url=os.environ.get("UVM_KERNEL_URL", ""),
            rootfs_url=os.environ.get("UVM_ROOTFS_URL", ""),
            firecracker_sha256=_optional_environment_value("UVM_FIRECRACKER_SHA256"),
            firecracker_binary_sha256=_optional_environment_value(
                "UVM_FIRECRACKER_BINARY_SHA256"
            ),
            kernel_sha256=_optional_environment_value("UVM_KERNEL_SHA256"),
            rootfs_sha256=_optional_environment_value("UVM_ROOTFS_SHA256"),
            allow_unverified_downloads=_boolean_environment_value(
                "UVM_ALLOW_UNVERIFIED_DOWNLOADS", default=False
            ),
            api_token=_optional_environment_value("UVM_API_TOKEN"),
            api_tls_cert=_optional_environment_path("UVM_API_TLS_CERT"),
            api_tls_key=_optional_environment_path("UVM_API_TLS_KEY"),
            known_hosts_files=_known_hosts_environment_paths(),
        )


def _optional_environment_value(name: str) -> str | None:
    value = os.environ.get(name)
    return value if value else None


def _optional_environment_path(name: str) -> Path | None:
    value = _optional_environment_value(name)
    return Path(value).expanduser() if value else None


def _known_hosts_environment_paths() -> tuple[Path, ...] | None:
    value = _optional_environment_value("UVM_KNOWN_HOSTS_FILES")
    if value is None:
        return None
    parts = value.split(os.pathsep)
    if any(not part for part in parts):
        raise ConfigurationError("UVM_KNOWN_HOSTS_FILES must be a colon-separated list of file paths")
    return tuple(Path(part).expanduser() for part in parts)


def _boolean_environment_value(name: str, *, default: bool) -> bool:
    value = os.environ.get(name)
    if value is None:
        return default
    normalized = value.strip().lower()
    if normalized in {"1", "true", "yes", "on"}:
        return True
    if normalized in {"0", "false", "no", "off"}:
        return False
    raise ConfigurationError(f"{name} must be one of true/false, yes/no, or 1/0")