XFESystem

"""Exercise real xHCI USB HID packets, with the emulated i8042 absent."""
import argparse
import json
from pathlib import Path
import shutil
import socket
import subprocess
import time
from PIL import Image
from TestBoot import ROOT, Qmp, digest, read_json, exercise_desktop
from IndependentConsoleEvidence import run_console, terminal_command_output
from IndependentBingEvidence import json_blocks


def verify(image, hub=False, bootstrap=False, edition='Desktop'):
    proof = ROOT / 'build/IndependentKernel' / ('UsbInput-' + str(time.time_ns()))
    proof.mkdir(parents=True)
    image = Path(image).resolve()
    result = {'Kind': 'XfeIndependentUsbInputProof', 'Passed': False, 'ImagePath': str(image),
              'ImageSha256': digest(image), 'Ps2ControllerPresent': False, 'Hub': hub, 'Edition': edition,
              'PhysicalCorsairVerified': False, 'GuestCudaVerified': False, 'QemuArguments': [],
              'SourceHashes': {str(p.relative_to(ROOT)).replace('\\', '/'): digest(p) for p in
                  [Path(__file__), *list((ROOT/'Source/XFESystem.NativeKernel').glob('*.cs'))]}}
    manifest=read_json(str(image)+'.json')
    build=manifest['IndependentKernelBuild']
    assert manifest['Edition']==edition and build['Edition']==edition and not build['BootstrapOnly'] or bootstrap
    result['KernelSha256']=manifest['KernelSha256']
    result['SourceHashes'].update(build['SourceSha256'])
    result['SourceHashes'].update(manifest['BootShimSourceSha256'])
    for relative in ('Tools/IndependentKernel/TestBoot.py','Tools/IndependentKernel/IndependentConsoleEvidence.py',
                     'Tools/IndependentKernel/IndependentDocumentEvidence.py','Tools/IndependentKernel/IndependentBingEvidence.py'):
        result['SourceHashes'][relative]=digest(ROOT/relative)
    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]
    usb = ['-device', 'qemu-xhci,id=XfeUsbXhci']
    bus = 'XfeUsbXhci.0'
    if hub:
        usb += ['-device', 'usb-hub,id=XfeInputHub,bus=' + bus + ',port=1']
        usb += ['-device', 'usb-kbd,id=XfeKeyboard,bus='+bus+',port=1.1', '-device', 'usb-mouse,id=XfeMouse,bus='+bus+',port=1.2']
    else:
        usb += ['-device', 'usb-kbd,id=XfeKeyboard,bus='+bus, '-device', 'usb-mouse,id=XfeMouse,bus='+bus]
    command = [str(ROOT/'.tools/qemu/qemu-system-x86_64.exe'), '-machine', 'q35,i8042=off', '-accel', 'tcg',
        '-cpu', 'max', '-smp', '8', '-m', '1024M', '-vga', 'std', '-display', 'none', '-net', 'none',
        '-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"}',
        '-qmp', f'tcp:127.0.0.1:{port},server=on,wait=off', '-no-reboot', '-no-shutdown', *usb]
    result['QemuArguments'] = command
    qemu = qmp = None
    def text():
        p=proof/'Serial.log'
        log=p.read_text(encoding='utf-8',errors='replace') if p.exists() else ''
        assert not any(x in log for x in ('XFE_NATIVE_FAIL','XFE_NATIVE_EXCEPTION','XFE_RUNTIME_FATAL','XFE_ZERO_FAILFAST')), log[-4096:]
        return log
    def wait(marker, timeout=60):
        end=time.monotonic()+timeout
        while time.monotonic()<end:
            assert qemu.poll() is None, 'QEMU exited'
            if marker in text(): return
            time.sleep(.1)
        raise AssertionError('USB timeout: '+marker+'\n'+text()[-4096:])
    try:
        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))
            end=time.monotonic()+15
            while qmp is None:
                assert qemu.poll() is None, (proof/'Qemu.log').read_text()
                try: qmp=Qmp(port)
                except OSError:
                    assert time.monotonic()<end
                    time.sleep(.1)
            wait('XFE_USB|INITIAL_ENUMERATION_COMPLETE')
            log=text()
            assert '|Kind|1|' in log and '|Kind|2|' in log, 'Both USB keyboard and mouse must be configured: '+log[-4096:]
            if hub:
                assert '|HUB_CONTEXT_CONFIGURED' in log, 'The xHC must complete Configure Endpoint before enumerating hub children'
                result['HubContextReadbackVerified']='|HUB_CONTEXT_APPLIED' in log
                result['QemuHubContextNotModeled']='|HUB_CONTEXT_QEMU_NOT_MODELED' in log
            tree=qmp.call('human-monitor-command',{'command-line':'info qtree'})
            assert 'dev: i8042' not in tree and 'dev: ps2-kbd' not in tree and 'dev: ps2-mouse' not in tree
            assert 'dev: usb-kbd' in tree and 'dev: usb-mouse' in tree
            (proof/'Qtree.txt').write_text(tree,encoding='utf-8')
            result['QtreeSha256']=digest(proof/'Qtree.txt')
            qmp.call('human-monitor-command',{'command-line':'sendkey a'})
            qmp.call('input-send-event',{'events':[{'type':'rel','data':{'axis':'x','value':30}}]})
            wait('XFE_USB|FIRST_HARDWARE_KEY_QUEUED');wait('XFE_USB|FIRST_HARDWARE_POINTER_QUEUED')
            result['HardwareReportsQueued']=True
            if not bootstrap:
                wait('XFE_INDEPENDENT_CONSOLE_FIRST_FRAME' if edition=='Console' else 'XFE_INDEPENDENT_DESKTOP_FIRST_FRAME')
                # The initial probe above typed into Files, not an editor.
                # Wait for release, then use only actual USB packets end to end.
                time.sleep(.4)
                # Restore the known pointer starting point after our 30px probe.
                qmp.call('input-send-event',{'events':[{'type':'rel','data':{'axis':'x','value':-30}}]})
                if edition=='Console':
                    qmp.call('human-monitor-command',{'command-line':'sendkey ctrl-c'});time.sleep(.3)
                    result['ConsoleInputProof']=run_console(qmp,proof,text,input_kind='USB HID')
                    result['ActualUsbConsoleEditingPassed']=True
                else:
                    result['DesktopInputProof']=exercise_desktop(qmp,proof,text,False,False,False,input_kind='USB HID')
                    result['ActualUsbDesktopEditingPassed']=True
                # Wait for the real removal before reconnecting the keyboard.
                before=len(text());qmp.call('device_del',{'id':'XfeKeyboard'})
                end=time.monotonic()+15
                while '|DISCONNECTED' not in text()[before:]:
                    assert time.monotonic()<end,'USB unplug not observed';time.sleep(.1)
                before=len(text())
                qmp.call('device_add',{'driver':'usb-kbd','id':'XfeKeyboardReloaded','bus':bus,'port':'1.1' if hub else '1'})
                end=time.monotonic()+15
                while 'XFE_USB_HID_READY' not in text()[before:]:
                    assert time.monotonic()<end,'USB replug not enumerated';time.sleep(.1)
                before=len(text())
                for key in ['e','c','h','o','spc','u','s','b','minus','r','e','p','l','u','g','ret']:
                    qmp.call('human-monitor-command',{'command-line':'sendkey '+key});time.sleep(.15)
                end=time.monotonic()+10
                passed=False
                final_packet=None
                prefix='XFE_CONSOLE_SNAPSHOT' if edition=='Console' else 'XFE_TERMINAL_SNAPSHOT'
                while time.monotonic()<end and not passed:
                    for item in json_blocks(text(),prefix,before):
                        try:
                            passed='usb-replug' in terminal_command_output(item['Value'],'echo usb-replug')
                            if passed: final_packet=item
                        except AssertionError: pass
                    time.sleep(.1)
                assert passed,'Replugged USB keyboard did not submit and render its own command output'
                result['ActualKeyboardUnplugReplugPassed']=True
                result['RepluggedCommandReadback']=final_packet
            qmp.call('screendump',{'filename':str(proof/'Frame.ppm')})
            with Image.open(proof/'Frame.ppm') as image_frame: image_frame.convert('RGB').save(proof/'Frame.png')
            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 Exception: pass
            try: qemu.wait(timeout=10)
            except subprocess.TimeoutExpired: qemu.kill();qemu.wait(timeout=10);result['ForcedCleanup']=True
            result['QemuExitCode']=qemu.returncode
        if qmp: qmp.close()
        result['ImageUnchanged']=digest(image)==result['ImageSha256']
        if (proof/'Serial.log').exists(): result['SerialSha256']=digest(proof/'Serial.log')
        for relative,sha in result['SourceHashes'].items():
            if digest(ROOT/relative)!=sha: result['Passed']=False;result['Failure']='Sources changed: '+relative
        result['SourcesUnchanged']=all(digest(ROOT/p)==sha for p,sha in result['SourceHashes'].items())
        (proof/'Result.json').write_text(json.dumps(result,indent=2),encoding='utf-8')
    print(json.dumps({'Passed':result['Passed'],'Proof':str(proof/'Result.json'),'Failure':result.get('Failure')}))
    return result['Passed']

if __name__=='__main__':
    parser=argparse.ArgumentParser();parser.add_argument('--image',type=Path,required=True)
    parser.add_argument('--hub',action='store_true');parser.add_argument('--bootstrap',action='store_true')
    parser.add_argument('--edition',choices=['Console','Desktop'],default='Desktop')
    args=parser.parse_args();raise SystemExit(0 if verify(args.image,args.hub,args.bootstrap,args.edition) else 1)