import importlib.util import json from pathlib import Path import subprocess import tempfile import threading import unittest import urllib.error import urllib.request class BootConsole: """Captured efibootmgr format with observable guest NVRAM mutations.""" def __init__(self, entries=None, changed_order=False, drop_next=False, changed_create_order=False): self.order = '0000,0001,0002,0003' self.next = None self.changed_order, self.drop_next = changed_order, drop_next self.changed_create_order = changed_create_order self.entries = entries if entries is not None else ( 'Boot0000* UiApp\tFvVol(example)/FvFile(example)\n' 'Boot0001* UEFI QEMU DVD-ROM QM00001 \tPciRoot(0x0)/Pci(0x1f,0x2)/Sata(0,65535,0){auto_created_boot_option}\n' 'Boot0002* UEFI Misc Device\tPciRoot(0x0)/Pci(0x3,0x0)\n' 'Boot0003* EFI Internal Shell\tFvVol(example)/FvFile(example)\n' 'Boot0004* HydraOS\tHD(1,GPT,5d24ab30-9c80-4f5d-8594-f9f08fdf31fb,0x800,0x100000)/\\EFI\\HydraOS\\grubx64.efi\n') def command(self, command, timeout): if command == 'sudo -n blkid -s PARTUUID -o value /dev/vda1': return b'5d24ab30-9c80-4f5d-8594-f9f08fdf31fb\n' if command == "sudo -n efibootmgr --create-only --disk /dev/vda --part 1 --label HydraOS --loader '\\EFI\\HydraOS\\grubx64.efi'": self.entries += ('Boot0004* HydraOS\tHD(1,GPT,5d24ab30-9c80-4f5d-8594-f9f08fdf31fb,0x800,0x100000)/\\EFI\\HydraOS\\grubx64.efi\n') if self.changed_create_order: self.order = '0004,0000,0001,0002,0003' elif command.startswith('sudo -n efibootmgr --bootnext '): if not self.drop_next: self.next = command.split()[-1] if self.changed_order: self.order = '0001,0000,0002,0003' elif command != 'sudo -n efibootmgr': raise AssertionError('Unexpected firmware mutation: ' + command) return ('BootCurrent: 0004\nTimeout: 1 seconds\nBootOrder: ' + self.order + '\n' + ('BootNext: ' + self.next + '\n' if self.next else '') + self.entries).encode() SCRIPT = Path(__file__).resolve().parents[1] / 'dev/laptop-smoke.py' class LaptopSmokeTests(unittest.TestCase): @classmethod def setUpClass(cls): spec = importlib.util.spec_from_file_location('laptop_smoke', SCRIPT) cls.mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(cls.mod) def test_cli_cannot_accept_existing_disks_or_private_signing_keys(self): parser = self.mod.argument_parser() flags = {flag for action in parser._actions for flag in action.option_strings} self.assertFalse({'--disk', '--target', '--private-key', '--existing-disk'} & flags) self.assertTrue({'--iso', '--ovmf-code', '--ovmf-vars'} <= flags) def test_invariant_check_rejects_partition_foreign_data_efi_and_bootorder_changes(self): original = {'gpt': {'partitions': [{'start': 2048, 'size': 1024}]}, 'foreign_sha256': 'data', 'foreign_efi': {'EFI/BOOT/BOOTX64.EFI': 'fallback'}, 'boot_order': '0001,0002'} self.mod.verify_preserved(original, json.loads(json.dumps(original))) for key in original: changed = dict(original, **{key: 'changed'}) with self.subTest(key=key), self.assertRaises(self.mod.SmokeError): self.mod.verify_preserved(original, changed) def test_http_serves_only_explicit_bytes_with_no_filesystem_or_traversal(self): server = self.mod.artifact_server({'/public.pem': b'TEST public', '/one/manifest.json': b'{}'}) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() self.addCleanup(server.server_close) self.addCleanup(server.shutdown) url = 'http://127.0.0.1:' + str(server.server_port) self.assertEqual(urllib.request.urlopen(url + '/public.pem').read(), b'TEST public') for path in ('/', '/one/', '/public.pem?secret=1', '/../private.pem', '/%2e%2e/private.pem'): with self.subTest(path=path), self.assertRaises(urllib.error.HTTPError): urllib.request.urlopen(url + path) def test_test_artifacts_have_real_signatures_and_no_private_key_in_share(self): from hydra.update_engine import validate_manifest, validate_archive with tempfile.TemporaryDirectory() as directory: files = self.mod.test_artifacts(Path(directory), b'# runtime fixture\n', 'a' * 40) public = Path(directory) / 'public.pem' for name in ('one', 'two'): manifest = validate_manifest(files['/' + name + '/manifest.json'], files['/' + name + '/manifest.sig'], public) validate_archive(files['/' + name + '/runtime.zip'], manifest) self.assertEqual(set(files), {'/public.pem'} | { '/' + name + '/' + file for name in ('one', 'two') for file in ('manifest.json', 'manifest.sig', 'runtime.zip')}) self.assertFalse((Path(directory) / 'TEST-private.pem').exists()) def test_every_guest_probe_compiles(self): for mode in ('seed', 'preserved', 'baseline', 'upgrade', 'base-state', 'damage', 'restored', 'runtime', 'restore-interrupt'): with self.subTest(mode=mode): compile(self.mod.guest_script(mode, {'sentinel': 'fixture', 'port': 1234, 'checkpoint': '00000000-0000-0000-0000-000000000001', 'root_uuid': '00000000-0000-0000-0000-000000000002'}), mode, 'exec') def test_installer_only_formats_selected_root_and_registration_is_opt_in(self): command = self.mod.partition_installer_command('safe123') self.assertIn('--root-partition /dev/vda3', command) self.assertIn('--esp-partition /dev/vda1', command) self.assertIn('--confirm-format /dev/vda3', command) self.assertIn('--root-fs btrfs', command) self.assertIn('--register-boot-entry', command) self.assertNotIn('--confirm-erase', command) self.assertNotIn('--disk', command) with self.assertRaises(self.mod.SmokeError): self.mod.partition_installer_command('bad;token') def test_live_boot_is_selected_once_without_changing_bootorder(self): console = BootConsole() self.mod.select_live(console) self.assertEqual(console.next, '0001') self.assertEqual(console.order, '0000,0001,0002,0003') def test_live_boot_refuses_missing_ambiguous_or_inactive_dvd_entry(self): for entries in ('Boot0001* Unrelated DVD\tPciRoot(0x0)\n', 'Boot0001* UEFI QEMU DVD-ROM QM00001 Other\tPciRoot(0x0)\n', 'Boot0001 UEFI QEMU DVD-ROM QM00001 \tPciRoot(0x0)\n', 'Boot0001* UEFI QEMU DVD-ROM QM00001 \tPciRoot(0x0)\n' 'Boot0002* UEFI QEMU DVD-ROM QM00001 \tPciRoot(0x0)\n'): console = BootConsole(entries=entries) with self.subTest(entries=entries), self.assertRaises(self.mod.SmokeError): self.mod.select_live(console) self.assertIsNone(console.next) def test_boot_selection_verifies_effective_bootnext_and_unchanged_order(self): for selector in ('select_live', 'select_installed'): for fault in ('changed_order', 'drop_next'): with self.subTest(selector=selector, fault=fault), self.assertRaises(self.mod.SmokeError): getattr(self.mod, selector)(BootConsole(**{fault: True})) def test_installed_boot_recreates_pruned_owned_entry_without_changing_order(self): console = BootConsole() console.entries = console.entries[:console.entries.index('Boot0004')] with self.assertRaises(self.mod.SmokeError): self.mod.select_installed(console) self.assertIsNone(console.next, 'First selection must prove installer registration') self.mod.select_installed(console, recreate=True) self.assertEqual(console.next, '0004') self.assertEqual(console.order, '0000,0001,0002,0003') self.assertIn('\\EFI\\HydraOS\\grubx64.efi', console.entries) def test_recreated_entry_cannot_change_bootorder(self): console = BootConsole(changed_create_order=True) console.entries = console.entries[:console.entries.index('Boot0004')] with self.assertRaises(self.mod.SmokeError): self.mod.select_installed(console, recreate=True) self.assertIsNone(console.next) def test_boot_selection_rejects_correct_label_with_foreign_device_or_loader(self): for selector, before, after in ( ('select_live', 'Sata(0,65535,0)', 'Sata(2,65535,0)'), ('select_installed', '5d24ab30-9c80-4f5d-8594-f9f08fdf31fb', 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'), ('select_installed', '\\EFI\\HydraOS\\grubx64.efi', '\\EFI\\ForeignOS\\grubx64.efi')): console = BootConsole() console.entries = console.entries.replace(before, after) with self.subTest(selector=selector, wrong=after), self.assertRaises(self.mod.SmokeError): getattr(self.mod, selector)(console) self.assertIsNone(console.next) def test_base_state_compares_kernel_bytes_without_including_mounted_esp(self): # Offline /boot/efi is an empty mountpoint; on disk-only boot it contains # the mounted ESP. That mount must not produce a false base mismatch. namespace = {} exec(self.mod.GUEST_COMMON, namespace) with tempfile.TemporaryDirectory() as directory: root = Path(directory) / 'root' esp = Path(directory) / 'esp' for path, data in [('boot/vmlinuz-fixture', 'kernel'), ('boot/efi/EFI/BOOT/BOOTX64.EFI', 'mounted foreign fallback'), ('var/db/pkg/app-misc/hydra-acceptance-1.0/CONTENTS', 'package'), ('var/lib/portage/world', 'app-misc/hydra-acceptance'), ('usr/share/hydra/base-fixture.txt', '1.0')]: target = root / path target.parent.mkdir(parents=True, exist_ok=True) target.write_text(data) target = esp / 'EFI/HydraOS/grubx64.efi' target.parent.mkdir(parents=True) target.write_text('owned loader') result = namespace['base_state'](root, esp) self.assertEqual(set(result['boot']), {'vmlinuz-fixture'}) (root / 'boot/vmlinuz-fixture').write_text('damaged') self.assertNotEqual(namespace['base_state'](root, esp)['boot'], result['boot']) def test_real_portage_install_is_executed_in_guest_network_namespace(self): import ast function = next(node for node in ast.parse(self.mod.GUEST_COMMON).body if isinstance(node, ast.FunctionDef) and node.name == 'fixture') commands = [node for node in ast.walk(function) if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == 'run'] emerge = next(node for node in commands if any(isinstance(arg, ast.Constant) and arg.value == 'emerge' for arg in node.args)) constants = [arg.value for arg in emerge.args if isinstance(arg, ast.Constant)] self.assertEqual(constants[:3], ['unshare', '--net', '--']) self.assertNotIn('--offline', constants, 'emerge has no --offline option') self.assertIn('--ignore-default-opts', constants) self.assertIn('--nodeps', constants) def test_guest_mount_explicitly_selects_known_filesystem(self): namespace = {} exec(self.mod.GUEST_COMMON, namespace) calls = [] namespace['run'] = lambda *args: calls.append(args) for device, filesystem, options in (('/dev/vda1', 'vfat', 'ro'), ('/dev/vda2', 'ext4', 'ro'), ('/dev/vda3', 'btrfs', 'ro,subvol=@root')): with self.subTest(device=device): with namespace['mounted'](device, filesystem, options) as directory: self.assertEqual(calls[-1], ('mount', '-t', filesystem, '-o', options, device, str(directory))) self.assertEqual(calls[-1], ('umount', str(directory))) def test_every_guest_mount_call_names_the_partition_filesystem(self): import ast expected = {'/dev/vda1': 'vfat', '/dev/vda2': 'ext4', '/dev/vda3': 'btrfs'} sources = [self.mod.GUEST_COMMON, *self.mod.GUEST_BODIES.values()] for source in sources: for node in ast.walk(ast.parse(source)): if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id == 'mounted': with self.subTest(call=ast.unparse(node)): self.assertGreaterEqual(len(node.args), 2) self.assertEqual(ast.literal_eval(node.args[1]), expected[ast.literal_eval(node.args[0])])