from __future__ import annotations import ast import hashlib import io import shlex import tarfile import tempfile import unittest from pathlib import Path from types import SimpleNamespace from unittest.mock import 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") ast.parse(argv[3], feature_version=(3, 10)) 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"))