diff --git a/tests/test_tool_catalog.py b/tests/test_tool_catalog.py new file mode 100644 index 0000000..715647d --- /dev/null +++ b/tests/test_tool_catalog.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import hashlib +import io +import os +import shlex +import tarfile +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from uvm import guest_tools +from uvm.errors import ValidationError +from uvm.tool_catalog import CATALOG, ToolSelection, all_selections, normalize_tools, recipe +from uvm.tools import guest_command + + +class ToolCatalogTests(unittest.TestCase): + def test_requested_catalog_and_every_guest_payload_fit_the_agent_protocol(self): + self.assertEqual({tool["id"] for tool in CATALOG}, + {"git", "nodejs", "python", "jdk", "wget", "curl", "htop", "php", "conda", "micromamba"}) + for arch in ("x86_64", "aarch64"): + recipes = [recipe(item, arch) for item in all_selections()] + command = guest_command({"action": "probe", "tools": recipes, "username": "worker", "architecture": arch}) + self.assertLess(len(command.encode()), 65536) + argv = shlex.split(command) + self.assertEqual(argv[:3], ["/usr/bin/python3", "-I", "-c"]) + compile(argv[3], "guest-tools", "exec") + for item in recipes: + if "url" in item: + self.assertTrue(item["url"].startswith("https://")) + self.assertNotIn("/latest", item["url"]) + self.assertRegex(item["sha256"], r"^[0-9a-f]{64}$") + self.assertEqual(len(all_selections()), 12) + + def test_selection_validation_is_allowlisted_strict_and_deduplicated(self): + expected = (ToolSelection("git"), ToolSelection("python", "3.12")) + self.assertEqual(normalize_tools([{"id": "python", "version": "3.12"}, {"id": "git"}, {"id": "git"}]), expected) + for value in (None, "git", [], ["git"], [{"id": "git; id"}], [{"id": "python"}], + [{"id": "python", "version": "3.14"}], [{"id": "git", "version": "latest"}], + [{"id": "git", "command": "id"}], [{"id": ["git"]}], [{"id": "git"}] * 13): + with self.subTest(value=value), self.assertRaises(ValidationError): + normalize_tools(value) + self.assertEqual(normalize_tools([], allow_empty=True), ()) + + def test_versioned_python_commands_and_jdk_compiler_are_verified(self): + for version in ("3.11", "3.12", "3.13"): + item = recipe(ToolSelection("python", version), "aarch64") + self.assertEqual(set(item["links"]), {"python" + version}) + self.assertIn("venv", item["checks"][0][-1]) + self.assertEqual(item["checks"][1][:3], ["python" + version, "-m", "pip"]) + jdk = recipe(ToolSelection("jdk"), "aarch64") + self.assertEqual(jdk["checks"], [["javac", "-version"], ["java", "-version"]]) + self.assertIn("arm64", jdk["java_home"]) + + +class GuestToolRecipeTests(unittest.TestCase): + def setUp(self): + self.temporary = self.enterContext(tempfile.TemporaryDirectory()) + self.root = Path(self.temporary) + + def test_download_mismatch_removes_artifact_and_https_redirect_is_checked(self): + target = self.root / "download" + response = io.BytesIO(b"publisher payload") + response.geturl = lambda: "https://publisher.test/tool" + with patch.object(guest_tools.urllib.request, "urlopen", return_value=response): + with self.assertRaisesRegex(guest_tools.ToolError, "checksum"): + guest_tools.download("https://publisher.test/tool", target, "0" * 64) + self.assertFalse(target.exists()) + response = io.BytesIO(b"publisher payload") + response.geturl = lambda: "https://publisher.test/tool" + with patch.object(guest_tools.urllib.request, "urlopen", return_value=response): + guest_tools.download("https://publisher.test/tool", target, hashlib.sha256(b"publisher payload").hexdigest()) + self.assertEqual(target.read_bytes(), b"publisher payload") + with self.assertRaisesRegex(guest_tools.ToolError, "HTTPS"): + guest_tools.download("http://publisher.test/tool", target) + response = io.BytesIO(b"publisher payload") + response.geturl = lambda: "http://publisher.test/tool" + with patch.object(guest_tools.urllib.request, "urlopen", return_value=response): + with self.assertRaisesRegex(guest_tools.ToolError, "HTTPS"): + guest_tools.download("https://publisher.test/tool", target) + + def test_archive_traversal_links_and_devices_cannot_escape_staging(self): + cases = [("../outside", tarfile.REGTYPE, ""), ("/outside", tarfile.REGTYPE, ""), + ("link", tarfile.SYMTYPE, "../../outside"), ("link", tarfile.LNKTYPE, "../outside"), + ("device", tarfile.CHRTYPE, "")] + for name, kind, link in cases: + with self.subTest(name=name, kind=kind): + archive = self.root / "archive.tar" + with tarfile.open(archive, "w") as stream: + item = tarfile.TarInfo(name) + item.type = kind + item.linkname = link + stream.addfile(item) + output = self.root / "output" + output.mkdir(exist_ok=True) + with self.assertRaises(guest_tools.ToolError): + guest_tools.extract_archive(archive, output) + self.assertFalse((self.root / "outside").exists()) + + def test_internal_archive_symlinks_are_supported(self): + archive = self.root / "archive.tar" + with tarfile.open(archive, "w") as stream: + item = tarfile.TarInfo("python/bin/python3.12") + item.size = 4 + item.mode = 0o755 + stream.addfile(item, io.BytesIO(b"test")) + link = tarfile.TarInfo("python/bin/python3") + link.type = tarfile.SYMTYPE + link.linkname = "python3.12" + stream.addfile(link) + output = self.root / "output" + output.mkdir() + guest_tools.extract_archive(archive, output) + self.assertEqual((output / "python/bin/python3").read_bytes(), b"test") + + def test_probe_uses_guest_account_and_rejects_wrong_runtime_or_missing_companion(self): + node = recipe(ToolSelection("nodejs"), "x86_64") + account = SimpleNamespace(pw_dir="/home/worker", pw_uid=1001, pw_gid=1002) + with (patch.object(guest_tools.pwd, "getpwnam", return_value=account), + patch.object(guest_tools.os, "getgrouplist", return_value=[1002, 1003]), + patch.object(guest_tools.shutil, "which", return_value="/usr/local/bin/node"), + patch.object(guest_tools.subprocess, "run") as run): + run.return_value = SimpleNamespace(returncode=0, stdout=b"v22.10.0\n") + result = guest_tools.probe(node, "worker") + self.assertFalse(result["installed"]) + self.assertEqual(result["detected_version"], "22.10.0") + self.assertEqual(run.call_args.kwargs["user"], 1001) + self.assertEqual(run.call_args.kwargs["env"]["HOME"], "/home/worker") + run.side_effect = [SimpleNamespace(returncode=0, stdout=b"v24.21.0\n"), + SimpleNamespace(returncode=1, stdout=b"broken npm")] + self.assertFalse(guest_tools.probe(node, "worker")["installed"]) + + def test_managed_files_preserve_edits_and_can_finish_interrupted_publication(self): + installer = guest_tools.Installer("root") + path = self.root / "profile" + # Isolated directory fixtures stand in for the root-owned guest hierarchy. + with (patch.object(guest_tools, "ROOT", self.root), + patch.object(guest_tools, "safe_directory"), + patch.object(guest_tools, "regular_file")): + installer.write_managed(path, b"initial") + installer.write_managed(path, b"updated") + self.assertEqual(path.read_bytes(), b"updated") + path.write_bytes(b"guest edit") + with self.assertRaisesRegex(guest_tools.ToolError, "customized"): + installer.write_managed(path, b"next") + self.assertEqual(path.read_bytes(), b"guest edit") + path.write_bytes(b"updated") + installer.remember(path, hashlib.sha256(b"next").hexdigest()) + installer.write_managed(path, b"next") + self.assertEqual(path.read_bytes(), b"next") + + def test_log_capture_retains_a_bounded_tail(self): + installer = guest_tools.Installer("root") + installer.append(b"a" * 20000) + installer.append(b"last failure") + self.assertEqual(len(installer.log), guest_tools.LOG_LIMIT) + self.assertTrue(installer.truncated) + self.assertTrue(installer.log.endswith(b"last failure")) diff --git a/tests/test_tool_jobs.py b/tests/test_tool_jobs.py new file mode 100644 index 0000000..e1a7b23 --- /dev/null +++ b/tests/test_tool_jobs.py @@ -0,0 +1,151 @@ +from __future__ import annotations + +import json +import os +import tempfile +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace +from pathlib import Path +from unittest.mock import Mock, patch + +from tests.test_server import FakeLifecycle +from uvm.config import Settings +from uvm.errors import ConflictError, GuestExecutionError, NotFoundError, StateError +from uvm.state import StateStore +from uvm.tool_catalog import ToolSelection +from uvm.tool_store import ToolJobStore +from uvm.tools import ToolService + + +class ToolJobTests(unittest.TestCase): + def setUp(self): + temporary = self.enterContext(tempfile.TemporaryDirectory()) + self.settings = replace(Settings(), base=Path(temporary), tool_workers=1) + self.state = StateStore(self.settings) + self.vm = replace(FakeLifecycle().vm, runtime_owned=True, pid=123, process_start_time="1", boot_id="host-boot") + self.settings.vm_dir(self.vm.id).mkdir(parents=True, mode=0o700) + with self.state.transaction() as state: + state.vms[self.vm.id] = self.vm + self.store = ToolJobStore(self.settings, self.state) + + def enqueue(self, selections=None, **options): + with self.state.operation_lock(): + return self.store.enqueue_locked(self.vm.id, selections or [ToolSelection("git")], **options) + + def test_duplicate_concurrent_submissions_have_one_durable_job(self): + with ThreadPoolExecutor(max_workers=4) as executor: + jobs = list(executor.map(lambda _: self.enqueue(request_id="request-1"), range(4))) + self.assertEqual(len({job["id"] for job in jobs}), 1) + self.assertEqual(len(self.store.read(self.vm.id)["jobs"]), 1) + path = self.settings.vm_dir(self.vm.id) / "tools.json" + self.assertEqual(path.stat().st_mode & 0o777, 0o600) + self.assertNotIn("guest-secret", path.read_text()) + with self.assertRaises(ConflictError): + self.enqueue([ToolSelection("htop")]) + + def test_history_is_vm_scoped_and_deletion_is_never_undone(self): + job = self.enqueue() + with self.assertRaises(NotFoundError): + self.store.get_job(self.vm.name, job["id"]) + with self.state.transaction() as state: + del state.vms[self.vm.id] + with self.assertRaises(NotFoundError): + self.store.update_job(self.vm.id, job["id"], lambda current, data: current.update(status="succeeded")) + self.assertNotIn(self.vm.id, self.state.load().vms) + + def test_recovery_preserves_queued_work_and_marks_inflight_outcomes_unknown(self): + job = self.enqueue() + self.store.recover() + self.assertEqual(self.store.get_job(self.vm.id, job["id"])["status"], "queued") + self.store.claim() + self.store.update_job(self.vm.id, job["id"], lambda current, data: current["items"][0].update(status="installing")) + self.store.recover() + recovered = self.store.get_job(self.vm.id, job["id"]) + self.assertEqual(recovered["status"], "interrupted") + self.assertTrue(recovered["outcome_unknown"]) + self.assertEqual(recovered["items"][0]["status"], "interrupted") + self.assertIsNone(self.store.claim()) + + def test_symlinks_and_foreign_history_are_rejected(self): + outside = self.settings.base / "outside" + outside.write_text("private data") + path = self.settings.vm_dir(self.vm.id) / "tools.json" + path.symlink_to(outside) + with self.assertRaises((StateError, OSError)): + self.store.read(self.vm.id) + self.assertEqual(outside.read_text(), "private data") + path.unlink() + self.enqueue() + data = json.loads(path.read_text()) + data["jobs"][0]["vm_id"] = "vm-other" + path.write_text(json.dumps(data)) + with self.assertRaises(StateError): + self.store.read(self.vm.id) + + def test_only_one_worker_can_own_this_registry_and_shutdown_releases_it(self): + first = ToolService(self.settings, self.state, Mock()) + second = ToolService(self.settings, self.state, Mock()) + self.addCleanup(first.shutdown) + self.addCleanup(second.shutdown) + first.start() + with self.assertRaises(ConflictError): + second.start() + first.shutdown() + second.start() + + def make_service(self): + lifecycle = Mock() + lifecycle.prepare_tool_install.return_value = self.vm + service = ToolService(self.settings, self.state, lifecycle) + return service + + @staticmethod + def observation(selection, installed=False): + return {**selection.to_dict(), "key": selection.key, "installed": installed, + "detected_version": "2.0.0" if installed else None, "usage": "tool --version", "error": None} + + def test_worker_reports_partial_failure_and_retains_successful_tools(self): + selections = [ToolSelection("git"), ToolSelection("htop")] + job = self.enqueue(selections) + claimed = self.store.claim() + service = self.make_service() + results = [ + {"ok": True, "busy": False, "boot_id": "guest-boot", "tools": [self.observation(item) for item in selections]}, + {"ok": True, "status": "installed", "tool": self.observation(selections[0], True), "output": "git installed"}, + {"ok": False, "error": "mirror unavailable", "output": "apt download failed"}, + ] + with patch.object(service, "_execute", side_effect=results) as execute: + service._run_job(claimed) + result = self.store.get_job(self.vm.id, job["id"]) + self.assertEqual(result["status"], "failed") + self.assertEqual([item["status"] for item in result["items"]], ["installed", "failed"]) + self.assertEqual(result["items"][1]["output"], "apt download failed") + self.assertEqual(self.state.get(self.vm.id).status, "running") + self.assertEqual(execute.call_args.args[1]["boot_id"], "guest-boot") + + def test_lost_result_is_never_replayed_and_remaining_tools_are_skipped(self): + selections = [ToolSelection("git"), ToolSelection("htop")] + job = self.enqueue(selections) + service = self.make_service() + with patch.object(service, "_execute", side_effect=[ + {"ok": True, "busy": False, "boot_id": "boot", "tools": [self.observation(item) for item in selections]}, + GuestExecutionError("connection lost", outcome_unknown=True), + ]) as execute: + service._run_job(self.store.claim()) + result = self.store.get_job(self.vm.id, job["id"]) + self.assertEqual(execute.call_count, 2) + self.assertEqual(result["status"], "interrupted") + self.assertTrue(result["outcome_unknown"]) + self.assertEqual([item["status"] for item in result["items"]], ["interrupted", "skipped"]) + + def test_prior_guest_installation_blocks_a_retry_before_submission(self): + job = self.enqueue() + service = self.make_service() + with patch.object(service, "_execute", return_value={"ok": True, "busy": True}) as execute: + service._run_job(self.store.claim()) + self.assertEqual(execute.call_count, 1) + result = self.store.get_job(self.vm.id, job["id"]) + self.assertEqual(result["status"], "interrupted") + self.assertFalse(result["outcome_unknown"]) diff --git a/tests/test_tool_lifecycle.py b/tests/test_tool_lifecycle.py new file mode 100644 index 0000000..076a3ed --- /dev/null +++ b/tests/test_tool_lifecycle.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +from unittest.mock import Mock + +from tests.test_lifecycle import LifecycleTestCase +from uvm.errors import ConflictError, NotFoundError, UvmError, ValidationError +from uvm.tool_catalog import ToolSelection + + +class ToolLifecycleTests(LifecycleTestCase): + def test_create_persists_tool_intent_before_boot(self): + def configure(config): + vm_id = next(iter(self.store.load().vms)) + jobs = self.service._tool_store._read(vm_id)["jobs"] + self.assertEqual(jobs[0]["selections"], [{"id": "python", "version": "3.12"}]) + self.assertEqual(jobs[0]["status"], "queued") + original = self.service._client_factory + def client(socket, timeout): + result = original(socket, timeout) + result.configure_and_start = configure + return result + self.service._client_factory = client + vm = self.create_vm(tools=(ToolSelection("python", "3.12"),)) + self.assertEqual(vm.status, "running") + self.assertEqual(self.service._tool_store.read(vm.id)["jobs"][0]["source"], "create") + + def test_invalid_tools_fail_before_reserving_a_vm_and_failed_boot_removes_intent(self): + with self.assertRaises(ValidationError): + self.create_vm(tools=({"id": "invalid"},)) + self.assertEqual(self.store.load().vms, {}) + self.host.failures["client.configure", None] = UvmError("failed boot") + with self.assertRaises(UvmError): + self.create_vm(tools=(ToolSelection("git"),)) + self.assertEqual(list(self.settings.vms_dir.iterdir()), []) + + def test_stopped_installation_starts_only_when_requested_and_aliases_are_rejected(self): + vm = self.create_vm(name="worker") + self.service.stop(vm.id) + for alias in (vm.name, vm.guest_ip): + with self.assertRaises(NotFoundError): + self.service.queue_tools(alias, [ToolSelection("git")], start_if_stopped=True) + with self.assertRaises(ConflictError): + self.service.queue_tools(vm.id, [ToolSelection("git")]) + job = self.service.queue_tools(vm.id, [ToolSelection("git")], start_if_stopped=True) + self.assertEqual(self.store.get(vm.id).status, "stopped") + ready = self.service.prepare_tool_install(vm.id, start_if_stopped=job["start_if_stopped"]) + self.assertEqual(ready.status, "running") + self.assertEqual(ready.guest_ip, vm.guest_ip) + + def test_a_tool_step_cannot_reconnect_to_a_restarted_vm(self): + vm = self.create_vm() + identity = vm.pid, vm.process_start_time, vm.boot_id + self.service.restart(vm.id) + self.service._guest_connector = Mock() + with self.assertRaisesRegex(ConflictError, "restarted"): + self.service.execute(vm.id, "id", expected_identity=identity) + self.service._guest_connector.assert_not_called() diff --git a/uvm/app.py b/uvm/app.py index ce9ec9b..6676190 100644 --- a/uvm/app.py +++ b/uvm/app.py @@ -14,6 +14,7 @@ from .lifecycle import LifecycleService from .network import NetworkManager from .state import StateStore from .system import CommandRunner +from .tools import ToolService @dataclass(slots=True) @@ -24,6 +25,7 @@ class Application: state_store: StateStore installer: Installer lifecycle: LifecycleService + tools: ToolService | None = None def build_application(emit: Callable[[str], None] | None = print) -> Application: @@ -41,4 +43,5 @@ def build_application(emit: Callable[[str], None] | None = print) -> Application state_store=state_store, installer=installer, lifecycle=lifecycle, + tools=ToolService(settings, state_store, lifecycle), ) diff --git a/uvm/config.py b/uvm/config.py index 1d37543..261b9d3 100644 --- a/uvm/config.py +++ b/uvm/config.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +import math from dataclasses import dataclass, field from ipaddress import IPv4Address, IPv4Network from pathlib import Path @@ -46,6 +47,9 @@ class Settings: 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() @@ -54,6 +58,13 @@ class Settings: 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 math.isfinite(value) or not 0 < value <= 3600: + 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: diff --git a/uvm/domain.py b/uvm/domain.py index 6906ef9..8ab8f8a 100644 --- a/uvm/domain.py +++ b/uvm/domain.py @@ -11,6 +11,7 @@ from typing import Any from .config import MAX_STORAGE_MIB, MIN_STORAGE_MIB from .errors import StateError, ValidationError +from .tool_catalog import ToolSelection STATE_SCHEMA_VERSION = 3 @@ -55,6 +56,7 @@ class VmSpec: name: str | None = None ports: tuple[PortRule, ...] = (PortRule(22),) storage_mib: int = MIN_STORAGE_MIB + tools: tuple[ToolSelection, ...] = () @dataclass(frozen=True, slots=True) diff --git a/uvm/guest_exec.py b/uvm/guest_exec.py index af0dff5..0d3abd8 100644 --- a/uvm/guest_exec.py +++ b/uvm/guest_exec.py @@ -111,4 +111,10 @@ class GuestSession: ) from error def close(self) -> None: + # Wake a thread waiting for a long-running tool command on server shutdown. + # The guest still owns its bounded command and package-manager cleanup. + try: + self._connection.shutdown(socket.SHUT_RDWR) + except OSError: + pass self._connection.close() diff --git a/uvm/guest_tools.py b/uvm/guest_tools.py new file mode 100644 index 0000000..0a9ad6c --- /dev/null +++ b/uvm/guest_tools.py @@ -0,0 +1,394 @@ +"""Standalone guest-only tool installer, submitted through the execution agent. + +Importing this module on the host has no side effects. It uses only Python 3.10 +stdlib; all package commands, downloads, and filesystem changes run in the guest. +""" + +from __future__ import annotations + +import fcntl +import hashlib +import json +import os +import pwd +import re +import shlex +import shutil +import stat +import subprocess +import sys +import tarfile +import tempfile +import time +import urllib.request +from pathlib import Path + + +ROOT = Path("/opt/uvm-tools") +LOCK = Path("/run/uvm-tools.lock") +BIN_DIR = Path("/usr/local/bin") +PROFILE_DIR = Path("/etc/profile.d") +PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" +LOG_LIMIT = 16384 +PHP_KEY = "B8DC7E53946656EFBCE4C1DD71DAEAAB4AD4CAB6" + + +class ToolError(Exception): + pass + + +def atomic_write(path, data, mode=0o644): + descriptor, name = tempfile.mkstemp(prefix=".uvm-tools-", dir=path.parent) + try: + with os.fdopen(descriptor, "wb") as stream: + os.fchmod(stream.fileno(), mode) + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + os.replace(name, path) + directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + finally: + Path(name).unlink(missing_ok=True) + + +def safe_directory(path): + for parent in reversed((path, *path.parents)): + if not parent.exists(): + parent.mkdir(mode=0o755) + metadata = parent.lstat() + if not stat.S_ISDIR(metadata.st_mode) or metadata.st_uid != os.geteuid() or metadata.st_mode & 0o022: + raise ToolError(f"tool installation needs a root-owned directory: {parent}") + + +def regular_file(path): + metadata = path.lstat() + if not stat.S_ISREG(metadata.st_mode) or metadata.st_uid != os.geteuid() or metadata.st_nlink != 1 or metadata.st_mode & 0o022: + raise ToolError(f"refusing to replace an unmanaged tool file: {path}") + + +def download(url, destination, digest=None, *, max_bytes=1024 * 1024 * 1024): + if not url.startswith("https://"): + raise ToolError("tool downloads require HTTPS") + checksum = hashlib.sha256() + size = 0 + with urllib.request.urlopen(url, timeout=30) as response, destination.open("wb") as output: + if not response.geturl().startswith("https://"): + raise ToolError("tool download redirected away from HTTPS") + while True: + chunk = response.read(65536) + if not chunk: + break + size += len(chunk) + if size > max_bytes: + raise ToolError("tool download exceeds its size limit") + checksum.update(chunk) + output.write(chunk) + if digest is not None and checksum.hexdigest() != digest: + destination.unlink(missing_ok=True) + raise ToolError("tool download checksum did not match the pinned release") + + +def extract_archive(archive_path, destination): + """Bound extraction and check each path, including previously extracted links.""" + root = destination.resolve() + with tarfile.open(archive_path) as archive: + members = archive.getmembers() + if len(members) > 50000 or sum(member.size for member in members) > 4 * 1024**3: + raise ToolError("tool archive exceeds its extraction limit") + for member in members: + name = Path(member.name) + target = destination / name + if name.is_absolute() or ".." in name.parts or not target.resolve().is_relative_to(root): + raise ToolError("unsafe path in tool archive") + if not (member.isfile() or member.isdir() or member.issym() or member.islnk()): + raise ToolError("unsupported file in tool archive") + if member.issym() or member.islnk(): + link = (target.parent if member.issym() else destination) / member.linkname + if Path(member.linkname).is_absolute() or not link.resolve().is_relative_to(root): + raise ToolError("unsafe link in tool archive") + member.mode &= 0o755 + member.uid = member.gid = 0 + member.uname = member.gname = "root" + options = {"filter": "data"} if hasattr(tarfile, "data_filter") else {} + archive.extract(member, destination, **options) + + +def platform_info(): + values = {} + for line in Path("/etc/os-release").read_text().splitlines(): + if "=" in line: + key, value = line.split("=", 1) + values[key] = value.strip('"') + arch = os.uname().machine + return {"os": values.get("PRETTY_NAME", "Unknown OS"), "architecture": arch, + "supported": values.get("ID") == "ubuntu" and values.get("VERSION_ID") == "22.04" + and arch in ("x86_64", "aarch64"), + "boot_id": Path("/proc/sys/kernel/random/boot_id").read_text().strip()} + + +def guest_busy(): + try: + descriptor = os.open(LOCK, os.O_RDONLY | os.O_NOFOLLOW) + except FileNotFoundError: + return False + try: + try: + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + return False + except BlockingIOError: + return True + finally: + os.close(descriptor) + + +def probe(tool, username): + account = pwd.getpwnam(username) + environment = {"PATH": PATH, "HOME": account.pw_dir, "USER": username, "LOGNAME": username, + "LANG": "C.UTF-8", "LC_ALL": "C.UTF-8", "PYTHONDONTWRITEBYTECODE": "1"} + result = {"key": tool["key"], "id": tool["id"], "version": tool["version"], "installed": False, + "detected_version": None, "error": None, "usage": tool["usage"]} + for index, command in enumerate(tool["checks"]): + if shutil.which(command[0], path=PATH) is None: + return result + try: + process = subprocess.run(command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, timeout=3, env=environment, cwd=account.pw_dir, + user=account.pw_uid, group=account.pw_gid, + extra_groups=os.getgrouplist(username, account.pw_gid)) + output = process.stdout[:4096].decode("utf-8", errors="replace") + except (OSError, subprocess.TimeoutExpired) as error: + result["error"] = f"could not verify {command[0]}: {error}"[:512] + return result + if process.returncode != 0: + result["error"] = f"{command[0]} verification failed: {output}"[:512] + return result + match = re.search(tool["match"], output) + if index == 0: + if not match: + result["error"] = f"could not determine the installed {tool['id']} version" + return result + result["detected_version"] = match.group(1) + if not result["detected_version"].startswith(tool["prefix_match"]): + return result + elif tool["id"] == "jdk" and (not match or not match.group(1).startswith("21.")): + return result + result["installed"] = True + return result + + +class Installer: + def __init__(self, username): + self.username = username + self.log = bytearray() + self.truncated = False + self.lock = None + self.managed = {} + + def append(self, data): + self.log.extend(data) + if len(self.log) > LOG_LIMIT: + del self.log[:-LOG_LIMIT] + self.truncated = True + + def run(self, command): + self.append(("+ " + shlex.join(map(str, command)) + "\n").encode()) + environment = {**os.environ, "PATH": PATH, "LC_ALL": "C.UTF-8", "DEBIAN_FRONTEND": "noninteractive"} + with subprocess.Popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, env=environment, pass_fds=(self.lock,)) as process: + while True: + chunk = process.stdout.read(8192) + if not chunk: + break + self.append(chunk) + if process.wait() != 0: + raise ToolError(f"{command[0]} failed with exit code {process.returncode}; see the installation output") + + def remember(self, path, value): + name = str(path) + self.managed[name] = list(dict.fromkeys([*self.managed.get(name, []), value]))[-2:] + atomic_write(ROOT / "managed.json", json.dumps(self.managed).encode(), 0o600) + + def write_managed(self, path, data, mode=0o644): + safe_directory(path.parent) + digest = hashlib.sha256(data).hexdigest() + if path.exists() or path.is_symlink(): + regular_file(path) + current = hashlib.sha256(path.read_bytes()).hexdigest() + if current == digest: + return + if current not in self.managed.get(str(path), []): + raise ToolError(f"customized file needs attention before installing: {path}") + # Record intent first so an interrupted publication can be retried. + self.remember(path, digest) + atomic_write(path, data, mode) + + def link(self, name, target): + path = BIN_DIR / name + safe_directory(path.parent) + target = str(target) + if path.exists() or path.is_symlink(): + if path.is_symlink() and os.readlink(path) == target: + return + if not path.is_symlink() or os.readlink(path) not in self.managed.get(str(path), []): + raise ToolError(f"customized executable needs attention before installing: {path}") + self.remember(path, target) + temporary = path.with_name(".uvm-" + name) + temporary.unlink(missing_ok=True) + temporary.symlink_to(target) + os.replace(temporary, path) + + def apt(self, packages, *, refresh=False): + stamp = ROOT / "apt-updated" + if refresh or not stamp.exists() or time.time() - stamp.stat().st_mtime > 300: + self.run(["/usr/bin/apt-get", "-o", "DPkg::Lock::Timeout=300", "-o", "Acquire::Retries=2", + "-o", "APT::Update::Error-Mode=any", "update"]) + atomic_write(stamp, b"updated\n", 0o600) + pending = ROOT / "apt-pending" + if pending.exists(): + # Complete configuration left by an interrupted tool installation. + self.run(["/usr/bin/dpkg", "--configure", "-a"]) + atomic_write(pending, b"pending\n", 0o600) + self.run(["/usr/bin/apt-get", "-o", "DPkg::Lock::Timeout=300", "-o", "Acquire::Retries=2", + "-o", "Dpkg::Options::=--force-confold", "-y", "--no-install-recommends", "install", *packages]) + pending.unlink() + + def php_repository(self, temporary): + self.apt(["ca-certificates", "gnupg"]) + key = temporary / "php.asc" + download(f"https://keyserver.ubuntu.com/pks/lookup?op=get&search=0x{PHP_KEY}", key, max_bytes=65536) + result = subprocess.run(["gpg", "--batch", "--homedir", str(temporary), "--with-colons", "--show-keys", str(key)], + capture_output=True, check=True) + fingerprints = [line.split(":")[9] for line in result.stdout.decode().splitlines() if line.startswith("fpr:")] + if not fingerprints or fingerprints[0] != PHP_KEY: + raise ToolError("PHP repository signing key did not match its pinned fingerprint") + binary_key = temporary / "php.gpg" + self.run(["gpg", "--batch", "--homedir", str(temporary), "--dearmor", "--output", str(binary_key), str(key)]) + self.write_managed(Path("/etc/apt/keyrings/uvm-php.gpg"), binary_key.read_bytes()) + self.write_managed(Path("/etc/apt/sources.list.d/uvm-php.list"), + b"deb [signed-by=/etc/apt/keyrings/uvm-php.gpg] https://ppa.launchpadcontent.net/ondrej/php/ubuntu jammy main\n") + + def payload(self, tool, temporary): + prefix = ROOT / tool["release"] + marker = prefix / ".uvm-tools.json" + identity = {"release": tool["release"], "sha256": tool["sha256"]} + if prefix.exists() or prefix.is_symlink(): + safe_directory(prefix) + if not marker.exists(): + raise ToolError(f"installation directory is already in use: {prefix}") + regular_file(marker) + previous = json.loads(marker.read_bytes()) + if any(previous.get(key) != value for key, value in identity.items()): + raise ToolError(f"installation directory belongs to a different release: {prefix}") + if tool["kind"] != "conda" or previous.get("complete"): + return prefix + artifact = temporary / "download" + self.append(f"Downloading {tool['release']} and verifying SHA-256…\n".encode()) + download(tool["url"], artifact, tool["sha256"]) + if tool["kind"] == "archive": + extraction = temporary / "extract" + extraction.mkdir() + extract_archive(artifact, extraction) + source = extraction / tool["archive_root"] + if not source.is_dir() or source.is_symlink(): + raise ToolError("tool archive has an unexpected layout") + atomic_write(source / ".uvm-tools.json", json.dumps(identity).encode(), 0o600) + os.replace(source, prefix) + else: + prefix.mkdir(exist_ok=True, mode=0o755) + atomic_write(marker, json.dumps(identity).encode(), 0o600) + if tool["kind"] == "conda": + self.run(["/bin/bash", str(artifact), "-b", "-f", "-p", str(prefix)]) + else: + artifact.chmod(0o755) + os.replace(artifact, prefix / "micromamba") + return prefix + + def configure(self, tool, prefix): + for name, target in tool["links"].items(): + self.link(name, Path(target) if target.startswith("/") else prefix / target) + if tool["id"] == "python": + executable = prefix / "bin" / f"python{tool['version']}" + self.write_managed(BIN_DIR / f"pip{tool['version']}", + f'#!/bin/sh\nexec {shlex.quote(str(executable))} -m pip "$@"\n'.encode(), 0o755) + elif tool["id"] == "jdk": + self.write_managed(PROFILE_DIR / "uvm-java.sh", + f"export JAVA_HOME={shlex.quote(tool['java_home'])}\n".encode()) + elif tool["id"] == "conda": + for key, value in (("auto_activate_base", "false"),): + self.run([str(prefix / "bin/conda"), "config", "--system", "--set", key, value]) + for key, value in (("envs_dirs", "~/.conda/envs"), ("pkgs_dirs", "~/.conda/pkgs")): + self.run([str(prefix / "bin/conda"), "config", "--system", "--prepend", key, value]) + self.write_managed(PROFILE_DIR / "uvm-conda.sh", + f". {shlex.quote(str(prefix / 'etc/profile.d/conda.sh'))}\n".encode()) + elif tool["id"] == "micromamba": + self.write_managed(PROFILE_DIR / "uvm-micromamba.sh", + b'export MAMBA_ROOT_PREFIX="${MAMBA_ROOT_PREFIX:-$HOME/micromamba}"\n' + b'eval "$(/usr/local/bin/micromamba shell hook --shell posix)"\n') + + def install(self, tool): + safe_directory(LOCK.parent) + self.lock = os.open(LOCK, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600) + try: + metadata = os.fstat(self.lock) + if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1 or metadata.st_uid != os.geteuid(): + raise ToolError("unsafe guest tool installation lock") + try: + fcntl.flock(self.lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + raise ToolError("another installation is still running inside this guest; wait before retrying") + existing = probe(tool, self.username) + if existing["installed"]: + return {"status": "already_installed", "tool": existing} + safe_directory(ROOT) + manifest = ROOT / "managed.json" + if manifest.exists(): + regular_file(manifest) + self.managed = json.loads(manifest.read_bytes()) + with tempfile.TemporaryDirectory(prefix=".install-", dir=ROOT) as name: + temporary = Path(name) + if tool["kind"] == "php": + self.php_repository(temporary) + self.apt(tool["packages"], refresh=tool["kind"] == "php") + prefix = self.payload(tool, temporary) if "url" in tool else None + self.configure(tool, prefix) + installed = probe(tool, self.username) + if not installed["installed"]: + raise ToolError(installed["error"] or f"{tool['key']} did not pass its version check after installation") + if prefix is not None: + atomic_write(prefix / ".uvm-tools.json", json.dumps({"release": tool["release"], + "sha256": tool["sha256"], "complete": True}).encode(), 0o600) + return {"status": "installed", "tool": installed} + finally: + os.close(self.lock) + self.lock = None + + +def main(payload): + installer = Installer(payload["username"]) + try: + if os.geteuid() != 0: + raise ToolError("tool installation and inspection require guest root") + info = platform_info() + if not info["supported"] or info["architecture"] != payload["architecture"]: + raise ToolError("one-click tools require an Ubuntu 22.04 guest on the selected x86_64/ARM64 architecture") + if payload.get("boot_id") and payload["boot_id"] != info["boot_id"]: + raise ToolError("the guest rebooted during installation; retry on the current boot") + if payload["action"] == "probe": + result = {**info, "busy": guest_busy(), "tools": [probe(tool, payload["username"]) for tool in payload["tools"]]} + elif payload["action"] == "install": + result = installer.install(payload["tools"][0]) + else: + raise ToolError("unsupported tool operation") + result["ok"] = True + except (ToolError, OSError, ValueError, KeyError, tarfile.TarError, subprocess.SubprocessError) as error: + result = {"ok": False, "error": str(error)[:2048]} + result.update(output=installer.log.decode("utf-8", errors="replace"), truncated=installer.truncated) + print(json.dumps(result, ensure_ascii=True)) + + +if __name__ == "__main__": + main(json.loads(sys.argv[1])) diff --git a/uvm/lifecycle.py b/uvm/lifecycle.py index 4fe5319..7d864f5 100644 --- a/uvm/lifecycle.py +++ b/uvm/lifecycle.py @@ -26,6 +26,8 @@ from .known_hosts import KnownHostsManager from .network import NetworkManager from .state import StateStore from .system import check_kvm, ensure_data_directories, require_root +from .tool_catalog import normalize_tools +from .tool_store import ToolJobStore from .validation import parse_name, parse_password, parse_ports, parse_timeout, parse_username @@ -75,6 +77,7 @@ class LifecycleService: self._guest_connector = guest_connector self._terminal_connector = terminal_connector self._clock = clock + self._tool_store = ToolJobStore(settings, state_store, clock=clock) def create(self, spec: VmSpec) -> VmRecord: require_root() @@ -84,6 +87,7 @@ class LifecycleService: if spec.name is not None: parse_name(spec.name) ports = self._validated_ports(spec.ports) + tools = normalize_tools(spec.tools, allow_empty=True) ensure_data_directories(self._settings) with self._state_store.operation_lock(): self._check_boot() @@ -100,6 +104,8 @@ class LifecycleService: self._images.provision_guest_agent(disk, vm.id, vm.username) vm.disk_ready = True self._save(vm) + if tools: + self._tool_store.enqueue_locked(vm.id, tools, source="create") self._boot(vm, assets.kernel) return vm except BaseException as error: @@ -506,6 +512,8 @@ class LifecycleService: def execute( self, vm_id: str, command: str, *, sudo: bool = False, timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, + expected_identity: tuple | None = None, + on_connect: Callable[[GuestSession], None] | None = None, ) -> ExecResult: """Pin a running VM's guest connection under the lock, then wait unlocked.""" require_root() @@ -516,14 +524,52 @@ class LifecycleService: if (vm.status != "running" or vm.operation is not None or vm.pending_resize is not None or self._process.observe(vm) != "alive"): raise ConflictError("guest execution requires a running VM with no pending lifecycle operation") + if expected_identity is not None and expected_identity != (vm.pid, vm.process_start_time, vm.boot_id): + raise ConflictError("the VM restarted during tool installation; retry on the current boot") session = self._guest_connector(self._settings.guest_socket_path(vm.id), vm.id, vm.username, vm.pid) try: + if on_connect is not None: + on_connect(session) # This session stays attached to that VMM even if the VM is stopped, # deleted, or restarted concurrently. Never reconnect/retry a command. return session.execute(command, sudo=sudo, timeout_seconds=timeout_seconds) finally: session.close() + def _validate_tool_target(self, vm: VmRecord, *, start_if_stopped: bool, restart: bool) -> str: + self._not_deleting(vm) + observation = self._process.observe(vm) + if vm.operation is not None or vm.pending_resize is not None or not vm.disk_ready: + raise ConflictError("finish pending lifecycle operations before installing tools") + if vm.status == "running" and observation == "alive": + return "running" + if vm.status == "stopped" and observation == "gone" and (start_if_stopped or restart): + self._validate_resume(vm) + return "stopped" + raise ConflictError("tool installation requires a running VM; use Start & install for a stopped VM") + + def queue_tools(self, vm_id: str, selections, *, start_if_stopped=False, restart=False, request_id=None) -> dict: + require_root() + selections = normalize_tools(selections) + with self._state_store.operation_lock(): + vm = self._load_for_operation(vm_id, exact_id=True) + self._validate_tool_target(vm, start_if_stopped=start_if_stopped, restart=restart) + return self._tool_store.enqueue_locked(vm.id, selections, start_if_stopped=start_if_stopped, + restart=restart, request_id=request_id) + + def prepare_tool_install(self, vm_id: str, *, start_if_stopped=False, restart=False) -> VmRecord: + """Perform requested startup/agent refresh atomically before a tool worker connects.""" + require_root() + with self._state_store.operation_lock(): + vm = self._load_for_operation(vm_id, exact_id=True) + status = self._validate_tool_target(vm, start_if_stopped=start_if_stopped, restart=restart) + if status == "running" and restart: + self._stop_locked(vm) + if status == "stopped" or restart: + # The normal retained-disk boot also provisions the current agent. + vm = self._start_locked(vm) + return vm + def open_terminal(self, vm_id: str, *, rows: int = 24, cols: int = 80) -> tuple[GuestTerminal, str]: """Connect to the exact running VMM under the lock, then open its guest PTY.""" require_root() diff --git a/uvm/tool_store.py b/uvm/tool_store.py new file mode 100644 index 0000000..8e9b057 --- /dev/null +++ b/uvm/tool_store.py @@ -0,0 +1,195 @@ +"""Atomic per-VM tool job records, serialized with lifecycle filesystem changes.""" + +from __future__ import annotations + +import copy +import json +import os +import re +import stat +import tempfile +import time +import uuid +from contextlib import contextmanager + +from .config import Settings +from .errors import ConflictError, NotFoundError, StateError +from .state import StateStore +from .tool_catalog import normalize_tools + + +ACTIVE_JOB_STATUSES = frozenset({"queued", "starting", "waiting", "installing"}) +MAX_RETAINED_JOBS = 10 +MAX_DOCUMENT_BYTES = 16 * 1024 * 1024 + + +class ToolJobStore: + """Only short transactions hold operations.lock; guest execution never does.""" + + def __init__(self, settings: Settings, state_store: StateStore, *, clock=time.time) -> None: + self.settings = settings + self.state = state_store + self.clock = clock + + def _path(self, vm_id): + vm = self.state.get(vm_id) # Exact ID only, including while creating a VM. + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,80}", vm.id): + raise StateError("invalid VM ID in tool history") + directory = self.settings.vm_dir(vm.id) + try: + metadata = directory.lstat() + except FileNotFoundError as error: + raise NotFoundError("VM runtime directory is no longer available") from error + if (not stat.S_ISDIR(metadata.st_mode) or metadata.st_uid != os.geteuid() + or metadata.st_mode & 0o077 or not vm.runtime_owned): + raise StateError("tool history requires a private, owned VM runtime directory") + return directory / "tools.json" + + @staticmethod + def _empty(): + return {"schema_version": 1, "jobs": [], "inventory": [], "checked_at": None, + "agent_ready": False, "guest_busy": False, "supported": None, "error": None} + + def _read(self, vm_id): + path = self._path(vm_id) + try: + descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) + except FileNotFoundError: + return self._empty() + try: + with os.fdopen(descriptor, "rb") as stream: + metadata = os.fstat(stream.fileno()) + if (not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1 + or metadata.st_uid != os.geteuid() or metadata.st_mode & 0o077): + raise StateError("unsafe tool history file") + raw = stream.read(MAX_DOCUMENT_BYTES + 1) + if len(raw) > MAX_DOCUMENT_BYTES: + raise StateError("tool history exceeds its size limit") + data = json.loads(raw) + if (not isinstance(data, dict) or data.get("schema_version") != 1 + or not isinstance(data.get("jobs"), list) or not isinstance(data.get("inventory"), list) + or len(data["jobs"]) > MAX_RETAINED_JOBS): + raise StateError("invalid tool history document") + for job in data["jobs"]: + if (not isinstance(job, dict) or job.get("vm_id") != vm_id + or not re.fullmatch(r"tools-[0-9a-f]{32}", str(job.get("id", ""))) + or not isinstance(job.get("items"), list) or len(job["items"]) > 12): + raise StateError("invalid tool installation job") + return data + except (OSError, ValueError) as error: + raise StateError("could not read tool installation history") from error + + def _write(self, vm_id, data): + path = self._path(vm_id) # Never recreate a deleted VM directory. + descriptor, name = tempfile.mkstemp(prefix=".tools-", suffix=".json", dir=path.parent) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as stream: + os.fchmod(stream.fileno(), 0o600) + json.dump(data, stream, ensure_ascii=True, allow_nan=False) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + os.replace(name, path) + self.state._fsync_parent(path) + except OSError as error: + raise StateError("could not save tool installation history") from error + finally: + try: + os.unlink(name) + except FileNotFoundError: + pass + + @contextmanager + def transaction(self, vm_id): + with self.state.operation_lock(): + data = self._read(vm_id) + yield data + self._write(vm_id, data) + + def read(self, vm_id): + with self.state.operation_lock(): + return self._read(vm_id) + + def enqueue_locked(self, vm_id, selections, *, start_if_stopped=False, restart=False, + request_id=None, source="user"): + """Called by lifecycle create/submit with operations.lock already held.""" + selections = normalize_tools(selections) + data = self._read(vm_id) + wanted = [item.to_dict() for item in selections] + for previous in reversed(data["jobs"]): + same = (previous["selections"] == wanted and previous["start_if_stopped"] == start_if_stopped + and previous.get("restart", False) == restart) + if request_id is not None and previous.get("request_id") == request_id: + if not same: + raise ConflictError("this installation request ID was already used for different tools") + return copy.deepcopy(previous) + if previous["status"] in ACTIVE_JOB_STATUSES: + if same: + return copy.deepcopy(previous) + raise ConflictError("a tool installation is already queued or running for this VM") + now = self.clock() + job = {"id": "tools-" + uuid.uuid4().hex, "vm_id": vm_id, "status": "queued", "source": source, + "selections": wanted, "start_if_stopped": start_if_stopped, "restart": restart, + "request_id": request_id, "created_at": now, "updated_at": now, "completed_at": None, + "error": None, "outcome_unknown": False, + "items": [{**item.to_dict(), "key": item.key, "status": "queued", "detected_version": None, + "error": None, "output": "", "truncated": False} for item in selections]} + data["jobs"] = [*data["jobs"], job][-MAX_RETAINED_JOBS:] + self._write(vm_id, data) + return copy.deepcopy(job) + + def get_job(self, vm_id, job_id): + for job in self.read(vm_id)["jobs"]: + if job["id"] == job_id: + return job + raise NotFoundError("tool installation job not found for this VM") + + def update_job(self, vm_id, job_id, update): + with self.transaction(vm_id) as data: + for job in data["jobs"]: + if job["id"] == job_id: + update(job, data) + job["updated_at"] = self.clock() + return copy.deepcopy(job) + raise NotFoundError("tool installation job not found for this VM") + + def claim(self): + with self.state.operation_lock(): + for vm in self.state.load().vms.values(): + # A failed/interrupted create may not own a runtime directory. + if not vm.runtime_owned or not self.settings.vm_dir(vm.id).is_dir(): + continue + data = self._read(vm.id) + for job in data["jobs"]: + if job["status"] == "queued": + job["status"] = "starting" if vm.status == "stopped" or job.get("restart") else "waiting" + job["updated_at"] = self.clock() + self._write(vm.id, data) + return copy.deepcopy(job) + return None + + @staticmethod + def interrupt(job, message, now, *, outcome_unknown=True): + job.update(status="interrupted", error=message, outcome_unknown=outcome_unknown, + updated_at=now, completed_at=now) + for item in job["items"]: + if item["status"] in {"installing", "queued"}: + item["status"] = "interrupted" if item["status"] == "installing" else "skipped" + item["error"] = message + + def recover(self): + """The worker owner lock proves that previous in-flight host workers exited.""" + with self.state.operation_lock(): + for vm in self.state.load().vms.values(): + if not vm.runtime_owned or not self.settings.vm_dir(vm.id).is_dir(): + continue + data = self._read(vm.id) + changed = False + for job in data["jobs"]: + if job["status"] in ACTIVE_JOB_STATUSES - {"queued"}: + self.interrupt(job, "Management server restarted. Refresh tool status before retrying; " + "the last installation may have completed inside the guest.", self.clock()) + changed = True + if changed: + data.update(agent_ready=False, guest_busy=False) + self._write(vm.id, data) diff --git a/uvm/tools.py b/uvm/tools.py new file mode 100644 index 0000000..bafeb4b --- /dev/null +++ b/uvm/tools.py @@ -0,0 +1,319 @@ +"""Bounded background installation workers using the existing guest exec channel.""" + +from __future__ import annotations + +import fcntl +import json +import logging +import os +import re +import shlex +import stat +import threading +import time +from functools import lru_cache +from pathlib import Path + +from . import guest_tools +from .errors import ConflictError, GuestAgentUnavailableError, GuestExecutionError, NotFoundError, UvmError, ValidationError +from .guest_exec import validate_exec +from .system import require_root +from .tool_catalog import all_selections, normalize_tools, recipe +from .tool_store import ToolJobStore + + +@lru_cache(maxsize=1) +def _guest_source(): + return Path(guest_tools.__file__).read_text(encoding="utf-8") + + +def guest_command(payload: dict) -> str: + command = shlex.join(["/usr/bin/python3", "-I", "-c", _guest_source(), json.dumps(payload, ensure_ascii=True)]) + validate_exec(command, True, 60) + return command + + +def _identity(vm): + return vm.pid, vm.process_start_time, vm.boot_id + + +def _text(value, limit=16384): + return str(value or "").encode("utf-8", errors="replace")[-limit:].decode("utf-8", errors="replace") + + +class ToolService: + def __init__(self, settings, state_store, lifecycle) -> None: + self.settings = settings + self.state = state_store + self.lifecycle = lifecycle + self.store = ToolJobStore(settings, state_store) + self._mutex = threading.RLock() + self._wake = threading.Condition() + self._stop = threading.Event() + self._threads = [] + self._sessions = {} + self._owner = None + self._workers_left = 0 + + def start(self): + with self._mutex: + if self._stop.is_set(): + raise ConflictError("the tool installer is shutting down") + if self._owner is not None: + return + with self.state.operation_lock(): + descriptor = os.open(self.settings.base / "tool-worker.lock", os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600) + try: + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1 or metadata.st_uid != os.geteuid(): + raise UvmError("unsafe tool worker ownership file") + os.fchmod(descriptor, 0o600) + fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BaseException as error: + os.close(descriptor) + if isinstance(error, BlockingIOError): + raise ConflictError("another management server is already processing tool installations for this registry") from error + raise + self._owner = descriptor + try: + self.store.recover() + self._workers_left = self.settings.tool_workers + self._threads = [threading.Thread(target=self._worker, name=f"uvm-tools-{index}", daemon=True) + for index in range(self.settings.tool_workers)] + for thread in self._threads: + thread.start() + except BaseException: + self._release_owner() + raise + + def _release_owner(self): + if self._owner is not None: + os.close(self._owner) + self._owner = None + + def notify(self): + with self._wake: + self._wake.notify_all() + + def shutdown(self): + self._stop.set() + self.notify() + with self._mutex: + sessions = list(self._sessions.values()) + for session in sessions: + session.close() + deadline = time.monotonic() + 8 + for thread in self._threads: + thread.join(max(0, deadline - time.monotonic())) + # A worker finishing a host start keeps ownership until that start ends. + # Its guest command is never submitted after shutdown was requested. + with self._mutex: + if not self._workers_left: + self._release_owner() + + def submit(self, vm_id, selections, *, start_if_stopped=False, restart=False, request_id=None): + require_root() + selections = normalize_tools(selections) + if type(start_if_stopped) is not bool or type(restart) is not bool: + raise ValidationError("start_if_stopped and restart must be booleans") + if request_id is not None and (not isinstance(request_id, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", request_id)): + raise ValidationError("request_id must contain 1–80 letters, numbers, underscores, or hyphens") + self.start() + job = self.lifecycle.queue_tools(vm_id, selections, start_if_stopped=start_if_stopped, + restart=restart, request_id=request_id) + self.notify() + return job + + def _execute(self, vm, payload, timeout): + if self._stop.is_set(): + raise InterruptedError("management server is shutting down") + thread_id = threading.get_ident() + + def connected(session): + with self._mutex: + self._sessions[thread_id] = session + if self._stop.is_set(): + session.close() + + try: + result = self.lifecycle.execute(vm.id, guest_command(payload), sudo=True, timeout_seconds=timeout, + expected_identity=_identity(vm), on_connect=connected) + finally: + with self._mutex: + self._sessions.pop(thread_id, None) + if result.timed_out: + return {"ok": False, "error": f"installation/verification timed out after {timeout:g} seconds; retry to finish it", + "output": _text(result.stderr), "truncated": result.truncated} + try: + value = json.loads(result.stdout) + if result.exit_code != 0 or result.truncated or not isinstance(value, dict) or type(value.get("ok")) is not bool: + raise ValueError("invalid result") + return value + except (ValueError, TypeError) as error: + raise GuestExecutionError("guest returned an incomplete tool result; refresh status before retrying", + outcome_unknown=payload["action"] == "install") from error + + def _payload(self, vm, action, selections, *, boot_id=None): + return {"action": action, "username": vm.username, "architecture": self.settings.architecture, + "boot_id": boot_id, "tools": [recipe(selection, self.settings.architecture) for selection in selections]} + + @staticmethod + def _inventory(result, selections): + expected = {item.key: item for item in selections} + records = result.get("tools") + if not isinstance(records, list) or len(records) != len(expected): + raise GuestExecutionError("invalid guest tool inventory") + inventory = [] + for value in records: + if (not isinstance(value, dict) or value.get("key") not in expected + or type(value.get("installed")) is not bool): + raise GuestExecutionError("invalid guest tool inventory") + selection = expected.pop(value["key"]) + inventory.append({**selection.to_dict(), "key": selection.key, "installed": value["installed"], + "detected_version": _text(value.get("detected_version"), 120) or None, + "error": _text(value.get("error"), 512) or None, + "usage": _text(value.get("usage"), 512)}) + return inventory + + def snapshot(self, vm_id, *, refresh=False): + require_root() + vm = self.state.get(vm_id) + data = self.store.read(vm_id) + running = self.lifecycle.inspect(vm.id).observed_status == "running" + if refresh and running: + try: + result = self._execute(vm, self._payload(vm, "probe", all_selections()), 90) + if not result["ok"]: + raise UvmError(_text(result.get("error"), 2048)) + inventory = self._inventory(result, all_selections()) + with self.store.transaction(vm_id) as current: + current.update(inventory=inventory, checked_at=time.time(), error=None, agent_ready=True, + guest_busy=result.get("busy") is True, supported=result.get("supported") is True) + data = self.store.read(vm_id) + except UvmError as error: + with self.store.transaction(vm_id) as current: + current.update(error=_text(error, 2048), agent_ready=not isinstance(error, GuestAgentUnavailableError)) + data = self.store.read(vm_id) + if not running: + data.update(agent_ready=False, guest_busy=False) + return {"vm_id": vm_id, **{key: data[key] for key in ( + "inventory", "checked_at", "agent_ready", "guest_busy", "supported", "error")}, + "stale": not running or data["checked_at"] is None or time.time() - data["checked_at"] > 60, + "latest_job": data["jobs"][-1] if data["jobs"] else None, + "jobs": [{key: job[key] for key in ("id", "status", "created_at", "completed_at")} + for job in reversed(data["jobs"])]} + + def _worker(self): + try: + while not self._stop.is_set(): + try: + job = self.store.claim() + if job is not None: + self._run_job(job) + continue + except (UvmError, OSError): + logging.getLogger(__name__).exception("Could not process a tool installation job") + with self._wake: + self._wake.wait(timeout=1) + finally: + with self._mutex: + self._workers_left -= 1 + if self._workers_left == 0 and self._stop.is_set(): + self._release_owner() + + def _update(self, job, update): + return self.store.update_job(job["vm_id"], job["id"], update) + + def _wait_ready(self, vm, selections): + deadline = time.monotonic() + self.settings.tool_ready_timeout_s + while not self._stop.is_set(): + try: + result = self._execute(vm, self._payload(vm, "probe", selections), 90) + if not result["ok"]: + raise UvmError(_text(result.get("error"), 2048)) + if result.get("busy"): + raise ConflictError("an earlier installation is still running inside this VM; wait and refresh before retrying") + if not isinstance(result.get("boot_id"), str) or not result["boot_id"]: + raise GuestExecutionError("guest did not report its boot identity") + return result + except GuestAgentUnavailableError: + if time.monotonic() >= deadline: + raise GuestAgentUnavailableError("Guest agent did not become ready. Restart & retry to apply agent setup, " + "then check the VM log if startup still fails.") + self._stop.wait(min(2, max(0, deadline - time.monotonic()))) + raise InterruptedError("management server is shutting down") + + def _run_job(self, job): + submitted = False + try: + if self._stop.is_set(): + raise InterruptedError("management server is shutting down") + selections = normalize_tools(job["selections"]) + vm = self.lifecycle.prepare_tool_install(job["vm_id"], start_if_stopped=job["start_if_stopped"], + restart=job.get("restart", False)) + self._update(job, lambda current, data: current.update(status="waiting")) + info = self._wait_ready(vm, selections) + initial = self._inventory(info, selections) + + def ready(current, data): + observed = {item["key"]: item for item in data["inventory"]} + observed.update({item["key"]: item for item in initial}) + data.update(inventory=list(observed.values()), checked_at=time.time(), agent_ready=True, + guest_busy=False, supported=True, error=None) + self._update(job, ready) + for index, selection in enumerate(selections): + if self._stop.is_set(): + raise InterruptedError("management server is shutting down") + + def begin(current, data): + current["status"] = "installing" + current["items"][index]["status"] = "installing" + self._update(job, begin) + submitted = True + result = self._execute(vm, self._payload(vm, "install", [selection], boot_id=info["boot_id"]), + self.settings.tool_install_timeout_s) + observed = None + if result["ok"]: + observed, = self._inventory({"tools": [result.get("tool")]}, [selection]) + if not observed["installed"] or result.get("status") not in {"installed", "already_installed"}: + raise GuestExecutionError("guest installation did not verify the selected tool", outcome_unknown=True) + + def finish_item(current, data): + current["items"][index].update( + status=result["status"] if result["ok"] else "failed", + detected_version=observed["detected_version"] if observed else None, + error=None if result["ok"] else _text(result.get("error"), 2048), + output=_text(result.get("output")), truncated=result.get("truncated") is True) + inventory = {item["key"]: item for item in data["inventory"]} + if observed: + inventory[selection.key] = observed + elif selection.key in inventory: + inventory[selection.key].update(installed=False, error="Refresh to verify after the failed installation.") + data.update(inventory=list(inventory.values()), checked_at=time.time()) + self._update(job, finish_item) + submitted = False + + def finish(current, data): + failures = sum(item["status"] == "failed" for item in current["items"]) + current.update(status="failed" if failures else "succeeded", completed_at=time.time(), + error=f"{failures} tool installation(s) failed; see the per-tool output and retry." if failures else None) + self._update(job, finish) + except NotFoundError: + return # Deletion owns removal of the whole runtime, including history. + except (InterruptedError, GuestExecutionError, ConflictError) as error: + try: + self._update(job, lambda current, data: self.store.interrupt( + current, _text(error, 2048), time.time(), + outcome_unknown=getattr(error, "outcome_unknown", submitted))) + except NotFoundError: + pass + except (UvmError, OSError, ValueError) as error: + def failed(current, data): + self.store.interrupt(current, _text(error, 2048), time.time(), outcome_unknown=submitted) + current["status"] = "failed" if not submitted else "interrupted" + if isinstance(error, GuestAgentUnavailableError): + data.update(agent_ready=False, error=_text(error, 2048)) + try: + self._update(job, failed) + except NotFoundError: + pass