130 lines
7.0 KiB
Python
Executable File
130 lines
7.0 KiB
Python
Executable File
#!/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')
|