XFESystem

"""Measure an existing private QEMU's real pointer-to-present software cadence.

Attaches only to a supplied loopback QMP UUID and verified image backing. It
never starts/stops a VM, copies an image/ELF, edits a guest page, or substitutes
host cursor/screen pixels. F12 starts/stops the guest's bounded timing probe.
"""
import argparse
import hashlib
import json
import pathlib
import re
import shutil
import socket
import time
from PointerCadenceEvidence import parse_timing_dump, summarize_timing, cross_check_host_clock


ROOT = pathlib.Path(__file__).resolve().parents[1]
FAILURES = ('CPU EXCEPTION:', '[FAILFAST]', '*** UNHANDLED EXCEPTION', 'XFE_RUNTIME_OOM|', 'XFE_RUNTIME_STACK_OOM|')


def sha256(path):
    digest = hashlib.sha256()
    with pathlib.Path(path).open('rb') as source:
        for block in iter(lambda: source.read(1024 * 1024), b''): digest.update(block)
    return digest.hexdigest()


class QmpClient:
    def __init__(self, port):
        self.socket = socket.create_connection(('127.0.0.1', port), timeout=5)
        self.socket.settimeout(5)
        self.stream = self.socket.makefile('rwb')
        self.sequence = 0
        greeting = json.loads(self.stream.readline(1_048_576))
        if 'QMP' not in greeting: raise AssertionError('The supplied port is not QMP')
        self.call('qmp_capabilities')

    def call(self, command, arguments=None):
        self.sequence += 1
        self.stream.write((json.dumps({'execute': command, 'arguments': arguments or {}, 'id': self.sequence}) + '\n').encode())
        self.stream.flush()
        while True:
            raw = self.stream.readline(1_048_576)
            if not raw: raise AssertionError('Private QMP connection closed')
            reply = json.loads(raw)
            if 'event' in reply: continue
            if reply.get('id') != self.sequence or 'error' in reply or 'return' not in reply:
                raise AssertionError('Unexpected QMP reply: ' + str(reply)[:512])
            return reply['return']

    def close(self):
        self.stream.close(); self.socket.close()


def require_backing(blocks, image):
    expected, matches = pathlib.Path(image).resolve(), []
    for block in blocks:
        node = block.get('inserted', {}).get('image', {})
        while node:
            filename = node.get('filename')
            if filename and pathlib.Path(filename).resolve() == expected:
                if node.get('format') != 'raw' or node.get('virtual-size') != expected.stat().st_size:
                    raise AssertionError('The private QMP backing geometry differs from the verified raw image')
                matches.append(block); break
            node = node.get('backing-image', {})
    if len(matches) != 1: raise AssertionError('QMP must identify exactly one verified source image backing device')


