"""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|', 'XFE_DIAGNOSTIC_EVIDENCE_LOST|') 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, motion_pattern='zigzag'): """Reusable with an already authenticated harness QMP callback; no VM operations.""" if not 2 <= seconds <= 35 or not 240 <= input_rate <= 2000 or not 20 <= target_hz <= 1000: raise ValueError('Bounded test requires 2..35 seconds, 240..2000 input events/s and 20..1000 target Hz') if motion_pattern not in ('zigzag', 'square'): raise ValueError('Unknown bounded pointer motion pattern') 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, 'MotionPattern': motion_pattern, '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) # A square avoids the old one-axis reversal alias: a frame around # a reversal can have fresh input but the exact previous position. # Preserve the legacy default for existing evidence/harnesses. dx, dy = ((4, 0), (0, 4), (-4, 0), (0, -4))[(index // 16) % 4] if motion_pattern == 'square' else (4 if (index // 16) % 2 == 0 else -4, 0) queued = time.perf_counter(); move_relative(dx, dy); acknowledged = time.perf_counter() dispatches.append([index, queued - started, acknowledged - started, dx, dy]) 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))