XFESystem

"""Private 4K independent-kernel software-cadence measurement, never physical scanout."""
import argparse
import hashlib
import json
from pathlib import Path
import re
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'))
from TestPointerCadence import QmpClient, require_backing, run_pointer_cadence
sys.path.insert(0, str(ROOT/'Tools/CSharpBoot'))
from ReadImage import inspect_image, verify_single_edition


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


def run(args):
    output = args.output.resolve(); output.mkdir(parents=True, exist_ok=False)
    image = output/'System.img'
    result = {'Kind':'XfeIndependentDesktopSoftwareCadence', 'Completed':False, 'WindowsFluidityParityPassed':False,
              'PhysicalRefreshHz':None, 'Physical240HzVerified':False, 'KernelSha256':digest(args.kernel),
              'RequestedWidth':args.width, 'RequestedHeight':args.height, 'Accel':args.accel,
              'ImageSourceUnchanged':False, 'ExecutedManagedCpus':1}
    harnesses=('Tools/DesktopFluidity/RunNative.py','Tools/TestPointerCadence.py','Tools/PointerCadenceEvidence.py',
               'Tools/CSharpBoot/ReadImage.py')
    result['HarnessSha256']={relative:digest(ROOT/relative) for relative in harnesses}
    process = qmp = None
    try:
        receipt=json.loads(Path(str(args.kernel)+'.json').read_text(encoding='utf-8-sig'))
        assert receipt['Passed'] and receipt['KernelSha256']==result['KernelSha256']
        for relative, expected in receipt['SourceSha256'].items():
            assert digest(ROOT/relative)==expected, 'Compiled source changed: '+relative
        with (output/'PackImage.log').open('w',encoding='utf-8') as log:
            subprocess.run(['dotnet','run','--project',str(ROOT/'Tools/CSharpBoot/PackImage.csproj'),'-c','Release','--no-build','--',
                str(args.kernel.resolve()),str(image),'0x0','--independent-kernel'],cwd=ROOT,stdout=log,stderr=subprocess.STDOUT,check=True,timeout=90)
        inspection,files=inspect_image(image)
        manifest=json.loads(Path(str(image)+'.json').read_text(encoding='utf-8-sig'))
        assert files['System/Kernel.elf']==args.kernel.read_bytes() and manifest['Resolution']=='0x0'
        result['ImageSha256']=inspection['ImageSha256']
        result['SingleEditionImageProof']=verify_single_edition(inspection,files,manifest)
        shutil.copyfile(ROOT/'.tools/firmware/edk2-i386-vars.fd',output/'Vars.fd')
        with socket.socket() as reserve:
            reserve.bind(('127.0.0.1',0));port=reserve.getsockname()[1]
        command=[str(ROOT/'.tools/qemu/qemu-system-x86_64.exe'),'-machine','q35,i8042=off' if args.input=='usb' else 'q35','-accel',args.accel,'-cpu','max','-smp','8','-m','1024M',
            '-vga','std','-global','VGA.vgamem_mb=64','-global','VGA.edid=on',
            '-global',f'VGA.xres={args.width}','-global',f'VGA.yres={args.height}',
            '-global',f'VGA.xmax={args.width}','-global',f'VGA.ymax={args.height}',
            '-drive',f'if=pflash,format=raw,readonly=on,file={ROOT/".tools/firmware/edk2-x86_64-code.fd"}',
            '-drive',f'if=pflash,format=raw,file={output/"Vars.fd"}',
            '-drive',f'if=ide,format=raw,snapshot=on,file={image}',
            '-display','none','-net','none','-serial',f'file:{output/"Serial.log"}',
            '-qmp',f'tcp:127.0.0.1:{port},server=on,wait=off','-no-reboot','-no-shutdown']
        if args.input=='usb':
            command+=['-device','qemu-xhci,id=Usb',
                      '-device',f'usb-kbd,bus=Usb.0,usb_version={args.usb_version}',
                      '-device',f'usb-mouse,bus=Usb.0,usb_version={args.usb_version}']
        result['QemuArguments']=command
        with (output/'Qemu.log').open('wb') as log:
            process=subprocess.Popen(command,cwd=ROOT,stdout=log,stderr=subprocess.STDOUT,
                                     creationflags=getattr(subprocess,'CREATE_NO_WINDOW',0))
        deadline=time.monotonic()+180
        while time.monotonic()<deadline:
            if process.poll() is not None: raise RuntimeError('Private QEMU exited during startup')
            try:qmp=QmpClient(port);break
            except (OSError,TimeoutError):time.sleep(.1)
        assert qmp is not None,'Private QMP did not become ready'
        require_backing(qmp.call('query-block'),image)
        def text():
            value=(output/'Serial.log').read_text(errors='replace') if (output/'Serial.log').exists() else ''
            assert process.poll() is None,'Private QEMU exited'
            assert not any(x in value for x in ('XFE_DIAGNOSTIC_EVIDENCE_LOST|','XFE_NATIVE_FAIL','XFE_NATIVE_EXCEPTION','XFE_RUNTIME_FATAL',
                'XFE_ZERO_FAILFAST','XFE_INDEPENDENT_APPLICATION_FAILED','XFE_RUNTIME_OOM|')),value[-4000:]
            return value
        while time.monotonic()<deadline:
            value=text()
            match=re.search(r'XFE_INDEPENDENT_DESKTOP_FIRST_FRAME\|Width\|(\d+)\|Height\|(\d+)\|ScalePercent\|(\d+)',value)
            if match and 'XFE_DESKTOP_FRAME_POLICY|' in value:break
            time.sleep(.1)
        else:raise TimeoutError('Native desktop did not become ready')
        width,height,scale=map(int,match.groups())
        assert (width,height)==(args.width,args.height),'Final framebuffer differs from requested native test monitor'
        result.update(ActualFramebufferWidth=width,ActualFramebufferHeight=height,ScalePercent=scale,
                      TimerConfiguration=re.findall(r'XFE_TIMER_CONFIGURATION[^\r\n]*',value))
        time.sleep(5)
        def screenshot(name):
            ppm=output/(name+'.ppm');png=output/(name+'.png')
            qmp.call('screendump',{'filename':str(ppm)})
            with Image.open(ppm) as frame:
                assert frame.size==(width,height)
                frame.convert('RGB').save(png)
            return {'Path':str(png),'Sha256':digest(png)}
        probe=output/'Pointer';probe.mkdir()
        cadence=run_pointer_cadence(qmp.call,text,screenshot,width,height,[width//2,height//2],probe,
                                    seconds=args.seconds,input_rate=args.input_rate,target_hz=240,motion_pattern='square')
        result['Pointer']=cadence
        assert cadence.get('EvidenceValid') and cadence.get('FramebufferPixelsChangedAfterMouseRequest'), 'Native timing or actual pointer pixel evidence is invalid'
        result['SoftwareCadenceGoalMet']=cadence.get('SoftwareCadenceGoalMet',False)
        qmp.call('quit');process.wait(timeout=15)
        result['QemuExitCode']=process.returncode
        assert process.returncode==0, 'Private QEMU exit was not clean'
        result['ImageSourceUnchanged']=digest(image)==result['ImageSha256']
        assert result['ImageSourceUnchanged']
        assert all(digest(ROOT/path)==expected for path,expected in result['HarnessSha256'].items()), 'Probe source changed during measurement'
        assert all(digest(ROOT/path)==expected for path,expected in receipt['SourceSha256'].items()), 'Compiled source changed during measurement'
        result['Completed']=True
    except Exception as error:
        result['Completed']=False
        result['Failure']=str(error)
    finally:
        if qmp:
            try:qmp.close()
            except OSError:pass
        if process and process.poll() is None:
            process.terminate()
            try:process.wait(timeout=10)
            except subprocess.TimeoutExpired:process.kill();process.wait(timeout=10)
        (output/'Result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf-8')
    print(json.dumps({key:result.get(key) for key in ('Completed','ActualFramebufferWidth','ActualFramebufferHeight','SoftwareCadenceGoalMet','Failure')}))
    return 0 if result['Completed'] else 1


if __name__=='__main__':
    parser=argparse.ArgumentParser(description=__doc__)
    parser.add_argument('kernel',type=Path);parser.add_argument('output',type=Path)
    parser.add_argument('--accel',choices=('tcg','whpx'),default='whpx')
    parser.add_argument('--input',choices=('ps2','usb'),default='ps2')
    parser.add_argument('--usb-version',type=int,choices=(1,2,3),default=2)
    parser.add_argument('--width',type=int,default=3840);parser.add_argument('--height',type=int,default=2160)
    parser.add_argument('--seconds',type=float,default=5);parser.add_argument('--input-rate',type=int,default=1000)
    raise SystemExit(run(parser.parse_args()))