#!/usr/bin/env python3 """Opt-in bounded private XTest keyboard and optional relative-look experiment. Not a launcher or gameplay proof.""" import argparse import ctypes as C import hashlib import json import multiprocessing import math import os from pathlib import Path import re import signal import stat import subprocess import sys import time import headless_display_probe as display NOTIFY = None def notify(value): if NOTIFY is not None: NOTIFY.send(value) class Refused(ValueError): pass class KeyStateBusy(Refused): pass def option(args, name): if args.count(name) != 1: raise Refused('Missing or repeated native option: ' + name) index = args.index(name) if index + 1 >= len(args): raise Refused('Missing native option value') return args[index + 1] def gate_options(native, args): # Two exact read-only options; a prefix match must never admit an alias, # duplicate, foreign nonce or extra diagnostic in keyboard-only mode. tokens = [value for value in native if value.startswith('-errOrdinaryGate')] if any(value not in {'-errOrdinaryGateObserve', '-errOrdinaryGateChoices'} for value in tokens): raise Refused('Unknown native gate observer option') approach = getattr(args, 'enable_private_approach_proof', False) if getattr(args, 'enable_private_selection_proof', False) and '-errOrdinaryGateChoices' not in tokens: raise Refused('Selection requires matching native gate choices') if approach and '-errOrdinaryGateObserve' not in tokens: raise Refused('Missing native gate observer nonce') if '-errOrdinaryGateChoices' in tokens and not approach: raise Refused('Native gate choices require explicit approach proof') for name in set(tokens): if option(native, name) != args.nonce: raise Refused('Native gate observer nonce mismatch') def game_arguments(owner, args): native = display.process(owner['pid']) if display.identity(native) != owner['process']: raise Refused('Game process changed') if option(native['args'], '-errOrdinaryInputObserve') != args.nonce: raise Refused('Native observer nonce mismatch') if getattr(args, 'enable_private_look_proof', False) and option(native['args'], '-errOrdinaryLookObserve') != args.nonce: raise Refused('Native look observer nonce mismatch') gate_options(native['args'], args) if option(native['args'], '-errOrdinaryScenario') not in ('core', 'chat'): raise Refused('Not an ordinary successor scenario') if option(native['args'], '-profile') != 'Z:' + str(args.profile): raise Refused('Native profile mismatch') NUMBER = r'[-+]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][-+]?\d+)?' VECTOR = r'<\s*(' + NUMBER + r')\s*,\s*(' + NUMBER + r')\s*,\s*(' + NUMBER + r')\s*>' LOOK_HEADER = re.compile(r'ERR LOOK OBSERVE: run=([A-Za-z0-9]{1,32}) sample=(\d+) real=(\d+) (.+)$') LOOK_SAMPLE = re.compile('SAMPLE direction=' + VECTOR + ' camera=' + VECTOR + ' body=' + VECTOR + ' bodyForward=' + VECTOR + ' turn=' + VECTOR) INPUT_SAMPLE = re.compile('SAMPLE origin=' + VECTOR + ' forward=(' + NUMBER + ') input=' + VECTOR + ' velocity=' + VECTOR + r' inventory=(0|1|true|false)') def parse_look_record(line, nonce): """Strict live handshake parser; final qualification independently reparses logs.""" if 'ERR LOOK OBSERVE:' not in line: return None match = LOOK_HEADER.search(line) if not match or match[1] != nonce or len(line) > 2048: raise Refused('Malformed or foreign look observation') sample, real, detail = int(match[2]), int(match[3]), match[4] if not 0 <= sample <= 450 or not 0 <= real <= 2147483647: raise Refused('Look observation bounds invalid') record = {'sample': sample, 'real_ms': real, 'raw': line} if detail == 'BASIS native-player-camera-world-transform': return {**record, 'kind': 'basis'} if re.fullmatch(r'CANCELLED [a-z-]+', detail): return {**record, 'kind': 'cancelled', 'reason': detail.split(' ', 1)[1]} values = LOOK_SAMPLE.fullmatch(detail) if not values: raise Refused('Malformed native look sample') numbers = [float(value) for value in values.groups()] if not all(math.isfinite(value) for value in numbers): raise Refused('Nonfinite native look sample') result = {**record, 'kind': 'sample'} for index, name in enumerate(('direction', 'camera', 'body', 'bodyForward', 'turn')): result[name] = numbers[index*3:index*3+3] for name in ('direction', 'bodyForward'): length = math.sqrt(sum(value*value for value in result[name])) if not .95 <= length <= 1.05 or math.hypot(result[name][0], result[name][2]) <= .1: raise Refused('Native look direction is degenerate or not normalized') return result def parse_stationary_sample(sample, real, detail): match = INPUT_SAMPLE.fullmatch(detail) if not match: raise Refused('Malformed input sample during look observation') values = [float(value) for value in match.groups()[:-1]] if not all(math.isfinite(value) for value in values): raise Refused('Nonfinite input sample during look observation') return {'sample': sample, 'real_ms': real, 'origin': values[:3], 'forward': values[3], 'input': values[4:7], 'velocity': values[7:10], 'inventory': match.groups()[-1] in ('1', 'true')} def yaw(direction): return math.degrees(math.atan2(direction[0], direction[2])) def yaw_delta(first, second): return (yaw(second) - yaw(first) + 180) % 360 - 180 class RelativeMouse: def __init__(self, query, keyboard, owner, enable_approach=False, enable_alignment=False): self.query, self.keyboard, self.owner = query, keyboard, owner self.enable_approach = enable_approach self.enable_alignment = enable_alignment self.xt = keyboard.xt self.xt.XTestFakeRelativeMotionEvent.restype = C.c_int self.xt.XTestFakeRelativeMotionEvent.argtypes = [C.c_void_p, C.c_int, C.c_int, C.c_ulong] query.x.XQueryPointer.restype = C.c_int query.x.XQueryPointer.argtypes = [C.c_void_p, C.c_ulong, C.POINTER(C.c_ulong), C.POINTER(C.c_ulong), C.POINTER(C.c_int), C.POINTER(C.c_int), C.POINTER(C.c_int), C.POINTER(C.c_int), C.POINTER(C.c_uint)] def pointer(self): root, child = C.c_ulong(), C.c_ulong() rx, ry, wx, wy = C.c_int(), C.c_int(), C.c_int(), C.c_int() mask = C.c_uint() if not self.query.x.XQueryPointer(self.query.d, self.owner['window'], C.byref(root), C.byref(child), C.byref(rx), C.byref(ry), C.byref(wx), C.byref(wy), C.byref(mask)): raise Refused('Private pointer is unavailable or on another screen') return {'root': root.value, 'child': child.value, 'root_x': rx.value, 'root_y': ry.value, 'mask': mask.value} def validate_pointer(self, target, dx, dy, allowed_mask=0): position = self.pointer() geometry = target['geometry']; left, top, right, bottom = target['focused_clipped_bounds'] if position['root'] != geometry['root'] or position['mask'] & ~allowed_mask != 0: raise Refused('Private pointer root or button/modifier state invalid') for x, y in ((position['root_x'], position['root_y']), (position['root_x']+dx, position['root_y']+dy)): if not left <= x < right or not top <= y < bottom: raise Refused('Relative pointer step leaves owned clipped bounds') child = position['child'] if child and self.query.server_pid(child) != self.owner['pid']: raise Refused('Pointer child belongs to a foreign native client') return position def move(self, dx, dy, guard): if type(dx) is not int or dx not in (-24, 24) or dy != 0 or type(dy) is not int: raise Refused('Unsupported relative look step') self._move(dx, dy, guard) def approach_move(self, dx, dy, guard, *, deadline, clock=time.monotonic): if (self.enable_approach is not True or type(dx) is not int or not 1 <= abs(dx) <= 128 or type(dy) is not int or dy != 0): raise Refused('Unsupported or unenabled approach relative step') def within_lease(): if clock() >= deadline: raise Refused("Original approach deadline exceeded before relative input") within_lease() self._move(dx, dy, guard, before_send=within_lease) def alignment_move(self, dx, dy, guard, *, deadline, clock=time.monotonic): if (self.enable_alignment is not True or self.enable_approach is not True or type(dx) is not int or not 1 <= abs(dx) <= 128 or type(dy) is not int or dy != 0): raise Refused('Unsupported or unenabled alignment relative step') def within_lease(): if clock() >= deadline: raise Refused('Original alignment deadline exceeded before relative input') within_lease() self._move(dx, dy, guard, before_send=within_lease) def _move(self, dx, dy, guard, before_send=None): target = guard.check() self.keyboard.require_idle() self.validate_pointer(target, dx, dy) # Recheck process/focus after the pointer roundtrip, with no focus changes. target = guard.check() self.validate_pointer(target, dx, dy) if before_send is not None: before_send() if not self.xt.XTestFakeRelativeMotionEvent(self.query.d, dx, dy, 0): raise Refused('Native relative motion request failed') self.query.x.XFlush(self.query.d) def perform_look(guard, keyboard, observer, mouse, report, clock=time.monotonic, sleep=time.sleep): proof = {'requested': True, 'completed': False, 'started_monotonic': clock(), 'finished_monotonic': None, 'baseline_samples': [], 'response_sample': None, 'return_sample': None, 'motions': []} report['look_proof'] = proof deadline = proof['started_monotonic'] + 4 notify({'kind': 'look-pending'}) ready_sample = observer.ready_sample baseline = None def current(): if clock() >= deadline: raise Refused('Native look proof exceeded four seconds') observer.fresh(); guard.check(); keyboard.require_idle() if observer.events != ['READY'] or any(r['kind'] == 'cancelled' for r in observer.look_records): raise Refused('Native look context ended or inventory changed') if sum(r['kind'] == 'basis' for r in observer.look_records) != 1: return [] poses = [r for r in observer.look_records if r['kind'] == 'sample' and r['sample'] >= ready_sample] inputs = {r['sample']: r for r in observer.input_samples} joined = [] for pose in poses: sample = inputs.get(pose['sample']) if sample is None: continue if sample['real_ms'] != pose['real_ms'] or math.dist(sample['origin'], pose['body']) > .01: raise Refused('Native look/input body or clock mismatch') ready_input = inputs.get(ready_sample) if ready_input is None: raise Refused('Missing native READY body sample') origin = ready_input['origin'] if (sample['inventory'] or abs(sample['forward']) > .01 or math.sqrt(sum(x*x for x in sample['input'])) > .01 or math.sqrt(sum(x*x for x in sample['velocity'])) > .05 or math.dist(pose['body'], origin) > .05): raise Refused('Player moved before look proof completed') joined.append(pose) for previous, following in zip(joined, joined[1:]): if following['sample'] != previous['sample'] + 1 or not 0 < following['real_ms'] - previous['real_ms'] <= 500: raise Refused('Native look sampling gap or repeated clock') if joined and observer.input_samples[-1]['real_ms'] - joined[-1]['real_ms'] > 500: raise Refused('Native look pose is stale relative to input') return joined def wait(predicate): while clock() < deadline: poses = current() if poses and predicate(poses): return poses[-1] sleep(.025) raise Refused('Native view rotation acknowledgement missing') def motion(dx, after): poses = current() if not poses: raise Refused('Native look sample unavailable before motion') watermark = poses[-1]['sample'] report['input_attempted'] = True mouse.move(dx, 0, guard) proof['motions'].append({'dx': dx, 'dy': 0, 'monotonic': clock(), 'after_sample': after['sample'], 'after_real_ms': after['real_ms']}) report['input_sent'] = True return watermark try: def stable(poses): return len(poses) >= 2 and poses[-1]['sample'] == poses[-2]['sample']+1 and abs(yaw_delta(poses[-2]['direction'], poses[-1]['direction'])) <= .1 wait(stable) poses = current() if not stable(poses): raise Refused('Initial native view no longer stable') baseline = poses[-1] proof['baseline_samples'] = [poses[-2]['sample'], baseline['sample']] outbound_after = motion(24, baseline) response = wait(lambda p: p[-1]['sample'] > outbound_after and .25 <= abs(yaw_delta(baseline['direction'], p[-1]['direction'])) <= 30) proof['response_sample'] = response['sample'] displacement = abs(yaw_delta(baseline['direction'], response['direction'])) return_after = motion(-24, response) returned = wait(lambda p: p[-1]['sample'] > return_after and abs(yaw_delta(baseline['direction'], p[-1]['direction'])) <= .35 and abs(yaw_delta(baseline['direction'], p[-1]['direction'])) < displacement/2) proof['return_sample'] = returned['sample'] current() if clock() >= deadline: raise Refused('Native look proof exceeded four seconds') proof['completed'] = True finally: proof['finished_monotonic'] = clock() notify({'kind': 'look-finished'}) class Observer: """Only records appended after this exact owned console was opened count.""" pattern = re.compile(r'ERR INPUT OBSERVE: run=([A-Za-z0-9]{1,32}) sample=(\d+) real=(\d+) world=(\d+(?:\.\d+)?) (.+)$') def __init__(self, path, nonce, clock=time.monotonic, enable_look=False, enable_approach=False, enable_selection=False): self.path, self.nonce, self.clock = path, nonce, clock self.fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) info = os.fstat(self.fd) if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid(): os.close(self.fd) raise Refused('Console is not a regular same-user file') self.identity = info.st_dev, info.st_ino self.start_offset = os.lseek(self.fd, 0, os.SEEK_END) self.offset = self.start_offset self.pending = b'' self.sample = self.real = self.world = -1 self.last_sample_at = None self.events = [] self.raw = [] self.bytes = 0 self.terminal = False if enable_selection and not (enable_approach and enable_look): os.close(self.fd) raise Refused("Selection observation requires approach and look") if enable_approach and not enable_look: os.close(self.fd) raise Refused("Approach observation requires native look") self.enable_look = enable_look self.enable_approach = enable_approach self.enable_selection = enable_selection self.choice_previous = 0; self.choice_real = -1; self.selection_previous = None self.approach_lines = []; self.gate_records = []; self.gate_received = {} self.gate_complete_end = 0; self.approach_previous = None; self.alignment_previous = None self.ready_at = self.ready_real = self.queued_real = None self.look_records = []; self.input_samples = []; self.ready_sample = None def close(self): os.close(self.fd) def poll(self): info = self.path.lstat() if (info.st_dev, info.st_ino) != self.identity or not stat.S_ISREG(info.st_mode) or info.st_size < self.offset: raise Refused('Console replaced or truncated') data = os.read(self.fd, 65536) self.offset += len(data); self.bytes += len(data) if self.bytes > 8 * 1024 * 1024: raise Refused('Console observation byte budget exceeded') self.pending += data if len(self.pending) > 131072: raise Refused('Console line budget exceeded') lines = self.pending.split(b'\n'); self.pending = lines.pop() for raw in lines: line = raw.decode('utf-8', 'strict').rstrip('\r') if self.enable_approach and any(prefix in line for prefix in (('ERR INPUT OBSERVE:', 'ERR LOOK OBSERVE:', 'ERR GATE OBSERVE:', 'ERR GATE CHOICE:') if self.enable_selection else ('ERR INPUT OBSERVE:', 'ERR LOOK OBSERVE:', 'ERR GATE OBSERVE:'))): if len(self.approach_lines) >= (2400 if self.enable_selection else 1900): raise Refused('Native approach record bound exceeded') self.approach_lines.append(line) if self.enable_selection and 'ERR GATE CHOICE:' in line: from ordinary_gate_choices import parse_choice_record record = parse_choice_record(line, self.nonce) if record['sample'] != self.choice_previous + 1 or record['real_ms'] <= self.choice_real: raise Refused('Native choice records missing, repeated or reversed') if not self.gate_records or self.gate_records[-1]['kind'] != 'geometry' or any(record[k] != self.gate_records[-1][k] for k in ('sample', 'real_ms')): raise Refused('Native choices lack exact preceding gate geometry') self.choice_previous = record['sample']; self.choice_real = record['real_ms'] self.gate_complete_end = len(self.approach_lines) self.gate_received.setdefault(record['sample'], self.clock()) continue if self.enable_approach and 'ERR GATE OBSERVE:' in line: from ordinary_gate_evidence import parse_gate_record record = parse_gate_record(line, self.nonce) self.gate_records.append(record) if len(self.gate_records) > 902: raise Refused('Native gate record bound exceeded') if record['kind'] == 'geometry' and not self.enable_selection: self.gate_complete_end = len(self.approach_lines) self.gate_received.setdefault(record['sample'], self.clock()) continue if self.enable_look and 'ERR LOOK OBSERVE:' in line: record = parse_look_record(line, self.nonce) if len(self.look_records) >= 453: raise Refused('Native look record bound exceeded') if self.look_records and (record['sample'] < self.look_records[-1]['sample'] or record['real_ms'] < self.look_records[-1]['real_ms']): raise Refused('Native look records reversed') if any(r['kind'] == 'cancelled' for r in self.look_records): raise Refused('Native look records after cancellation') if record['kind'] == 'basis' and (self.look_records or record['sample'] != 1): raise Refused('Repeated or misplaced native look basis') if record['kind'] == 'sample' and not self.look_records: raise Refused('Missing native look basis') if record['kind'] == 'sample' and any(r['kind'] == 'sample' and r['sample'] == record['sample'] for r in self.look_records): raise Refused('Duplicate native look sample') self.look_records.append(record) continue if 'ERR INPUT OBSERVE: run=' + self.nonce + ' ' not in line: continue match = self.pattern.search(line) if not match or match[1] != self.nonce: raise Refused('Malformed matching observer record') sample, real, world = int(match[2]), int(match[3]), float(match[4]) if not math.isfinite(world) or real > 2147483647: raise Refused('Invalid native observer clock') if sample < self.sample or real < self.real or world < self.world: raise Refused('Native observer clock/sample reversed') if sample > 450 or len(self.raw) >= 512: raise Refused('Native observer record budget exceeded') detail = match[5] if detail.startswith('SAMPLE '): if sample <= self.sample: raise Refused('Repeated native sample') self.last_sample_at = self.clock() if self.enable_look: self.input_samples.append(parse_stationary_sample(sample, real, detail)) self.sample, self.real, self.world = sample, real, world self.raw.append(line) if self.enable_approach and detail == 'QUEUED': if self.queued_real is not None: raise Refused('Repeated native queued clock') self.queued_real = real if detail == 'READY': if self.events or self.last_sample_at is None: raise Refused('Repeated or unbound READY') self.events.append('READY') self.ready_sample = sample self.ready_at = self.clock(); self.ready_real = real elif detail in ('INVENTORY_OPEN', 'INVENTORY_CLOSED'): expected = ['READY'] if detail == 'INVENTORY_OPEN' else ['READY', 'INVENTORY_OPEN'] if self.events != expected: raise Refused('Unexpected native inventory transition') self.events.append(detail) elif detail.startswith(('CANCELLED ', 'FINISHED ')): self.terminal = True if self.terminal: raise Refused('Native observer ended') def fresh(self): self.poll() if self.last_sample_at is None or self.clock() - self.last_sample_at > 1: raise Refused('Native observer samples are stale') def approach_data(self): """Reparse complete raw triples, retaining incomplete final pairs for next poll.""" from ordinary_gate_evidence import join_gate_snapshots, validate_gate_geometry self.fresh() if (not self.enable_approach or self.events != ['READY'] or not self.gate_complete_end or any(r['kind'] == 'cancelled' for r in self.gate_records + self.look_records)): raise Refused('Native approach context is unavailable or ended') content = '\n'.join(self.approach_lines[:self.gate_complete_end]) if self.enable_selection: from ordinary_gate_choices import join_choice_snapshots data = join_choice_snapshots(content, self.nonce) else: data = join_gate_snapshots(content, self.nonce) index = max(data['snapshots']) geometry = validate_gate_geometry(content, self.nonce, data['ready']['sample'], index) if not geometry['verified']: raise Refused('Native gate geometry refused: ' + '; '.join(geometry['errors'])) for item in data['snapshots'].values(): action = item['action'] if item['sample'] >= data['ready']['sample'] and (action['pressed'] != 0 or action['performing'] or action['progress'] != 0): raise Refused('Gate action input or progress changed during approach-only observation') if (self.clock() - self.gate_received[index] > .5 or self.input_samples[-1]['real_ms'] - data['snapshots'][index]['real_ms'] > 500): raise Refused('Complete native gate snapshot is stale') return content, data def approach_snapshots(self): return self._snapshot_batch('approach_previous') def alignment_snapshots(self): return self._snapshot_batch('alignment_previous') def selection_snapshots(self): return self._snapshot_batch('selection_previous') def _snapshot_batch(self, cursor): _, data = self.approach_data() snapshots = data['snapshots']; latest = max(snapshots) previous = getattr(self, cursor) first = latest if previous is None else min(previous + 1, latest) if previous is not None and latest < previous: raise Refused('Native approach snapshot index reversed') result = [] for index in range(first, latest + 1): item = snapshots[index]; geo, action, native, pose = item['geometry'], item['action'], item['input'], item['look'] result.append({'sample': index, 'real_ms': item['real_ms'], 'received_monotonic': self.gate_received[index], 'origin': native['origin'], 'direction': pose['direction'], 'body_forward': pose['body_forward'], 'camera': pose['camera'], **{key: action[key] for key in ('selected', 'context_selected', 'cached_performable', 'progress_bound', 'pressed', 'progress', 'performing', 'duration')}, 'input': native['input'], 'forward': native['forward'], 'velocity': native['velocity'], 'inventory': native['inventory'], 'phase': action['phase'], 'door_state': geo['door_state'], 'door_control': geo['door_control'], 'geometry': {'nonce': self.nonce, 'basis': 'registered-prison-lock-native-interaction', **{key: geo[key] for key in ('prison', 'gate', 'gate_forward', 'context', 'radius')}}}) if self.enable_selection: choice = item['choices'] if cursor != 'approach_previous' and action['context_selected']: from ordinary_gate_choices import exact_prison_choices choice = exact_prison_choices(item) result[-1]['choices'] = {key: choice[key] for key in ('nonce','sample','real_ms','selected_id','selected_class','actions')} setattr(self, cursor, latest) return result class Keyboard: """Only W/Tab key events on the pre-existing explicit owned connection.""" def __init__(self, query): self.query = query self.xt = C.CDLL('libXtst.so.6') self.xt.XTestQueryExtension.restype = C.c_int self.xt.XTestQueryExtension.argtypes = [C.c_void_p] + [C.POINTER(C.c_int)] * 4 self.xt.XTestFakeKeyEvent.restype = C.c_int self.xt.XTestFakeKeyEvent.argtypes = [C.c_void_p, C.c_uint, C.c_int, C.c_ulong] for name, restype, argtypes in ( ('XKeysymToKeycode', C.c_ubyte, [C.c_void_p, C.c_ulong]), ('XKeycodeToKeysym', C.c_ulong, [C.c_void_p, C.c_ubyte, C.c_int]), ('XQueryKeymap', C.c_int, [C.c_void_p, C.POINTER(C.c_char)]), ('XFlush', C.c_int, [C.c_void_p])): fn = getattr(query.x, name); fn.restype = restype; fn.argtypes = argtypes values = [C.c_int() for _ in range(4)] if not self.xt.XTestQueryExtension(query.d, *(C.byref(v) for v in values)): raise Refused('Private XTest extension unavailable') self.codes = {} for key, keysym in (('W', 0x77), ('Tab', 0xff09)): code = int(query.x.XKeysymToKeycode(query.d, keysym)) if not 8 <= code <= 255 or query.x.XKeycodeToKeysym(query.d, code, 0) != keysym: raise Refused('Default unmodified key mapping unavailable') self.codes[key] = code if len(set(self.codes.values())) != 2: raise Refused('Ambiguous key mapping') def require_idle(self): state = C.create_string_buffer(32) if not self.query.x.XQueryKeymap(self.query.d, state): raise Refused('Native private keyboard state query failed') if any(state.raw): raise KeyStateBusy('A private keyboard key is still held') def send(self, key, down): if key not in self.codes or type(down) is not bool: raise Refused('Unsupported input event') if not self.xt.XTestFakeKeyEvent(self.query.d, self.codes[key], int(down), 0): raise Refused('Native XTest key request failed') self.query.x.XFlush(self.query.d) class Guard: def __init__(self, args, bound, query, owner): self.args, self.bound, self.query, self.owner = args, bound, query, owner def check(self): if display.binding(self.args.host_display, self.args.prefix) != self.bound: raise Refused('Original display binding changed') # A new per-check read budget, still bounded by the experiment deadline. self.query.query_budget = 128 current = {'window': self.owner['window'], 'pid': self.query.prop(self.owner['window'], '_NET_WM_PID'), 'server_pid': self.query.server_pid(self.owner['window'])} verified = display.owned_window(current, self.bound, self.args.prefix, self.args.game_executable) if verified is None or verified['process'] != self.owner['process'] or verified['pid'] != self.owner['pid']: raise Refused('Original game window or process changed') game_arguments(verified, self.args) target = display.input_target(self.query, verified) if target.get('verified') is not True: raise Refused('Owned existing focus/mapping unavailable: ' + target.get('reason', 'unknown')) if display.binding(self.args.host_display, self.args.prefix) != self.bound: raise Refused('Display binding changed during target check') return target def confirm_release(guard, keyboard, report, key, clock=time.monotonic, sleep=time.sleep): """Bounded read-only state polling; never sends/retries an input event.""" start = clock(); deadline = start + .35 observation = {'key': key, 'start': start, 'attempts': 0, 'confirmed': False} report.setdefault('release_observations', []).append(observation) try: while True: guard.check() if clock() >= deadline: raise Refused('Private key release state did not settle within 350ms') observation['attempts'] += 1 try: keyboard.require_idle() if clock() >= deadline: raise Refused('Private key release state query exceeded 350ms bound') observation['confirmed'] = True return except KeyStateBusy as error: observation['last_busy_error'] = str(error) remaining = deadline - clock() if remaining <= 0: raise Refused('Private key release state did not settle within 350ms') from error sleep(min(.025, remaining)) finally: observation['end'] = clock() def sequence(guard, keyboard, observer, report, clock=time.monotonic, sleep=time.sleep, mouse=None, approach=False, alignment=False, selection=False): deadline = clock() + 70 held = None approach_completed = False if selection and (not alignment or not approach or mouse is None or not getattr(observer, "enable_selection", False)): raise Refused("Selection requires explicit alignment and choice capabilities") if alignment and (not approach or mouse is None): raise Refused("Alignment requires approach and look capabilities") report.update(events=[], paired_release_errors=[], sequence_completed=False) def check(native=True): if clock() >= deadline: raise Refused('Input sequence deadline exceeded') if native: observer.fresh() return guard.check() def wait_for(event, seconds): end = min(deadline, clock() + seconds) while clock() < end: observer.poll() guard.check() if event in observer.events: observer.fresh() return sleep(0.05) raise Refused('Native transition timeout: ' + event) def key(key, down): nonlocal held check() if down: keyboard.require_idle() # Mark uncertain commit before the FFI call, so an exception also releases. held = key notify({'kind': 'press-pending', 'key': key}) report['input_attempted'] = True keyboard.send(key, down) report['events'].append({'key': key, 'down': down, 'monotonic': clock()}) report['input_sent'] = True if not down: confirm_release(guard, keyboard, report, key, clock, sleep) held = None notify({'kind': 'release-confirmed', 'key': key}) def pulse(key_name, duration): key(key_name, True) end = clock() + duration while clock() < end: sleep(min(0.05, end - clock())) # The key-up path performs a full fresh guard immediately before # sending. Avoid duplicating that guard once the hold has elapsed. if clock() < end: check() key(key_name, False) try: wait_for('READY', 45) deadline = clock() + 12 if approach: if mouse is None or not observer.enable_approach or observer.ready_at is None: raise Refused('Approach requires explicit look and gate observation') deadline = observer.ready_at + 35 if clock() >= deadline: raise Refused('Original approach deadline exceeded before calibration') notify({'kind': 'approach-pending', 'deadline_monotonic': deadline}) if observer.events != ['READY']: raise Refused('Inventory was already changed before input') if mouse is not None: perform_look(guard, keyboard, observer, mouse, report, clock, sleep) if not approach: deadline = clock() + 12 if approach: run_approach(guard, keyboard, observer, mouse, report, clock, sleep) if clock() >= deadline: raise Refused("Original approach deadline exceeded") keyboard.require_idle() approach_completed = True if alignment: notify({'kind': 'approach-finished', 'successful': True}) if selection: run_selection(guard, keyboard, observer, mouse, report, clock, sleep) else: run_alignment(guard, keyboard, observer, mouse, report, clock, sleep) report["sequence_completed"] = True return pulse('W', 0.30) check() pulse('Tab', 0.05) wait_for('INVENTORY_OPEN', 4) open_until = clock() + 3.0 while clock() < open_until: sleep(min(.05, open_until - clock())) check() if observer.events != ['READY', 'INVENTORY_OPEN']: raise Refused('Inventory closed before the observed open interval') pulse('Tab', 0.05) wait_for('INVENTORY_CLOSED', 4) check() keyboard.require_idle() report['sequence_completed'] = True finally: if held is not None: try: # Release needs live display/window/focus authority, but not a live # observer; cancellation must not suppress an otherwise safe key-up. guard.check() keyboard.send(held, False) confirm_release(guard, keyboard, report, held, clock, sleep) notify({'kind': 'release-confirmed', 'key': held}) report['events'].append({'key': held, 'down': False, 'finally': True, 'monotonic': clock()}) except (OSError, ValueError) as error: report['paired_release_errors'].append(str(error)) report['unconfirmed_held_key'] = held if report['paired_release_errors'] else None if approach: route = report.get('approach_proof', {}) report['paired_release_errors'].extend(route.get('paired_release_errors', [])) report['unconfirmed_held_key'] = route.get('unconfirmed_held_key') notify({'kind': 'approach-finished', 'successful': approach_completed if alignment else report['sequence_completed']}) report['all_attempted_presses_released'] = not report.get('unconfirmed_held_key') and not report['paired_release_errors'] and not report.get('unconfirmed_held_button') and not report.get('wheel_release_errors') class ApproachKeyboard: """W-only capability for the separate approach module; never F/Tab.""" def __init__(self, keyboard, report, clock, deadline): self.keyboard, self.report, self.clock, self.deadline = keyboard, report, clock, deadline def require_idle(self): self.keyboard.require_idle() def send(self, key, down): if key != 'W' or type(down) is not bool: raise Refused('Approach permits W key pairs only') if down and self.clock() >= self.deadline: raise Refused('Original approach deadline exceeded before W press') self.report['input_attempted'] = True self.keyboard.send(key, down) self.report['events'].append({'key': key, 'down': down, 'monotonic': self.clock()}) self.report['input_sent'] = True class ApproachMouse: def __init__(self, mouse, report, clock, deadline): self.mouse, self.report, self.clock, self.deadline = mouse, report, clock, deadline def move(self, dx, dy, guard): self.report['input_attempted'] = True self.mouse.approach_move(dx, dy, guard, deadline=self.deadline, clock=self.clock) self.report['input_sent'] = True def run_approach(guard, keyboard, observer, mouse, report, clock, sleep): from ordinary_look_evidence import validate_look_evidence from prison_approach_proof import perform_approach content, data = observer.approach_data() validated = validate_look_evidence(content, observer.nonce, report['look_proof']) if not validated['verified']: raise Refused('Native approach calibration refused: ' + '; '.join(validated['errors'])) poses = {r['sample']: r for r in observer.look_records if r['kind'] == 'sample'} proof = report['look_proof'] if observer.queued_real is None or observer.ready_at is None or observer.ready_real is None: raise Refused('Original native observation clocks missing') calibration = {'nonce': observer.nonce, 'completed': True, 'pixels': 24, 'baseline_direction': poses[proof['baseline_samples'][-1]]['direction'], 'response_direction': poses[proof['response_sample']]['direction'], 'ready_real_ms': observer.ready_real, 'ready_monotonic': observer.ready_at, 'observer_start_real_ms': observer.queued_real} geometry = observer.approach_snapshots()[-1]['geometry'] # The adapter has only read here. The module's initial snapshot can repeat # this latest sample; every subsequent native sample is returned in order. def route_notify(event): # Only the sequence owner can mark completion after its final checks. if event.get("kind") != "approach-finished": notify(event) deadline = observer.ready_at + 35 perform_approach(guard, ApproachKeyboard(keyboard, report, clock, deadline), ApproachMouse(mouse, report, clock, deadline), observer.approach_snapshots, geometry, calibration, report, clock=clock, sleep=sleep, notify=route_notify, key_state_busy=KeyStateBusy) if not report.get('approach_proof', {}).get('completed'): raise Refused('Approach module did not complete') class AlignmentMouse: """Explicit yaw-only capability. Caller cannot extend its original lease.""" def __init__(self, mouse, report, clock, deadline): self.mouse, self.report, self.clock, self.deadline = mouse, report, clock, deadline def move(self, dx, dy, guard, *, deadline): if type(deadline) not in (int, float) or not math.isfinite(deadline) or deadline > self.deadline: raise Refused('Alignment module attempted to extend original deadline') self.report['input_attempted'] = True self.mouse.alignment_move(dx, dy, guard, deadline=deadline, clock=self.clock) self.report['input_sent'] = True class SelectionWheel: """One button5 pair on the existing private connection; no other input API.""" def __init__(self, mouse, report, clock, sleep, deadline, enabled=False): self.mouse, self.report, self.clock, self.sleep = mouse, report, clock, sleep self.deadline, self.enabled, self.attempted = deadline, enabled, False if enabled is not True: raise Refused('Selection wheel capability is not enabled') self.xt = mouse.xt self.xt.XTestFakeButtonEvent.restype = C.c_int self.xt.XTestFakeButtonEvent.argtypes = [C.c_void_p, C.c_uint, C.c_int, C.c_ulong] def pulse(self, button, guard, *, deadline): if (self.enabled is not True or self.attempted or type(button) is not int or button != 5 or not math.isfinite(deadline) or deadline > self.deadline or self.clock() >= deadline): raise Refused('Unsupported, repeated or expired wheel trial') self.attempted = True report = self.report report.setdefault('wheel_events', []); report.setdefault('wheel_release_errors', []) up_attempted = False; pending = False; down_at = up_at = None mask = 1 << 12 # X11 Button5Mask; no other modifier/button may be held. def owned(releasing=False): target = guard.check(); self.mouse.keyboard.require_idle() self.mouse.validate_pointer(target, 0, 0, allowed_mask=mask if releasing else 0) target = guard.check() self.mouse.validate_pointer(target, 0, 0, allowed_mask=mask if releasing else 0) def send(down, cleanup=False): nonlocal up_attempted, pending, down_at, up_at owned(not down) if not cleanup and self.clock() >= deadline: raise Refused('Original facing lease expired before wheel request') if down: pending = True; report['unconfirmed_held_button'] = button notify({'kind':'button-pending','button':button}) else: up_attempted = True event = {'button':button,'down':down,'monotonic':self.clock(),'sent':False} if cleanup: event['finally'] = True report['wheel_events'].append(event) report['input_attempted'] = True if not self.xt.XTestFakeButtonEvent(self.mouse.query.d, button, int(down), 0): raise Refused('Native wheel request refused') self.mouse.query.x.XFlush(self.mouse.query.d) event['sent'] = True; report['input_sent'] = True if down: down_at = event['monotonic'] else: up_at = event['monotonic'] def confirm(): nonlocal pending end = self.clock() + .35 while True: target = guard.check(); self.mouse.keyboard.require_idle() position = self.mouse.validate_pointer(target, 0, 0, allowed_mask=mask) if self.clock() >= end: raise Refused('Private wheel release confirmation exceeded bound') if position['mask'] == 0: pending = False; report['unconfirmed_held_button'] = None notify({'kind':'button-release-confirmed','button':button}) return if position['mask'] != mask or self.clock() >= end: raise Refused('Private wheel release was not confirmed') self.sleep(min(.025, end-self.clock())) try: send(True) send(False) confirm() if self.clock() >= deadline: raise Refused('Original facing lease expired before wheel confirmation') return {'button':button,'down_monotonic':down_at,'up_monotonic':up_at,'released':True} except Exception as error: report['wheel_release_errors'].append(str(error)) raise finally: if pending and not up_attempted: try: # Expiry permits only cleanup of our already attempted button; # current private ownership remains mandatory, without native state. send(False, cleanup=True); confirm() except Exception as error: report['wheel_release_errors'].append(str(error)) if pending: report['unconfirmed_held_button'] = button def run_alignment(guard, keyboard, observer, mouse, report, clock, sleep): from prison_alignment_proof import perform_alignment from prison_approach_evidence import validate_approach_evidence if observer.ready_at is None or observer.ready_real is None or observer.queued_real is None: raise Refused('Original native alignment clocks missing') original_end = observer.ready_at + (45000 - (observer.ready_real - observer.queued_real))/1000 - .5 started = clock() deadline = min(started + 8, original_end) if deadline <= started: raise Refused('Original observation lifetime unavailable for alignment') notify({'kind': 'alignment-pending', 'deadline_monotonic': deadline}) successful = False try: content, data = observer.approach_data() route = report.get('approach_proof') evidence = validate_approach_evidence(content, observer.nonce, report['look_proof'], route) if not evidence['verified']: raise Refused('Native approach prerequisite refused: ' + '; '.join(evidence['errors'])) if clock() >= deadline: raise Refused('Original alignment deadline exceeded during prerequisite validation') poses = {r['sample']: r for r in observer.look_records if r['kind'] == 'sample'} look = report['look_proof'] calibration = {'nonce': observer.nonce, 'completed': True, 'pixels': 24, 'baseline_direction': poses[look['baseline_samples'][-1]]['direction'], 'response_direction': poses[look['response_sample']]['direction'], 'ready_real_ms': observer.ready_real, 'ready_monotonic': observer.ready_at, 'observer_start_real_ms': observer.queued_real, 'observer_deadline_monotonic': original_end, 'alignment_deadline_monotonic': deadline} observer.alignment_previous = evidence['last_sample'] - 1 def alignment_notify(event): if event.get('kind') != 'alignment-finished': notify(event) # Native ownership checks remain live, but no keyboard-sending capability # is handed to this stage. It may only assert that all keys are idle. perform_alignment(guard, keyboard.require_idle, AlignmentMouse(mouse, report, clock, deadline), observer.alignment_snapshots, route['geometry'], calibration, evidence, report, clock=clock, sleep=sleep, notify=alignment_notify) if not report.get('alignment_proof', {}).get('completed'): raise Refused('Alignment module did not complete') guard.check(); keyboard.require_idle() observer.approach_data() if clock() >= deadline: raise Refused('Original alignment deadline exceeded at final observation') successful = True finally: notify({'kind': 'alignment-finished', 'successful': successful}) def run_selection(guard, keyboard, observer, mouse, report, clock, sleep): from prison_alignment_proof import perform_facing from prison_alignment_evidence import validate_facing_evidence from prison_action_selection_proof import perform_selection from prison_approach_evidence import validate_approach_evidence if observer.ready_at is None or observer.ready_real is None or observer.queued_real is None: raise Refused('Original native alignment clocks missing') original_end = observer.ready_at + (45000 - (observer.ready_real - observer.queued_real))/1000 - .5 started = clock() deadline = min(started + 8, original_end) if deadline <= started: raise Refused('Original observation lifetime unavailable for alignment') notify({'kind': 'alignment-pending', 'deadline_monotonic': deadline}) successful = False try: content, data = observer.approach_data() route = report.get('approach_proof') evidence = validate_approach_evidence(content, observer.nonce, report['look_proof'], route) if not evidence['verified']: raise Refused('Native approach prerequisite refused: ' + '; '.join(evidence['errors'])) if clock() >= deadline: raise Refused('Original alignment deadline exceeded during prerequisite validation') poses = {r['sample']: r for r in observer.look_records if r['kind'] == 'sample'} look = report['look_proof'] calibration = {'nonce': observer.nonce, 'completed': True, 'pixels': 24, 'baseline_direction': poses[look['baseline_samples'][-1]]['direction'], 'response_direction': poses[look['response_sample']]['direction'], 'ready_real_ms': observer.ready_real, 'ready_monotonic': observer.ready_at, 'observer_start_real_ms': observer.queued_real, 'observer_deadline_monotonic': original_end, 'alignment_deadline_monotonic': deadline} observer.alignment_previous = evidence['last_sample'] - 1 def alignment_notify(event): if event.get('kind') in ('facing-pending', 'action-selection-pending'): notify({'kind':'alignment-pending','deadline_monotonic':event['deadline_monotonic']}) # Native ownership checks remain live, but no keyboard-sending capability # is handed to this stage. It may only assert that all keys are idle. perform_facing(guard, keyboard.require_idle, AlignmentMouse(mouse, report, clock, deadline), observer.alignment_snapshots, route['geometry'], calibration, evidence, report, clock=clock, sleep=sleep, notify=alignment_notify) facing = report.get('facing_proof', {}) if not facing.get('completed'): raise Refused('Facing module did not complete') deadline = min(deadline, facing['deadline_monotonic']) content, data = observer.approach_data() facing_evidence = validate_facing_evidence(content, observer.nonce, look, route, facing) if not facing_evidence.get('verified'): raise Refused('Independent native facing prerequisite refused') if clock() >= deadline: raise Refused('Original facing lease expired during validation') observer.selection_previous = facing_evidence['last_sample'] - 1 wheel = SelectionWheel(mouse, report, clock, sleep, deadline, enabled=True) perform_selection(guard, keyboard.require_idle, wheel, observer.selection_snapshots, route['geometry'], calibration, facing_evidence, report, clock=clock, sleep=sleep, notify=alignment_notify) if not report.get('action_selection_proof', {}).get('completed'): raise Refused('Selection module did not complete') guard.check(); keyboard.require_idle() observer.approach_data() if clock() >= deadline: raise Refused('Original alignment deadline exceeded at final observation') successful = True finally: notify({'kind': 'alignment-finished', 'successful': successful}) def arguments(argv=None): p = argparse.ArgumentParser(description=__doc__) p.add_argument('--enable-private-keyboard-proof', action='store_true', required=True) p.add_argument('--enable-private-selection-proof', action='store_true', help='After facing, try one button5 wheel selection; never perform the action') p.add_argument('--enable-private-look-proof', action='store_true') p.add_argument('--enable-private-alignment-proof', action='store_true', help='After successful approach, attempt stationary native lock selection using horizontal relative mouse only') p.add_argument('--enable-private-approach-proof', action='store_true', help='After look calibration, replace W/Tab cycle with bounded interior approach only') for name in ('prefix', 'game-executable', 'profile', 'report'): p.add_argument('--' + name, type=Path, required=True) p.add_argument('--console', type=Path, help='Existing exact session; omit only in wrapper mode') p.add_argument('--host-display', required=True) p.add_argument('--nonce', required=True) p.add_argument('command', nargs=argparse.REMAINDER) args = p.parse_args(argv) if args.enable_private_selection_proof and not (args.enable_private_alignment_proof and args.enable_private_approach_proof and args.enable_private_look_proof): p.error('Selection proof requires explicit alignment, approach and look proof') if args.enable_private_alignment_proof and not (args.enable_private_approach_proof and args.enable_private_look_proof): p.error("Alignment proof requires explicit approach and look proof") if args.enable_private_approach_proof and not args.enable_private_look_proof: p.error("Approach proof requires explicit private look proof") return args def wrapper_command(args): command = args.command[1:] if args.command[:1] == ['--'] else args.command if not command: if args.console is None: raise Refused('Sidecar requires an exact existing console') return [] if args.console is not None: raise Refused('Wrapper discovers a new session; do not supply console') if (len(command) < 3 or Path(command[0]).resolve() != Path(sys.executable).resolve() or Path(command[1]) != Path(display.__file__).resolve()): raise Refused('Only the unchanged sibling read-only display wrapper is allowed') wrapped = display.parse_args(command[2:]) if (wrapped.query_worker or wrapped.prefix != args.prefix or wrapped.profile != args.profile or wrapped.game_executable != args.game_executable or wrapped.host_display != args.host_display or wrapped.report == args.report): raise Refused('Read-only wrapper context differs') display.validate_paths(wrapped) native = wrapped.command[1:] if wrapped.command[:1] == ['--'] else wrapped.command proton = Path.home() / '.local/share/Steam/steamapps/common/Proton - Experimental/proton' if len(native) < 3 or native[:3] != [str(proton), 'run', str(args.game_executable)]: raise Refused('Expected exact installed Proton/game command') if (option(native, '-errOrdinaryInputObserve') != args.nonce or option(native, '-profile') != 'Z:' + str(args.profile) or option(native, '-errOrdinaryScenario') not in ('core', 'chat')): raise Refused('Wrapped native observer context differs') if args.enable_private_look_proof and option(native, '-errOrdinaryLookObserve') != args.nonce: raise Refused('Wrapped native look observer context differs') gate_options(native, args) return command def sessions(profile): folder = profile / 'logs' if not folder.exists(): return set() if folder.is_symlink() or not folder.is_dir(): raise Refused('Invalid private native logs directory') return {path for path in folder.iterdir() if re.fullmatch(r'logs_\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}', path.name)} def console_path(args): path = args.console if (path != path.resolve() or path.name != 'console.log' or path.parent.parent != args.profile / 'logs' or not re.fullmatch(r'logs_\d{4}-\d{2}-\d{2}_\d{2}-\d{2}-\d{2}', path.parent.name)): raise Refused('Expected exact canonical native session console') APPROACH_MODULES = ('prison_approach_proof.py', 'ordinary_gate_evidence.py', 'ordinary_look_evidence.py', 'prison_approach_evidence.py') ALIGNMENT_MODULES = ('prison_alignment_proof.py', 'prison_alignment_evidence.py') SELECTION_MODULES = ('ordinary_gate_choices.py', 'prison_action_selection_proof.py', 'prison_action_selection_evidence.py') def approach_module_hashes(): return module_hashes(APPROACH_MODULES) def module_hashes(names): folder = Path(__file__).resolve().parent return {name: hashlib.sha256((folder / name).read_bytes()).hexdigest() for name in names} def report_for_write(report): """Keep raw console as pose authority; archive bounded references, not duplicate poses.""" if report.get('approach_requested') is not True: return report result = dict(report) names = ['approach_proof'] if report.get('alignment_requested') is True: names.append('alignment_proof') if report.get('selection_requested') is True: names += ['facing_proof', 'action_selection_proof'] for name in names: route = report.get(name) if isinstance(route, dict): result[name] = dict(route) samples = route.get('samples') if isinstance(samples, list): result[name]['samples'] = [ {key: sample.get(key) for key in ('sample', 'real_ms', 'received_monotonic')} if isinstance(sample, dict) else {'invalid_reference': True} for sample in samples] return result def main(argv=None): args = arguments(argv) args.observe_seconds = 25; args.query_worker = False display.validate_paths(args) if not re.fullmatch(r'[A-Za-z0-9]{1,32}', args.nonce): raise Refused('Invalid observer nonce') command = wrapper_command(args) if args.console is not None: console_path(args) bound = display.binding(args.host_display, args.prefix) report = {'schema': 1, 'kind': 'headless-private-keyboard-experiment', 'input_sent': False, 'look_requested': args.enable_private_look_proof, 'approach_requested': getattr(args, 'enable_private_approach_proof', False), 'alignment_requested': getattr(args, 'enable_private_alignment_proof', False), 'selection_requested': getattr(args, 'enable_private_selection_proof', False), 'sequence_completed': False, 'release_qualified': False, 'gameplay_qualified': False, 'hard_supervision': NOTIFY is not None, 'nonce': args.nonce, 'binding': bound, 'command': command, 'report': str(args.report), 'input_attempted': False, 'helper_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), 'ownership_helper_sha256': hashlib.sha256(Path(display.__file__).read_bytes()).hexdigest(), 'prefix': str(args.prefix), 'profile': str(args.profile), 'game_executable': str(args.game_executable)} if report['approach_requested']: report['approach_module_sha256'] = approach_module_hashes() if report['alignment_requested']: report['alignment_module_sha256'] = module_hashes(ALIGNMENT_MODULES) if report['selection_requested']: report['selection_module_sha256'] = module_hashes(SELECTION_MODULES) report.update(wheel_events=[], wheel_release_errors=[], unconfirmed_held_button=None) fd = os.open(args.report, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) info = os.fstat(fd) notify({'kind': 'report', 'path': str(args.report), 'device': info.st_dev, 'inode': info.st_ino, 'metadata': report}) query = observer = child = None before_sessions = sessions(args.profile) if command else set() deadline = time.monotonic() + 180 previous = {} def interrupted(signum, frame): if report.get('sequence_completed') and fd is None and child is not None: if child.poll() is None: child.send_signal(signum) return raise Refused('Input experiment interrupted by signal ' + str(signum)) try: for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP): previous[signum] = signal.signal(signum, interrupted) if command: child = subprocess.Popen(command) # inherit the existing private process group notify({'kind': 'child', 'identity': display.identity(display.process(child.pid)), 'command': command}) while time.monotonic() < deadline: if child.poll() is not None: raise Refused('Read-only wrapper exited before native session') if display.binding(args.host_display, args.prefix) != bound: raise Refused('Display binding changed during startup') new = sessions(args.profile) - before_sessions if len(new) > 1: raise Refused('Ambiguous new native sessions') if new: candidate = next(iter(new)) / 'console.log' if candidate.exists(): args.console = candidate; console_path(args); break time.sleep(.1) if args.console is None: raise Refused('New native console deadline exceeded') report['console'] = str(args.console) observer = Observer(args.console, args.nonce, enable_look=args.enable_private_look_proof, enable_approach=getattr(args, "enable_private_approach_proof", False), enable_selection=getattr(args, "enable_private_selection_proof", False)) query = display.X11(bound['display']) while time.monotonic() < deadline: if child is not None and child.poll() is not None: raise Refused('Read-only wrapper exited during startup') observer.poll() if display.binding(args.host_display, args.prefix) != bound: raise Refused('Display binding changed during window startup') # Bind only after the native mission's original player is ready. # Startup/splash windows may legitimately disappear before then; # after binding, the existing guard still forbids any replacement. if observer.events != ['READY']: time.sleep(.1) continue query.query_budget = display.MAX_WINDOWS owners = [owner for window in query.windows() if (owner := display.owned_window(window, bound, args.prefix, args.game_executable))] eligible = [owner for owner in owners[:16] if display.input_target(query, owner).get('verified') is True] exact = [owner for owner in eligible if query.focus()[0] == owner['window']] selected = exact or eligible if len(selected) == 1: break if len(selected) > 1: raise Refused('Ambiguous mapped focused owned game windows') time.sleep(.1) else: raise Refused('Mapped owned game window deadline exceeded') owner = selected[0] report['owner'] = {key: owner[key] for key in ('window', 'pid', 'server_pid', 'process')} guard = Guard(args, bound, query, owner) guard.check() keyboard = Keyboard(query) keyboard.require_idle() mouse = RelativeMouse(query, keyboard, owner, enable_approach=getattr(args, "enable_private_approach_proof", False), enable_alignment=getattr(args, "enable_private_alignment_proof", False)) if args.enable_private_look_proof else None sequence(guard, keyboard, observer, report, mouse=mouse, approach=getattr(args, "enable_private_approach_proof", False), alignment=getattr(args, "enable_private_alignment_proof", False), selection=getattr(args, "enable_private_selection_proof", False)) except (OSError, ValueError, UnicodeError, subprocess.SubprocessError) as error: report['error'] = str(error) finally: report['whole_process_group_cleanup_verified'] = False # outer runner owns it if observer: report['observer'] = {'inode': observer.identity, 'start_offset': observer.start_offset, 'end_offset': observer.offset, 'records': observer.raw, 'events': observer.events} if args.enable_private_look_proof: report['look_observation'] = observer.look_records if getattr(args, "enable_private_approach_proof", False): report["gate_observation"] = observer.gate_records observer.close() if query: query.close() with os.fdopen(fd, 'w') as stream: fd = None json.dump(report_for_write(report), stream, indent=2); stream.write('\n'); stream.flush(); os.fsync(stream.fileno()) notify({'kind': 'proof-finished', 'successful': report['sequence_completed'] and not report.get('error')}) child_code = 0 try: if child is not None: if report.get('error'): child.terminate() child_code = child.wait(timeout=max(.1, deadline - time.monotonic()) if not report.get('error') else 5) except subprocess.TimeoutExpired: child.terminate() try: child.wait(timeout=5) except subprocess.TimeoutExpired: child.kill(); child.wait(timeout=2) child_code = 1 finally: for signum, handler in previous.items(): signal.signal(signum, handler) return 0 if report['sequence_completed'] and not report.get('error') and child_code == 0 else 1 def worker_entry(argv, pipe): global NOTIFY NOTIFY = pipe try: code = main(argv) finally: pipe.close() raise SystemExit(code) def stop_exact_child(record, read=display.process, kill=os.kill): if record is None: return False try: current = read(record['identity']['pid']) if (display.identity(current) != record['identity'] or current['args'] != record['command'] or current['uid'] != os.getuid()): return False # Only the exact display wrapper created by our worker; no group signalling. kill(current['pid'], signal.SIGTERM) return True except (OSError, ValueError, KeyError): return False def supervision_failure(record, key, reason, button=None): """Amend only the exact exclusive report created by our now-stopped worker.""" if record is None: return fd = os.open(record['path'], os.O_RDWR | os.O_NOFOLLOW) try: info = os.fstat(fd) if (info.st_dev, info.st_ino) != (record['device'], record['inode']) or info.st_uid != os.getuid() or not stat.S_ISREG(info.st_mode): raise Refused('Worker report changed before timeout recording') raw = os.read(fd, 2 * 1024 * 1024) try: report = json.loads(raw) if raw else dict(record['metadata']) except (ValueError, UnicodeError): report = dict(record['metadata']) # Keep the worker's actual refusal; supervisor lifecycle is separate evidence. key = key or report.get('unconfirmed_held_key') button = button or report.get('unconfirmed_held_button') report.setdefault('error', reason) report.update(supervision_failed=True, supervision_error=reason, sequence_completed=False, all_attempted_presses_released=key is None and button is None and report.get('all_attempted_presses_released', True), unconfirmed_held_key=key) if key is not None: report['input_attempted'] = True report.setdefault('paired_release_errors', []).append('Supervisor could not confirm paired key release') if button is not None: report['unconfirmed_held_button'] = button report['input_attempted'] = True report.setdefault('wheel_release_errors', []).append('Supervisor could not confirm wheel button release') payload = (json.dumps(report, indent=2) + '\n').encode() os.lseek(fd, 0, os.SEEK_SET); os.write(fd, payload); os.ftruncate(fd, len(payload)); os.fsync(fd) finally: os.close(fd) def supervise(argv=None): # Fork before loading Xlib or opening any display. The worker owns the one # connection. Parent never opens a second connection or retries an event. context = multiprocessing.get_context('fork') parent, child_pipe = context.Pipe(duplex=False) child = context.Process(target=worker_entry, args=(argv, child_pipe)) previous = {}; received = [] def forward(signum, frame): received.append(signum) if child.pid and child.is_alive(): os.kill(child.pid, signum) for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP): previous[signum] = signal.signal(signum, forward) alignment_deadline = None; alignment_finished = False approach_deadline = None; approach_finished = False button_at = None; held_button = None look_at = None; held_at = None; held_key = None; owned_child = None; report_record = None; proof_finished = False; expired = False deadline = time.monotonic() + 190 try: child.start(); child_pipe.close() while child.is_alive(): if parent.poll(.05): try: event = parent.recv() except EOFError: event = None if event and event['kind'] == 'alignment-pending': proposed = event.get('deadline_monotonic') current = time.monotonic() if type(proposed) not in (int, float) or not math.isfinite(proposed) or proposed > current + 8: proposed = current alignment_deadline = proposed if alignment_deadline is None else min(alignment_deadline, proposed) elif event and event['kind'] == 'alignment-finished': alignment_finished = bool(event.get('successful')) elif event and event['kind'] == 'approach-pending': proposed = event.get('deadline_monotonic') current = time.monotonic() if type(proposed) not in (int, float) or not math.isfinite(proposed) or proposed > current + 35: proposed = current # malformed lease fails closed approach_deadline = proposed if approach_deadline is None else min(approach_deadline, proposed) elif event and event['kind'] == 'approach-finished': approach_finished = bool(event.get('successful')) elif event and event['kind'] == 'look-pending': look_at = time.monotonic() elif event and event['kind'] == 'look-finished': look_at = None elif event and event['kind'] == 'button-pending': if button_at is None: button_at = time.monotonic() held_button = event['button'] elif event and event['kind'] == 'button-release-confirmed': if event['button'] == held_button: button_at = None; held_button = None elif event and event['kind'] == 'press-pending': held_at = time.monotonic(); held_key = event['key'] elif event and event['kind'] == 'release-confirmed': held_at = None; held_key = None elif event and event['kind'] == 'child': owned_child = event elif event and event['kind'] == 'report': report_record = event elif event and event['kind'] == 'proof-finished': proof_finished = event['successful'] now = time.monotonic() if received or now >= deadline or (alignment_deadline is not None and not alignment_finished and now >= alignment_deadline) or (approach_deadline is not None and not approach_finished and now >= approach_deadline) or (button_at is not None and now - button_at > 1.5) or (held_at is not None and now - held_at > 1.5) or (look_at is not None and now - look_at > 4): expired = True child.terminate() # asks worker finally to release if still provable child.join(1) stop_exact_child(owned_child) if child.is_alive(): child.kill(); child.join(2) # Drain release acknowledgements emitted during cooperative finally. while parent.poll(): try: last = parent.recv() except EOFError: break if last['kind'] == 'release-confirmed': held_key = None elif last['kind'] == 'button-release-confirmed' and last['button'] == held_button: held_button = None elif last['kind'] == 'report': report_record = last elif last['kind'] == 'proof-finished': proof_finished = last['successful'] if not (received and proof_finished and held_key is None and held_button is None): supervision_failure(report_record, held_key, 'Private input worker timed out or was interrupted', button=held_button) break child.join(2) if child.exitcode != 0 and not expired: stop_exact_child(owned_child) if not (received and proof_finished and held_key is None and held_button is None): supervision_failure(report_record, held_key, 'Private input worker exited unsuccessfully', button=held_button) return 1 if expired or child.exitcode != 0 else 0 finally: if child.pid and child.is_alive(): child.terminate(); child.join(1) stop_exact_child(owned_child) if child.is_alive(): child.kill(); child.join(2) parent.close() for signum, handler in previous.items(): signal.signal(signum, handler) if __name__ == '__main__': raise SystemExit(supervise())