XFESystem

"""Actual USB boot, ordinary FAT program loading and two cold boots of the same writable medium."""
import argparse
import hashlib
import json
from pathlib import Path
import re
import shutil
import socket
import subprocess
import time
from PIL import Image
from TestBoot import ROOT, Qmp, digest, read_json
from IndependentBingEvidence import json_blocks
from IndependentConsoleEvidence import terminal_command_output, FAILURE_MARKERS


def verify_storage(image, edition=None):
    image=Path(image).resolve();manifest=read_json(str(image)+'.json');edition=edition or manifest['Edition']
    assert manifest.get('UsbWorkspace') is True and edition==manifest['Edition'] and edition in ('Console','Desktop')
    proof=ROOT/'build/IndependentKernel'/('UsbStorage-'+str(time.time_ns()));proof.mkdir()
    disk=proof/'WritableUsb.img';shutil.copyfile(image,disk)
    result={'Kind':'XfeNativeUsbPersistentWorkspaceProof','Passed':False,'Edition':edition,
        'SourceImagePath':str(image),'SourceImageSha256':digest(image),'SourceManifestSha256':digest(str(image)+'.json'),
        'KernelSha256':manifest['KernelSha256'],'Input':'Actual QMP USB HID packets only',
        'HostCommandExecution':False,'HostWorkspaceInjection':False,'PhysicalHardwareAcceptancePassed':False,
        'ImageWrites':'Private copy of the exact source image; original image remains unchanged',
        'Boots':[],'EvidenceFiles':[],
        'SourceHashes':{str(p.relative_to(ROOT)).replace('\\','/'):digest(p) for p in [Path(__file__),ROOT/'Tools/CSharpBoot/ReadImage.py']}}
    result['SourceHashes'].update(manifest['IndependentKernelBuild']['SourceSha256'])
    result['SourceHashes'].update(manifest['StorageSourceSha256'])
    token='UsbSaved'+str(time.time_ns())[-12:]
    def ref(path):return {'Path':path.relative_to(proof).as_posix(),'Sha256':digest(path)}
    try:
        for boot_number in (1,2):
            folder=proof/('Boot'+str(boot_number));folder.mkdir();shutil.copyfile(ROOT/'.tools/firmware/edk2-i386-vars.fd',folder/'Vars.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,i8042=off','-accel','tcg','-cpu','max','-smp','8','-m','1024M','-vga','std','-display','none','-net','none','-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={folder/"Vars.fd"}',
                '-device','qemu-xhci,id=Usb','-drive',f'if=none,id=BootUsb,format=raw,file={disk}',
                '-device','usb-storage,drive=BootUsb,bus=Usb.0,bootindex=1',
                '-device','usb-kbd,bus=Usb.0','-device','usb-mouse,bus=Usb.0',
                '-serial',f'file:{folder/"Serial.log"}','-qmp',f'tcp:127.0.0.1:{port},server=on,wait=off','-no-reboot','-no-shutdown']
            boot={'Number':boot_number,'QemuArguments':command,'Commands':[]};result['Boots'].append(boot)
            qemu=qmp=None;selected=[]
            def text():
                path=folder/'Serial.log';value=path.read_text(encoding='utf-8',errors='replace') if path.exists() else ''
                assert qemu.poll() is None,'USB guest exited unexpectedly'
                assert not any(marker in value for marker in FAILURE_MARKERS),value[-3000:]
                return value
            def wait(marker,after=0,timeout=50):
                end=time.monotonic()+timeout
                while time.monotonic()<end:
                    if marker in text()[after:]:return
                    time.sleep(.1)
                raise AssertionError('USB workspace timeout: '+marker+'\n'+text()[-2000:])
            def key(chord):qmp.call('human-monitor-command',{'command-line':'sendkey '+chord});time.sleep(.11)
            def type_text(value):
                special={':':'shift-semicolon','.':'dot','/':'slash','-':'minus','_':'shift-minus',' ':'spc'}
                for char in value:key(special.get(char,'shift-'+char.lower() if char.isupper() else char))
            def execute(value,required=()):
                after=len(text());type_text(value);key('ret');deadline=time.monotonic()+35;last=''
                while time.monotonic()<deadline:
                    packets=json_blocks(text(),'XFE_CONSOLE_SNAPSHOT' if edition=='Console' else 'XFE_TERMINAL_SNAPSHOT',after)
                    if packets:
                        packet=packets[-1]
                        try:
                            output=terminal_command_output(packet['Value'],value)
                            assert all(line in output for line in required),repr(output)
                            selected.append(packet);boot['Commands'].append({'Command':value,'PostCommandOutput':output,'Passed':True});return
                        except AssertionError as error:last=str(error)
                    time.sleep(.1)
                raise AssertionError('USB command did not render: '+value+': '+last)
            def screenshot(name):
                ppm=folder/(name+'.ppm');png=folder/(name+'.png');qmp.call('screendump',{'filename':str(ppm)})
                with Image.open(ppm) as frame:assert frame.size==(1280,800);frame.convert('RGB').save(png)
                result['EvidenceFiles'].append(ref(png));return ref(png)
            def terminal():
                if edition=='Desktop':
                    after=len(text());key('ctrl-esc');key('down');key('down');key('down');key('ret');wait('XFE_DESKTOP_ACTIVE|Terminal',after)
            try:
                with (folder/'Qemu.log').open('w') as log:qemu=subprocess.Popen(command,stdout=log,stderr=subprocess.STDOUT,creationflags=getattr(subprocess,'CREATE_NO_WINDOW',0))
                deadline=time.monotonic()+15
                while qmp is None:
                    assert qemu.poll() is None,(folder/'Qemu.log').read_text()
                    try:qmp=Qmp(port)
                    except OSError:assert time.monotonic()<deadline;time.sleep(.1)
                wait('XFE_INDEPENDENT_'+('CONSOLE' if edition=='Console' else 'DESKTOP')+'_FIRST_FRAME',timeout=90)
                log=text();assert 'XFE_WORKSPACE|Mode|USB|Persistent|1|' in log and 'XFE_USB_STORAGE_READY|Matches|1|' in log
                match=re.search(r'XFE_WORKSPACE\|Mode\|USB\|Persistent\|1\|Generation\|(\d+)',log);assert match
                boot['LoadedGeneration']=int(match[1]);boot['BootScreenshot']=screenshot('Boot')
                if boot_number==2:
                    assert boot['LoadedGeneration']>0,'Cold boot did not load a committed physical workspace'
                    if edition=='Desktop':assert 'XFE_DESKTOP_APPEARANCE|Dark|' in log and '|Persistent|1' in log
                if boot_number==1 and edition=='Desktop':
                    after=len(text());key('ctrl-esc')
                    for _ in range(4):key('down')
                    key('ret');wait('XFE_DESKTOP_ACTIVE|Settings',after);key('right');wait('XFE_DESKTOP_APPEARANCE|Dark|',after)
                    boot['SettingsScreenshot']=screenshot('Settings-Saved')
                terminal();execute('help',('Storage: persistent workspace',))
                if boot_number==1:
                    execute('mf UsbKeep.txt');execute('write UsbKeep.txt '+token)
                    execute('mv UsbKeep.txt UsbRenamed.txt');execute('cat UsbRenamed.txt',(token,))
                    execute('write /System/Configuration/UsbSetting.txt '+token)
                    execute('run /UsbFiles/UsbHello.dll',('XFE_USB_EXTERNAL_APP_OK','Managed ConsoleIL exit: 0'))
                    execute('cp /UsbFiles/UsbHello.dll UsbCopied.dll')
                else:
                    execute('cat UsbRenamed.txt',(token,));execute('cat /System/Configuration/UsbSetting.txt',(token,))
                    execute('run UsbCopied.dll',('XFE_USB_EXTERNAL_APP_OK','Managed ConsoleIL exit: 0'))
                    execute('run /UsbFiles/UsbHello.dll',('XFE_USB_EXTERNAL_APP_OK','Managed ConsoleIL exit: 0'))
                boot['FinalScreenshot']=screenshot('Workspace-Verified');boot['Passed']=True
                raw=folder/'RenderedCommands.log';raw.write_text('\n'.join(p['Raw'] for p in selected),encoding='utf-8');result['EvidenceFiles'].append(ref(raw))
                qmp.call('quit');qemu.wait(timeout=10);boot['QemuExitCode']=qemu.returncode;assert qemu.returncode==0
            finally:
                if qmp is not None:qmp.connection.close()
                if qemu is not None and qemu.poll() is None:qemu.terminate();qemu.wait(timeout=10)
                if (folder/'Serial.log').exists():result['EvidenceFiles'].append(ref(folder/'Serial.log'))
        original=image.read_bytes();actual=disk.read_bytes();offset=manifest['WorkspaceFirstLba']*512;size=manifest['WorkspaceBytes']
        assert len(original)==len(actual) and original[:offset+512]==actual[:offset+512] and original[offset+size:]==actual[offset+size:]
        assert original[offset+512:offset+size]!=actual[offset+512:offset+size],'No data changed on the actual USB disk'
        assert all(digest(ROOT/p)==h for p,h in result['SourceHashes'].items())
        result.update(Passed=True,SourceImageUnchanged=digest(image)==result['SourceImageSha256'],OnlyWorkspaceSectorsChanged=True,
            FilePersistedAcrossColdBoot=True,ConfigurationPersistedAcrossColdBoot=True,AppearancePersistedAcrossColdBoot=edition=='Desktop',
            StandardNetPeReadFromFatAndExecuted=True,ImportedBinaryCopiedToWorkspaceAndExecutedAfterReboot=True,
            WritableImageSha256=digest(disk),WritableImagePath=str(disk),SourcesUnchanged=True,ProofDirectory=str(proof))
    except Exception as error:
        result['Failure']=str(error)
    finally:
        (proof/'Result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf-8')
    return result,proof/'Result.json'


if __name__=='__main__':
    parser=argparse.ArgumentParser();parser.add_argument('image',type=Path);args=parser.parse_args()
    result,path=verify_storage(args.image);print(json.dumps({'Passed':result['Passed'],'Receipt':str(path),'Failure':result.get('Failure')},ensure_ascii=True))
    raise SystemExit(0 if result['Passed'] else 1)