def run_pointer_cadence(qmp, read_log, screenshot, width, height, pointer, output, *, seconds=5, input_rate=1000, target_hz=240):
    """Reusable with an already authenticated harness QMP callback; no VM operations."""
    if not 2 <= seconds <= 10 or not 240 <= input_rate <= 2000 or not 20 <= target_hz <= 1000:
        raise ValueError('Bounded test requires 2..10 seconds, 240..2000 input events/s and 20..1000 target Hz')
    output = pathlib.Path(output)
    result = {'Kind': 'NativeSoftwarePointerCadence', 'Passed': False, 'EvidenceValid': False,
              'SoftwareCadenceGoalMet': False, 'PhysicalMonitorRefreshMatched': False,
              'NativeScanoutHzMeasured': None, 'VSyncVerified': False,
              'InputRateRequested': input_rate, 'MotionSecondsRequested': seconds, 'TargetHz': target_hz,
              'ScreenshotsDuringTimedMotion': 0, 'HostScreenshotsUsed': False, 'DomInjectionUsed': False,
              'VmStartedByThisTool': False, 'ImageCopied': False, 'KernelCopied': False}
    dispatches, armed = [], False
    def healthy():
        log = read_log()
        if any(marker in log for marker in FAILURES): raise AssertionError('Native runtime failure; inspect the original serial log')
        return log
    def move_relative(dx, dy):
        events = []
        if dx: events.append({'type': 'rel', 'data': {'axis': 'x', 'value': dx}})
        if dy: events.append({'type': 'rel', 'data': {'axis': 'y', 'value': dy}})
        qmp('input-send-event', {'events': events})
        pointer[0] += dx; pointer[1] += dy
    def tap_f12():
        for down in (True, False):
            qmp('input-send-event', {'events': [{'type': 'key', 'data': {'down': down, 'key': {'type': 'qcode', 'data': 'f12'}}}]})
            time.sleep(.025)
    def capture(name):
        value = screenshot(name)
        path = pathlib.Path(value['Path'] if isinstance(value, dict) else value)
        return {'Path': str(path), 'Sha256': sha256(path), 'Source': 'QMP screendump of guest framebuffer; original resolution'}
    try:
        healthy()
        target = [width - 160, min(height - 100, max(100, height // 3))]
        for _ in range(200):
            if pointer == target: break
            move_relative(max(-64, min(64, target[0] - pointer[0])), max(-64, min(64, target[1] - pointer[1])))
            time.sleep(.006)
        if pointer != target: raise AssertionError('Bounded initial pointer positioning did not finish')
        time.sleep(.15)
        start_offset = len(healthy())
        tap_f12(); armed = True
        # No serial/screenshot polling in the measured hot motion path. QMP
        # acknowledgement means queued emulated input, never guest consumption.
        time.sleep(.12)
        started = time.perf_counter(); due = started; missed = 0
        maximum_dispatches = int(seconds * input_rate) + 1
        while len(dispatches) < maximum_dispatches:
            now = time.perf_counter()
            if now - started >= seconds: break
            delay = due - now
            if delay > .0005: time.sleep(delay - .0003)
            while time.perf_counter() < due: pass
            index = len(dispatches)
            dx = 4 if (index // 16) % 2 == 0 else -4
            queued = time.perf_counter(); move_relative(dx, 0); acknowledged = time.perf_counter()
            dispatches.append([index, queued - started, acknowledged - started, dx, 0])
            due += 1 / input_rate
            if acknowledged > due + 1 / input_rate:
                missed += 1; due = acknowledged + 1 / input_rate
        host_elapsed = time.perf_counter() - started
        tap_f12(); armed = False
        result['HostInputDispatch'] = {'Count': len(dispatches), 'Seconds': host_elapsed,
            'FirstQueuedToLastAcknowledgedSeconds': dispatches[-1][2] - dispatches[0][1] if dispatches else 0,
            'AcknowledgedQmpRequestsPerSecond': len(dispatches) / host_elapsed,
            'MissedHostSchedulingDeadlines': missed,
            'TimestampMeaning': 'Host monotonic queue/ack timestamps only; not guest input delivery, presentation or physical USB polling.'}
        deadline, dump = time.monotonic() + 45, None
        while time.monotonic() < deadline:
            current = healthy()
            dump = parse_timing_dump(current[start_offset:])
            if dump is not None: break
            time.sleep(.05)
        if dump is None: raise AssertionError('No sealed native F12 timing dump arrived; a target/EDID log alone cannot pass')
        result['NativeTiming'] = dump
        result.update(cross_check_host_clock(summarize_timing(dump, target_hz),
            result['HostInputDispatch']['FirstQueuedToLastAcknowledgedSeconds']))
        (output / 'NativeTiming.txt').write_text(current[start_offset:], encoding='utf-8')
        # Capture after export, then move the real emulated mouse and capture
        # again. Do not collect expensive 4K screenshots inside the timing span.
        time.sleep(.15)
        first = capture('PointerCadence-AfterMotion')
        move_relative(-48, 24); time.sleep(.2)
        second = capture('PointerCadence-AfterMove')
        from PIL import Image, ImageChops
        with Image.open(first['Path']) as first_image, Image.open(second['Path']) as second_image:
            if first_image.size != (width, height) or second_image.size != (width, height):
                raise AssertionError('Screenshot geometry differs from native framebuffer')
            difference = ImageChops.difference(first_image.convert('RGB'), second_image.convert('RGB'))
            bounds = difference.getbbox()
        result['Screenshots'] = [first, second]
        result['ScreenshotPixelChangeBounds'] = list(bounds) if bounds else None
        result['FramebufferPixelsChangedAfterMouseRequest'] = bounds is not None
        result['ScreenshotMeaning'] = 'Actual framebuffer comparison, including any clock/animation changes; cursor appearance requires visual inspection. Native motion is bound separately by frame position records.'
        result['Passed'] = result['EvidenceValid'] and result['SoftwareCadenceGoalMet'] and result['FramebufferPixelsChangedAfterMouseRequest']
        if not result['SoftwareCadenceGoalMet']:
            result['GoalFailure'] = 'Measured software presentation/input cadence did not meet every requested frame budget; physical scanout remains unmeasured.'
    except Exception as error:
        result['Error'] = str(error)
    finally:
        if armed:
            # Release the opt-in recorder even on injection failure; do not quit
            # or kill a QEMU owned by another harness.
            try: tap_f12()
            except Exception as error: result['ProbeStopFailure'] = str(error)
        (output / 'HostInputDispatches.json').write_text(json.dumps({'Columns': ['Index', 'QueuedSeconds', 'AcknowledgedSeconds', 'RequestedDx', 'RequestedDy'], 'Rows': dispatches}), encoding='utf-8')
        (output / 'Result.json').write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
    return result


def main(options):
    image = pathlib.Path(options.image).resolve(strict=True)
    receipt_path = pathlib.Path(options.boot_result).resolve(strict=True)
    serial_path = pathlib.Path(options.serial_log).resolve(strict=True)
    receipt = json.loads(receipt_path.read_text(encoding='utf-8-sig'))
    if not (receipt.get('Passed') is True and receipt.get('ImageUnchanged') is True and receipt.get('QemuExitCode') == 0):
        raise AssertionError('A complete prior passing boot receipt is required')
    if image.suffix.lower() != '.img' or not image.is_file(): raise AssertionError('Expected a verified file image, never a physical device')
    image_hash, receipt_hash = sha256(image), sha256(receipt_path)
    if image_hash != receipt['ImageSha256']: raise AssertionError('Image hash differs from prior verified boot')
    evidence = pathlib.Path(options.evidence_root).resolve() if options.evidence_root else ROOT / 'build/NativePointerCadence'
    parent = evidence
    while not parent.exists(): parent = parent.parent
    if shutil.disk_usage(parent).free < 10 * 1024 ** 3: raise AssertionError('Need 10 GiB free before any test artifact is created')
    out = evidence / ('Proof-' + str(time.time_ns())); out.mkdir(parents=True, exist_ok=False)
    connection = QmpClient(options.qmp_port)
    result = None
    try:
        qmp = connection.call
        if qmp('query-uuid').get('UUID', '').lower() != options.qemu_uuid.lower(): raise AssertionError('QMP UUID differs from the explicitly owned test VM')
        if qmp('query-status').get('status') != 'running': raise AssertionError('The private VM must already be running')
        require_backing(qmp('query-block'), image)
        mice = qmp('query-mice'); active = [mouse for mouse in mice if mouse.get('current')]
        if len(active) != 1: raise AssertionError('QMP must identify exactly one active emulated mouse')
        if active[0].get('absolute'): raise AssertionError('This bounded relative-motion test requires a relative mouse, not a tablet')
        if options.require_usb_mouse:
            tree = qmp('human-monitor-command', {'command-line': 'info qtree'})
            if len(mice) != 1 or tree.count('dev: usb-mouse,') != 1 or 'dev: i8042' in tree or 'dev: ps2-mouse' in tree:
                raise AssertionError('The unique active pointer topology is not USB-only; a device name is insufficient')
        def read_log():
            if serial_path.stat().st_size > 16 * 1024 * 1024: raise AssertionError('Native serial log exceeded its bounded size')
            return serial_path.read_text(encoding='utf-8', errors='replace')
        log = read_log()
        matches = list(re.finditer(r'XFE_DISPLAY_NATIVE\|([0-9]+)\|([0-9]+)\|ScalePercent\|([0-9]+)', log))
        if not matches or 'XFE_CSHARP_DESKTOP_READY' not in log: raise AssertionError('No real ready native framebuffer geometry in serial log')
        width, height, scale_percent = (int(value) for value in matches[-1].groups())
        if not 640 <= width <= 16384 or not 400 <= height <= 16384: raise AssertionError('Native display geometry is out of bounds')
        def screenshot(name):
            from PIL import Image
            ppm, png = out / (name + '.ppm'), out / (name + '.png')
            qmp('screendump', {'filename': str(ppm)})
            with Image.open(ppm) as source: source.save(png)
            ppm.unlink()  # Delete only this freshly generated raw screenshot.
            return png
        pointer = [options.pointer_x, options.pointer_y]
        result = run_pointer_cadence(qmp, read_log, screenshot, width, height, pointer, out,
            seconds=options.seconds, input_rate=options.input_rate, target_hz=options.target_hz)
        result.update({'ImagePath': str(image), 'ImageSha256': image_hash, 'PriorBootResult': str(receipt_path),
            'PriorBootResultSha256': receipt_hash, 'KernelSha256': receipt['KernelSha256'], 'HarnessSha256': sha256(__file__),
            'PrivateQemuUuid': options.qemu_uuid, 'PrivateQmpIdentityVerified': True, 'MouseDevices': mice,
            'UsbMousePathRequested': options.require_usb_mouse, 'PhysicalUsbMouseTested': False,
            'NativeWidth': width, 'NativeHeight': height, 'ScalePercent': scale_percent,
            'ImageUnchanged': sha256(image) == image_hash, 'PriorReceiptUnchanged': sha256(receipt_path) == receipt_hash,
            'QemuLeftRunning': qmp('query-status').get('status') == 'running'})
        result['Passed'] &= result['ImageUnchanged'] and result['PriorReceiptUnchanged']
        (out / 'Result.json').write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
    finally:
        connection.close()
        print(out)
    return 0 if result and result.get('Passed') else 1


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--qmp-port', type=int, required=True)
    parser.add_argument('--qemu-uuid', required=True)
    parser.add_argument('--image', required=True)
    parser.add_argument('--boot-result', required=True)
    parser.add_argument('--serial-log', required=True)
    parser.add_argument('--pointer-x', type=int, required=True, help='Known current native X from this harness, never guessed host coordinates')
    parser.add_argument('--pointer-y', type=int, required=True)
    parser.add_argument('--seconds', type=float, default=5)
    parser.add_argument('--input-rate', type=int, default=1000)
    parser.add_argument('--target-hz', type=int, default=240)
    parser.add_argument('--require-usb-mouse', action='store_true')
    parser.add_argument('--evidence-root')
    arguments = parser.parse_args()
    if not 1 <= arguments.qmp_port <= 65535: parser.error('A loopback QMP port in 1..65535 is required')
    raise SystemExit(main(arguments))