"""Serialized lifecycle and private-port policy changes with durable intent.""" from __future__ import annotations import os import re import shutil import time from collections.abc import Callable, Sequence from dataclasses import dataclass, replace from ipaddress import IPv4Network from pathlib import Path from .config import MIN_STORAGE_MIB, Settings from .domain import PortRule, VmRecord, VmResources, VmSpec, new_vm_id from .errors import ConflictError, NotFoundError, UvmError, ValidationError from .firecracker.api import FirecrackerClient from .firecracker.config import build_config, write_config from .firecracker.process import FirecrackerProcessManager, ProcessInfo from .images import ImageStore from .integrity import load_manifest, verify_file from .known_hosts import KnownHostsManager from .network import NetworkManager from .state import StateStore from .system import check_kvm, ensure_data_directories, require_root from .validation import parse_name, parse_password, parse_ports, parse_timeout, parse_username @dataclass(frozen=True, slots=True) class ListedVm: vm: VmRecord observed_status: str policy_status: str = "unknown" @dataclass(frozen=True, slots=True) class OperationResult: id: str action: str outcome: str error: str | None = None @dataclass(frozen=True, slots=True) class LogTail: content: str truncated: bool _MAX_LOG_TAIL_BYTES = 262_144 class LifecycleService: """All mutating entry points use operation lock -> short state transaction.""" def __init__( self, settings: Settings, state_store: StateStore, images: ImageStore, network: NetworkManager, process: FirecrackerProcessManager, *, known_hosts: KnownHostsManager, client_factory: Callable[[Path, float], FirecrackerClient] = FirecrackerClient, clock: Callable[[], float] = time.time, ) -> None: self._settings = settings self._state_store = state_store self._images = images self._network = network self._process = process self._known_hosts = known_hosts self._client_factory = client_factory self._clock = clock def create(self, spec: VmSpec) -> VmRecord: require_root() VmResources(spec.cpu, spec.ram_mib, spec.storage_mib) parse_username(spec.username) parse_password(spec.password) if spec.name is not None: parse_name(spec.name) ports = self._validated_ports(spec.ports) ensure_data_directories(self._settings) with self._state_store.operation_lock(): self._check_boot() assets = self._images.installed_assets() vm = self._reserve_vm(replace(spec, ports=ports)) try: runtime = self._settings.vm_dir(vm.id) runtime.mkdir(mode=0o700, exist_ok=False) vm.runtime_owned = True self._save(vm) disk = self._images.create_vm_disk(assets.rootfs, Path(vm.disk), storage_mib=spec.storage_mib) self._images.provision_credentials(disk, vm.username, vm.password) vm.disk_ready = True self._save(vm) self._boot(vm, assets.kernel) return vm except BaseException as error: cleanup_error = None try: if vm.operation != "preparing": self._cleanup_runtime(vm) self._remove_runtime(vm) self._remove_record(vm.id) except BaseException as cleanup: cleanup_error = cleanup message = f"failed to create {vm.id}: {error}" if cleanup_error is not None: message += f"; cleanup incomplete: {cleanup_error}" self._record_failure(vm, message) if isinstance(error, (KeyboardInterrupt, SystemExit)): raise raise UvmError(message) from error def _check_boot(self) -> None: if not any(self._settings.network.subnet_of(IPv4Network(network)) for network in ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16")): raise ValidationError("private ports require a network inside an RFC 1918 IPv4 range") check_kvm() self._verify_firecracker() self._network.firewall.check_available() def _verify_firecracker(self) -> None: manifest = load_manifest(self._settings.integrity_manifest_path) checksum = self._settings.firecracker_binary_sha256 if manifest.verified or self._settings.allow_unverified_downloads: checksum = checksum or manifest.checksums.get("firecracker") verify_file(self._settings.firecracker_binary, checksum, "Firecracker binary", "UVM_FIRECRACKER_BINARY_SHA256", allow_unverified=self._settings.allow_unverified_downloads) def _boot(self, vm: VmRecord, kernel: Path) -> None: vm.status, vm.operation = "starting", "network" vm.pid = vm.process_start_time = None vm.spawn_pending = False self._save(vm) self._network.ensure_bridge() self._network.prepare_tap(vm) self._save(vm) self._network.firewall.apply(vm) self._network.activate_tap(vm) config = build_config(self._settings, vm, kernel, Path(vm.disk)) write_config(Path(vm.config), config) # Persist the crash window before spawning, but give start() the already # verified process-free record. A restored pending spawn is never guessed gone. vm.operation = "boot" vm.boot_id = self._process.host_boot_id() self._save(replace(vm, operation="spawning", spawn_pending=True)) def spawned(info: ProcessInfo) -> None: vm.pid, vm.process_start_time, vm.boot_id = info.pid, info.start_time, info.boot_id vm.spawn_pending = False vm.operation = "configuring" self._save(vm) self._process.start(vm, on_spawn=spawned) client = self._client_factory(Path(vm.socket), self._settings.api_timeout_s) client.configure_and_start(config) if self._network.firewall.status(vm) != "enforced": raise UvmError("private-port enforcement could not be confirmed after boot") vm.status, vm.operation, vm.last_error = "running", None, None self._save(vm) def start(self, identifier: str) -> VmRecord: require_root() with self._state_store.operation_lock(): return self._start_locked(self._load_for_operation(identifier)) def _start_locked(self, vm: VmRecord, *, kernel: Path | None = None) -> VmRecord: self._not_deleting(vm) if vm.ports is None: raise ConflictError("legacy VM has no port policy; stop it, use ports set, then start") observation = self._process.observe(vm) if observation == "unknown": raise ConflictError(f"cannot start {vm.id}: process identity is unknown; inspect before retrying") if observation == "alive": if vm.pending_resize is not None: raise ConflictError(f"resize is pending for {vm.id}; retry resize or stop it before starting") if vm.status != "running" or vm.operation not in {None, "policy"}: raise ConflictError(f"{vm.id} has an interrupted boot/stop; stop it before starting") self._validate_retained_network(vm) vm = self._sync_ports_locked(vm) self._network.ensure_bridge() return vm self._validate_resume(vm) if kernel is None: self._check_boot() kernel = self._images.installed_kernel() try: self._cleanup_runtime(vm) self._apply_pending_resize(vm) vm.network, vm.bridge = str(self._settings.network), self._settings.bridge self._boot(vm, kernel) return vm except BaseException as error: message = f"failed to start {vm.id}: {error}" try: self._cleanup_runtime(vm) except BaseException as cleanup: message += f"; cleanup incomplete: {cleanup}" self._record_failure(vm, message) if isinstance(error, (KeyboardInterrupt, SystemExit)): raise raise UvmError(message) from error def stop(self, identifier: str, *, timeout: float | None = None) -> VmRecord: require_root() with self._state_store.operation_lock(): return self._stop_locked(self._load_for_operation(identifier), timeout=timeout) def _stop_locked(self, vm: VmRecord, *, timeout: float | None = None) -> VmRecord: deleting = vm.operation == "delete" or vm.status == "terminating" pending = "delete" if deleting else "resize" if vm.pending_resize is not None else None vm.status, vm.operation = "stopping", pending or "stop" self._save(vm) try: self._cleanup_runtime(vm, timeout=timeout) vm.status = "stopped" vm.operation = pending vm.last_error = None self._save(vm) return vm except BaseException as error: self._record_failure(vm, f"failed to stop {vm.id}: {error}") if isinstance(error, (KeyboardInterrupt, SystemExit)): raise raise UvmError(vm.last_error) from error def restart(self, identifier: str, *, timeout: float | None = None) -> VmRecord: require_root() with self._state_store.operation_lock(): vm = self._load_for_operation(identifier) self._not_deleting(vm) self._stop_locked(vm, timeout=timeout) return self._start_locked(vm) def resize( self, identifier: str, *, cpu: float | None = None, ram_mib: int | None = None, storage_mib: int | None = None, timeout: float | None = None, ) -> VmRecord: """Stop, apply resource changes to the retained disk, then boot under one lock.""" require_root() if cpu is None and ram_mib is None and storage_mib is None: raise ValidationError("resize requires at least one of CPU, RAM, or storage") VmResources(cpu if cpu is not None else 1, ram_mib if ram_mib is not None else 128, storage_mib if storage_mib is not None else MIN_STORAGE_MIB) if timeout is not None: timeout = parse_timeout(str(timeout)) with self._state_store.operation_lock(): vm = self._load_for_operation(identifier) self._not_deleting(vm) self._validate_resume(vm) if self._process.observe(vm) == "unknown": raise ConflictError(f"cannot resize {vm.id}: process identity is unknown") # Verify boot prerequisites before stopping an otherwise healthy VM. self._check_boot() kernel = self._images.installed_kernel() previous = vm.pending_resize capacity = previous.storage_mib if previous is not None else vm.storage_mib if capacity is None: capacity = max(MIN_STORAGE_MIB, (self._images.disk_size_bytes(Path(vm.disk)) + 1024 * 1024 - 1) // (1024 * 1024)) vm.pending_resize = VmResources( cpu=cpu if cpu is not None else previous.cpu if previous is not None else vm.cpu, ram_mib=ram_mib if ram_mib is not None else previous.ram_mib if previous is not None else vm.ram_mib, storage_mib=storage_mib if storage_mib is not None else capacity, ) vm.operation = "resize" self._save(vm) try: self._stop_locked(vm, timeout=timeout) return self._start_locked(vm, kernel=kernel) except UvmError as error: raise UvmError(f"failed to resize {vm.id}: {error}") from error def _validate_resume(self, vm: VmRecord) -> None: if vm.ports is None: raise ConflictError("legacy VM has no port policy; stop it, use ports set, then start") if not vm.disk_ready: raise ConflictError(f"{vm.id} has no completely provisioned private disk") disk = Path(vm.disk) if not vm.disk or not disk.is_file() or disk.is_symlink() or disk.stat().st_nlink != 1: raise UvmError(f"private VM disk is missing or unsafe: {disk}") self._validate_retained_network(vm) def _validate_retained_network(self, vm: VmRecord) -> None: if (vm.gateway != str(self._settings.gateway) or vm.network not in {None, str(self._settings.network)} or vm.bridge not in {None, self._settings.bridge}): raise ConflictError("configured network differs from the VM's retained network") def _apply_pending_resize(self, vm: VmRecord) -> None: target = vm.pending_resize if target is None: return if self._process.observe(vm) != "gone": raise ConflictError("cannot resize a disk without confirmed process exit") # A prior attempt may have published a resized image without saving # its allocation. Check the disk too, including when a retry changes # the target back to the previously recorded capacity. if (target.storage_mib != vm.storage_mib or self._images.disk_size_bytes(Path(vm.disk)) != target.storage_mib * 1024 * 1024): self._images.resize_vm_disk(Path(vm.disk), target.storage_mib) vm.cpu, vm.ram_mib, vm.storage_mib = target.cpu, target.ram_mib, target.storage_mib vm.pending_resize = None self._save(vm) def destroy(self, identifier: str, *, timeout: float | None = None) -> VmRecord: require_root() with self._state_store.operation_lock(): return self._delete_locked(self._load_for_operation(identifier), timeout=timeout) def _delete_locked(self, vm: VmRecord, *, timeout: float | None = None) -> VmRecord: vm.status, vm.operation = "terminating", "delete" self._save(vm) try: self._cleanup_runtime(vm, timeout=timeout) self._remove_runtime(vm) self._known_hosts.remove(vm) self._remove_record(vm.id) return vm except BaseException as error: self._record_failure(vm, f"failed to delete {vm.id}: {error}") if isinstance(error, (KeyboardInterrupt, SystemExit)): raise raise UvmError(vm.last_error) from error def _cleanup_runtime(self, vm: VmRecord, *, timeout: float | None = None) -> None: containment_errors = [] try: if self._network.has_tap(vm): try: self._network.firewall.apply(vm, quarantine=True) finally: self._network.deactivate_tap(vm) except UvmError as error: containment_errors.append(str(error)) try: self._process.terminate(vm, timeout=timeout) except UvmError as error: detail = f"; containment: {'; '.join(containment_errors)}" if containment_errors else "" raise UvmError(f"process exit unconfirmed: {error}{detail}") from error vm.pid = vm.process_start_time = None vm.spawn_pending = False if vm.operation == "spawning": vm.operation = "boot" self._network.delete_vm_tap(vm) self._network.firewall.cleanup_connections(vm) self._network.firewall.remove(vm) def set_ports(self, identifier: str, ports: tuple[PortRule, ...], *, mode: str = "set") -> VmRecord: """Mutate a frozen registry ID; the CLI resolves names before confirmation.""" require_root() ports = self._validated_ports(ports) if mode not in {"open", "close", "set"}: raise ValidationError(f"invalid port operation: {mode}") with self._state_store.operation_lock(): vm = self._load_for_operation(identifier, exact_id=True) self._not_deleting(vm) observation = self._process.observe(vm) if observation == "unknown": raise ConflictError("port changes require a known process identity") if vm.ports is None and (mode != "set" or observation != "gone" or vm.status != "stopped"): raise ConflictError("legacy VM policy must be explicitly set while stopped") if vm.operation not in {None, "policy"}: raise ConflictError(f"finish the pending {vm.operation} operation before editing ports") current = {port.port: port for port in vm.ports or ()} if mode == "set": current = {port.port: port for port in ports} elif mode == "open": for port in ports: current.setdefault(port.port, port) else: for port in ports: current.pop(port.port, None) vm.ports = tuple(current[number] for number in sorted(current)) vm.operation = "policy" self._save(vm) return self._sync_ports_locked(vm) def sync_ports(self, identifier: str) -> VmRecord: require_root() with self._state_store.operation_lock(): return self._sync_ports_locked(self._load_for_operation(identifier)) def _sync_ports_locked(self, vm: VmRecord) -> VmRecord: self._not_deleting(vm) if vm.ports is None: raise ConflictError("legacy VM has no managed policy; stop it and use ports set") if vm.operation not in {None, "policy"}: raise ConflictError(f"finish the pending {vm.operation} operation before syncing ports") observation = self._process.observe(vm) if observation == "unknown": raise ConflictError("cannot enforce policy for a VM with unknown process identity") vm.operation = "policy" self._save(vm) try: if observation == "alive": if not self._network.has_tap(vm): raise UvmError("running VM has no owned network attachment") self._network.firewall.apply(vm) self._network.activate_tap(vm) vm.status = "running" else: if vm.status != "stopped" or vm.tap_owned: self._cleanup_runtime(vm) vm.status = "stopped" vm.operation, vm.last_error = None, None self._save(vm) return vm except BaseException as error: # Never restore an older, more permissive policy after a failed close. containment_errors = [] for contain in ( lambda: self._network.deactivate_tap(vm), lambda: (self._network.firewall.apply(vm, quarantine=True) if self._network.has_tap(vm) else None), ): try: contain() except UvmError as containment_error: containment_errors.append(str(containment_error)) detail = f"; containment errors: {'; '.join(containment_errors)}" if containment_errors else "" self._record_failure(vm, f"policy pending for {vm.id}: {error}{detail}") if isinstance(error, (KeyboardInterrupt, SystemExit)): raise raise UvmError(vm.last_error) from error def resolve_targets(self, identifiers: Sequence[str], *, all_vms: bool = False) -> list[VmRecord]: if all_vms and identifiers: raise ValidationError("--all cannot be combined with explicit VM targets") state = self._state_store.load() if all_vms: return list(state.vms.values()) if not identifiers: raise ValidationError("specify a VM or --all") resolved = [self._state_store.resolve(state, identifier) for identifier in identifiers] return list({vm.id: vm for vm in resolved}.values()) def run_many(self, action: str, vm_ids: Sequence[str], *, timeout: float | None = None) -> list[OperationResult]: require_root() if action not in {"start", "stop", "restart", "delete", "ports-sync"}: raise ValidationError(f"invalid lifecycle action: {action}") results = [] with self._state_store.operation_lock(): for index, vm_id in enumerate(vm_ids): try: vm = self._load_for_operation(vm_id, exact_id=True) if action == "start": self._start_locked(vm) elif action == "stop": self._stop_locked(vm, timeout=timeout) elif action == "delete": self._delete_locked(vm, timeout=timeout) elif action == "ports-sync": self._sync_ports_locked(vm) else: self._not_deleting(vm) self._stop_locked(vm, timeout=timeout) self._start_locked(vm) results.append(OperationResult(vm_id, action, "succeeded")) except NotFoundError: results.append(OperationResult(vm_id, action, "skipped", "already absent")) except KeyboardInterrupt: results.append(OperationResult(vm_id, action, "interrupted", "interrupted")) results.extend(OperationResult(remaining, action, "skipped", "not attempted after interruption") for remaining in vm_ids[index + 1:]) break except (UvmError, OSError) as error: results.append(OperationResult(vm_id, action, "failed", str(error))) return results def list_vms(self) -> list[ListedVm]: return [self._listed(vm) for vm in self._state_store.load().vms.values()] def inspect(self, identifier: str) -> ListedVm: return self._listed(self._state_store.find_by_id_or_ip(identifier)) def _listed(self, vm: VmRecord) -> ListedVm: observation = self._process.observe(vm) status = vm.status if observation == "unknown": status = "unknown" elif observation == "alive" and status == "stopped": status = "running" elif observation == "gone" and status == "running": status = "dead" policy = ("inactive" if observation == "gone" and vm.status == "stopped" and vm.operation is None else self._network.firewall.status(vm)) if vm.ports is None and policy == "inactive": policy = "unmanaged" return ListedVm(vm, status, policy) def find_for_ssh(self, identifier: str) -> VmRecord: return self._state_store.find_by_id_or_ip(identifier) def log_path(self, identifier: str) -> Path: vm = self._state_store.find_by_id_or_ip(identifier) self._validate_paths(vm) return Path(vm.log) def read_log_tail(self, identifier: str, *, tail: int) -> LogTail: """Read a bounded tail from a validated VM log without following it.""" if type(tail) is not int or tail < 0: raise ValidationError("log tail must be a non-negative integer") path = self.log_path(identifier) try: with path.open("rb") as stream: end = stream.seek(0, os.SEEK_END) if tail == 0: return LogTail(content="", truncated=False) start = max(0, end - _MAX_LOG_TAIL_BYTES) previous = b"" if start: stream.seek(start - 1) previous = stream.read(1) stream.seek(start) content = stream.read(end - start) except OSError as error: raise UvmError("could not read VM log") from error truncated = False if start and previous != b"\n": newline = content.find(b"\n") if newline == -1: truncated = True else: content = content[newline + 1:] lines = content.splitlines(keepends=True) if start and len(lines) < tail: truncated = True return LogTail( content=b"".join(lines[-tail:]).decode("utf-8", errors="replace"), truncated=truncated, ) def diagnostics(self) -> list[dict]: checks = [("root", require_root), ("kvm", check_kvm), ("Firecracker", self._verify_firecracker), ("guest assets", self._images.installed_assets), ("private firewall", self._network.firewall.check_available)] results = [] for name, check in checks: try: check() results.append({"name": name, "ok": True, "message": "available"}) except (UvmError, OSError) as error: results.append({"name": name, "ok": False, "message": str(error)}) for command in ("ip", "iptables", "nft", "conntrack", "debugfs", "e2fsck", "resize2fs", "dumpe2fs", "cp", "openssl", "ssh-keygen", "ssh"): path = shutil.which(command) results.append({"name": command, "ok": path is not None, "message": path or "not installed"}) return results def _load_for_operation(self, identifier: str, *, exact_id: bool = False) -> VmRecord: vm = (self._state_store.get(identifier) if exact_id else self._state_store.find_by_id_or_ip(identifier)) self._validate_paths(vm) if self._process.adopt_legacy_identity(vm): self._save(vm) return vm def _validate_paths(self, vm: VmRecord) -> None: if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,80}", vm.id) is None: raise ConflictError("invalid persisted VM ID; refusing resource access") directory = self._settings.vm_dir(vm.id).absolute() if directory.is_symlink(): raise ConflictError("VM directory must not be a symbolic link") for field, filename in (("socket", "firecracker.sock"), ("config", "config.json"), ("log", "firecracker.log"), ("disk", "rootfs.ext4")): path = getattr(vm, field) if field == "disk" and not path: continue # Legacy records may have used a shared template, never delete it. if Path(path).absolute() != directory / filename or Path(path).is_symlink(): raise ConflictError(f"unsafe persisted VM {field} path") @staticmethod def _not_deleting(vm: VmRecord) -> None: if vm.operation == "delete" or vm.status == "terminating": raise ConflictError(f"deletion is pending for {vm.id}; retry delete instead") @staticmethod def _validated_ports(ports: tuple[PortRule, ...]) -> tuple[PortRule, ...]: if not isinstance(ports, tuple) or any(not isinstance(port, PortRule) for port in ports): raise ValidationError("ports must be a tuple of private TCP port rules") return parse_ports([f"{port.port}:{port.protocol}" for port in ports]) def _reserve_vm(self, spec: VmSpec) -> VmRecord: with self._state_store.transaction() as state: for existing in state.vms.values(): if (existing.guest_ip == str(self._settings.gateway) or existing.gateway != str(self._settings.gateway) or existing.network not in {None, str(self._settings.network)} or existing.bridge not in {None, self._settings.bridge}): raise ConflictError("configured topology conflicts with retained VMs; keep their network settings") if spec.name is not None and any(spec.name in {vm.name, vm.id} for vm in state.vms.values()): raise ConflictError(f"VM name already exists: {spec.name}") vm_id = new_vm_id() used_taps = {vm.tap for vm in state.vms.values()} while vm_id in state.vms or any(vm.name == vm_id for vm in state.vms.values()) or self._network.tap_name(vm_id) in used_taps: vm_id = new_vm_id() guest_ip = self._network.allocate_ip(state.vms.values(), spec.guest_ip) mac = self._network.mac_for(state.next_mac_index) state.next_mac_index += 1 directory = self._settings.vm_dir(vm_id) vm = VmRecord( id=vm_id, name=spec.name, cpu=spec.cpu, ram_mib=spec.ram_mib, storage_mib=spec.storage_mib, guest_ip=str(guest_ip), gateway=str(self._settings.gateway), network=str(self._settings.network), bridge=self._settings.bridge, tap=self._network.tap_name(vm_id), mac=mac, socket=str(directory / "firecracker.sock"), config=str(directory / "config.json"), log=str(directory / "firecracker.log"), disk=str(directory / "rootfs.ext4"), username=spec.username, password=spec.password, ports=spec.ports, status="starting", operation="preparing", created_at=int(self._clock()), updated_at=int(self._clock()), ) state.vms[vm.id] = vm return vm def _remove_runtime(self, vm: VmRecord) -> None: self._validate_paths(vm) directory = self._settings.vm_dir(vm.id) if directory.exists(): if not vm.runtime_owned: raise ConflictError(f"refusing to remove unowned VM directory {directory}") if self._process.observe(vm) != "gone": raise ConflictError("refusing to remove disk without confirmed process exit") shutil.rmtree(directory) vm.runtime_owned = False def _save(self, vm: VmRecord) -> None: vm.updated_at = int(self._clock()) with self._state_store.transaction() as state: if vm.id not in state.vms: raise NotFoundError(f"VM not found: {vm.id}") state.vms[vm.id] = vm def _record_failure(self, vm: VmRecord, message: str) -> None: vm.status, vm.last_error = "failed", message self._save(vm) def _remove_record(self, vm_id: str) -> None: with self._state_store.transaction() as state: state.vms.pop(vm_id, None)