XFESystem

"""Negative native guest proof: unavailable CPU entropy must refuse Bing HTTPS.

This creates its own bounded QEMU session and a fresh RAM-only image. A passing
negative proof is deliberately incompatible with the normal Open.ps1 receipt.
"""
import argparse
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, inspect_image, read_json
from IndependentBingEvidence import (HOME, REMOTE, RESPONSE, native_navigation_failure,
    native_network_profile, network_arguments, window_geometry)

CPU = 'max,-rdrand,-rdseed'
REFUSAL = 'No healthy CPU cryptographic random source is available; HTTPS is refused.'
FATAL = ('XFE_NATIVE_FAIL', 'XFE_NATIVE_EXCEPTION', 'XFE_RUNTIME_FATAL', 'XFE_ZERO_FAILFAST',
         'XFE_INDEPENDENT_APPLICATION_FAILED', 'XFE_BROWSER_PAINT_FAILED')


def validate_negative_profile(arguments):
    for option, value in (('-cpu', CPU), ('-accel', 'tcg'), ('-rtc', 'base=utc'), ('-smp', '1')):
        indices = [index for index, item in enumerate(arguments) if item == option]
        assert len(indices) == 1 and indices[0] + 1 < len(arguments) and arguments[indices[0] + 1] == value, 'Wrong explicit negative CPU/RTC profile'
    actual = []
    for index, item in enumerate(arguments):
        if item in ('-net', '-netdev', '-nic', '-device'):
            assert index + 1 < len(arguments)
            actual.extend((item, arguments[index + 1]))
    assert actual == network_arguments(True), 'Negative proof must retain the real e1000/user NAT profile'


def verify_entropy_refusal(serial, after, window_id):
    fresh = serial[after:]
    assert not any(marker in fresh for marker in FATAL), 'The guest crashed rather than refusing HTTPS'
    assert RESPONSE + '_BEGIN' not in fresh and REMOTE + '_BEGIN' not in fresh, 'An unavailable RNG still produced remote response/DOM evidence'
    errors = re.findall(r'XFE_NATIVE_WEB_ERROR_BEGIN\r?\n(.*?)XFE_NATIVE_WEB_ERROR_END', fresh, re.S)
    assert any(REFUSAL in error for error in errors), 'No actual complete native hardware entropy rejection diagnostic'
    failure = native_navigation_failure(serial, after, window_id)
    assert failure is not None and REFUSAL in failure['Text'], 'The current actual Bing window did not show the explicit entropy refusal'
    return {key: value for key, value in failure.items() if key != 'Raw'}


class DeadlineQmp(Qmp):
    def __init__(self, port, deadline):
        self.deadline = deadline
        self.connection = socket.create_connection(('127.0.0.1', port), timeout=self.remaining())
        self.stream = self.connection.makefile('rwb'); self.number = 0
        try:
            assert 'QMP' in json.loads(self.stream.readline()), 'No QMP greeting'
            self.call('qmp_capabilities')
        except Exception:
            self.close()
            raise

    def remaining(self):
        remaining = self.deadline - time.monotonic()
        if remaining <= 0: raise TimeoutError('The owned negative guest phase deadline expired')
        return min(2, remaining)

    def call(self, command, arguments=None):
        self.connection.settimeout(self.remaining())
        return super().call(command, arguments)


