"""Behavioral tests for the fail-closed VirtualBox target guard. These tests use a temporary synthetic VM inventory and a recording fake CLI. They never invoke VBoxManage, start a VM, or contact a host service. """ from __future__ import annotations import importlib.util import json import os from pathlib import Path import subprocess import sys import tempfile import unittest import uuid SPEC = importlib.util.spec_from_file_location("vm_guard", Path(__file__).with_name("guard.py")) assert SPEC and SPEC.loader guard = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(guard) class FakeVBox: def __init__(self, root: Path, *, mutate: dict[str, str] | None = None, listed_uuid: str | None = None, duplicate: bool = False, extra_list_rows: tuple[str, ...] = (), guest_marker: str = guard.EXPECTED_GUEST_MARKER): self.root = root self.mutate = mutate or {} self.listed_uuid = listed_uuid self.duplicate = duplicate self.extra_list_rows = extra_list_rows self.guest_marker = guest_marker self.calls: list[tuple[str, ...]] = [] self.target_uuid = str(uuid.UUID("1fd553db-3b0e-43e8-b407-244ea5f8c5c8")) def __call__(self, argv): args = tuple(argv) self.calls.append(args) if args[1:3] == ("list", "vms"): uid = self.listed_uuid or self.target_uuid rows = [f'"{guard.TARGET_NAME}" {{{uid}}}'] if self.duplicate: rows.append(f'"{guard.TARGET_NAME}" {{{self.target_uuid}}}') rows.extend(self.extra_list_rows) return subprocess.CompletedProcess(args, 0, "\n".join(rows) + "\n", "") if args[1] == "showvminfo": lines = { "name": guard.TARGET_NAME, "UUID": self.target_uuid, "description": f"{guard.TARGET_NAME}:{self.target_uuid}", "CfgFile": str(self.root / guard.TARGET_NAME / f"{guard.TARGET_NAME}.vbox"), "nic1": "nat", **{f"nic{n}": "none" for n in range(2, 9)}, "natpf1": "", "storagecontrollername0": "SATA", "SATA-0-0": str(self.root / guard.TARGET_NAME / f"{guard.TARGET_NAME}.vdi"), "SATA-1-0": str(self.root / guard.EXPECTED_ISO), "SATA-2-0": "none", "SATA-ImageUUID-0-0": "{synthetic-disk-uuid}", "SATA-hot-pluggable-0-0": "off", "SATA-nonrotational-0-0": "off", "SATA-discard-0-0": "off", "SATA-IsEjected-1-0": "off", "SATA-tempeject-1-0": "off", } lines.update(self.mutate) out = "\n".join( f'{json.dumps(key) if "-" in key else key}={json.dumps(value)}' for key, value in lines.items() ) + "\n" return subprocess.CompletedProcess(args, 0, out, "") if args[1:3] == ("guestcontrol", self.target_uuid): return subprocess.CompletedProcess(args, 0, self.guest_marker + "\n", "") return subprocess.CompletedProcess(args, 0, "", "") class GuardTests(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.vm_dir = self.root / guard.TARGET_NAME self.vm_dir.mkdir() (self.vm_dir / f"{guard.TARGET_NAME}.vbox").write_text("synthetic", encoding="ascii") (self.vm_dir / f"{guard.TARGET_NAME}.vdi").write_bytes(b"synthetic disk") (self.root / guard.EXPECTED_ISO).write_bytes(b"synthetic ISO marker") self.vbox = FakeVBox(self.root) marker = f"{guard.TARGET_NAME}:{self.vbox.target_uuid}" manifest = { "schema": 1, "name": guard.TARGET_NAME, "owner_uid": os.getuid(), "uuid": self.vbox.target_uuid, "host_marker": marker, } manifest_path = self.root / guard.IDENTITY_FILE manifest_path.write_text(json.dumps(manifest), encoding="utf-8") os.chmod(manifest_path, 0o600) def tearDown(self): self.temp.cleanup() def inspect(self, runner=None): return guard.inspect_target(lab_root=self.root, runner=runner or self.vbox) def test_normal_selected_identity_is_allowed(self): target_uuid, info = self.inspect() self.assertEqual(target_uuid, self.vbox.target_uuid) self.assertEqual(info["nic1"], "nat") self.assertEqual(self.vbox.calls, [ ("VBoxManage", "list", "vms"), ("VBoxManage", "showvminfo", self.vbox.target_uuid, "--machinereadable"), ]) def test_missing_manifest_fails_closed_before_inventory_or_start(self): (self.root / guard.IDENTITY_FILE).unlink() with self.assertRaisesRegex(guard.GuardError, "manifest is missing"): guard.start_target(lab_root=self.root, runner=self.vbox) self.assertEqual(self.vbox.calls, []) def test_duplicate_or_mismatched_name_uuid_fails_before_start(self): for fake in (FakeVBox(self.root, duplicate=True), FakeVBox(self.root, listed_uuid=str(uuid.uuid4()))): with self.subTest(duplicate=fake.duplicate, listed_uuid=fake.listed_uuid): with self.assertRaisesRegex(guard.GuardError, "duplicate|missing, duplicated"): guard.start_target(lab_root=self.root, runner=fake) self.assertEqual([call[1] for call in fake.calls], ["list"]) def test_malformed_or_ambiguous_vm_inventory_fails_before_start(self): cases = ( (FakeVBox(self.root, extra_list_rows=("not a VM row",)), "malformed"), (FakeVBox(self.root, extra_list_rows=(f'"other" {{{uuid.uuid4()}}}', f'"other" {{{uuid.uuid4()}}}')), "duplicate"), (FakeVBox(self.root, extra_list_rows=(f'"other" {{{self.vbox.target_uuid}}}',)), "duplicate"), ) for fake, expected in cases: with self.subTest(rows=fake.extra_list_rows): with self.assertRaisesRegex(guard.GuardError, expected): guard.start_target(lab_root=self.root, runner=fake) self.assertEqual([call[1] for call in fake.calls], ["list"]) def test_malformed_or_duplicate_machine_readable_values_refuse_before_start(self): for bad_line, expected in (("not-a-key-value", "malformed"), ('name="duplicate"', "duplicate")): class MalformedMachine(FakeVBox): def __call__(inner_self, argv): result = super(MalformedMachine, inner_self).__call__(argv) if argv[1] == "showvminfo": result.stdout += bad_line + "\n" return result fake = MalformedMachine(self.root) with self.subTest(line=bad_line): with self.assertRaisesRegex(guard.GuardError, expected): guard.start_target(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_quoted_storage_keys_and_port_metadata_match_observed_vbox_output(self): fake = FakeVBox(self.root) _, info = self.inspect(fake) self.assertEqual(info['SATA-0-0'], str(self.vm_dir / f"{guard.TARGET_NAME}.vdi")) self.assertEqual(info['SATA-2-0'], "none") self.assertEqual( {key for key in info if key.startswith("SATA-") and key not in { "SATA-0-0", "SATA-1-0", "SATA-2-0" }}, { "SATA-ImageUUID-0-0", "SATA-hot-pluggable-0-0", "SATA-nonrotational-0-0", "SATA-discard-0-0", "SATA-IsEjected-1-0", "SATA-tempeject-1-0", }, ) self.assertEqual(info["SATA-ImageUUID-0-0"], "{synthetic-disk-uuid}") def test_unknown_attachment_shaped_key_and_extra_media_are_rejected(self): cases = ( ({"SATA-unknown-property-1-0": "off"}, "unrecognized attached storage"), ({"SATA-3-0": str(self.vm_dir / f"{guard.TARGET_NAME}.vdi")}, "expected exactly one target VDI"), ) for mutations, expected in cases: with self.subTest(mutations=mutations): fake = FakeVBox(self.root, mutate=mutations) with self.assertRaisesRegex(guard.GuardError, expected): guard.start_target(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_extra_or_undeclared_hyphenated_controller_storage_fails_closed(self): cases = ( ({"storagecontrollername1": "IDE-Extra", "IDE-Extra-0-0": "/tmp/other.iso"}, "storage controller inventory"), ({"IDE-Extra-0-0": "/tmp/other.iso"}, "unrecognized attached storage"), ) for mutations, expected in cases: with self.subTest(mutations=mutations): fake = FakeVBox(self.root, mutate=mutations) with self.assertRaisesRegex(guard.GuardError, expected): guard.start_target(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_missing_storage_controller_inventory_fails_closed(self): class MissingController(FakeVBox): def __call__(inner_self, argv): result = super(MissingController, inner_self).__call__(argv) if argv[1] == "showvminfo": result.stdout = "\n".join( line for line in result.stdout.splitlines() if not line.startswith("storagecontrollername") ) + "\n" return result fake = MissingController(self.root) with self.assertRaisesRegex(guard.GuardError, "storage controller inventory"): guard.start_target(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_target_folder_symlink_outside_lab_fails_configuration_guard(self): outside_folder = Path(self.temp.name) / "outside-target" self.vm_dir.rename(outside_folder) self.vm_dir.symlink_to(outside_folder, target_is_directory=True) fake = FakeVBox(self.root) with self.assertRaisesRegex(guard.GuardError, "exact target folder must be a real directory"): guard.start_target(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_quoted_and_unquoted_duplicate_keys_are_rejected(self): class DuplicateQuotedKey(FakeVBox): def __call__(inner_self, argv): result = super(DuplicateQuotedKey, inner_self).__call__(argv) if argv[1] == "showvminfo": result.stdout += 'SATA-0-0="duplicate.vdi"\n' return result fake = DuplicateQuotedKey(self.root) with self.assertRaisesRegex(guard.GuardError, "duplicate.*key"): guard.start_target(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_malformed_quoted_key_is_rejected(self): class MalformedQuotedKey(FakeVBox): def __call__(inner_self, argv): result = super(MalformedQuotedKey, inner_self).__call__(argv) if argv[1] == "showvminfo": result.stdout += '"SATA-0-1=value\n' return result fake = MalformedQuotedKey(self.root) with self.assertRaisesRegex(guard.GuardError, "unrecognized.*key"): guard.start_target(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_virtualbox_recording_section_marker_is_ignored(self): class RecordingSection(FakeVBox): def __call__(inner_self, argv): result = super(RecordingSection, inner_self).__call__(argv) if argv[1] == "showvminfo": result.stdout = result.stdout.replace( 'name="' + guard.TARGET_NAME + '"\n', 'name="' + guard.TARGET_NAME + '"\n rec_screen0\n', ) return result fake = RecordingSection(self.root) target_uuid, info = self.inspect(fake) self.assertEqual(target_uuid, fake.target_uuid) self.assertEqual(info["name"], guard.TARGET_NAME) class UnknownSection(FakeVBox): def __call__(inner_self, argv): result = super(UnknownSection, inner_self).__call__(argv) if argv[1] == "showvminfo": result.stdout = result.stdout.replace( 'name="' + guard.TARGET_NAME + '"\n', 'name="' + guard.TARGET_NAME + '"\n rec_unknown0\n', ) return result unknown = UnknownSection(self.root) with self.assertRaisesRegex(guard.GuardError, "malformed VBoxManage"): self.inspect(unknown) def test_mismatched_identity_path_network_and_marker_all_refuse_before_start(self): cases = [ ({"name": "wga-other"}, "showvminfo identity"), ({"description": "wrong-marker"}, "host marker"), ({"CfgFile": "/tmp/not-the-lab/other.vbox"}, "configuration"), ({"nic1": "bridged"}, "only NIC 1 in NAT"), ({"nic2": "hostonly"}, "only NIC 1 in NAT"), ({"natpf1": "ssh,tcp,,2222,,22"}, "port forwarding"), ({"Forwarding(0)": "ssh,tcp,,2222,,22"}, "port forwarding"), ({"SharedFolderPathMachineMapping1": "/"}, "shared folders"), ({"SATA-0-0": "/tmp/other.vdi"}, "attached storage path"), ] for mutations, expected in cases: with self.subTest(mutations=mutations): fake = FakeVBox(self.root, mutate=mutations) with self.assertRaisesRegex(guard.GuardError, expected): guard.start_target(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_incomplete_nic_inventory_refuses_ambiguous_state(self): class MissingNic(FakeVBox): def __call__(self, argv): result = super().__call__(argv) if argv[1] == "showvminfo": result.stdout = "\n".join( line for line in result.stdout.splitlines() if not line.startswith('nic8=') ) + "\n" return result fake = MissingNic(self.root) with self.assertRaisesRegex(guard.GuardError, "inventory is incomplete"): guard.start_target(lab_root=self.root, runner=fake) self.assertNotIn("startvm", [call[1] for call in fake.calls]) def test_valid_start_is_scoped_to_exact_uuid_after_checks(self): result = guard.start_target(lab_root=self.root, runner=self.vbox) self.assertEqual(result, self.vbox.target_uuid) self.assertEqual(self.vbox.calls[-1], ( "VBoxManage", "startvm", self.vbox.target_uuid, "--type=gui")) def test_guest_marker_is_read_only_and_must_match_before_authorization(self): password = self.root / "guest-password" password.write_text("synthetic\n", encoding="ascii") os.chmod(password, 0o600) self.assertEqual(guard.authorize_guest( password_file=password, lab_root=self.root, runner=self.vbox), self.vbox.target_uuid) self.assertIn("--passwordfile", self.vbox.calls[-1]) self.assertIn("--wait-stdout", self.vbox.calls[-1]) self.assertEqual(self.vbox.calls[-1][self.vbox.calls[-1].index("--timeout") + 1], "15000") self.assertNotIn("synthetic", " ".join(self.vbox.calls[-1])) fake = FakeVBox(self.root, guest_marker="wrong") with self.assertRaisesRegex(guard.GuardError, "guest-local marker"): guard.authorize_guest(password_file=password, lab_root=self.root, runner=fake) self.assertEqual(fake.calls[-1][1], "guestcontrol") self.assertEqual(len(fake.calls), 3) def test_subprocess_capture_stops_at_limit(self): for stream in ("stdout", "stderr"): script = f"import sys; sys.{stream}.write('x' * {guard.MAX_CAPTURE + 1})" with self.subTest(stream=stream), self.assertRaisesRegex(guard.GuardError, "output exceeded"): guard.run_command((sys.executable, "-c", script)) def test_unsafe_lab_or_manifest_permissions_refuse(self): os.chmod(self.root, 0o755) with self.assertRaisesRegex(guard.GuardError, "mode 0700"): self.inspect() os.chmod(self.root, 0o700) os.chmod(self.root / guard.IDENTITY_FILE, 0o644) with self.assertRaisesRegex(guard.GuardError, "mode 0600"): self.inspect() if __name__ == "__main__": unittest.main()