From 798a5779ef3dd56a48866f72a03017bde07c82a7 Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 13 Oct 2026 21:57:42 +0300 Subject: [PATCH] test: add disposable laptop and base recovery acceptance harness --- dev/laptop-smoke.py | 477 +++++++++++++++++++++++++++++++++++++ docs/laptop-acceptance.md | 70 ++++++ tests/test_laptop_smoke.py | 119 +++++++++ 3 files changed, 666 insertions(+) create mode 100755 dev/laptop-smoke.py create mode 100644 docs/laptop-acceptance.md create mode 100644 tests/test_laptop_smoke.py diff --git a/dev/laptop-smoke.py b/dev/laptop-smoke.py new file mode 100755 index 0000000..da5ae8f --- /dev/null +++ b/dev/laptop-smoke.py @@ -0,0 +1,477 @@ +#!/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): + for name in ('gpt', 'foreign_sha256', 'foreign_efi', 'boot_order'): + 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()} + +@contextlib.contextmanager +def mounted(device, options=''): + import tempfile + with tempfile.TemporaryDirectory(prefix='hydra-acceptance-') as directory: + args = ['mount'] + (['-o', options] if options else []) + [device, directory] + run(*args) + try: yield Path(directory) + finally: run('umount', directory) + +def preserved(): + table = json.loads(run('sfdisk', '--json', '/dev/vda'))['partitiontable'] + # No filesystem UUIDs: selected root UUID must change on formatting. + with mounted('/dev/vda1', 'ro') 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 = { +'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') 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', 'ro,subvol=@root') as root: + with mounted('/dev/vda1', 'ro') as esp: + result = base_state(root, esp) +''', +'damage': r''' +with mounted('/dev/vda3', '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') 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 mounted('/dev/vda1', 'ro') 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', '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_installed(console): + # This is a deliberate harness action AFTER proving installer preserved order. + output = console.command('sudo -n efibootmgr', 30).decode() + entries = re.findall(r'^Boot([0-9A-Fa-f]{4})\*?\s+HydraOS\s', output, re.M) + if len(entries) != 1: + raise SmokeError('Expected one registered HydraOS entry: ' + output) + console.command('sudo -n efibootmgr --bootnext ' + entries[0], 30) + return entries[0] + + +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.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'} + 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) + 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) + 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)); shutdown(console) + console = boot(True, 'live-checkpoint') + 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))) + select_installed(console); shutdown(console) + console = boot(False, 'installed-upgrade'); record('real_portage_upgrade_current_home', probe(console, 'upgrade', context, 720)); shutdown(console) + console = boot(True, 'live-restore') + 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(before, preserved_after) + record('restored_base_foreign_invariants', {'base': base_after, 'preserved': preserved_after}) + select_installed(console); 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); 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()) diff --git a/docs/laptop-acceptance.md b/docs/laptop-acceptance.md new file mode 100644 index 0000000..d899a90 --- /dev/null +++ b/docs/laptop-acceptance.md @@ -0,0 +1,70 @@ +# Disposable laptop acceptance + +`dev/laptop-smoke.py` exercises the packaged installer, doctor, ASCII branding, +signed runtime updater and offline base recovery in one new UEFI/KVM VM. It accepts +only trusted ISO and firmware inputs. It creates a private scratch directory and a +new 40 GiB qcow2, copies both firmware files, and retains the disk, serial logs and +`report.json` after success or failure. It never accepts an existing disk or host +block device. Run it where QEMU/KVM, Python and OpenSSL are already available; on +this development machine that is the normal-user Ubuntu development workspace. + +```sh +python3 dev/laptop-smoke.py \ + --iso /absolute/path/to/new-hydra.iso \ + --ovmf-code /absolute/path/to/OVMF_CODE.fd \ + --ovmf-vars /absolute/path/to/OVMF_VARS.fd \ + --scratch-parent /tmp +``` + +The separate `dev/iso-smoke.py` whole-disk regression remains required. This harness +adds the following real guest operations: + +1. Seed fresh GPT with an ESP, foreign data partition and selected root. Record + partition GUIDs/boundaries, the complete foreign partition hash, foreign EFI and + fallback bytes, and firmware BootOrder. Install Btrfs into `/dev/vda3` while + retaining `/dev/vda1`, explicitly register a HydraOS entry, then verify those + invariants before the harness deliberately sets BootNext to choose that entry. +2. Boot without ISO, verify the packaged Fastfetch art and doctor, install version + 1.0 of a locally created EAPI 8 Portage package, and create a real qcow2 plus + profile/data sentinels under HOME. The fixture has no sources or dependencies; + `emerge --nodeps` runs in a private guest network namespace with no interfaces, + so it cannot fetch a package. +3. Temporarily pin a disposable TEST public key in that guest and verify signed + passive local artifacts through the packaged CLI. Apply and roll back a runtime + change, refuse a tampered archive without mutation, SIGKILL a child after an + actual runtime replacement, damage the CLI, then invoke the independently + installed `/usr/bin/hydra-update-recover` in a fresh process. Verify exact original + runtime bytes, the monotonic replay floor and unchanged VM data. Restore the + guest's original public key. The production signing key is never read. +4. Boot live with installed filesystems offline. Record kernel/boot, fixture Portage + database, world and owned EFI hashes, then create and list a real Btrfs checkpoint + through `hydra-system`. Boot the installed system, upgrade the fixture to 2.0 + through Portage in a private network namespace and modify HOME and world. +5. Boot live again, deliberately damage actual installed kernels and the owned EFI + loader, and SIGKILL a real restore after its durable `root-retained` transition. + Resume the same checkpoint through a fresh `hydra-system` process. Compare exact + base, package database, world and EFI hashes and all preserved foreign invariants. + Boot without ISO and verify version 1.0, current HOME and unchanged VM disk/profile + data. The previous failed root and checkpoint remain on the disposable disk. + +The loopback HTTP fixture serves only explicitly named in-memory TEST artifacts +and their public key. It exposes no host directory, build tree, private key or +credentials. The disposable private key is deleted before the VM boots. This HTTP +transport tests offline artifact delivery, not production HTTPS discovery or its +redirect policy. Neither runtime nor base recovery receives production signing +material or existing VM disks. + +Only an actual `report.json` with `status: "passed"`, input SHA-256 identities and +all guest phases establishes acceptance. Source tests verify harness construction, +real signatures, narrow sharing, byte comparisons and guest script compilation; +they do not prove UEFI, Btrfs, Portage or installed boot. This sequence restores an +exact kernel after damage and upgrades a real minimal fixture package; it does not +claim an upgrade between two kernel versions. Physical ASUS Wi-Fi, audio, suspend, +brightness and refresh-rate checks still require live laptop verification. + +Run the focused source checks without booting a VM: + +```sh +python3 -m unittest tests.test_laptop_smoke +python3 -m py_compile dev/laptop-smoke.py tests/test_laptop_smoke.py +``` diff --git a/tests/test_laptop_smoke.py b/tests/test_laptop_smoke.py new file mode 100644 index 0000000..ee73586 --- /dev/null +++ b/tests/test_laptop_smoke.py @@ -0,0 +1,119 @@ +import importlib.util +import json +from pathlib import Path +import subprocess +import tempfile +import threading +import unittest +import urllib.error +import urllib.request + + +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_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)