def verify(kernel, boot_timeout=60, negative_timeout=15):
    assert 0 < boot_timeout <= 60 and 0 < negative_timeout <= 15, 'Negative proof deadlines exceed their bounds'
    kernel = Path(kernel).resolve()
    build_path = Path(str(kernel) + '.json'); build = read_json(build_path)
    assert build['Kind'] == 'XfeIndependentKernelBuild' and build['Passed'] is True and build['BootstrapOnly'] is False
    assert build['ContainsCosmos'] is False and build['ManagedRuntimeLinked'] is True and build['Target'] == 'x86_64-none'
    assert build['KernelSha256'] == digest(kernel)
    assert 'Source/XFESystem.NativeRuntime/Application/NativeApplication.csproj' in build['SourceSha256'], 'A real desktop build is required'
    for relative, expected in build['SourceSha256'].items():
        source = (ROOT / relative).resolve()
        assert source.is_relative_to(ROOT) and digest(source) == expected, 'Source changed: ' + relative
    proof = ROOT / 'build/IndependentKernel' / ('TlsEntropyFailure-' + str(time.time_ns()))
    proof.mkdir(parents=True); image = proof / 'System.img'; serial_path = proof / 'Serial.log'
    profile = native_network_profile(True); profile.update(Cpu=CPU, ExpectedEntropyAvailable=False)
    result = {'Kind': 'XfeIndependentTlsEntropyFailureProof', 'Passed': False, 'NegativeProfile': True,
              'ExpectedRefusal': REFUSAL, 'SelectedCpu': CPU, 'DisabledCpuInstructions': ['rdrand', 'rdseed'],
              'NativeNetworkProfile': profile, 'BingAcceptancePassed': False, 'NativeHttpsBingLoaded': False,
              'NetworkAcceptancePassed': False, 'DesktopAcceptancePassed': False, 'FullReleaseQualified': False,
              'ContainsCosmos': False, 'RamOnly': True, 'StorageEnabled': False, 'HostHttpRequests': False,
              'HostRandomFallback': False, 'BootTimeoutSeconds': boot_timeout, 'NegativeTimeoutSeconds': negative_timeout,
              'KernelPath': str(kernel), 'KernelSha256': digest(kernel), 'BuildReceipt': str(build_path),
              'BuildReceiptSha256': digest(build_path), 'ProofDirectory': str(proof), 'Screenshots': []}
    qemu = None; qmp = None; sources = None

    def serial():
        return serial_path.read_text(encoding='utf-8', errors='replace') if serial_path.exists() else ''

    def healthy():
        assert qemu is not None and qemu.poll() is None, 'The owned QEMU exited unexpectedly'
        value = serial(); assert not any(marker in value for marker in FATAL), value[-4096:]
        return value

    def wait(marker, after, deadline):
        while time.monotonic() < deadline:
            if marker in healthy()[after:]: return
            time.sleep(.05)
        raise TimeoutError('Native negative-proof acknowledgement timeout: ' + marker)

    def key(chord):
        qmp.call('human-monitor-command', {'command-line': 'sendkey ' + chord})
        time.sleep(.15)

    def screenshot(name):
        ppm = proof / (name + '.ppm'); png = proof / (name + '.png')
        try:
            qmp.call('screendump', {'filename': str(ppm)})
            with Image.open(ppm) as frame:
                assert frame.size == (1280, 800), 'Unexpected actual guest framebuffer geometry'
                frame.convert('RGB').save(png)
            capture = {'Path': png.name, 'Sha256': digest(png), 'NativeDimensions': [1280, 800]}
            result['Screenshots'].append(capture); return capture
        finally:
            if ppm.exists(): ppm.unlink()

    try:
        with (proof / '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(kernel), str(image), '1280x800', '--independent-kernel'], cwd=ROOT, check=True, stdout=log, stderr=subprocess.STDOUT, timeout=60)
        inspected, files = inspect_image(image)
        assert image.stat().st_size == 128 * 1024 * 1024 and files['System/Kernel.elf'] == kernel.read_bytes()
        assert read_json(str(image) + '.json')['KernelImplementation'] == 'XFESystem independent kernel'
        result.update(ImagePath=str(image), ImageSha256=digest(image), ImageVerification=inspected)
        sources = {**build['SourceSha256'], **inspected['BootShimSourceSha256']}
        for relative in ('Tools/IndependentKernel/TestTlsEntropyFailure.py', 'Tools/IndependentKernel/TestBoot.py', 'Tools/IndependentKernel/IndependentBingEvidence.py'):
            sources[relative] = digest(ROOT / relative)
        snapshot = proof / 'SourceSnapshot.json'; snapshot.write_text(json.dumps(sources, indent=2), encoding='utf-8')
        result.update(SourceSnapshotPath=str(snapshot), SourceSnapshotSha256=digest(snapshot))
        firmware_code = ROOT / '.tools/firmware/edk2-x86_64-code.fd'
        firmware_vars = ROOT / '.tools/firmware/edk2-i386-vars.fd'
        lock = read_json(ROOT / 'Tools/CSharpBoot/Dependencies.json')['Firmware']
        assert digest(firmware_code) == lock[firmware_code.name] and digest(firmware_vars) == lock[firmware_vars.name], 'Pinned firmware changed'
        shutil.copy2(firmware_vars, 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', CPU, '-smp', '1', '-m', '1024M',
                   '-vga', 'std', '-global', 'VGA.vgamem_mb=64', '-display', 'none', *network_arguments(True), '-rtc', 'base=utc',
                   '-drive', f'if=pflash,format=raw,readonly=on,file={firmware_code}',
                   '-drive', f'if=pflash,format=raw,file={proof / "FirmwareVars.fd"}',
                   '-drive', f'file={image},format=raw,if=ide,snapshot=on', '-serial', 'file:' + str(serial_path),
                   '-qmp', f'tcp:127.0.0.1:{port},server=on,wait=off', '-no-reboot', '-no-shutdown']
        validate_negative_profile(command); result['QemuArguments'] = command
        with (proof / 'Qemu.log').open('w', encoding='utf-8') as log:
            qemu = subprocess.Popen(command, cwd=ROOT, stdout=log, stderr=subprocess.STDOUT, creationflags=getattr(subprocess, 'CREATE_NO_WINDOW', 0))
            result['OwnedQemuProcessId'] = qemu.pid
            deadline = time.monotonic() + boot_timeout
            while qmp is None:
                healthy()
                try: qmp = DeadlineQmp(port, deadline)
                except (OSError, ConnectionError):
                    assert time.monotonic() < deadline, 'Negative-profile QMP boot deadline expired'
                    time.sleep(.05)
            for marker in ('XFE_NATIVE_SELFTEST_OK', 'XFE_APPLICATION_RUNTIME_CHECK_OK', 'XFE_INDEPENDENT_DESKTOP_FIRST_FRAME', 'XFE_DESKTOP_ACTIVE|Files'):
                wait(marker, 0, deadline)
            screenshot('Desktop-Before-Entropy-Rejection')
            after = len(serial()); deadline = time.monotonic() + negative_timeout; qmp.deadline = deadline
            for chord in ('ctrl-esc', 'down', 'down', 'ret'): key(chord)
            wait('XFE_DESKTOP_ACTIVE|Browser', after, deadline)
            error = None
            while time.monotonic() < deadline:
                value = healthy()
                try:
                    window_id, _ = window_geometry(value, 'Browser')
                    error = verify_entropy_refusal(value, after, window_id)
                    break
                except AssertionError:
                    time.sleep(.05)
            assert error is not None, 'Guest did not explicitly refuse missing CPU entropy within the negative deadline'
            result['ActualNavigationRefusal'] = error
            screenshot('Browser-Entropy-Refused')
            close_start = len(serial()); key('alt-f4'); wait('XFE_DESKTOP_ACTIVE|Files', close_start, deadline)
            restored = screenshot('Desktop-After-Entropy-Rejection')
            assert restored['Sha256'] != result['Screenshots'][-2]['Sha256'], 'Alt+F4 produced no changed actual guest framebuffer'
            verify_entropy_refusal(healthy(), after, error['WindowId'])
            result.update(EntropyRefusalPassed=True, NoRemoteResponseOrDom=True, AltF4ReturnedToResponsiveFiles=True, 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.deadline = time.monotonic() + 2; qmp.call('quit')
                except (OSError, RuntimeError, TimeoutError): pass
            try: qemu.wait(timeout=5)
            except subprocess.TimeoutExpired:
                qemu.terminate(); qemu.wait(timeout=5); result['ForcedCleanup'] = True
            result['QemuExitCode'] = qemu.returncode
        if qmp is not None: qmp.close()
        if serial_path.exists(): result['SerialEvidence'] = {'Path': serial_path.name, 'Sha256': digest(serial_path)}
        if image.exists(): result['ImageUnchanged'] = digest(image) == result.get('ImageSha256')
        result['KernelAndReceiptUnchanged'] = digest(kernel) == result['KernelSha256'] and digest(build_path) == result['BuildReceiptSha256']
        result['SourcesUnchanged'] = sources is not None and all((ROOT / relative).is_file() and digest(ROOT / relative) == expected for relative, expected in sources.items())
        result['Passed'] = (result['Passed'] and result.get('QemuExitCode') == 0 and not result.get('ForcedCleanup') and
                            result.get('ImageUnchanged') is True and result['KernelAndReceiptUnchanged'] and result['SourcesUnchanged'])
        (proof / 'Result.json').write_text(json.dumps(result, ensure_ascii=False, 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('--boot-timeout', type=int, choices=range(1, 61), default=60)
    parser.add_argument('--negative-timeout', type=int, choices=range(1, 16), default=15)
    args = parser.parse_args()
    raise SystemExit(0 if verify(args.kernel, args.boot_timeout, args.negative_timeout) else 1)