#!/usr/bin/env python3 """Exercise production stage0 and the Pi kernel using isolated ARM VM disks.""" import io import json import os from pathlib import Path import selectors import socket import subprocess import sys import tarfile import time project = Path(__file__).resolve().parents[2] sys.path.insert(0, str(project/'tools')) from image_formats import digest, gpt, LINUX_FILESYSTEM work = Path(sys.argv[1]).resolve(strict=True) system = work/'system/system.img' recovery = work/'recovery.img' gpt(recovery, [('FDS_RECOVERY', LINUX_FILESYSTEM, work/'system/system.erofs')]) invalid = work/'invalid.img' (work/'invalid.payload').write_bytes(bytes(1048576)) gpt(invalid, [('FDS_SYSTEM', LINUX_FILESYSTEM, work/'invalid.payload')]) foreign = work/'foreign.img' with tarfile.open(work/'foreign.tar', 'w') as archive: payload = b'ID=another-os\n' info = tarfile.TarInfo('usr/lib/os-release') info.mode = 0o644 info.size = len(payload) archive.addfile(info, io.BytesIO(payload)) subprocess.run([str(project/'tools/in-image-tools'), 'mkfs.erofs', '-T', '0', '--tar=f', str(work/'foreign.erofs'), str(work/'foreign.tar')], check=True, stdout=subprocess.DEVNULL) gpt(foreign, [('FDS_SYSTEM', LINUX_FILESYSTEM, work/'foreign.erofs')]) images = {'system': system, 'recovery': recovery, 'invalid': invalid, 'foreign': foreign} before = {str(path): digest(path) for path in images.values()} def run_case(name, suffix='', disks=('system',), command_line='', hot_insert=False, recover=False, wait_message=b'MULTIPLE FDS_SYSTEM CARTRIDGES (2)'): qmp_path = work/(name+'.qmp') log = work/(name+'.log') command = [str(project/'tools/in-void'), 'qemu-system-aarch64', '-machine', 'virt', '-cpu', 'max', '-accel', 'tcg', '-m', '1024', '-smp', '2', '-nodefaults', '-display', 'none', '-serial', 'stdio', '-nic', 'none', '-no-reboot', '-qmp', f'unix:{qmp_path},server=on,wait=off', '-kernel', str(project/'out/kernel/boot/kernel_2712.img'), '-initrd', str(project/'out/initramfs'/('initramfs.cpio'+suffix)), '-append', 'console=ttyAMA0 rdinit=/init panic=-1 fds.debug=1 '+command_line] for number, disk in enumerate(disks): image = images[disk] command += ['-drive', f'file={image},if=none,id=disk{number},format=raw,readonly=on', '-device', f'virtio-blk-pci,drive=disk{number}'] if hot_insert: command += ['-device', 'qemu-xhci,id=xhci', '-drive', f'file={system},if=none,id=inserted,format=raw,readonly=on'] data = bytearray() transition = False child = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) selector = selectors.DefaultSelector() selector.register(child.stdout, selectors.EVENT_READ) deadline = time.monotonic()+120 try: with log.open('wb') as stream: while selector.get_map(): remaining = deadline-time.monotonic() assert remaining > 0, f'{name}: boot deadline; inspect {log}' for key, _ in selector.select(remaining): chunk = os.read(key.fd, 65536) if not chunk: selector.unregister(key.fileobj) continue stream.write(chunk) stream.flush() data.extend(chunk) assert b'Kernel panic' not in data and b'FDS_STAGE0_ERROR' not in data, f'{name}: boot failure; {log}' assert b'FDS_M2_FAIL' not in data and b'FDS_M4_FAIL' not in data, f'{name}: guest check failed; {log}' if hot_insert and not transition and b'FDS_STAGE0_WAIT' in data: assert b'FDS_SYSTEM MEDIA NOT PRESENT' in data and b'FDS_STAGE0_HANDOFF' not in data # The prompt proves initial discovery is finished. Now inject a USB event. with socket.socket(socket.AF_UNIX) as control: control.settimeout(10) control.connect(str(qmp_path)) with control.makefile('rwb') as protocol: assert 'QMP' in json.loads(protocol.readline()) def request(execute, arguments=None): protocol.write((json.dumps({'execute': execute, 'arguments': arguments or {}})+'\n').encode()) protocol.flush() while True: reply = json.loads(protocol.readline()) if 'event' in reply: continue assert 'return' in reply, reply return reply request('qmp_capabilities') request('device_add', {'driver': 'usb-storage', 'drive': 'inserted', 'bus': 'xhci.0'}) transition = True if recover and not transition and b'FDS_STAGE0_WAIT' in data: assert wait_message in data, f'{name}: missing expected refusal; {log}' assert b'FDS_STAGE0_HANDOFF' not in data child.stdin.write(b'recovery\n') child.stdin.flush() transition = True assert child.wait(timeout=10) == 0, f'{name}: QEMU failed' finally: selector.close() if child.poll() is None: child.terminate() try: child.wait(timeout=10) except subprocess.TimeoutExpired: child.kill() child.wait() for marker in (b'FDS_STAGE0_ROOT', b'FDS_STAGE0_HANDOFF', b'FDS_M4_PASS', b'FDS_M2_PASS', b'Power down'): assert marker in data, f'{name}: missing {marker!r}; inspect {log}' if hot_insert or recover: assert transition if recover or 'fds.boot=recovery' in command_line: assert b'FDS_STAGE0_ROOT: FDS_RECOVERY' in data print(f'PASS: {name}', flush=True) for name, suffix in [('uncompressed', ''), ('gzip', '.gz'), ('lz4', '.lz4'), ('zstandard', '.zst')]: run_case(name, suffix) run_case('missing-then-insert', disks=(), hot_insert=True) run_case('multiple-then-recovery', disks=('system', 'system', 'recovery'), recover=True) run_case('explicit-recovery', disks=('system', 'recovery'), command_line='fds.boot=recovery') run_case('invalid-filesystem', disks=('invalid', 'recovery'), recover=True, wait_message=b'SYSTEM CANNOT BE USED: mount:') run_case('foreign-root', disks=('foreign', 'recovery'), recover=True, wait_message=b'Selected root is not an FDS image') assert all(digest(Path(path)) == sha for path, sha in before.items()), 'VM altered an input disk' (work/'disk-digests.json').write_text(json.dumps(before, indent=2)+'\n') print('PASS: all stage0 cases; input images unchanged') print('SKIP: Pi firmware, NVMe/RP1, physical USB/display behavior and real boot timings')