XFESystem

"""Validate bounded native read-only PCI discovery logs; never assert GPU execution.

Does not launch QEMU or touch hardware. TestBoot may call validate_discovery on
the actual serial text after the normal native self-test; existing image/source
hash binding remains TestBoot's responsibility.
"""
import argparse
import hashlib
import json
from pathlib import Path


def fields(line):
    parts = line.strip().split('|')
    if len(parts) < 3 or len(parts) % 2 != 1:
        raise AssertionError('Malformed GPU discovery field pairs')
    result = {}
    for key, value in zip(parts[1::2], parts[2::2]):
        if key in result or not key or not value:
            raise AssertionError('Duplicate or empty GPU discovery field')
        result[key] = value
    return result


def number(value, key):
    try:
        result = int(value[key], 0)
    except (KeyError, ValueError, TypeError) as error:
        raise AssertionError('Missing or invalid GPU field ' + key) from error
    if result < 0:
        raise AssertionError('Negative GPU discovery field ' + key)
    return result


def zeros(value, *keys):
    for key in keys:
        if number(value, key) != 0:
            raise AssertionError('Read-only discovery must not claim ' + key)


def validate_discovery(serial, expect_absent=False, require_firmware_handoff=False):
    lines = serial.splitlines()
    begins = [i for i, line in enumerate(lines) if line.startswith('XFE_GPU_DISCOVERY_BEGIN|')]
    ends = [i for i, line in enumerate(lines) if line.startswith('XFE_GPU_DISCOVERY_END|')]
    if len(begins) != 1 or len(ends) != 1 or begins[0] >= ends[0]:
        raise AssertionError('Expected one complete native GPU discovery transaction')
    begin, end = fields(lines[begins[0]]), fields(lines[ends[0]])
    if number(begin, 'Version') != 1 or number(begin, 'ReadOnly') != 1:
        raise AssertionError('Unsupported GPU discovery schema or policy')
    zeros(begin, 'BarAccess', 'GpuAcceleration', 'Cuda')
    zeros(end, 'BarMaps', 'BarReads', 'ConfigWrites', 'GpuAcceleration', 'Cuda', 'UnmapFailures')
    if number(end, 'State') != 2:
        raise AssertionError('Native GPU discovery did not complete')
    count, matched = number(end, 'Devices'), number(end, 'MatchedDevices')
    if count > 8 or matched < count or number(end, 'Omitted') != matched - count:
        raise AssertionError('GPU discovery count/capacity mismatch')
    if not 8192 <= number(end, 'ConfigReads') <= 200000:
        raise AssertionError('Missing or unbounded native PCI enumeration')
    if number(end, 'EcamPageMaps') > count:
        raise AssertionError('Too many ECAM mappings for bounded GPU discovery')
    platforms, devices, bars = [], {}, {}
    for line in lines[begins[0] + 1:ends[0]]:
        if line.startswith('XFE_GPU_DISCOVERY_PLATFORM|'):
            platforms.append(fields(line))
        elif line.startswith('XFE_GPU_DEVICE|'):
            value = fields(line); index = number(value, 'Index')
            if index in devices or index >= count:
                raise AssertionError('Duplicate or out-of-range GPU record')
            if (number(value, 'Vendor') != 0x10de or number(value, 'Class') >> 16 != 3 or
                number(value, 'DeviceId') > 65535 or number(value, 'Bus') > 255 or
                number(value, 'Device') > 31 or number(value, 'Function') > 7):
                raise AssertionError('Non-NVIDIA/display or invalid PCI identity')
            candidate = number(value, 'DeviceId') == 0x2b85 and number(value, 'PcieStatus') == 1
            if number(value, 'Sm120Candidate') != int(candidate):
                raise AssertionError('SM120 candidate must use exact PCI identity and PCIe')
            if (number(value, 'PcieStatus') > 2 or number(value, 'ExtendedStatus') > 3 or
                number(value, 'RebarStatus') > 3):
                raise AssertionError('Invalid capability state')
            devices[index] = value
        elif line.startswith('XFE_GPU_BAR|'):
            value = fields(line); index, bar = number(value, 'DeviceIndex'), number(value, 'Index')
            if index >= count or bar > 5 or (index, bar) in bars:
                raise AssertionError('Duplicate or out-of-range GPU BAR')
            if number(value, 'MappingDenied') != 1 or value.get('Reason') != 'DiscoveryOnly':
                raise AssertionError('Discovery must never map or use a GPU BAR')
            known, flags = number(value, 'SizeKnown'), number(value, 'Flags')
            address, size, source = number(value, 'Address'), number(value, 'Bytes'), number(value, 'SizeSource')
            if known not in (0, 1) or known != int(bool(flags & 128)):
                raise AssertionError('BAR known-size flags disagree')
            if not known:
                if size != 0 or source != 0:
                    raise AssertionError('Unknown BAR size must remain zero with no source')
            elif (source not in (1, 2) or flags & (1 | 2 | 16 | 32 | 64) != 1 or
                  size < 16 or size & (size - 1) or address == 0 or address & (size - 1) or
                  address >= 1 << 52 or size > (1 << 52) - address or
                  (not flags & 4 and address + size > 1 << 32)):
                raise AssertionError('BAR size lacks validated bounded memory resource')
            bars[index, bar] = value
    if len(platforms) != 1 or number(platforms[0], 'McfgStatus') > 2 or number(platforms[0], 'McfgEntries') > 32:
        raise AssertionError('Missing or invalid platform resource observation')
    if number(platforms[0], 'Segment') != 0 or number(platforms[0], 'RecordLimit') != 8:
        raise AssertionError('Unsupported PCI segment or record capacity')
    if len(devices) != count or len(bars) != count * 6:
        raise AssertionError('Missing GPU device/BAR records')
    if expect_absent and (count != 0 or matched != 0):
        raise AssertionError('This QEMU profile must not pretend to expose an NVIDIA GPU')
    firmware = [fields(line) for line in lines if line.startswith('XFE_GPU_FIRMWARE_HANDOFF|')]
    if len(firmware) > 1:
        raise AssertionError('Duplicate firmware resource handoff observation')
    if firmware:
        zeros(firmware[0], 'Cuda', 'GpuRasterization')
        if number(firmware[0], 'ResourceOnly') != 1 or number(firmware[0], 'Valid') not in (0, 1) or number(firmware[0], 'Count') > 8:
            raise AssertionError('Invalid firmware resource handoff record')
    if require_firmware_handoff and (not firmware or number(firmware[0], 'Valid') != 1):
        raise AssertionError('Missing validated native UEFI resource handoff')
    if expect_absent and firmware and number(firmware[0], 'Count') != 0:
        raise AssertionError('Unexpected NVIDIA firmware resources in no-GPU QEMU')
    return {'Kind': 'XfeIndependentReadOnlyGpuDiscoveryEvidence', 'Passed': True,
            'NativeProbeObserved': True, 'NvidiaDeviceCount': count, 'MatchedDeviceCount': matched,
            'Exact5090CandidateObserved': any(number(value, 'Sm120Candidate') == 1 for value in devices.values()),
            'ExpectedNvidiaAbsent': bool(expect_absent), 'FirmwareHandoffRequired': bool(require_firmware_handoff),
            'GpuExecutionVerified': False, 'CudaExecutionVerified': False, 'GpuRasterizationVerified': False,
            'Platform': platforms[0], 'FirmwareHandoff': firmware[0] if firmware else None,
            'Devices': list(devices.values()), 'Bars': list(bars.values()), 'Summary': end}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--serial', type=Path, required=True)
    parser.add_argument('--expect-absent', action='store_true')
    parser.add_argument('--require-firmware-handoff', action='store_true')
    args = parser.parse_args()
    raw = args.serial.read_bytes()
    result = validate_discovery(raw.decode('utf-8', errors='replace'), args.expect_absent, args.require_firmware_handoff)
    result.update(SerialPath=str(args.serial.resolve()), SerialSha256=hashlib.sha256(raw).hexdigest())
    print(json.dumps(result, indent=2))


if __name__ == '__main__':
    main()