aussie/EscapistsReintroduced
1
0
Fork 0
forked from escapists/EscapistsReintroduced
Code Pull Requests Activity

Verify private relative mouse input against native player camera

Browse Source
This commit is contained in:
jakub
2026-09-19 18:42:55 +02:00
parent 67de3752fb
commit e10b8799e9
60 changed files with 14049 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/headless_input_proof.py
+236 -8
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Opt-in bounded private XTest keyboard experiment. Not a launcher or gameplay proof."""
"""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
@@ -49,17 +49,210 @@ def game_arguments(owner, args):
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')
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):
self.query, self.keyboard, self.owner = query, keyboard, owner
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):
position = self.pointer()
geometry = target['geometry']; left, top, right, bottom = target['focused_clipped_bounds']
if position['root'] != geometry['root'] or position['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')
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')
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 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):
def __init__(self, path, nonce, clock=time.monotonic, enable_look=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)
@@ -76,6 +269,8 @@ class Observer:
self.raw = []
self.bytes = 0
self.terminal = False
self.enable_look = enable_look
self.look_records = []; self.input_samples = []; self.ready_sample = None
def close(self):
os.close(self.fd)
@@ -94,6 +289,22 @@ class Observer:
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_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)
@@ -111,12 +322,15 @@ class Observer:
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 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
elif detail in ('INVENTORY_OPEN', 'INVENTORY_CLOSED'):
expected = ['READY'] if detail == 'INVENTORY_OPEN' else ['READY', 'INVENTORY_OPEN']
if self.events != expected:
@@ -225,7 +439,7 @@ def confirm_release(guard, keyboard, report, key, clock=time.monotonic, sleep=ti
observation['end'] = clock()
def sequence(guard, keyboard, observer, report, clock=time.monotonic, sleep=time.sleep):
def sequence(guard, keyboard, observer, report, clock=time.monotonic, sleep=time.sleep, mouse=None):
deadline = clock() + 70
held = None
report.update(events=[], paired_release_errors=[], sequence_completed=False)
@@ -281,6 +495,9 @@ def sequence(guard, keyboard, observer, report, clock=time.monotonic, sleep=time
deadline = clock() + 12
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)
deadline = clock() + 12
pulse('W', 0.30)
check()
pulse('Tab', 0.05)
@@ -315,6 +532,7 @@ def sequence(guard, keyboard, observer, report, clock=time.monotonic, sleep=time
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-look-proof', action='store_true')
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')
@@ -349,6 +567,8 @@ def wrapper_command(args):
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')
return command
@@ -380,6 +600,7 @@ def main(argv=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,
'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(),
@@ -421,7 +642,7 @@ def main(argv=None):
if args.console is None:
raise Refused('New native console deadline exceeded')
report['console'] = str(args.console)
observer = Observer(args.console, args.nonce)
observer = Observer(args.console, args.nonce, enable_look=args.enable_private_look_proof)
query = display.X11(bound['display'])
while time.monotonic() < deadline:
if child is not None and child.poll() is not None:
@@ -448,7 +669,8 @@ def main(argv=None):
guard.check()
keyboard = Keyboard(query)
keyboard.require_idle()
sequence(guard, keyboard, observer, report)
mouse = RelativeMouse(query, keyboard, owner) if args.enable_private_look_proof else None
sequence(guard, keyboard, observer, report, mouse=mouse)
except (OSError, ValueError, UnicodeError, subprocess.SubprocessError) as error:
report['error'] = str(error)
finally:
@@ -456,6 +678,8 @@ def main(argv=None):
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
observer.close()
if query:
query.close()
@@ -549,7 +773,7 @@ def supervise(argv=None):
os.kill(child.pid, signum)
for signum in (signal.SIGINT, signal.SIGTERM, signal.SIGHUP):
previous[signum] = signal.signal(signum, forward)
held_at = None; held_key = None; owned_child = None; report_record = None; proof_finished = False; expired = False
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()
@@ -559,7 +783,11 @@ def supervise(argv=None):
event = parent.recv()
except EOFError:
event = None
if event and event['kind'] == 'press-pending':
if 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'] == 'press-pending':
held_at = time.monotonic(); held_key = event['key']
elif event and event['kind'] == 'release-confirmed':
held_at = None; held_key = None
@@ -570,7 +798,7 @@ def supervise(argv=None):
elif event and event['kind'] == 'proof-finished':
proof_finished = event['successful']
now = time.monotonic()
if received or now >= deadline or (held_at is not None and now - held_at > 1.5):
if received or now >= deadline 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)