"""Read-only image boot proof with no emulated serial controller. This test proves that two framebuffer captures reach the already verified idle desktop. It does not claim guest serial markers, physical hardware support, complete editor stability, or GPU/CUDA acceleration. """ import argparse import hashlib import json import pathlib import re import shutil import socket import subprocess import time from PIL import Image, ImageChops, ImageStat from ReadImage import inspect_image ROOT = pathlib.Path(__file__).resolve().parents[2] GIB = 1024 ** 3 def require(condition, message): if not condition: raise ValueError(message) def sha256(path): digest = hashlib.sha256() with pathlib.Path(path).open('rb') as stream: for block in iter(lambda: stream.read(1024 * 1024), b''): digest.update(block) return digest.hexdigest() def absolute_file(value, name): path = pathlib.Path(value) require(path.is_absolute(), name + ' must be an absolute file path') path = path.resolve(strict=True) require(path.is_file(), name + ' must identify a file') return path def close_rgb_fraction(reference, actual): if reference.size != actual.size: return 0.0 difference = ImageChops.difference(reference, actual) red, green, blue = difference.split() maximum = ImageChops.lighter(ImageChops.lighter(red, green), blue) histogram = maximum.histogram() return sum(histogram[:4]) / (reference.width * reference.height) def regions_for(width, height, scale): # Match the current DesktopSession's five launcher bounds and 48px taskbar. work_height = height // scale - 48 if work_height >= 554: top, bottom = 84, 84 + 4 * 94 + 74 else: icon_height = max(40, min(74, (work_height - 80) // 5)) gap = max(4, min(12, (work_height - 32 - icon_height * 5) // 4)) top, bottom = 16, 16 + 4 * (icon_height + gap) + icon_height return { 'DesktopEntries': (22 * scale, top * scale, 122 * scale, min(height - 48 * scale, bottom * scale)), 'Taskbar': (0, height - 48 * scale, width, height), } def verify_reference(reference, regions): for name, box in regions.items(): require(box[2] > box[0] and box[3] > box[1], 'Invalid ' + name + ' comparison region') with reference.crop(box) as crop: ranges = [high - low for low, high in crop.getextrema()] require(max(ranges) >= 32 and max(ImageStat.Stat(crop).stddev) >= 4, name + ' reference lacks visible interface detail; a black/background frame cannot qualify') def converted_arguments(original, source_image, variables, port, acceleration): require(isinstance(original, list) and original and all(isinstance(value, str) for value in original), 'Verified boot receipt has no QEMU argument list') executable = absolute_file(original[0], 'QEMU executable') require(executable == (ROOT / '.tools/qemu/qemu-system-x86_64.exe').resolve(), 'Only this project QEMU executable may run') result = [str(executable)] counts = {'-serial': 0, '-qmp': 0, '-display': 0, '-accel': 0, 'Image': 0, 'Vars': 0, 'Code': 0} cursor = 1 while cursor < len(original): option = original[cursor] if option in ('-serial', '-qmp', '-display', '-accel', '-drive'): require(cursor + 1 < len(original), 'Incomplete QEMU option: ' + option) value = original[cursor + 1] cursor += 2 if option == '-drive': fields = dict(piece.split('=', 1) for piece in value.split(',') if '=' in piece) path = pathlib.Path(fields.get('file', '')).resolve() if path == source_image: counts['Image'] += 1 require(fields.get('format') == 'raw' and fields.get('snapshot') == 'on', 'The verified source image must already use raw snapshot mode') # snapshot=on opens the source read-only and gives the IDE # frontend a temporary writable overlay. readonly=on here # would make that frontend read-only and QEMU rejects it. require('readonly' not in fields, 'Unexpected frontend read-only override') value = ','.join(key + '=' + item for key, item in fields.items()) elif fields.get('if') == 'pflash' and fields.get('readonly') == 'on': counts['Code'] += 1 require(path.is_file(), 'Verified firmware code is unavailable') elif fields.get('if') == 'pflash': counts['Vars'] += 1 fields['file'] = str(variables) value = ','.join(key + '=' + item for key, item in fields.items()) else: raise ValueError('Unexpected writable/extra drive in verified QEMU recipe') else: counts[option] += 1 value = {'-serial': 'none', '-qmp': f'tcp:127.0.0.1:{port},server=on,wait=off', '-display': 'none', '-accel': acceleration}[option] result.extend((option, value)) else: require(option not in ('-chardev', '-readconfig', '-writeconfig', '-S', '-incoming'), 'Unexpected diagnostic/device override in verified recipe') if option == '-device': require(cursor + 1 < len(original) and not re.search(r'serial|uart', original[cursor + 1], re.I), 'An explicit serial/UART device cannot qualify') result.append(option) cursor += 1 require(all(value == 1 for value in counts.values()), 'Expected exactly one image, firmware pair, serial/display/accel/QMP option') return result class Qmp: def __init__(self, connection): self.connection = connection self.stream = connection.makefile('rwb') greeting = self.stream.readline() require(greeting and 'QMP' in json.loads(greeting), 'Invalid QMP greeting') self.call('qmp_capabilities') def call(self, command, arguments=None): self.stream.write((json.dumps({'execute': command, 'arguments': arguments or {}}) + '\n').encode()) self.stream.flush() while True: line = self.stream.readline() require(bool(line), 'QMP connection closed') message = json.loads(line) if 'error' in message: raise RuntimeError(str(message['error'])) if 'return' in message: return message['return'] def close(self): self.stream.close() self.connection.close() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--verified-boot-result', required=True) parser.add_argument('--reference-screenshot', required=True) parser.add_argument('--accel', choices=('whpx', 'tcg'), required=True) options = parser.parse_args() receipt_path = absolute_file(options.verified_boot_result, 'Verified boot receipt') screenshot_path = absolute_file(options.reference_screenshot, 'Reference screenshot') receipt = json.loads(receipt_path.read_text(encoding='utf-8-sig')) require(receipt.get('Kind') == 'BareMetalNativeAotBootProof' and receipt.get('Passed') is True and receipt.get('ImageUnchanged') is True and receipt.get('QemuExitCode') == 0 and receipt.get('Booted') is True, 'A passing, cleanly closed native boot receipt is required') image = absolute_file(receipt['ImagePath'], 'Verified source image') source_hash = sha256(image) require(source_hash == receipt['ImageSha256'], 'Source image differs from the verified boot receipt') inspection, payloads = inspect_image(image) require(inspection['ImageSha256'] == source_hash and hashlib.sha256(payloads['System/Kernel.elf']).hexdigest() == receipt['KernelSha256'], 'Embedded kernel does not match the verified boot receipt') del payloads snapshot = absolute_file(receipt['SourceSnapshotPath'], 'Verified source snapshot') require(sha256(snapshot) == receipt['SourceSnapshotSha256'], 'Source snapshot differs from the verified boot receipt') require(pathlib.Path(receipt['DefaultAppearance']['Screenshot']).resolve() == screenshot_path and receipt['DefaultAppearance'].get('Passed') is True, 'The reference must be the passing idle Desktop screenshot from this same boot receipt') with Image.open(screenshot_path) as source: reference = source.convert('RGB') require(reference.size == (receipt['NativeWidth'], receipt['NativeHeight']), 'Reference/native resolution mismatch') scale_percent = receipt['ScalePercent'] require(scale_percent in (100, 200), 'The comparison supports the actual 100/200 percent startup profiles') regions = regions_for(reference.width, reference.height, scale_percent // 100) verify_reference(reference, regions) require(shutil.disk_usage(ROOT).free >= 10 * GIB, 'At least 10 GiB free space is required; no test artifacts were created') out = ROOT / 'build/CSharpBoot' / f'NoSerial-{time.time_ns()}' out.mkdir(parents=True, exist_ok=False) variables = out / 'FirmwareVars.fd' shutil.copyfile(ROOT / '.tools/firmware/edk2-i386-vars.fd', variables) with socket.socket() as probe: probe.bind(('127.0.0.1', 0)) port = probe.getsockname()[1] arguments = converted_arguments(receipt['QemuArgs'], image, variables, port, options.accel) result = { 'Kind': 'NativeNoSerialFramebufferBootProof', 'Passed': False, 'VerifiedBootResult': str(receipt_path), 'VerifiedBootResultSha256': sha256(receipt_path), 'ImagePath': str(image), 'ImageSha256': source_hash, 'KernelSha256': receipt['KernelSha256'], 'ImageCopied': False, 'SourceDriveReadOnly': False, 'SourceDriveSnapshot': True, 'SourceSnapshotSha256': receipt['SourceSnapshotSha256'], 'ReferenceScreenshot': str(screenshot_path), 'ReferenceScreenshotSha256': sha256(screenshot_path), 'QemuArgs': arguments, 'Acceleration': options.accel, 'TestHarnessSha256': sha256(__file__), 'SerialArgument': 'none', 'SerialControllersAbsent': False, 'GuestSerialMarkersObserved': False, 'CompleteStabilityAcceptance': False, 'PhysicalHardwareAcceptance': False, 'CudaExecutionVerified': False, 'GpuRasterizationVerified': False, 'PixelTolerancePerRgbChannel': 3, 'FullFrameRequiredMatchingFraction': 0.99, 'StrictRegionsRequiredMatchingFraction': 1.0, 'StrictRegions': regions, 'RequiredConsecutiveMatches': 2, 'TimeoutSeconds': 90, 'Samples': [], } process, qmp = None, None consecutive = 0 kept_frames = [] started = time.monotonic() try: with (out / 'QemuStdout.log').open('xb') as stdout, (out / 'QemuStderr.log').open('xb') as stderr: process = subprocess.Popen(arguments, cwd=ROOT, stdout=stdout, stderr=stderr) result['OwnedProcessId'] = process.pid deadline = started + 90 while time.monotonic() < deadline: if process.poll() is not None: raise RuntimeError('QEMU exited before framebuffer readiness: ' + str(process.returncode)) try: qmp = Qmp(socket.create_connection(('127.0.0.1', port), timeout=2)) break except ConnectionRefusedError: time.sleep(0.1) require(qmp is not None, 'QMP startup timeout') tree = qmp.call('human-monitor-command', {'command-line': 'info qtree'}) require(isinstance(tree, str) and 'bus: main-system-bus' in tree and 'dev:' in tree, 'QMP did not return a complete device tree') (out / 'Qtree.txt').write_text(tree, encoding='utf-8') require(not re.search(r'\bdev:\s*(?:isa-serial|serial-mm)\b', tree, re.I), 'A serial controller exists despite -serial none') result['SerialControllersAbsent'] = True block_nodes = qmp.call('query-named-block-nodes') require(isinstance(block_nodes, list), 'QMP did not return block node details') source_nodes = [node for node in block_nodes if node.get('drv') == 'raw' and isinstance(node.get('image'), dict) and pathlib.Path(node['image'].get('filename', '')).resolve() == image] require(source_nodes and all(node.get('ro') is True for node in source_nodes), 'The exact source image is not attached as a read-only raw backing node') result['SourceDriveReadOnly'] = True result['SourceBlockNodes'] = source_nodes last_capture = started - 1 while time.monotonic() < deadline: if process.poll() is not None: raise RuntimeError('QEMU exited before two matching desktop frames') remaining = 1 - (time.monotonic() - last_capture) if remaining > 0: time.sleep(min(remaining, max(0, deadline - time.monotonic()))) if time.monotonic() >= deadline: break require(shutil.disk_usage(out).free >= 10 * GIB, 'Free space fell below 10 GiB; no additional screenshot was created') last_capture = time.monotonic() index = len(result['Samples']) ppm, png = out / f'Frame-{index:03}.ppm', out / f'Frame-{index:03}.png' qmp.call('screendump', {'filename': str(ppm)}) try: with Image.open(ppm) as frame: actual = frame.convert('RGB') actual.save(png) finally: if ppm.exists(): ppm.unlink() fractions = {'FullFrame': close_rgb_fraction(reference, actual)} if actual.size == reference.size: for name, box in regions.items(): with reference.crop(box) as expected_region, actual.crop(box) as actual_region: fractions[name] = close_rgb_fraction(expected_region, actual_region) else: fractions.update({name: 0.0 for name in regions}) actual.close() matched = fractions['FullFrame'] >= 0.99 and all(fractions[name] == 1.0 for name in regions) consecutive = consecutive + 1 if matched else 0 result['Samples'].append({'ElapsedSeconds': round(last_capture - started, 3), 'Screenshot': str(png), 'MatchingPixelFractions': fractions, 'Matched': matched, 'ConsecutiveMatches': consecutive}) kept_frames.append(png) if len(kept_frames) > 2: kept_frames.pop(0).unlink() if consecutive >= 2: result['FramebufferReady'] = True result['DesktopReadyElapsedSeconds'] = round(time.monotonic() - started, 3) break require(consecutive >= 2, 'No serial hardware: desktop framebuffer did not match twice within 90 seconds') except Exception as error: result['Failure'] = type(error).__name__ + ': ' + str(error) finally: if process is not None: if qmp is not None and process.poll() is None: try: qmp.call('quit') result['QmpQuitSent'] = True except Exception as error: result['QmpQuitError'] = str(error) try: process.wait(timeout=10) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=10) result['OwnedProcessKilledAfterTimeout'] = True result['QemuExitCode'] = process.returncode result['OwnedProcessExited'] = process.poll() is not None if qmp is not None: qmp.close() result['ImageSha256After'] = sha256(image) result['ImageUnchanged'] = result['ImageSha256After'] == source_hash result['Passed'] = bool(result.get('FramebufferReady') and result['SerialControllersAbsent'] and result['SourceDriveReadOnly'] and result.get('QmpQuitSent') and result.get('QemuExitCode') == 0 and result['ImageUnchanged']) result['ElapsedSeconds'] = round(time.monotonic() - started, 3) result['RetainedScreenshots'] = [str(path) for path in kept_frames] for sample in result['Samples']: sample['ScreenshotRetained'] = pathlib.Path(sample['Screenshot']).exists() (out / 'Result.json').write_text(json.dumps(result, indent=2), encoding='utf-8') reference.close() print(json.dumps({'Passed': result['Passed'], 'Result': str(out / 'Result.json'), 'Failure': result.get('Failure')}, ensure_ascii=False)) return 0 if result['Passed'] else 1 if __name__ == '__main__': raise SystemExit(main())