XFESystem

"""Verify a source-bound independent kernel in a private QEMU/UEFI guest."""
import argparse
import hashlib
import json
import re
from pathlib import Path
import shutil
import socket
import subprocess
import sys
import time
from PIL import Image

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / 'Tools' / 'CSharpBoot'))
sys.path.insert(0, str(ROOT / 'Tools'))
from ReadImage import inspect_image
from TestBrowserPipeline import _monaco_frame_stats, _monaco_fourth_line_delta
from IndependentDocumentEvidence import PRESENTED, run_documents, seed_documents, window_geometry
from IndependentBingEvidence import json_blocks, native_network_profile, network_arguments, run_bing, validate_network_arguments


def digest(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def read_json(path):
    return json.loads(Path(path).read_text(encoding='utf-8-sig'))


def terminal_command_output(snapshot, command):
    """Read only output following the latest exact echoed command, never history arguments."""
    assert snapshot.get('Busy') is False and snapshot.get('Input') == '', 'The actual terminal is still busy or has unsubmitted input'
    assert snapshot.get('LastCommand') == command, 'The actual terminal submitted another command'
    prompt = snapshot.get('Prompt')
    assert isinstance(prompt, str) and prompt.endswith('> ') and snapshot.get('Version'), 'The actual terminal has no versioned prompt'
    lines = [line.rstrip() for line in snapshot.get('Text', '').replace('\r', '').split('\n')]
    matches = [index for index, line in enumerate(lines) if line == prompt + command]
    assert matches, 'No exact actual echoed terminal command in the rendered snapshot'
    output = lines[matches[-1] + 1:]
    while output and not output[-1]: output.pop()
    assert not any(line.startswith(('Error:', 'Unknown command:')) for line in output), 'The actual terminal command failed: ' + repr(output)
    return output


def run_terminal_commands(proof, text, healthy, screenshot, key, type_text):
    """Real PS/2 CLI acceptance. No host storage reads or injected shell execution."""
    started = time.monotonic(); beginning = len(text()); selected = []
    window_id, _ = window_geometry(healthy(), 'Terminal')
    result = {'Kind': 'XfeIndependentClassicTerminalWorkflow', 'Passed': False,
              'Input': 'Actual QMP PS2 keyboard packets only', 'WindowId': window_id,
              'HostCommandExecution': False, 'HostWorkspaceReads': False, 'Commands': []}
    output_path = proof / 'TerminalWorkflow.json'

    def wait_command(after, command, required_lines=()):
        deadline = time.monotonic() + 30
        last_error = 'No fresh complete rendered terminal snapshot'
        while time.monotonic() < deadline:
            packets = [item for item in json_blocks(healthy(), 'XFE_TERMINAL_SNAPSHOT', after)
                       if item['Value'].get('WindowId') == window_id]
            if packets:
                packet = packets[-1]
                try:
                    lines = terminal_command_output(packet['Value'], command)
                    assert all(required in lines for required in required_lines), 'Missing actual post-command output: ' + repr(required_lines)
                    selected.append(packet)
                    result['Commands'].append({'Command': command, 'Passed': True, 'SerialOffset': packet['SerialOffset'],
                        'Prompt': packet['Value']['Prompt'], 'Version': packet['Value']['Version'], 'Busy': False,
                        'PostCommandOutput': lines})
                    return packet['Value'], lines
                except AssertionError as error:
                    last_error = str(error)
            time.sleep(.1)
        raise AssertionError('Actual terminal command timeout: ' + command + ': ' + last_error)

    def execute(command, required_lines=()):
        after = len(text()); type_text(command); key('ret')
        return wait_command(after, command, required_lines)

    try:
        snapshot, help_lines = execute('help')
        assert any('md Path | mf Path | cat Path | write Path Text' in line for line in help_lines), 'The actual CLI help lacks real file commands'
        assert any('Up/Down: history' in line for line in help_lines), 'The actual CLI help lacks history/caret editing'
        assert 'Run Demo' not in snapshot['Text'], 'The removed demo-only terminal content is still present'
        result['HelpPassed'] = True
        execute('md CliProof')
        execute('mf CliProof/Command.txt')
        write_command = 'write CliProof/Command.txt NativeTerminalReady'
        execute(write_command)
        execute('cat CliProof/Command.txt', ('NativeTerminalReady',))
        result['RamDirectoryCreateWriteReadPassed'] = True

        # Go beyond the most recent command. Executing the recalled command
        # must produce its exact identity after the actual native render.
        after = len(text()); key('up'); key('up'); key('ret')
        wait_command(after, write_command)
        result['HistoryBeyondPreviousPassed'] = True
        after = len(text()); type_text('echo draft'); key('up'); key('down'); key('ret')
        wait_command(after, 'echo draft', ('draft',))
        result['HistoryDraftRestorePassed'] = True

        # Home/right/delete insert a missing scalar; left plus overwrite mode
        # replaces the last scalar. The submitted command and its own output
        # both need to match, rather than searching previous echo history.
        after = len(text()); type_text('echo aXc'); key('home')
        for _ in range(6): key('right')
        key('delete'); type_text('b'); key('right'); key('left'); key('insert'); type_text('d'); key('insert'); key('end'); key('ret')
        wait_command(after, 'echo abd', ('abd',))
        result['CaretDeleteInsertEditingPassed'] = True
        _, history_lines = execute('history')
        assert any(write_command in line for line in history_lines) and any('echo abd' in line for line in history_lines), 'Actual history omitted executed commands'
        result['HistoryListingPassed'] = True
        result['Screenshot'] = {'Path': 'Terminal-ClassicCLI.png', 'Sha256': screenshot('Terminal-ClassicCLI')}
        result['NoDemoOnlyTerminalContent'] = True
        result['Passed'] = True
    except Exception as error:
        result['Failure'] = str(error)
        raise
    finally:
        raw = proof / 'TerminalCommandEvidence.log'
        raw.write_text('\n'.join(item['Raw'] for item in selected), encoding='utf-8')
        result['RawGuestEvidence'] = {'Path': raw.name, 'Sha256': digest(raw), 'SelectedRenderedSnapshots': len(selected), 'StartSerialOffset': beginning}
        result['ElapsedHostSeconds'] = time.monotonic() - started
        output_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
    return result


class Qmp:
    def __init__(self, port):
        self.connection = socket.create_connection(('127.0.0.1', port), timeout=10)
        self.stream = self.connection.makefile('rwb')
        self.number = 0
        greeting = json.loads(self.stream.readline())
        assert 'QMP' in greeting, greeting
        self.call('qmp_capabilities')

    def call(self, command, arguments=None):
        self.number += 1
        item = {'execute': command, 'id': self.number}
        if arguments is not None:
            item['arguments'] = arguments
        self.stream.write(json.dumps(item).encode() + b'\n')
        self.stream.flush()
        while True:
            line = self.stream.readline()
            if not line:
                raise RuntimeError('QEMU closed QMP unexpectedly')
            result = json.loads(line)
            if result.get('id') == self.number:
                if 'error' in result:
                    raise RuntimeError(result['error'])
                return result.get('return')

    def close(self):
        self.stream.close()
        self.connection.close()


def exercise_desktop(qmp, proof, text, browser_required, documents_required=False, bing_required=False):
    pointer = [640, 400]
    def healthy():
        log = text()
        assert not any(marker in log for marker in ('XFE_NATIVE_FAIL', 'XFE_NATIVE_EXCEPTION', 'XFE_RUNTIME_FATAL',
                                                     'XFE_ZERO_FAILFAST', 'XFE_INDEPENDENT_APPLICATION_FAILED',
                                                     'XFE_BROWSER_PAINT_FAILED', 'Rendering failed:',
                                                     'XFE_EDITOR_BACKGROUND|SessionError|')), log[-4096:]
        return log
    def wait(marker, start=0, seconds=90):
        end = time.monotonic() + seconds
        while time.monotonic() < end:
            log = healthy()
            if marker in log[start:]:
                return
            time.sleep(.1)
        raise AssertionError('Desktop acknowledgement timeout: ' + marker + '\n' + healthy()[-4096:])
    def screenshot(name):
        path = proof / (name + '.ppm')
        qmp.call('screendump', {'filename': str(path)})
        with Image.open(path) as frame:
            frame.convert('RGB').save(proof / (name + '.png'))
        return digest(proof / (name + '.png'))
    def key(chord):
        qmp.call('human-monitor-command', {'command-line': 'sendkey ' + chord})
        time.sleep(.15)
    def type_text(value):
        special = {':': 'shift-semicolon', '.': 'dot', '/': 'slash', '-': 'minus', '_': 'shift-minus', ' ': 'spc', '\n': 'ret'}
        for char in value:
            key(special.get(char, 'shift-' + char.lower() if char.isupper() else char))
    def move(x, y):
        while pointer != [x, y]:
            dx, dy = max(-80, min(80, x-pointer[0])), max(-80, min(80, y-pointer[1]))
            qmp.call('input-send-event', {'events': [
                {'type': 'rel', 'data': {'axis': 'x', 'value': dx}},
                {'type': 'rel', 'data': {'axis': 'y', 'value': dy}}]})
            pointer[0] += dx; pointer[1] += dy
            time.sleep(.025)
    def click(x, y):
        move(x, y)
        for down in (True, False):
            qmp.call('input-send-event', {'events': [{'type': 'btn', 'data': {'down': down, 'button': 'left'}}]})
            time.sleep(.1)
    def wheel(count):
        for _ in range(count):
            for down in (True, False):
                qmp.call('input-send-event', {'events': [{'type': 'btn', 'data': {'down': down, 'button': 'wheel-down'}}]})
                time.sleep(.1)
    if 'XFE_DESKTOP_ACTIVE|Terminal' in healthy():
        start = len(text()); key('alt-f4'); wait('XFE_DESKTOP_ACTIVE|Files', start)
    else:
        wait('XFE_DESKTOP_ACTIVE|Files')
    files_hash = screenshot('Desktop-Files')
    start = len(text()); click(370, 116)
    wait('XFE_DESKTOP_ACTIVE|Editor', start)
    type_text('hello')
    start = len(text()); key('ctrl-s')
    wait('XFE_DESKTOP_SAVED|X:\\Users\\Default\\Documents\\Document1.txt|5', start)
    wait('XFE_DESKTOP_SAVED_TEXT_BEGIN\nhello\nXFE_DESKTOP_SAVED_TEXT_END', start)
    editor_hash = screenshot('Desktop-Editor-Saved')
    assert files_hash != editor_hash, 'Editor input produced no changed guest framebuffer'
    start = len(text()); key('alt-f4'); wait('XFE_DESKTOP_ACTIVE|Files', start)
    start = len(text()); click(220, 170); wait('XFE_DESKTOP_ACTIVE|Editor', start)
    start = len(text()); key('ctrl-s')
    wait('XFE_DESKTOP_SAVED|X:\\Users\\Default\\Documents\\Document1.txt|5', start)
    wait('XFE_DESKTOP_SAVED_TEXT_BEGIN\nhello\nXFE_DESKTOP_SAVED_TEXT_END', start)
    start = len(text()); key('alt-f4'); wait('XFE_DESKTOP_ACTIVE|Files', start)
    start = len(text()); key('ctrl-esc'); wait('XFE_DESKTOP_FLAGS|1|0|', start)
    key('down'); key('down'); key('ret')
    wait('XFE_DESKTOP_ACTIVE|Browser', start)
    wait('XFE_DESKTOP_STATE_END', start)
    bing_proof = None
    if bing_required:
        # The actual menu input above opened the product's unchanged Bing
        # default. Never substitute a host download or offline page here.
        bing_proof = run_bing(proof, text, healthy, wait, screenshot, key, type_text, move,
                              open_browser=False, start_offset=start)
    browser_id, _ = window_geometry(healthy(), 'Browser')
    # Let the genuine default navigation publish before editing its address;
    # its completion writes the page URL back to the same address model.
    wait('XFE_EDITOR_BACKGROUND|Ready|Window|' + str(browser_id) + '|', start, seconds=180)
    # The merged desktop preserves its Bing default URL. Exercise the built-in
    # JavaScript page through the actual address bar instead of assuming it is
    # the newly opened page or changing the product default for the harness.
    start = len(text()); key('ctrl-l'); type_text('about:Start'); key('ret')
    wait('JavaScript ready: 12', start)
    browser_hash = screenshot('Browser-JavaScript')
    assert browser_hash not in (files_hash, editor_hash), 'Browser did not present a distinct guest frame'
    details = {'MouseCreatedEditor': True, 'KeyboardSavedRamFile': True, 'SavedRamFileReopened': True, 'AltF4ClosedEditor': True,
               'StartMenuKeyboardNavigation': True, 'JavaScriptDomResult': 'JavaScript ready: 12',
               'Screenshots': {'Files': files_hash, 'EditorSaved': editor_hash, 'Browser': browser_hash}}
    if bing_required:
        assert bing_proof['Passed'] is True and bing_proof['NativeHttpsBingLoaded'] is True
        details['BingEvidence'] = {'Path': 'BingEvidence.json', 'Sha256': digest(proof / 'BingEvidence.json'),
                                  'Passed': True, 'NativeHttpsBingLoaded': True}
    if browser_required:
        start = len(text()); key('ctrl-l'); type_text('about:Monaco'); key('ret')
        model_pattern = re.compile(r'XFE_MONACO_VALUE_BEGIN\r?\n(.*?)\r?\nXFE_MONACO_VALUE_END', re.S)
        # The background protocol publishes model snapshots rather than the
        # page's DOM status label. Require the actual three-line initial model.
        expected_initial = '// XFESystem | XFEstudio\nconst answer = 6 * 7;\nconsole.log(answer);'
        deadline = time.monotonic() + 180
        models = []
        while time.monotonic() < deadline:
            models = model_pattern.findall(healthy()[start:])
            if models and models[-1] == expected_initial: break
            time.sleep(.1)
        assert models and models[-1] == expected_initial, 'The actual initial Monaco model is unavailable or differs'
        details['MonacoInitialModel'] = models[-1]
        # Background page owners publish immutable paint snapshots. The UI
        # must acknowledge actual scanout before glyph evidence can qualify.
        window_id, _ = window_geometry(healthy(), 'Browser')
        deadline = time.monotonic() + 90
        presented = None
        while time.monotonic() < deadline:
            matches = [item for item in PRESENTED.finditer(healthy(), start)
                       if int(item[1]) == window_id and '/Monaco/' in item[3] and int(item[5]) > 0 and int(item[6]) > 0]
            if matches:
                presented = matches[-1]; break
            time.sleep(.1)
        assert presented is not None, 'The actual Monaco snapshot was not presented to scanout'
        details['MonacoPresentation'] = {'WindowId': window_id, 'Address': presented[3],
            'SnapshotSequence': int(presented[4]), 'PresentationSequence': int(presented[5]), 'PresentationTicks': int(presented[6])}
        details['MonacoRasterPaintCompleted'] = True
        windows = re.findall(r'XFE_WINDOW_STATE\|\d+\|Browser\|(-?\d+)\|(-?\d+)\|(\d+)\|(\d+)\|', text())
        assert windows, 'Actual browser window geometry is missing'
        left, top, width, height = map(int, windows[-1])
        viewport = (left + 2, top + 67, width - 4, height - 96)
        move(left + width - 20, top + 50)
        def paint_frame(name, edited=False, before=None):
            end = time.monotonic() + 60
            attempt = 0
            while time.monotonic() < end:
                healthy(); attempt += 1
                screenshot(name + '-' + str(attempt))
                with Image.open(proof / (name + '-' + str(attempt) + '.png')) as pixels:
                    frame = pixels.copy()
                stats = _monaco_frame_stats(frame, viewport, 1)
                if edited:
                    delta = _monaco_fourth_line_delta(before, frame, stats['Lines'][3]['NativeBounds'])
                    stats['FourthLineDelta'] = delta
                    visible = stats['Lines'][3]['HasTextGlyphDistribution'] and delta['NewInkPixels'] >= 35 and delta['NewInkColumns'] >= 24
                else:
                    visible = stats['ReadyPainted']
                if visible:
                    return frame, stats
                time.sleep(.25)
            raise AssertionError('Actual Monaco glyph/syntax pixels did not appear: ' + repr(stats))
        initial_frame, initial_stats = paint_frame('Browser-Monaco-Ready')
        details['MonacoReadyPaint'] = initial_stats
        start = len(text()); click(viewport[0] + 160, viewport[1] + 70)
        key('ctrl-end'); key('ret'); type_text('xfenativeedit')
        expected_edited = expected_initial + '\nxfenativeedit'
        end = time.monotonic() + 180
        while time.monotonic() < end:
            models = model_pattern.findall(healthy()[start:])
            if models and models[-1] == expected_edited: break
            time.sleep(.1)
        assert models and models[-1] == expected_edited, 'PS/2 input did not update the actual Monaco model exactly'
        details['MonacoEditedModel'] = models[-1]
        move(left + width - 20, top + 50)
        _, edited_stats = paint_frame('Browser-Monaco-Edited', True, initial_frame)
        details['MonacoEditedPaint'] = edited_stats
        details['MonacoKeyboardEditedModel'] = True
        details['Screenshots']['Monaco'] = screenshot('Browser-Monaco')
        details['OfficialMonacoReady'] = True
    # Keep the actual event loop alive after input and verify it responds again.
    time.sleep(2)
    start = len(text()); key('alt-f4'); wait('XFE_DESKTOP_ACTIVE|Files', start)
    screenshot('Desktop-After-Browser-Close')
    if documents_required:
        details['DocumentWorkflowProof'] = run_documents(proof, text, healthy, wait, screenshot,
            key, type_text, move, click, wheel)
    _, files_bounds = window_geometry(healthy(), 'Files')
    managed_row = seed_documents(True)['Names'].index('ManagedHello.dll')
    start = len(text()); click(files_bounds[0] + 70, files_bounds[1] + 110 + managed_row * 28)
    wait('XFE_INDEPENDENT_CONSOLE_IL_OK', start)
    wait('XFE_DESKTOP_ACTIVE|Terminal', start)
    wait('XFE_INDEPENDENT_CONSOLE_PRESENTED|Path|X:\\Users\\Default\\Documents\\ManagedHello.dll|Window|', start)
    details['ActualConsoleILSamplePassed'] = True
    details['Screenshots']['ConsoleIL'] = screenshot('Terminal-ConsoleIL')
    # Real terminal command dispatch replaces the removed demo-only toolbar.
    # The sample itself must produce its two actual ANSI/XFEConsole markers;
    # merely accepting a queued command cannot qualify program execution.
    start = len(text()); type_text('run ManagedHello.dll'); key('ret')
    wait('XFE_INDEPENDENT_CONSOLE_IL_OK', start)
    wait('XFE_INDEPENDENT_CONSOLE_PRESENTED|Path|X:\\Users\\Default\\Documents\\ManagedHello.dll|Window|', start)
    details['TerminalRunManagedSamplePassed'] = True
    details['ActualAnsiAndXfeConsoleDemoPassed'] = True
    details['Screenshots']['ConsoleDemo'] = screenshot('Terminal-Console-Demo')
    details['TerminalWorkflowProof'] = run_terminal_commands(proof, text, healthy, screenshot, key, type_text)
    assert details['TerminalWorkflowProof']['Screenshot']['Sha256'] != details['Screenshots']['ConsoleDemo'], 'Actual CLI commands produced no changed guest framebuffer'
    healthy()
    details['NoRenderFailures'] = True
    return details


def verify_boot(kernel_path, cpus, serial_enabled, runtime_required, timeout, desktop_required=False, browser_required=False, documents_required=False, bing_required=False):
    build = read_json(str(kernel_path) + '.json')
    assert build['Kind'] == 'XfeIndependentKernelBuild' and build['Passed'] is True
    assert build['ContainsCosmos'] is False and build['KernelSha256'] == digest(kernel_path)
    assert build['Target'] == 'x86_64-none'
    if runtime_required:
        assert build['BootstrapOnly'] is False, 'A bootstrap build cannot pass managed-runtime acceptance'
        assert serial_enabled, 'Runtime checks need independent serial evidence'
    if desktop_required:
        assert not build['BootstrapOnly'] and serial_enabled, 'Actual managed desktop acceptance requires a full build and serial evidence'
        assert not runtime_required, 'Runtime-probe acceptance and managed-application acceptance target separate entry points'
    for relative, expected in build['SourceSha256'].items():
        path = (ROOT / relative).resolve()
        assert path.is_relative_to(ROOT) and digest(path) == expected, f'Source changed: {relative}'
    proof = ROOT / 'build' / 'IndependentKernel' / f'Proof-{time.time_ns()}'
    proof.mkdir(parents=True)
    image = proof / 'System.img'
    result = {'Kind': 'XfeIndependentKernelBootProof', 'Passed': False, 'ContainsCosmos': False,
              'VirtualCpus': cpus, 'ExecutingCpus': 1, 'SerialDevice': serial_enabled,
              'RuntimeAcceptanceRequired': runtime_required, 'DesktopAcceptanceRequired': desktop_required,
              'BrowserAcceptanceRequired': browser_required, 'ManagedRuntimeAcceptancePassed': False,
              'DocumentAcceptanceRequired': documents_required, 'DocumentAcceptancePassed': False,
              'BingAcceptanceRequired': bing_required, 'BingAcceptancePassed': False,
              'NativeNetworkProfile': native_network_profile(bing_required),
              'DesktopAcceptancePassed': False, 'BrowserAcceptancePassed': False,
              'PhysicalHardwareAcceptancePassed': False, 'FullReleaseQualified': False,
              'KernelSha256': digest(kernel_path), 'BuildReceipt': str(kernel_path) + '.json',
              'BuildReceiptSha256': digest(str(kernel_path) + '.json'),
              'RamOnly': True, 'StorageEnabled': False, 'ProofDirectory': str(proof)}
    qemu = None
    qmp = None
    try:
        with (proof / 'PackImage.log').open('w', encoding='utf-8') as pack_log:
            subprocess.run(['dotnet', 'run', '--project', str(ROOT / 'Tools/CSharpBoot/PackImage.csproj'),
                        '-c', 'Release', '--no-build', '--', str(kernel_path), str(image),
                        '1280x800', '--independent-kernel'], cwd=ROOT, check=True,
                       stdout=pack_log)
        inspected, files = inspect_image(image)
        assert files['System/Kernel.elf'] == kernel_path.read_bytes()
        assert read_json(str(image) + '.json')['KernelImplementation'] == 'XFESystem independent kernel'
        result.update(ImagePath=str(image), ImageSha256=digest(image), ImageVerification=inspected,
                      BootShimSha256=inspected['Files']['EFI/BOOT/BOOTX64.EFI'])
        sources = dict(build['SourceSha256'])
        sources.update(inspected['BootShimSourceSha256'])
        for harness in ('Tools/IndependentKernel/TestBoot.py', 'Tools/IndependentKernel/IndependentDocumentEvidence.py',
                        'Tools/IndependentKernel/IndependentBingEvidence.py',
                        'Tools/TestBrowserPipeline.py', 'Tools/TestDesktopDocuments.py', 'Tools/TestWebsiteDocument.py',
                        'Tools/NativeCSharpEvidence.py'):
            sources[harness] = digest(ROOT / harness)
        snapshot = proof / 'SourceSnapshot.json'
        snapshot.write_text(json.dumps(sources, indent=2), encoding='utf-8')
        result.update(SourceSnapshotPath=str(snapshot), SourceSnapshotSha256=digest(snapshot))
        shutil.copy2(ROOT / '.tools/firmware/edk2-i386-vars.fd', proof / 'FirmwareVars.fd')
        with socket.socket() as reservation:
            reservation.bind(('127.0.0.1', 0))
            port = reservation.getsockname()[1]
        command = [str(ROOT / '.tools/qemu/qemu-system-x86_64.exe'), '-machine', 'q35',
                   '-accel', 'tcg', '-cpu', 'max', '-smp', str(cpus), '-m', '1024M',
                   '-vga', 'std', '-global', 'VGA.vgamem_mb=64', '-display', 'none', *network_arguments(bing_required), '-rtc', 'base=utc',
                   '-drive', f'if=pflash,format=raw,readonly=on,file={ROOT / ".tools/firmware/edk2-x86_64-code.fd"}',
                   '-drive', f'if=pflash,format=raw,file={proof / "FirmwareVars.fd"}',
                   '-drive', f'file={image},format=raw,if=ide,snapshot=on',
                   '-serial', f'file:{proof / "Serial.log"}' if serial_enabled else 'none',
                   '-qmp', f'tcp:127.0.0.1:{port},server=on,wait=off', '-no-reboot', '-no-shutdown']
        assert validate_network_arguments(command, bing_required) == result['NativeNetworkProfile']
        result['QemuArguments'] = command
        with (proof / 'Qemu.log').open('w', encoding='utf-8') as log:
            qemu = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT,
                                    creationflags=getattr(subprocess, 'CREATE_NO_WINDOW', 0))
            deadline = time.monotonic() + timeout
            while qmp is None:
                assert qemu.poll() is None, 'QEMU exited before QMP was available'
                try:
                    qmp = Qmp(port)
                except (OSError, ConnectionError):
                    assert time.monotonic() < deadline, 'QMP connection timeout'
                    time.sleep(0.1)
            ready = False
            serial = ''
            frame = proof / 'Frame.ppm'
            while time.monotonic() < deadline:
                assert qemu.poll() is None, 'Guest/QEMU exited during boot'
                if serial_enabled:
                    serial = (proof / 'Serial.log').read_text(encoding='utf-8', errors='replace')
                    assert not any(marker in serial for marker in ('XFE_NATIVE_FAIL', 'XFE_NATIVE_EXCEPTION', 'XFE_RUNTIME_FATAL',
                                                                   'XFE_ZERO_FAILFAST', 'XFE_INDEPENDENT_APPLICATION_FAILED')), serial[-4096:]
                    ready = 'XFE_NATIVE_SELFTEST_OK' in serial
                    if runtime_required:
                        ready = ready and 'XFE_DOTNET_RUNTIME_PROBE_OK' in serial
                    if desktop_required:
                        ready = ready and 'XFE_INDEPENDENT_DESKTOP_FIRST_FRAME' in serial and 'XFE_APPLICATION_RUNTIME_CHECK_OK' in serial
                else:
                    qmp.call('screendump', {'filename': str(frame)})
                    with Image.open(frame) as rendered:
                        rgb = rendered.convert('RGB')
                        ready = rgb.size == (1280, 800) and all(rgb.getpixel((x, 4)) == (56, 199, 164)
                                                               for x in (0, 320, 640, 960, 1279))
                if ready:
                    break
                time.sleep(0.2)
            assert ready, 'Independent kernel boot/self-test timeout: ' + serial[-4096:]
            if serial_enabled:
                for marker in ('XFE_CPU_OK', 'XFE_MEMORY_MAP_OK', 'XFE_PAGES_OK', 'XFE_LIBC_OK',
                               'XFE_STDIO_SELFTEST_OK', 'XFE_LOADED_ELF_SELFTEST_OK', 'XFE_THREADS_OK', 'XFE_INPUT_READY',
                               'XFE_DESCRIPTORS_OK', 'XFE_VIRTUAL_MEMORY_OK', 'XFE_CLOCK_OK',
                               'XFE_FRAMEBUFFER_OK', 'XFE_NATIVE_SELFTEST_OK'):
                    assert marker in serial, marker
                result['GuestMarkers'] = [line for line in serial.splitlines() if line.startswith('XFE_')]
            qmp.call('screendump', {'filename': str(frame)})
            with Image.open(frame) as rendered:
                rgb = rendered.convert('RGB')
                assert rgb.size == (1280, 800)
                rgb.save(proof / 'Frame.png')
                result['ScreenshotSha256'] = digest(proof / 'Frame.png')
                result['FramebufferSize'] = list(rgb.size)
            qmp.call('query-status')
            if desktop_required:
                result['DesktopInputProof'] = exercise_desktop(qmp, proof,
                    lambda: (proof / 'Serial.log').read_text(encoding='utf-8', errors='replace'), browser_required, documents_required, bing_required)
                result['DesktopAcceptancePassed'] = True
                result['BrowserAcceptancePassed'] = browser_required
                result['DocumentAcceptancePassed'] = documents_required
                result['BingAcceptancePassed'] = bing_required and result['DesktopInputProof']['BingEvidence']['Passed']
            result['ManagedRuntimeAcceptancePassed'] = runtime_required or desktop_required
            result['Passed'] = True
    except Exception as error:
        result['Failure'] = str(error)
    finally:
        if qemu is not None:
            if qemu.poll() is None and qmp is not None:
                try:
                    qmp.call('quit')
                except (OSError, RuntimeError):
                    pass
            try:
                qemu.wait(timeout=10)
            except subprocess.TimeoutExpired:
                qemu.terminate()
                qemu.wait(timeout=10)
                result['ForcedCleanup'] = True
            result['QemuExitCode'] = qemu.returncode
        if qmp is not None:
            qmp.close()
        if image.exists():
            result['ImageUnchanged'] = digest(image) == result.get('ImageSha256')
        result['Passed'] = result['Passed'] and result.get('QemuExitCode') == 0 and result.get('ImageUnchanged') is True
        (proof / 'Result.json').write_text(json.dumps(result, indent=2), encoding='utf-8')
    print(json.dumps(result, ensure_ascii=True))
    return result['Passed']


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('kernel', type=Path)
    parser.add_argument('--virtual-cpus', type=int, choices=(1, 8), default=1)
    parser.add_argument('--no-serial', action='store_true')
    parser.add_argument('--runtime', action='store_true')
    parser.add_argument('--desktop', action='store_true')
    parser.add_argument('--browser', action='store_true')
    parser.add_argument('--documents', action='store_true')
    parser.add_argument('--bing', action='store_true', help='Require real guest e1000/NAT HTTPS Bing, immutable snapshot presentation and actual framebuffer evidence')
    parser.add_argument('--timeout', type=int, default=90)
    args = parser.parse_args()
    raise SystemExit(0 if verify_boot(args.kernel.resolve(), args.virtual_cpus, not args.no_serial,
                                     args.runtime, args.timeout, args.desktop or args.browser or args.documents or args.bing,
                                     args.browser or args.documents, args.documents, args.bing) else 1)