"""Behavioral tests for the credential-redacting M9 unattended preparation.""" from __future__ import annotations import hashlib import importlib.util import os 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_prepare", HERE / "install_prepare.py") assert SPEC and SPEC.loader prepare = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(prepare) class PrepareTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.lab = self.root / "lab" self.vm_dir = self.lab / prepare.guard.TARGET_NAME self.vm_dir.mkdir(parents=True) os.chmod(self.lab, 0o700) self.disk = self.vm_dir / f"{prepare.guard.TARGET_NAME}.vdi" self.disk.write_bytes(b"synthetic VDI") self.iso = self.lab / prepare.guard.EXPECTED_ISO self.iso.write_bytes(b"synthetic installer") self.additions = self.root / "VBoxGuestAdditions.iso" self.additions.write_bytes(b"synthetic additions") self.password_dir = self.root / "private" self.password_dir.mkdir(mode=0o700) os.chmod(self.password_dir, 0o700) self.user_password = self.password_dir / "wga-password" self.root_password = self.password_dir / "root-password" self.user_password.write_text("synthetic-user-secret", encoding="ascii") self.root_password.write_text("synthetic-root-secret", encoding="ascii") os.chmod(self.user_password, 0o600) os.chmod(self.root_password, 0o600) self.iso_digest = hashlib.sha256(self.iso.read_bytes()).hexdigest() self.additions_digest = hashlib.sha256(self.additions.read_bytes()).hexdigest() self.info = { "VMState": "poweroff", "SATA-0-0": str(self.disk), "SATA-1-0": str(self.iso), } self.info.update({f"SATA-{port}-0": "none" for port in range(2, 30)}) self.calls: list[tuple[str, ...]] = [] self.constants = mock.patch.multiple( prepare, LAB_ROOT=self.lab, ISO_PATH=self.iso, ISO_SHA256=self.iso_digest, ADDITIONS_ISO=self.additions, ADDITIONS_SHA256=self.additions_digest, PASSWORD_DIR=self.password_dir, USER_PASSWORD_FILE=self.user_password, ROOT_PASSWORD_FILE=self.root_password, ) self.constants.start() self.addCleanup(self.constants.stop) self.guard_digest = mock.patch.object(prepare.guard, "EXPECTED_ISO_SHA256", self.iso_digest) self.guard_digest.start() self.addCleanup(self.guard_digest.stop) self.guard_inspect = mock.patch.object( prepare.guard, "inspect_target", return_value=(prepare.TARGET_UUID, self.info), ) self.guard_inspect.start() self.addCleanup(self.guard_inspect.stop) def tearDown(self): self.temp.cleanup() def runner(self, argv): args = tuple(argv) self.calls.append(args) return subprocess.CompletedProcess(args, 0, "synthetic output", "") def test_runs_dry_run_then_exact_target_prepare_without_start(self): selected = prepare.prepare_target(runner=self.runner) self.assertEqual(selected, prepare.TARGET_UUID) self.assertEqual(len(self.calls), 2) dry, actual = self.calls self.assertEqual(dry[:4], ("VBoxManage", "unattended", "install", prepare.TARGET_UUID)) self.assertIn("--dry-run", dry) self.assertNotIn("--dry-run", actual) for args in (dry, actual): self.assertIn(f"--iso={self.iso}", args) self.assertIn("--user=wga", args) self.assertIn("--hostname=wga-m9-debian137.invalid", args) self.assertIn("--locale=en_US", args) self.assertIn("--time-zone=UTC", args) self.assertIn("--country=US", args) self.assertIn("--package-selection-adjustment=minimal", args) self.assertIn("--install-additions", args) self.assertIn(f"--additions-iso={self.additions}", args) self.assertIn("--start-vm=none", args) self.assertIn(f"--user-password-file={self.user_password}", args) self.assertIn(f"--admin-password-file={self.root_password}", args) self.assertNotIn(f"--root-password-file={self.root_password}", args) self.assertFalse(any("startvm" in value for value in args)) self.assertNotIn("synthetic-user-secret", args) self.assertNotIn("synthetic-root-secret", args) def test_bad_private_credentials_refuse_before_guard_or_vbox(self): os.chmod(self.user_password, 0o644) with self.assertRaisesRegex(prepare.PrepareError, "mode 0600"): prepare.prepare_target(runner=self.runner) self.assertEqual(self.calls, []) def test_guard_rejects_extra_nonempty_sata_media_before_vbox(self): self.info["SATA-2-0"] = str(self.additions) with self.assertRaisesRegex(prepare.PrepareError, "target media"): prepare.prepare_target(runner=self.runner) self.assertEqual(self.calls, []) def test_guard_accepts_emptydrive_placeholders_for_extra_sata_slots(self): self.info["SATA-2-0"] = "emptydrive" self.info["SATA-3-0"] = "none" prepare.prepare_target(runner=self.runner) self.assertEqual(len(self.calls), 2) def test_guard_power_media_or_digest_refusal_makes_no_mutating_calls(self): failures = ( ("running", {"VMState": "running"}, "powered off"), ("media", {"SATA-1-0": str(self.additions)}, "target media"), ) for name, update, expected in failures: with self.subTest(name=name): self.info.update(update) with self.assertRaisesRegex(prepare.PrepareError, expected): prepare.prepare_target(runner=self.runner) self.assertEqual(self.calls, []) self.info.update({"VMState": "poweroff", "SATA-1-0": str(self.iso)}) with mock.patch.object(prepare, "ADDITIONS_SHA256", "0" * 64): with self.assertRaisesRegex(prepare.PrepareError, "Guest Additions ISO SHA-256"): prepare.prepare_target(runner=self.runner) self.assertEqual(self.calls, []) def test_dry_run_failure_redacts_output_and_does_not_prepare(self): secret = "SENSITIVE-PLAINTEXT-DRY-RUN" def failing(argv): self.calls.append(tuple(argv)) return subprocess.CompletedProcess(argv, 1, secret, secret) with self.assertRaises(prepare.PrepareError) as raised: prepare.prepare_target(runner=failing) self.assertNotIn(secret, str(raised.exception)) self.assertEqual(len(self.calls), 1) self.assertIn("unattended dry run", str(raised.exception)) def test_actual_failure_reports_partial_target_and_suppresses_output(self): secret = "SENSITIVE-PLAINTEXT-ACTUAL" def failing(argv): self.calls.append(tuple(argv)) code = 0 if "--dry-run" in argv else 9 return subprocess.CompletedProcess(argv, code, secret, secret) with self.assertRaises(prepare.PrepareError) as raised: prepare.prepare_target(runner=failing) self.assertNotIn(secret, str(raised.exception)) self.assertIn("partially completed", str(raised.exception)) self.assertIn(prepare.TARGET_UUID, str(raised.exception)) self.assertEqual(len(self.calls), 2) if __name__ == "__main__": unittest.main()