#!/usr/bin/env python3 import fcntl import hashlib import http.server import json import os from pathlib import Path import shutil import signal import socket import ssl import struct import subprocess import sys import tempfile import threading import time def digest(path): checksum = hashlib.sha256() with open(path, 'rb') as stream: for block in iter(lambda: stream.read(65536), b''): checksum.update(block) return checksum.hexdigest() def error(message, status): print('holy-qemu: ' + message, file=sys.stderr) raise SystemExit(status) def accelerator(): choice = os.environ.get('QEMU_ACCEL', 'auto') if choice not in ('auto', 'kvm', 'tcg'): error('QEMU_ACCEL must be auto, kvm or tcg', 2) if choice == 'tcg': return choice try: fd = os.open('/dev/kvm', os.O_RDWR | os.O_CLOEXEC) try: available = fcntl.ioctl(fd, 0xAE00) == 12 finally: os.close(fd) except OSError: available = False if available: return 'kvm' if choice == 'kvm': error('KVM unavailable', 6) return 'tcg' def quit_qemu(path): with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as channel: channel.settimeout(2) channel.connect(str(path)) with channel.makefile('rwb') as stream: if 'QMP' not in json.loads(stream.readline()): raise ValueError('missing QMP greeting') for command in ('qmp_capabilities', 'quit'): stream.write(json.dumps({'execute': command}).encode() + b'\n') stream.flush() while True: reply = json.loads(stream.readline()) if 'error' in reply: raise ValueError(str(reply['error'])) if 'return' in reply: break def boot_frames(lines, disk): if not disk: return [set(lines)], True frames = [] for line in lines: if line.startswith('HOLY-BOOT-1 boot '): if len(frames) == 2 or line != f'HOLY-BOOT-1 boot {len(frames) + 1}': return frames, False frames.append(set()) if frames: frames[-1].add(line) return frames, True def disk_format(qemu_img, path): details = json.loads(subprocess.check_output( [qemu_img, 'info', '--output=json', str(path)], text=True)) kind = details.get('format') if kind not in ('raw', 'qcow2') or details.get('backing-filename'): error('ROOT_DISK must be a self-contained raw or qcow2 image', 6) return kind def main(): network = os.environ.get('NETWORK_RECOVERY', 'off') if network == 'fixture' and os.environ.get('HOLY_QEMU_NETNS') != '1': if not shutil.which('unshare') or not shutil.which('ip'): error('unshare and ip required for isolated fixture network', 6) descriptor, resolver = tempfile.mkstemp(prefix='holy-qemu-resolv-') with os.fdopen(descriptor, 'w') as stream: stream.write('nameserver 127.0.0.1\n') environment = os.environ.copy() environment['HOLY_QEMU_NETNS'] = '1' environment['HOLY_QEMU_HOST_NETNS'] = str(os.stat('/proc/self/ns/net').st_ino) environment['HOLY_QEMU_RESOLV'] = resolver os.execvpe('unshare', ['unshare', '--map-root-user', '--net', '--mount', '--propagation', 'private', '--', sys.executable, str(Path(__file__).resolve())], environment) namespace_evidence = None if network == 'fixture': host_namespace = os.environ.get('HOLY_QEMU_HOST_NETNS', '') current_namespace = str(os.stat('/proc/self/ns/net').st_ino) if not host_namespace or host_namespace == current_namespace: error('fixture network namespace was not isolated', 6) subprocess.run(['ip', 'link', 'set', 'lo', 'up'], check=True) resolver = os.environ.get('HOLY_QEMU_RESOLV', '') if not resolver or not Path(resolver).is_file(): error('fixture resolver file missing', 6) subprocess.run(['mount', '--bind', resolver, '/etc/resolv.conf'], check=True) os.unlink(resolver) routes = subprocess.check_output(['ip', '-4', 'route', 'show'], text=True).splitlines() if routes: error('fixture network namespace has an external route', 6) namespace_evidence = {'host_inode': host_namespace, 'fixture_inode': current_namespace, 'fixture_routes': routes} arch = os.environ.get('ARCH', '') if arch not in ('i686', 'x86_64'): error('ARCH must be i686 or x86_64', 2) qemu = shutil.which('qemu-system-' + ('i386' if arch == 'i686' else arch)) media = os.environ.get('BOOT_MEDIA', 'iso') if media not in ('iso', 'disk'): error('BOOT_MEDIA must be iso or disk', 2) iso = Path(os.environ.get('ISO', '')) if not qemu or (media == 'iso' and not iso.is_file()): error('QEMU and selected readable boot input required', 6) if media == 'disk' and os.environ.get('ISO'): error('disk boot must not provide an ISO', 2) plan = os.environ.get('BOOT_PLAN', '') if len(plan) != 64 or any(c not in '0123456789abcdef' for c in plan): error('BOOT_PLAN must be SHA-256', 2) limit = os.environ.get('QEMU_TIMEOUT', '120') if not limit.isascii() or not limit.isdecimal() or not 1 <= int(limit) <= 600: error('QEMU_TIMEOUT must be 1..600 seconds', 2) probe_limit = os.environ.get('QEMU_PROBE_TIMEOUT', limit) if not probe_limit.isascii() or not probe_limit.isdecimal() or not 1 <= int(probe_limit) <= 600: error('QEMU_PROBE_TIMEOUT must be 1..600 seconds', 2) firmware = os.environ.get('FIRMWARE', 'bios') if firmware not in ('bios', 'uefi'): error('FIRMWARE must be bios or uefi', 2) profile = os.environ.get('IMAGE_PROFILE', 'dual-libc') state = os.environ.get('LIBC_BOOT_STATE', 'present') if profile not in ('static-core', 'dual-libc') or state not in ('present', 'glibc', 'musl', 'both', 'remove-both'): error('unsupported image profile or libc boot state', 2) if profile == 'static-core' and state != 'present': error('libc boot state requires dual-libc profile', 2) module_probe = os.environ.get('KERNEL_MODULE_PROBE', 'off') if module_probe not in ('on', 'off'): error('KERNEL_MODULE_PROBE must be on or off', 2) if network not in ('off', 'fixture'): error('NETWORK_RECOVERY must be off or fixture', 2) if network == 'fixture' and (profile != 'dual-libc' or state != 'both' or media != 'iso'): error('network fixture requires dual-libc, both absent and ISO boot', 2) disk_path = os.environ.get('ROOT_DISK', '') if state == 'remove-both' and (not disk_path or network != 'off'): error('remove-both requires persistent root without network fixture', 2) if media == 'disk' and not disk_path: error('disk boot requires ROOT_DISK', 6) qemu_img = shutil.which('qemu-img') if disk_path and (profile != 'dual-libc' or not Path(disk_path).is_file() or not qemu_img): error('disk recovery requires dual-libc, a regular ROOT_DISK and qemu-img', 6) keep_value = os.environ.get('QEMU_KEEP', '0') if keep_value not in ('0', '1'): error('QEMU_KEEP must be 0 or 1', 2) keep = keep_value == '1' firmware_files = {} if firmware == 'uefi': for field in ('UEFI_CODE', 'UEFI_VARS'): path = Path(os.environ.get(field, '')) if not path.is_file(): error(field + ' must name a firmware file', 6) firmware_files[field] = path accel = accelerator() report_dir = Path(os.environ.get('REPORT_DIR', tempfile.gettempdir())) if not report_dir.is_dir(): error('missing report directory', 6) run = Path(tempfile.mkdtemp(prefix='holy-qemu-' + arch + '-', dir=report_dir)).resolve() inputs = {} if media == 'iso': inputs['iso'] = {'source': str(iso.resolve())} shutil.copyfile(iso, run / 'input.iso') os.chmod(run / 'input.iso', 0o444) inputs['iso']['sha256'] = digest(run / 'input.iso') if disk_path: format_name = disk_format(qemu_img, disk_path) base = run / ('root-base.' + format_name) shutil.copyfile(disk_path, base) os.chmod(base, 0o444) if disk_format(qemu_img, base) != format_name: error('copied root disk changed format', 6) inputs['root_disk'] = {'source': str(Path(disk_path).resolve()), 'sha256': digest(base), 'format': format_name, 'copy_consistency': 'unverified'} subprocess.run([qemu_img, 'create', '-q', '-f', 'qcow2', '-F', format_name, '-b', str(base), str(run / 'root.qcow2')], check=True) for field in ('KERNEL_IMAGE', 'INITRAMFS', 'ROOT_IMAGE'): if field in os.environ: path = Path(os.environ[field]) inputs[field.lower()] = {'source': str(path.resolve()), 'sha256': digest(path)} requests = [] dns_queries = [] dns_socket = None server = None network_inputs = {} network_args = ['-net', 'none'] if network == 'fixture': dns_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) dns_socket.bind(('127.0.0.1', 53)) dns_socket.settimeout(0.2) dns_stopped = threading.Event() def serve_dns(): while not dns_stopped.is_set(): try: data, peer = dns_socket.recvfrom(4096) except socket.timeout: continue except OSError: return if len(data) < 17 or struct.unpack('!H', data[4:6])[0] != 1: continue cursor, labels = 12, [] while cursor < len(data) and 0 < data[cursor] <= 63: size = data[cursor] if cursor + size + 1 >= len(data): break labels.append(data[cursor + 1:cursor + size + 1]) cursor += size + 1 if cursor + 5 > len(data) or data[cursor] != 0: continue kind, group = struct.unpack('!HH', data[cursor + 1:cursor + 5]) name = b'.'.join(labels).lower() if name != b'fixture.holy.test' or group != 1: continue dns_queries.append({'name': name.decode(), 'type': kind}) question = data[12:cursor + 5] answer = b'' if kind == 1: answer = b'\xc0\x0c' + struct.pack('!HHIH', 1, 1, 0, 4) + socket.inet_aton('10.0.2.2') response = data[:2] + struct.pack('!HHHHH', 0x8180, 1, bool(answer), 0, 0) dns_socket.sendto(response + question + answer, peer) dns_thread = threading.Thread(target=serve_dns, daemon=True) dns_thread.start() source_dir = Path(os.environ.get('NETWORK_DIR', '')) for name in ('ca.pem', 'key.pem', 'glibc.holy', 'musl.holy'): if not (source_dir / name).is_file(): error('network fixture missing ' + name, 6) network_dir = run / 'network' network_dir.mkdir(mode=0o700) for name in ('ca.pem', 'key.pem', 'glibc.holy', 'musl.holy'): shutil.copyfile(source_dir / name, network_dir / name) os.chmod(network_dir / name, 0o600 if name == 'key.pem' else 0o444) class Handler(http.server.SimpleHTTPRequestHandler): def __init__(self, *arguments, **keywords): super().__init__(*arguments, directory=str(network_dir), **keywords) def do_GET(self): if self.path not in ('/glibc.holy', '/musl.holy'): self.send_error(404) return super().do_GET() def do_HEAD(self): self.send_error(404) def log_message(self, *arguments): pass def send_response(self, code, message=None): requests.append({'method': self.command, 'path': self.path, 'status': code}) return super().send_response(code, message) server = http.server.ThreadingHTTPServer(('127.0.0.1', 8443), Handler) tls = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) tls.load_cert_chain(network_dir / 'ca.pem', network_dir / 'key.pem') server.socket = tls.wrap_socket(server.socket, server_side=True) server_thread = threading.Thread(target=server.serve_forever, daemon=True) server_thread.start() network_args = ['-netdev', 'user,id=holy-net,ipv6=off', '-device', 'virtio-net-pci,netdev=holy-net'] network_inputs = {name: digest(network_dir / name) for name in ('ca.pem', 'glibc.holy', 'musl.holy')} args = [qemu, '-accel', accel, '-m', '1024', '-display', 'none', '-monitor', 'none', *network_args, '-boot', 'c' if media == 'disk' else 'd', '-serial', 'file:' + str(run / 'serial.log')] if media == 'iso': args += ['-cdrom', str(run / 'input.iso')] if disk_path: args += ['-blockdev', json.dumps({'driver': 'qcow2', 'node-name': 'holy-root', 'file': {'driver': 'file', 'filename': str(run / 'root.qcow2')}}), '-device', 'virtio-blk-pci,drive=holy-root' + (',bootindex=1' if media == 'disk' else '')] else: args += ['-no-reboot'] for field, path in firmware_files.items(): dest = run / (field.lower() + '.fd') shutil.copyfile(path, dest) inputs[field.lower()] = {'source': str(path.resolve()), 'sha256': digest(dest)} args += ['-drive', 'if=pflash,format=raw,' + ('readonly=on,' if field == 'UEFI_CODE' else '') + 'file=' + str(dest)] expected = {f'HOLY-BOOT-1 plan {plan}', f'HOLY-BOOT-1 arch {arch}', 'HOLY-BOOT-1 pid1 dinit', 'HOLY-BOOT-1 shell busybox', 'HOLY-BOOT-1 pkg holypkg', 'HOLY-BOOT-1 static-core verified', 'HOLY-BOOT-1 installer static-cli', 'HOLY-BOOT-1 docs installed-man-bundle', 'HOLY-BOOT-1 device mdevd-coldplug', 'HOLY-BOOT-1 transaction install-check-remove', 'HOLY-BOOT-1 installer root-plan-apply', 'HOLY-BOOT-1 result pass'} if profile == 'dual-libc': expected.update({f'HOLY-BOOT-1 profile {profile}', f'HOLY-BOOT-1 libc-initial {state}', 'HOLY-BOOT-1 libc-probes glibc-musl-pipe'}) if state != 'remove-both': expected.add(f'HOLY-BOOT-1 libc-recovery {state}') for abi in ('glibc', 'musl'): if state in (abi, 'both'): expected.update({f'HOLY-BOOT-1 missing-libc {abi}', f'HOLY-BOOT-1 restored-libc {abi}'}) if network == 'fixture': expected.add(f'HOLY-BOOT-1 downloaded-libc {abi}') if network == 'fixture': expected.add('HOLY-BOOT-1 network fixture-static-ip') expected.add('HOLY-BOOT-1 network fixture-dns') if 'KERNEL_VERSION' in os.environ: expected.add('HOLY-BOOT-1 kernel ' + os.environ['KERNEL_VERSION']) if module_probe == 'on': expected.add('HOLY-BOOT-1 kernel-module dummy-loaded') if media == 'disk': expected.add('HOLY-BOOT-1 esp mounted-writable') expected_boots = [expected] if disk_path: first = expected - {'HOLY-BOOT-1 result pass'} first.update({'HOLY-BOOT-1 boot 1', 'HOLY-BOOT-1 root ext4', 'HOLY-BOOT-1 first-boot pass', 'HOLY-BOOT-1 reboot requested'}) if state == 'remove-both': first.update({f'HOLY-BOOT-1 removed-libc {abi}' for abi in ('glibc', 'musl')}) second = expected - {'HOLY-BOOT-1 libc-initial remove-both'} second.update({'HOLY-BOOT-1 boot 2', 'HOLY-BOOT-1 root ext4', 'HOLY-BOOT-1 libc-initial removed-both', 'HOLY-BOOT-1 libc-recovery removed-both'}) for abi in ('glibc', 'musl'): second.update({f'HOLY-BOOT-1 missing-libc {abi}', f'HOLY-BOOT-1 reinstalled-libc {abi}', f'HOLY-BOOT-1 restored-libc {abi}'}) else: second = {marker for marker in expected if not marker.startswith(( 'HOLY-BOOT-1 libc-initial ', 'HOLY-BOOT-1 libc-recovery ', 'HOLY-BOOT-1 missing-libc ', 'HOLY-BOOT-1 restored-libc '))} second.update({'HOLY-BOOT-1 boot 2', 'HOLY-BOOT-1 root ext4', 'HOLY-BOOT-1 libc-initial restored', 'HOLY-BOOT-1 libc-recovery restored'}) expected_boots = [first, second] expected |= first | second cancelled = [] def request_stop(number, frame): if not cancelled: cancelled.append(number) signal.signal(signal.SIGTERM, request_stop) signal.signal(signal.SIGINT, request_stop) started = time.monotonic() reason = 'boot-timeout' result = 'fail' seen = set() frames = [] first_completed = None serial_lines = [] probe_history = [] active_probe = None probe_deadline = None probe_boot = 1 started_stages = set() deadline = started + int(limit) with tempfile.TemporaryDirectory(prefix='holy-qmp-') as control: qmp = Path(control) / 'control' args += ['-qmp', 'unix:' + str(qmp) + ',server=on,wait=off'] with open(run / 'qemu.stdout', 'wb') as stdout, open(run / 'qemu.stderr', 'wb') as stderr: process = subprocess.Popen(args, stdin=subprocess.DEVNULL, stdout=stdout, stderr=stderr) try: while time.monotonic() < deadline: if cancelled: reason = 'cancelled' break serial = run / 'serial.log' if serial.exists(): content = serial.read_text(errors='replace') complete = content.rsplit('\n', 1)[0] if '\n' in content else '' lines = complete.splitlines() if lines[:len(serial_lines)] != serial_lines: reason = 'serial-log-changed' break now = time.monotonic() for line in lines[len(serial_lines):]: if line == 'HOLY-BOOT-1 boot 2': if active_probe: active_probe['duration_seconds'] = now - active_probe['started_at'] active_probe['status'] = 'pass' active_probe = None probe_deadline = None probe_boot = 2 if not line.startswith('HOLY-BOOT-1 stage '): continue name = line.removeprefix('HOLY-BOOT-1 stage ') if name not in ('identity', 'login', 'esp', 'static-core', 'installer', 'devices', 'kernel-module', 'libc-recovery', 'network-setup', 'documentation', 'packages', 'installer-transaction', 'libc-removal', 'reboot', 'result'): reason = 'invalid-probe-stage' break if (probe_boot, name) in started_stages: continue started_stages.add((probe_boot, name)) if active_probe: active_probe['duration_seconds'] = now - active_probe['started_at'] active_probe['status'] = 'pass' active_probe = {'boot': probe_boot, 'stage': name, 'started_at': now, 'duration_seconds': None, 'status': 'unknown'} probe_history.append(active_probe) probe_deadline = now + int(probe_limit) if reason == 'invalid-probe-stage': break serial_lines = lines seen = set(lines) frames, valid = boot_frames(lines, bool(disk_path)) if not valid: reason = 'invalid-boot-sequence' break if 'HOLY-BOOT-1 result fail' in seen: reason = 'guest-failure' if active_probe: active_probe['duration_seconds'] = time.monotonic() - active_probe['started_at'] active_probe['status'] = 'fail' break if disk_path and first_completed is None and frames and expected_boots[0] <= frames[0]: first_completed = time.monotonic() - started deadline = time.monotonic() + int(limit) reason = 'reboot-timeout' if len(frames) == len(expected_boots) and all( wanted <= actual for wanted, actual in zip(expected_boots, frames)): reason = 'probes-complete' if active_probe: active_probe['duration_seconds'] = time.monotonic() - active_probe['started_at'] active_probe['status'] = 'pass' break if probe_deadline and time.monotonic() >= probe_deadline: reason = 'probe-timeout' if active_probe: active_probe['duration_seconds'] = time.monotonic() - active_probe['started_at'] break if process.poll() is not None: reason = 'early-exit' break time.sleep(0.1) if process.poll() is None: try: quit_qemu(qmp) except (OSError, ValueError) as exc: reason = 'qmp-failure: ' + str(exc) process.terminate() try: code = process.wait(timeout=5) except subprocess.TimeoutExpired: reason = 'shutdown-timeout' process.kill() code = process.wait() if reason == 'probes-complete' and code == 0: result = 'pass' finally: if process.poll() is None: process.kill() process.wait() if active_probe and active_probe['duration_seconds'] is None: active_probe['duration_seconds'] = time.monotonic() - active_probe['started_at'] base_unchanged = not disk_path or digest(base) == inputs['root_disk']['sha256'] if server: server.shutdown() server.server_close() server_thread.join(timeout=5) dns_stopped.set() dns_socket.close() dns_thread.join(timeout=5) wanted_requests = {f'/{abi}.holy' for abi in ('glibc', 'musl')} served = {entry['path'] for entry in requests if entry['status'] == 200} if not wanted_requests <= served or not any(item['type'] == 1 for item in dns_queries): reason, result = 'network-fixture-incomplete', 'fail' if not base_unchanged: reason, result = 'changed-read-only-base', 'fail' boots = [] for index, wanted in enumerate(expected_boots): actual = frames[index] if index < len(frames) else set() boots.append({'boot': index + 1, 'missing_markers': sorted(wanted - actual), 'checks': {marker: 'pass' if marker in actual else 'unknown' for marker in sorted(wanted)}}) report = {'schema': 'holy-qemu-report-2', 'arch': arch, 'accelerator': accel, 'firmware': firmware, 'profile': profile, 'libc_boot_state': state, 'kernel_module_probe': module_probe, 'network': 'private-loopback-fixture-https' if server else 'disabled', 'plan': plan, 'inputs': inputs, 'network_inputs': network_inputs, 'network_requests': requests, 'dns_queries': dns_queries, 'network_namespace': namespace_evidence, 'argv': args, 'pid': process.pid, 'exit': code, 'cancel_signal': cancelled[0] if cancelled else None, 'elapsed_seconds': time.monotonic() - started, 'boot_timeout_seconds': int(limit), 'probe_timeout_seconds': int(probe_limit), 'shutdown_timeout_seconds': 5, 'probes': [{'boot': probe['boot'], 'stage': probe['stage'], 'started_seconds': probe['started_at'] - started, 'duration_seconds': probe['duration_seconds'], 'status': probe['status']} for probe in probe_history], 'timed_out_probe': active_probe['stage'] if reason == 'probe-timeout' else None, 'timed_out_boot': active_probe['boot'] if reason == 'probe-timeout' else None, 'root_storage': 'ext4-overlay' if disk_path else 'ram', 'boots': boots, 'boot_media': media, 'first_boot_completed_seconds': first_completed, 'reason': reason, 'result': result, 'missing_markers': ([f'boot-{b["boot"]}: {marker}' for b in boots for marker in b['missing_markers']] if disk_path else sorted(expected - seen)), 'checks': {marker: 'pass' if marker in seen else 'unknown' for marker in sorted(expected)}, 'not_tested': (['libc-recovery'] if profile != 'dual-libc' or state == 'present' else []) + ([] if disk_path else ['libc-recovery-reboot']) + (['i686-libc'] if arch != 'i686' or profile != 'dual-libc' else []) + (['kernel-module-load'] if module_probe == 'off' else []) + ['installer', 'hardware', 'public-network' if server else 'network', 'kernel-update']} if disk_path: report['overlay'] = {'path': str(run / 'root.qcow2'), 'sha256': digest(run / 'root.qcow2'), 'base_unchanged': base_unchanged, 'backing_format': inputs['root_disk']['format'], 'retained': keep} report['temporary_inputs_retained'] = keep (run / 'report.json').write_text(json.dumps(report, indent=2) + '\n') (run / 'report').write_text( f'format holy-qemu-report-2\narch {arch}\nboot-media {media}\niso-sha256 {inputs.get("iso", {}).get("sha256", "none")}\n' f'plan {plan}\naccelerator {accel}\nfirmware {firmware}\nexit {code}\n' f'reason {reason}\nresult {result}\n') if not keep: for temporary in ('root.qcow2', 'root-base.raw', 'root-base.qcow2', 'uefi_code.fd', 'uefi_vars.fd', 'input.iso'): (run / temporary).unlink(missing_ok=True) print(f'holy-qemu: {result} report {run}/report') return 0 if result == 'pass' else 4 if __name__ == '__main__': try: sys.exit(main()) except (OSError, ValueError, subprocess.CalledProcessError) as exc: error(str(exc), 1)