"""Synthetic behavioral checks for the pinned installer-state guard.""" from __future__ import annotations import importlib.util import json import os import contextlib import io from pathlib import Path import subprocess import sys import tempfile import unittest from unittest import mock HERE = Path(__file__).parent sys.path.insert(0, str(HERE)) SPEC = importlib.util.spec_from_file_location("vm_install_boot", HERE / "install_boot.py") assert SPEC and SPEC.loader boot = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(boot) class FakeVBox: def __init__(self, root: Path, *, mutations: dict[str, str] | None = None, extra_vms: tuple[str, ...] = (), start_result: int = 0): self.root = root self.mutations = mutations or {} self.extra_vms = extra_vms self.start_result = start_result self.calls: list[tuple[str, ...]] = [] def __call__(self, argv): args = tuple(argv) self.calls.append(args) if args[1:3] == ("list", "vms"): rows = [f'"{boot.TARGET_NAME}" {{{boot.TARGET_UUID}}}', *self.extra_vms] return subprocess.CompletedProcess(args, 0, "\n".join(rows) + "\n", "") if args[1] == "showvminfo": values = { "name": boot.TARGET_NAME, "UUID": boot.TARGET_UUID, "description": f"{boot.TARGET_NAME}:{boot.TARGET_UUID}", "CfgFile": str(self.root / boot.TARGET_NAME / f"{boot.TARGET_NAME}.vbox"), "VMState": "poweroff", "storagecontrollername0": "SATA", "SATA-0-0": str(self.root / boot.TARGET_NAME / f"{boot.TARGET_NAME}.vdi"), "SATA-1-0": str(self.root / boot.TARGET_NAME / f"Unattended-{boot.TARGET_UUID}-aux-iso.viso"), **{f"SATA-{port}-0": "none" for port in range(2, 30)}, **{f"nic{slot}": "nat" if slot == 1 else "none" for slot in range(1, 9)}, "natpf1": "", "SharedFolderNameMachineMapping1": "", "SATA-ImageUUID-0-0": "{synthetic-disk-uuid}", "SATA-IsEjected-1-0": "off", } values.update(self.mutations) out = "\n".join( f'{json.dumps(key) if "-" in key else key}={json.dumps(value)}' for key, value in values.items() ) + "\n" return subprocess.CompletedProcess(args, 0, out, "") if args[1] == "startvm": return subprocess.CompletedProcess(args, self.start_result, "", "") return subprocess.CompletedProcess(args, 0, "", "") class InstallBootTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) / "lab" self.root.mkdir(mode=0o700) os.chmod(self.root, 0o700) self.folder = self.root / boot.TARGET_NAME self.folder.mkdir(mode=0o700) os.chmod(self.folder, 0o700) self.cfg = self.folder / f"{boot.TARGET_NAME}.vbox" self.disk = self.folder / f"{boot.TARGET_NAME}.vdi" self.viso = self.folder / f"Unattended-{boot.TARGET_UUID}-aux-iso.viso" self.cfg.write_text("synthetic vbox", encoding="ascii") self.disk.write_bytes(b"synthetic disk") self.viso.write_bytes(b"synthetic viso; contents intentionally never read") os.chmod(self.viso, 0o600) manifest = { "schema": 1, "name": boot.TARGET_NAME, "owner_uid": os.getuid(), "uuid": boot.TARGET_UUID, "host_marker": f"{boot.TARGET_NAME}:{boot.TARGET_UUID}", } (self.root / "target-identity.json").write_text(json.dumps(manifest), encoding="utf-8") os.chmod(self.root / "target-identity.json", 0o600) self.constants = mock.patch.multiple( boot, LAB_ROOT=self.root, TARGET_FOLDER=self.folder, EXPECTED_VDI=self.disk, EXPECTED_VISO=self.viso, ) self.constants.start() self.addCleanup(self.constants.stop) self.fake = FakeVBox(self.root) def tearDown(self): self.temp.cleanup() def test_exact_installer_state_checks_read_only(self): uid, info = boot.inspect_installer_state(lab_root=self.root, runner=self.fake) self.assertEqual(uid, boot.TARGET_UUID) self.assertEqual(info["SATA-1-0"], str(self.viso)) self.assertEqual([call[1] for call in self.fake.calls], ["list", "showvminfo"]) def test_start_uses_only_pinned_headless_target_after_inspection(self): uid = boot.start_installer(lab_root=self.root, runner=self.fake) self.assertEqual(uid, boot.TARGET_UUID) self.assertEqual(self.fake.calls[-1], ( "VBoxManage", "startvm", boot.TARGET_UUID, "--type=headless", )) def test_identity_media_power_network_and_forwarding_mismatches_refuse_before_start(self): cases = ( ({"description": "wrong marker"}, "host marker"), ({"SATA-2-0": "/tmp/extra.iso"}, "SATA media"), ({"SATA-1-0": "/tmp/elsewhere.viso"}, "SATA media"), ({"VMState": "running"}, "powered off"), ({"nic2": "nat"}, "NAT-only"), ({"nic9": "nat"}, "extra devices"), ({"natpf1": "guest,tcp,,2222,,22"}, "forwarding"), ({"SharedFolderNameMachineMapping1": "hostshare"}, "shared folders"), ({"IDE-0-0": "/tmp/extra.iso"}, "storage controller or media"), ) for mutations, message in cases: with self.subTest(mutations=mutations): fake = FakeVBox(self.root, mutations=mutations) with self.assertRaisesRegex(boot.InstallBootError, message): boot.start_installer(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_extra_registered_vm_row_is_rejected_without_start(self): fake = FakeVBox(self.root, extra_vms=( '"wga-m9-debian137-20261025" {12345678-1234-1234-1234-123456789012}', )) with self.assertRaisesRegex(boot.InstallBootError, "missing, duplicated, or mismatched"): boot.start_installer(lab_root=self.root, runner=fake) self.assertEqual([call[1] for call in fake.calls], ["list"]) def test_target_folder_and_attached_viso_symlinks_refuse(self): outside = Path(self.temp.name) / "outside" self.folder.rename(outside) self.folder.symlink_to(outside, target_is_directory=True) fake = FakeVBox(self.root) with self.assertRaisesRegex(boot.InstallBootError, "target folder"): boot.start_installer(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_viso_mode_and_missing_media_are_checked_without_reading_contents(self): os.chmod(self.viso, 0o644) fake = FakeVBox(self.root) with self.assertRaisesRegex(boot.InstallBootError, "mode 0600"): boot.start_installer(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_target_vdi_hard_link_refuses_before_start(self): other_vm_disk = Path(self.temp.name) / "other-vm.vdi" os.link(self.disk, other_vm_disk) fake = FakeVBox(self.root) with self.assertRaisesRegex(boot.InstallBootError, "one hard link"): boot.start_installer(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_generated_private_file_symlink_is_rejected(self): secret = self.folder / "preseed.cfg" secret.symlink_to(self.viso) fake = FakeVBox(self.root) with self.assertRaisesRegex(boot.InstallBootError, "symlink"): boot.start_installer(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_failed_start_reports_may_have_started_and_never_retries(self): fake = FakeVBox(self.root, start_result=1) with self.assertRaisesRegex(boot.InstallBootError, "may have started"): boot.start_installer(lab_root=self.root, runner=fake) self.assertEqual([call[1] for call in fake.calls].count("startvm"), 1) def test_start_timeout_is_bounded_and_reported_as_ambiguous(self): fake = FakeVBox(self.root) start_attempts = [] def timed_out(argv): if tuple(argv)[1] == "startvm": start_attempts.append(tuple(argv)) raise subprocess.TimeoutExpired(argv, 15) return fake(argv) with self.assertRaisesRegex(boot.InstallBootError, "may have started"): boot.start_installer(lab_root=self.root, runner=timed_out) self.assertEqual(start_attempts, [( "VBoxManage", "startvm", boot.TARGET_UUID, "--type=headless", )]) def test_cli_has_no_target_or_path_override(self): with mock.patch.object(boot, "inspect_installer_state") as inspect: with contextlib.redirect_stderr(io.StringIO()): with self.assertRaises(SystemExit) as raised: boot.main(["check", "--lab-root", "/tmp/other"]) self.assertEqual(raised.exception.code, 2) inspect.assert_not_called() if __name__ == "__main__": unittest.main()