176 lines
7.6 KiB
Python
176 lines
7.6 KiB
Python
"""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()
|