#!/usr/bin/env python3 """Real laptop acceptance in a NEW disposable UEFI/KVM VM; retains scratch evidence. Only ISO/firmware inputs are accepted. Never attach host devices or existing VM images. A private HTTP server exposes exact TEST artifact bytes, never a directory. """ from __future__ import annotations import argparse import hashlib import http.server import importlib.util import io import json import os from pathlib import Path import re import secrets import shlex import shutil import stat import subprocess import tempfile import threading import time import zipfile _ROOT = Path(__file__).resolve().parents[1] _spec = importlib.util.spec_from_file_location('hydra_iso_smoke', _ROOT / 'dev/iso-smoke.py') iso_smoke = importlib.util.module_from_spec(_spec) _spec.loader.exec_module(iso_smoke) SmokeError = iso_smoke.SmokeError def argument_parser(): parser = argparse.ArgumentParser(description=__doc__) for option in ('iso', 'ovmf-code', 'ovmf-vars'): parser.add_argument('--' + option, required=True, type=Path) parser.add_argument('--qemu', default='qemu-system-x86_64') parser.add_argument('--qemu-img', default='qemu-img') parser.add_argument('--scratch-parent', type=Path, default=Path(tempfile.gettempdir())) parser.add_argument('--memory', type=int, default=4096) parser.add_argument('--boot-timeout', type=int, default=360) parser.add_argument('--install-timeout', type=int, default=1800) return parser def verify_preserved(before, after, *, firmware_reboot=False): fields = ('gpt', 'foreign_sha256', 'foreign_efi') if not firmware_reboot: fields += ('boot_order',) for name in fields: if before[name] != after[name]: raise SmokeError('Preserved installation changed ' + name) def partition_installer_command(token): if not re.fullmatch('[a-zA-Z0-9]+', token): raise SmokeError('Invalid installer token') return ('sudo -n hydra-install --root-partition /dev/vda3 --esp-partition /dev/vda1 ' '--root-fs btrfs --user hydratest --execute --confirm-format /dev/vda3 ' '--register-boot-entry; rc=$?; ' f"printf '\\036{token}:INSTALL_EXIT:%s\\037' \"$rc\"") def artifact_server(files): payload = dict(files) class Handler(http.server.BaseHTTPRequestHandler): def do_GET(self): data = payload.get(self.path) if data is None: self.send_error(404) return self.send_response(200) self.send_header('Content-Type', 'application/octet-stream') self.send_header('Content-Length', str(len(data))) self.end_headers() self.wfile.write(data) def log_message(self, *args): pass return http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler) def test_artifacts(scratch, runtime, commit): """Generate disposable TEST trust; never include private bytes in the share.""" private = scratch / 'TEST-private.pem' public = scratch / 'public.pem' subprocess.run(['openssl', 'genpkey', '-algorithm', 'ED25519', '-out', str(private)], check=True, capture_output=True, timeout=15) private.chmod(0o600) shared = {} try: subprocess.run(['openssl', 'pkey', '-in', str(private), '-pubout', '-out', str(public)], check=True, capture_output=True, timeout=15) shared['/public.pem'] = public.read_bytes() for sequence, name in ((1, 'one'), (2, 'two')): path = 'usr/lib/hydra/hydra/cli.py' content = runtime + ('\n# Disposable laptop acceptance TEST sequence ' + str(sequence) + '\n').encode() stream = io.BytesIO() with zipfile.ZipFile(stream, 'w', zipfile.ZIP_DEFLATED) as archive: info = zipfile.ZipInfo(path, (1980, 1, 1, 0, 0, 0)) info.create_system = 3 info.external_attr = (stat.S_IFREG | 0o644) << 16 info.compress_type = zipfile.ZIP_DEFLATED archive.writestr(info, content) payload = stream.getvalue() digest = lambda data: hashlib.sha256(data).hexdigest() manifest = {'format': 1, 'product': 'Hydra OS', 'scope': 'runtime', 'channel': 'stable', 'architecture': 'x86_64', 'sequence': sequence, 'version': '0.0.0-test-' + str(sequence), 'source_commit': commit, 'archive': {'name': 'runtime.zip', 'size': len(payload), 'sha256': digest(payload)}, 'files': {path: {'size': len(content), 'sha256': digest(content), 'mode': 0o644}}} raw = (json.dumps(manifest, sort_keys=True, separators=(',', ':')) + '\n').encode() input_path, signature = scratch / 'TEST-manifest.json', scratch / 'TEST-manifest.sig' input_path.write_bytes(raw) subprocess.run(['openssl', 'pkeyutl', '-sign', '-rawin', '-inkey', str(private), '-in', str(input_path), '-out', str(signature)], check=True, capture_output=True, timeout=15) for file, data in (('manifest.json', raw), ('manifest.sig', signature.read_bytes()), ('runtime.zip', payload)): shared['/' + name + '/' + file] = data input_path.unlink(); signature.unlink() finally: private.unlink(missing_ok=True) return shared GUEST_COMMON = r''' import contextlib, hashlib, json, os, re, runpy, shutil, signal, subprocess, sys, time, urllib.request from pathlib import Path def run(*args, timeout=120): p = subprocess.run(args, capture_output=True, text=True, timeout=timeout) if p.returncode: raise RuntimeError('command failed: ' + repr(args) + '\n' + p.stdout + p.stderr) return p.stdout def sha(path): h = hashlib.sha256() with open(path, 'rb') as source: for block in iter(lambda: source.read(1024 * 1024), b''): h.update(block) return h.hexdigest() def tree(path): return {str(p.relative_to(path)): sha(p) for p in sorted(path.rglob('*')) if p.is_file()} EFI_GUID = '8be4df61-93ca-11d2-aa0d-00e098032b8c' DVD_LABEL = 'UEFI QEMU DVD-ROM QM00001' def efi_description(data): # The file starts with efivarfs attributes, EFI_LOAD_OPTION attributes and # FilePathListLength, followed by a bounded UTF-16 description. for end in range(10, min(len(data) - 1, 522), 2): if data[end:end + 2] == b'\0\0': try: return data[10:end].decode('utf-16-le').strip(), end + 2 except UnicodeDecodeError: break return None, None def validate_dvd_variable(data): label, offset = efi_description(data) path = bytes.fromhex('02010c00d041030a0000000001010600021f03120a000000ffff00007fff0400') automatic = bytes.fromhex('4eac0881119f594d850ee21a522c59b2') if (len(data) > 512 or data[:10] != bytes.fromhex('07000000010000002000') or label != DVD_LABEL or data[offset:offset + 32] != path or data[offset + 32:] not in (b'', automatic)): raise RuntimeError('Invalid saved SATA0 DVD EFI load option') def dvd_variables(variables): matches = [] for path in variables.glob('Boot????-' + EFI_GUID): if not re.fullmatch('Boot[0-9A-Fa-f]{4}-' + EFI_GUID, path.name): continue data = path.read_bytes() if efi_description(data)[0] == DVD_LABEL: validate_dvd_variable(data) matches.append((path.name[4:8], data)) if len(matches) > 1: raise RuntimeError('Ambiguous DVD EFI load options') return matches def capture_dvd_variable(variables): matches = dvd_variables(variables) if len(matches) != 1: raise RuntimeError('Expected one active SATA0 DVD EFI load option') entry, data = matches[0] current = (variables / ('BootCurrent-' + EFI_GUID)).read_bytes() if len(current) != 6 or int.from_bytes(current[4:], 'little') != int(entry, 16): raise RuntimeError('Initial live system did not boot from the captured DVD entry') return {'entry': entry, 'raw_hex': data.hex(), 'sha256': hashlib.sha256(data).hexdigest()} def restore_dvd_variable(variables, data): validate_dvd_variable(data) order = variables / ('BootOrder-' + EFI_GUID) before = order.read_bytes() matches = dvd_variables(variables) created = not matches if matches: entry, existing = matches[0] if existing != data: raise RuntimeError('Existing DVD option differs from captured bytes') else: occupied = {int(p.name[4:8], 16) for p in variables.glob('Boot????-' + EFI_GUID) if re.fullmatch('Boot[0-9A-Fa-f]{4}-' + EFI_GUID, p.name)} free = next((number for number in range(65536) if number not in occupied), None) if free is None: raise RuntimeError('No unused UEFI boot variable') entry = f'{free:04X}' path = variables / ('Boot' + entry + '-' + EFI_GUID) # Never overwrite an existing entry, including a concurrently created one. fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) with os.fdopen(fd, 'wb', buffering=0) as target: if target.write(data) != len(data): raise RuntimeError('Short EFI variable write') if path.read_bytes() != data: raise RuntimeError('Saved DVD variable readback mismatch') if order.read_bytes() != before: raise RuntimeError('DVD variable creation changed BootOrder') return {'entry': entry, 'created': created, 'sha256': hashlib.sha256(data).hexdigest(), 'boot_order_sha256': hashlib.sha256(before).hexdigest()} @contextlib.contextmanager def mounted(device, filesystem, options=''): import tempfile with tempfile.TemporaryDirectory(prefix='hydra-acceptance-') as directory: args = ['mount', '-t', filesystem] + (['-o', options] if options else []) + [device, directory] run(*args) try: yield Path(directory) finally: run('umount', directory) def installed_esp(marker): mounts = json.loads(run('findmnt', '--json', '--mountpoint', '/boot/efi', '--output', 'SOURCE,TARGET,FSTYPE,UUID,MAJ:MIN'))['filesystems'] devices = json.loads(run('lsblk', '--json', '--nodeps', '--output', 'NAME,MAJ:MIN,FSTYPE,UUID', '/dev/vda1'))['blockdevices'] if len(mounts) != 1 or len(devices) != 1: raise RuntimeError('Expected one mounted installed ESP and one test ESP device') mount, device = mounts[0], devices[0] if (mount.get('source') != '/dev/vda1' or mount.get('target') != '/boot/efi' or mount.get('fstype') != 'vfat' or device.get('name') != 'vda1' or device.get('fstype') != 'vfat' or not device.get('maj:min') or mount.get('maj:min') != device['maj:min'] or not device.get('uuid') or mount.get('uuid') != device['uuid'] or marker.get('esp_uuid') != device['uuid']): raise RuntimeError('Installed ESP mount/device identity does not match the storage marker') return Path('/boot/efi') @contextlib.contextmanager def readonly_esp(): if Path('/etc/hydra/live-mode/active').exists(): with mounted('/dev/vda1', 'vfat', 'ro') as esp: yield esp else: # The installed ESP is already mounted rw by fstab. Verify it, then only # read hashes; a second ro mount would conflict with its superblock state. yield installed_esp(json.loads(Path('/etc/hydra/storage.json').read_text())) def preserved(): table = json.loads(run('sfdisk', '--json', '/dev/vda'))['partitiontable'] # No filesystem UUIDs: selected root UUID must change on formatting. with readonly_esp() as esp: foreign = {k: v for k, v in tree(esp / 'EFI').items() if not k.startswith('HydraOS/')} order = re.search(r'^BootOrder:\s*(.*)$', run('efibootmgr'), re.M) assert order, 'UEFI BootOrder was not observable' return {'gpt': table, 'foreign_sha256': sha('/dev/vda2'), 'foreign_efi': foreign, 'boot_order': order.group(1) if order else None} def base_state(root, esp): pkg = root / 'var/db/pkg/app-misc' records = {p.name: tree(p) for p in sorted(pkg.glob('hydra-acceptance-*')) if p.is_dir()} return {'boot': {k: v for k, v in tree(root / 'boot').items() if not k.startswith('efi/')}, 'portage_fixture': records, 'world': sha(root / 'var/lib/portage/world'), 'owned_efi': tree(esp / 'EFI/HydraOS'), 'fixture': (root / 'usr/share/hydra/base-fixture.txt').read_text()} def fixture(version): repo = Path('/var/lib/hydra-acceptance-repo') (repo / 'profiles').mkdir(parents=True, exist_ok=True) (repo / 'metadata').mkdir(exist_ok=True) (repo / 'profiles/repo_name').write_text('hydra-acceptance\n') (repo / 'metadata/layout.conf').write_text('masters =\nauto-sync = false\nthin-manifests = true\n') conf = Path('/etc/portage/repos.conf') conf.mkdir(exist_ok=True) (conf / 'hydra-acceptance.conf').write_text('[hydra-acceptance]\nlocation = ' + str(repo) + '\nauto-sync = no\n') package = repo / 'app-misc/hydra-acceptance' package.mkdir(parents=True, exist_ok=True) ebuild = package / ('hydra-acceptance-' + version + '.ebuild') ebuild.write_text('EAPI=8\nDESCRIPTION="Offline recovery acceptance fixture"\nHOMEPAGE="https://example.invalid"\nLICENSE="MIT"\nSLOT="0"\nKEYWORDS="amd64"\nS="${WORKDIR}"\nsrc_install() {\n insinto /usr/share/hydra\n printf "%s\\n" "${PV}" > base-fixture.txt\n doins base-fixture.txt\n}\n') run('unshare', '--net', '--', 'ebuild', str(ebuild), 'manifest') output = run('unshare', '--net', '--', 'emerge', '--ignore-default-opts', '--nodeps', '--autounmask=n', '--usepkg=n', '--buildpkg=n', '--getbinpkg=n', '=app-misc/hydra-acceptance-' + version, timeout=600) assert Path('/usr/share/hydra/base-fixture.txt').read_text().strip() == version assert Path('/var/db/pkg/app-misc/hydra-acceptance-' + version).is_dir() return output def stop_child(source): ready = Path('/run/hydra-acceptance-child-ready') ready.unlink(missing_ok=True) process = subprocess.Popen([sys.executable, '-I', '-c', source], stdout=subprocess.PIPE, stderr=subprocess.PIPE) try: deadline = time.monotonic() + 120 while not ready.exists() and process.poll() is None and time.monotonic() < deadline: time.sleep(.05) if not ready.exists(): raise RuntimeError('child never reached durable mutation: ' + repr(process.communicate(timeout=5))) process.kill() process.communicate(timeout=10) assert process.returncode == -signal.SIGKILL finally: if process.poll() is None: process.kill(); process.communicate(timeout=10) ready.unlink(missing_ok=True) ''' GUEST_BODIES = { 'capture-live-boot': "result = capture_dvd_variable(Path('/sys/firmware/efi/efivars'))\n", 'restore-live-boot': "result = restore_dvd_variable(Path('/sys/firmware/efi/efivars'), bytes.fromhex(context['dvd_variable']))\n", 'seed': r''' # Guest-only target is the new empty disk attached by this harness. assert os.geteuid() == 0 and Path('/etc/hydra/live-mode/active').exists() assert not Path('/sys/class/block/vda/partition').exists() assert json.loads(run('lsblk', '-J', '-o', 'NAME,TYPE,MOUNTPOINT', '/dev/vda'))['blockdevices'][0].get('children') is None run('parted', '-s', '/dev/vda', 'mklabel', 'gpt', 'mkpart', 'ESP', 'fat32', '1MiB', '513MiB', 'set', '1', 'esp', 'on', 'mkpart', 'foreign', '513MiB', '577MiB', 'mkpart', 'HydraRoot', '577MiB', '100%') run('udevadm', 'settle') run('mkfs.vfat', '-F', '32', '/dev/vda1') with open('/dev/vda2', 'r+b', buffering=0) as foreign: foreign.write((context['sentinel'] * 1024).encode()); os.fsync(foreign.fileno()) with mounted('/dev/vda1', 'vfat') as esp: for name in ('EFI/ForeignOS/loader.efi', 'EFI/BOOT/BOOTX64.EFI'): path = esp / name; path.parent.mkdir(parents=True, exist_ok=True) path.write_text(context['sentinel'] + ':' + name) run('sync') result = preserved() ''', 'preserved': 'result = preserved()\n', 'baseline': r''' assert not Path('/etc/hydra/live-mode/active').exists() output = fixture('1.0') home = Path('/home/hydratest') (home / 'acceptance-home').write_text('before-' + context['sentinel']) vm = home / '.local/share/hydra/acceptance-guest' vm.mkdir(parents=True, exist_ok=True) run('qemu-img', 'create', '-f', 'qcow2', str(vm / 'disk.qcow2'), '64M') (vm / 'guest-data-sentinel').write_text('VM-DATA-' + context['sentinel']) (vm / 'profile.json').write_text('{"fixture":"preserve"}\n') run('chown', '-R', 'hydratest:hydratest', str(vm)) run('chown', 'hydratest:hydratest', str(home / 'acceptance-home')) result = {'fixture_emerge': output, 'vm_data': tree(vm), 'doctor': json.loads(run('hydra', 'doctor', '--json'))} ''', 'upgrade': r''' output = fixture('2.0') Path('/home/hydratest/acceptance-home').write_text('current-' + context['sentinel']) with Path('/var/lib/portage/world').open('a') as world: world.write('# acceptance post-checkpoint world\n') result = {'fixture_emerge': output, 'home': Path('/home/hydratest/acceptance-home').read_text()} ''', 'base-state': r''' with mounted('/dev/vda3', 'btrfs', 'ro,subvol=@root') as root: with mounted('/dev/vda1', 'vfat', 'ro') as esp: result = base_state(root, esp) ''', 'damage': r''' with mounted('/dev/vda3', 'btrfs', 'subvol=@root') as root: kernels = list((root / 'boot').glob('vmlinuz*')) assert kernels, 'No actual installed kernel to damage' for kernel in kernels: kernel.write_bytes(b'DELIBERATE ACCEPTANCE KERNEL DAMAGE\n') with mounted('/dev/vda1', 'vfat') as esp: loader = esp / 'EFI/HydraOS/grubx64.efi' assert loader.is_file(); loader.write_bytes(b'DELIBERATE ACCEPTANCE EFI DAMAGE\n') run('sync') result = {'damaged_kernels': [p.name for p in kernels], 'damaged_owned_loader': 'EFI/HydraOS/grubx64.efi'} ''', 'restored': r''' marker = json.loads(Path('/etc/hydra/storage.json').read_text()) assert marker['root_subvolume'] == '@root' assert Path('/home/hydratest/acceptance-home').read_text() == 'current-' + context['sentinel'] assert not Path('/var/db/pkg/app-misc/hydra-acceptance-2.0').exists() assert Path('/usr/share/hydra/base-fixture.txt').read_text().strip() == '1.0' with readonly_esp() as esp: result = base_state(Path('/'), esp) result['home'] = Path('/home/hydratest/acceptance-home').read_text() result['vm_data'] = tree(Path('/home/hydratest/.local/share/hydra/acceptance-guest')) result['root_mount'] = run('findmnt', '-n', '-o', 'SOURCE,OPTIONS', '/') ''', 'runtime': r''' base = Path('/var/lib/hydra-acceptance-runtime') base.mkdir(mode=0o700) url = 'http://10.0.2.2:' + str(context['port']) def download(path): with urllib.request.urlopen(url + path, timeout=15) as response: return response.read(16 * 1024 * 1024) for name in ('one', 'two'): directory = base / name; directory.mkdir() for file in ('manifest.json', 'manifest.sig', 'runtime.zip'): (directory / file).write_bytes(download('/' + name + '/' + file)) target = Path('/usr/lib/hydra/hydra/cli.py') original = target.read_bytes() key = Path('/etc/hydra/update.pub'); production_key = key.read_bytes() vm = Path('/home/hydratest/.local/share/hydra/acceptance-guest'); original_vm = tree(vm) key.write_bytes(download('/public.pem')); key.chmod(0o644) try: checked = json.loads(run('hydra', 'update', 'check', '--artifacts', str(base / 'one'))) assert checked['available']['sequence'] == 1 applied = json.loads(run('hydra', 'update', 'apply', '--artifacts', str(base / 'one'))) assert target.read_bytes() != original rollback = json.loads(run('/usr/bin/hydra-update', 'rollback', applied['transaction'])) assert target.read_bytes() == original tamper = base / 'tampered'; shutil.copytree(base / 'two', tamper) with (tamper / 'runtime.zip').open('ab') as file: file.write(b'TAMPER') prior_status = run('/usr/bin/hydra-update', 'status') refused = subprocess.run(['/usr/bin/hydra-update', 'apply', '--artifacts', str(tamper)], capture_output=True, text=True) assert refused.returncode != 0 and target.read_bytes() == original assert run('/usr/bin/hydra-update', 'status') == prior_status source = """import os, runpy, signal from pathlib import Path m = runpy.run_path('/usr/lib/hydra-update/engine.py') write = m['atomic_write'] def pause(path, *args, **kwargs): write(path, *args, **kwargs) if path == Path('/usr/lib/hydra/hydra/cli.py'): Path('/run/hydra-acceptance-child-ready').touch() os.kill(os.getpid(), signal.SIGSTOP) m['Updater'].apply.__globals__['atomic_write'] = pause m['Updater']().apply(Path(ARTIFACTS)) """.replace('ARTIFACTS', repr(str(base / 'two'))) stop_child(source) assert json.loads(run('/usr/bin/hydra-update', 'status'))['pending'] target.write_bytes(b'DELIBERATELY INVALID PYTHON !!!\n') recovered = json.loads(run('/usr/bin/hydra-update-recover')) assert target.read_bytes() == original status = json.loads(run('/usr/bin/hydra-update', 'status')) assert not status['pending'] and status['highest_sequence'] == 2 assert tree(vm) == original_vm run('hydra', 'doctor', '--json') result = {'signed_check': checked, 'apply': applied, 'rollback': rollback, 'tamper_refusal': refused.stderr, 'fresh_process_recovery': recovered, 'status': status, 'unchanged_vm_data': original_vm, 'trust': 'disposable TEST key only'} finally: key.write_bytes(production_key); key.chmod(0o644) ''', 'restore-interrupt': r''' # Instrument a real live CLI process, after its durable journal and real rename. # No fake filesystems or production fault hooks are involved. source = """import os, signal, sys from pathlib import Path sys.path.insert(0, '/usr/lib/hydra') from hydra import system_recovery as recovery save = recovery.RecoverySession.save_journal def pause(self, value, phase): save(self, value, phase) if phase == 'root-retained': Path('/run/hydra-acceptance-child-ready').touch() os.kill(os.getpid(), signal.SIGSTOP) recovery.RecoverySession.save_journal = pause raise SystemExit(recovery.main(ARGUMENTS)) """.replace('ARGUMENTS', repr(['restore', '--root-uuid', context['root_uuid'], '--checkpoint', context['checkpoint'], '--execute', '--confirm-root-uuid', context['root_uuid']])) stop_child(source) with mounted('/dev/vda3', 'btrfs', 'ro,subvolid=5') as top: journal = json.loads((top / '@snapshots/restore.json').read_text()) assert journal['phase'] == 'root-retained' assert not (top / '@root').exists() and (top / journal['failed_root']).exists() result = {'sigkill_after_real_rename': True, 'durable_journal': journal} ''', } def guest_script(mode, context): return GUEST_COMMON + '\ncontext = ' + repr(context) + '\n' + GUEST_BODIES[mode] + '\nprint(json.dumps(result, sort_keys=True))\n' def probe(console, mode, context, timeout=180): source = guest_script(mode, context) return json.loads(console.command("sudo -n python3 - <<'HYDRA_LAPTOP_PROBE'\n" + source + '\nHYDRA_LAPTOP_PROBE', timeout)) def select_boot_entry(console, label, expected_path, create=None): # Harness-only one-shot selection: never rewrite BootOrder or firmware VARS. output = console.command('sudo -n efibootmgr', 30).decode() before = re.findall(r'^BootOrder:\s*([^\r\n]+)', output, re.M) if len(before) != 1: raise SmokeError('Expected one observable UEFI BootOrder: ' + output) pattern = r'^Boot([0-9A-Fa-f]{4})(\*?)[ \t]+' + re.escape(label) + r' *\t([^\r\n]+)' entries = re.findall(pattern, output, re.M) if not entries and create: # OVMF may prune create-only entries that are outside BootOrder after # consuming BootNext. Recreate only our known loader; never reorder it. console.command(create, 30) output = console.command('sudo -n efibootmgr', 30).decode() if re.findall(r'^BootOrder:\s*([^\r\n]+)', output, re.M) != before: raise SmokeError('Create-only registration changed UEFI BootOrder') entries = re.findall(pattern, output, re.M) if len(entries) != 1 or entries[0][1] != '*': raise SmokeError('Expected one active ' + label + ' entry: ' + output) entry, _, path = entries[0] if not re.fullmatch(expected_path, path, re.I): raise SmokeError('Unexpected ' + label + ' boot device or loader: ' + path) console.command('sudo -n efibootmgr --bootnext ' + entry, 30) after = console.command('sudo -n efibootmgr', 30).decode() if re.findall(r'^BootOrder:\s*([^\r\n]+)', after, re.M) != before: raise SmokeError('BootNext selection changed UEFI BootOrder') if re.findall(r'^BootNext:\s*([0-9A-Fa-f]{4})\s*$', after, re.M) != [entry]: raise SmokeError('UEFI did not retain the requested BootNext') return entry def select_installed(console, *, recreate=False): # Deliberate action AFTER proving the installer preserved the original order. esp = console.command('sudo -n blkid -s PARTUUID -o value /dev/vda1', 30).decode().strip() if not re.fullmatch(r'[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}', esp): raise SmokeError('Expected GPT identity of the disposable ESP') path = re.escape('HD(1,GPT,' + esp + ',0x800,0x100000)/\\EFI\\HydraOS\\grubx64.efi') create = ("sudo -n efibootmgr --create-only --disk /dev/vda --part 1 " "--label HydraOS --loader '\\EFI\\HydraOS\\grubx64.efi'") return select_boot_entry(console, 'HydraOS', path, create if recreate else None) def select_live(console): path = re.escape('PciRoot(0x0)/Pci(0x1f,0x2)/Sata(0,65535,0)') + r'(?:\{auto_created_boot_option\})?' return select_boot_entry(console, 'UEFI QEMU DVD-ROM QM00001', path) def main(argv=None): parser = argument_parser(); args = parser.parse_args(argv) for path in (args.iso, args.ovmf_code, args.ovmf_vars): if not path.is_file(): parser.error('Missing regular input: ' + str(path)) if args.memory < 2048 or min(args.boot_timeout, args.install_timeout) <= 0: parser.error('Memory >=2048 MiB and positive timeouts required') args.keep_cdrom_drive = True args.iso, args.ovmf_code, args.ovmf_vars = (p.resolve() for p in (args.iso, args.ovmf_code, args.ovmf_vars)) scratch = Path(tempfile.mkdtemp(prefix='hydra-laptop-smoke-', dir=args.scratch_parent)) if len(str(scratch / 'serial.sock').encode()) >= 104: parser.error('Use shorter scratch parent such as /tmp') print('Acceptance artifacts: ' + str(scratch), flush=True) inputs = {str(p): iso_smoke.digest(p) for p in (args.iso, args.ovmf_code, args.ovmf_vars)} report = {'schema': 1, 'status': 'running', 'inputs_sha256': inputs, 'checks': {}, 'physical_hardware': 'pending', 'kernel_scope': 'exact kernel restoration; no two-kernel upgrade claim', 'trust_scope': 'disposable TEST key temporarily pinned only in this disposable installed guest', 'firmware_scope': 'BootOrder checked around each installer/checkpoint/restore/selector operation; firmware may prune empty-media entries between boots'} context = {'sentinel': secrets.token_hex(32)} password = secrets.token_urlsafe(24); vm = None; server = None def record(name, evidence): report['checks'][name] = evidence (scratch / 'report.json').write_text(json.dumps(report, indent=2) + '\n') print('Passed: ' + name, flush=True) def boot(live, phase): nonlocal vm vm = iso_smoke.VM(args, scratch, disk, args.iso if live else None, password, phase) console = vm.console if live: console.expect(rb'hydra-live[^\r\n]*[#$] ', args.boot_timeout) # Keep asynchronous mount diagnostics out of framed JSON on ttyS0; # they remain available in the guest's dmesg/journal buffers. console.command('sudo -n dmesg --console-level 3', 30) else: console.expect(rb'login: ', args.boot_timeout); console.send('hydratest\n') console.expect(rb'Password: ', 30); console.send(password + '\n') console.expect(rb'hydratest[^\r\n]*[#$] ', 60) iso_smoke.installed_admin(console, password) return console def shutdown(console): nonlocal vm console.send('sudo -n poweroff\n'); vm.wait_poweroff(90); vm.close(); vm = None try: disk = scratch / 'fresh-laptop.qcow2' iso_smoke.reserve_disk(disk) subprocess.run([args.qemu_img, 'create', '-f', 'qcow2', str(disk), '40G'], check=True, timeout=30, stdout=subprocess.DEVNULL) # Both code and variables copied; original firmware never attached writable. for source, name in ((args.ovmf_code, 'OVMF_CODE.fd'), (args.ovmf_vars, 'OVMF_VARS.fd')): shutil.copyfile(source, scratch / name); (scratch / name).chmod(0o600) args.ovmf_code = scratch / 'OVMF_CODE.fd' commit = subprocess.run(['git', '-C', str(_ROOT), 'rev-parse', 'HEAD'], check=True, capture_output=True, text=True).stdout.strip() files = test_artifacts(scratch, (_ROOT / 'hydra/cli.py').read_bytes(), commit) server = artifact_server(files); context['port'] = server.server_port threading.Thread(target=server.serve_forever, daemon=True).start() console = boot(True, 'live-install') before = probe(console, 'seed', context); record('seeded_foreign_invariants', before) live_boot = probe(console, 'capture-live-boot', context) context['dvd_variable'] = live_boot['raw_hex']; record('captured_live_boot_option', live_boot) token = secrets.token_hex(12); console.buffer = b'' console.send(partition_installer_command(token) + '\n') console.expect_installer(rb'Installed account password: ', 120, token); console.send(password + '\n') console.expect_installer(rb'Repeat password: ', 30, token); console.send(password + '\n') console.expect_installer(rb'Installation completed\. Diagnostic log: [^\r\n]+', args.install_timeout, token) console.finish_installer(token) after = probe(console, 'preserved', context); verify_preserved(before, after) record('preserved_partition_install', after) record('harness_bootnext_selection', select_installed(console)); shutdown(console) console = boot(False, 'installed-baseline') # The existing UX probe runs as the normal account. console.send('exit\n'); console.expect(rb'hydratest[^\r\n]*[#$] ', 30) record('packaged_ascii_branding', json.loads(iso_smoke.installed_terminal_ux_probe(console))) iso_smoke.installed_admin(console, password) baseline = probe(console, 'baseline', context, 720); record('real_portage_baseline_doctor_vmdata', baseline) record('signed_runtime_update_recovery', probe(console, 'runtime', context, 300)) record('harness_checkpoint_dvd_option', probe(console, 'restore-live-boot', context)) record('harness_checkpoint_bootnext', select_live(console)); shutdown(console) console = boot(True, 'live-checkpoint') checkpoint_preserved = probe(console, 'preserved', context) verify_preserved(before, checkpoint_preserved, firmware_reboot=True) base_before = probe(console, 'base-state', context); record('base_before_checkpoint', base_before) context['root_uuid'] = console.command('sudo -n blkid -s UUID -o value /dev/vda3').decode().strip() esp_partuuid = console.command('sudo -n blkid -s PARTUUID -o value /dev/vda1').decode().strip() command = ('sudo -n hydra-system checkpoint --root-uuid ' + shlex.quote(context['root_uuid']) + ' --esp-partuuid ' + shlex.quote(esp_partuuid) + ' --name laptop-before-upgrade') record('checkpoint_plan', json.loads(console.command(command, 180))) cp = json.loads(console.command(command + ' --execute --confirm-root-uuid ' + shlex.quote(context['root_uuid']), 300)) context['checkpoint'] = cp['id']; record('real_offline_checkpoint', cp) record('checkpoint_list', json.loads(console.command('sudo -n hydra-system list --root-uuid ' + shlex.quote(context['root_uuid']), 180))) checkpoint_after = probe(console, 'preserved', context) verify_preserved(checkpoint_preserved, checkpoint_after) record('checkpoint_preserved_firmware_and_data', checkpoint_after) select_installed(console, recreate=True); shutdown(console) console = boot(False, 'installed-upgrade'); record('real_portage_upgrade_current_home', probe(console, 'upgrade', context, 720)) record('harness_restore_dvd_option', probe(console, 'restore-live-boot', context)) record('harness_restore_bootnext', select_live(console)); shutdown(console) console = boot(True, 'live-restore') restore_preserved = probe(console, 'preserved', context) verify_preserved(before, restore_preserved, firmware_reboot=True) record('deliberate_kernel_efi_damage', probe(console, 'damage', context)) record('restore_sigkill_real_btrfs', probe(console, 'restore-interrupt', context, 300)) restore = ('sudo -n hydra-system restore --root-uuid ' + shlex.quote(context['root_uuid']) + ' --checkpoint ' + shlex.quote(context['checkpoint']) + ' --execute --confirm-root-uuid ' + shlex.quote(context['root_uuid'])) record('fresh_process_restore_resume', json.loads(console.command(restore, 300))) base_after = probe(console, 'base-state', context) if base_after != base_before: raise SmokeError('Offline restore did not restore exact base/Portage/kernel/ownedEFI hashes') preserved_after = probe(console, 'preserved', context) verify_preserved(restore_preserved, preserved_after) verify_preserved(before, preserved_after, firmware_reboot=True) record('restored_base_foreign_invariants', {'base': base_after, 'preserved': preserved_after}) select_installed(console, recreate=True); shutdown(console) console = boot(False, 'installed-restored'); restored = probe(console, 'restored', context) if any(restored[k] != v for k, v in base_before.items()): raise SmokeError('Disk-only boot did not retain restored base') if restored['vm_data'] != baseline['vm_data']: raise SmokeError('VM disk/profile bytes changed') record('disk_only_restored_boot_current_home_vmdata', restored) final_preserved = probe(console, 'preserved', context) verify_preserved(before, final_preserved, firmware_reboot=True) record('final_disk_boot_foreign_invariants', final_preserved); shutdown(console) if any(iso_smoke.digest(Path(path)) != digest for path, digest in inputs.items()): raise SmokeError('Input ISO/firmware changed') report['status'] = 'passed'; return 0 except (SmokeError, OSError, ValueError, subprocess.SubprocessError, iso_smoke.QMPError) as exc: report['status'] = 'failed'; report['error'] = str(exc).replace(password, '[REDACTED]') print('Acceptance failed: ' + report['error'], flush=True); return 1 finally: if vm: vm.close() if server: server.shutdown(); server.server_close() report['completed_utc'] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()) (scratch / 'report.json').write_text(json.dumps(report, indent=2) + '\n') print('Report: ' + str(scratch / 'report.json'), flush=True) if __name__ == '__main__': raise SystemExit(main())