366 lines
17 KiB
Python
366 lines
17 KiB
Python
"""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()
|