#!/usr/bin/env python3 """Private-display observer policy tests; native libraries/processes are substituted.""" import ctypes as C import json import os from pathlib import Path import signal import shutil import subprocess import sys import tempfile from types import SimpleNamespace import unittest from unittest.mock import Mock, patch sys.path.insert(0, str(Path(__file__).resolve().parents[1]/'tools')) import headless_display_probe as probe def proc(pid, parent=1, exe='/usr/bin/helper', args=None, start=10): return {'pid':pid,'ppid':parent,'exe':exe,'args':args or [exe], 'start':start,'uid':os.getuid()} class FakeWindowQuery: def __init__(self): self.root_id=2;self.focused=(20,2) self.attrs={2:self.attribute(1280,720),20:self.attribute(800,600),21:self.attribute(200,100)} self.positions={20:(20,30),21:(40,50)};self.parents={20:2,21:20};self.pids={20:90,21:90} def attribute(self,w,h):return {'root':2,'window_class':1,'map_state':2,'width':w,'height':h} def root(self):return self.root_id def attributes(self,window):return dict(self.attrs[window]) def translate(self,window,root):return self.positions[window] def focus(self):return self.focused def server_pid(self,window):return self.pids.get(window) def tree(self,window):return 2,self.parents[window],[] class DisplayTests(unittest.TestCase): def setUp(self): self.temp=tempfile.TemporaryDirectory(prefix='erri-display-test-') self.root=Path(self.temp.name) self.profile=self.root/'profile';self.profile.mkdir() (self.profile/'.erri-game-profile.json').write_text(json.dumps(probe.PROFILE_ID)) self.prefix=self.root/'pfx';self.prefix.mkdir() self.exe=self.root/'ArmaReforgerSteam.exe';self.exe.write_bytes(b'test-fixture-not-executable') self.proton=self.root/'proton';self.proton.write_text('test-fixture-not-executable') self.report=self.profile/'display-ownership-fixture.json' self.bound={'display':':2','gamescope':{'pid':40,'start':10,'uid':os.getuid()},'xwayland':{'pid':41,'start':11,'uid':os.getuid()},'sockets':[{'path':'@/tmp/.X11-unix/X2','inode':100}], 'relationship':{'mode':'descendant','descendant':True}} self.args=SimpleNamespace(host_display=':0',prefix=self.prefix,game_executable=self.exe,profile=self.profile, report=self.report,observe_seconds=20,query_worker=False) def tearDown(self): self.temp.cleanup() def cli(self): return ['--host-display',':0','--prefix',str(self.prefix),'--game-executable',str(self.exe),'--profile',str(self.profile), '--report',str(self.report),'--observe-seconds','20','--',str(self.proton),'run',str(self.exe),'-noBackend'] def test_display_refuses_host_alias_remote_empty_or_malformed(self): for display,host in [(':0',':0.1'),('unix:0.0',':0'),(':0','localhost:0'),('',':0'),('host:2',':0'),(':2;evil',':0'),(':2/0',':0')]: with self.subTest(display=display,host=host),self.assertRaises(ValueError):probe.validate_display(display,host) self.assertEqual(probe.validate_display(':2.1',':0'),2) self.assertEqual(probe.validate_display('unix:2',''),2) def test_ancestor_requires_native_same_user_headless_before_separator(self): table={50:proc(50,40),40:proc(40,1,'/usr/bin/gamescope',['gamescope','--backend','headless','--','proton'])} self.assertEqual(probe.ancestor(50,table.__getitem__)['pid'],40) for args in (['gamescope','--','--backend','headless'],['gamescope','--backend','wayland','--'],['gamescope','--backend','headless','--backend=drm','--']): table[40]['args']=args with self.assertRaises(ValueError):probe.ancestor(50,table.__getitem__) table[40]['uid']=os.getuid()+1 with self.assertRaises(ValueError):probe.ancestor(50,table.__getitem__) def binding_fixture(self,extra=None): table={50:proc(50,40),40:proc(40,1,'/usr/bin/gamescope',['gamescope','--backend','headless','--','proton']), 41:proc(41,40,'/usr/bin/Xwayland',['Xwayland',':2','-rootless'],11)} if extra:extra(table) folder=self.root/'proc';folder.mkdir(exist_ok=True) for p in table: (folder/str(p)).mkdir(exist_ok=True) (folder/'net').mkdir(exist_ok=True) (folder/'net/unix').write_text('header\n0000: 00000002 00000000 00010000 0001 01 100 @/tmp/.X11-unix/X2\n') fds=folder/'41/fd';fds.mkdir(exist_ok=True) if not (fds/'7').is_symlink(): (fds/'7').symlink_to('socket:[100]') with patch.dict(os.environ,{'DISPLAY':':2','WINEPREFIX':str(self.prefix)}),patch.object(probe.os,'getppid',return_value=50): return probe.binding(':0',self.prefix,read=table.__getitem__,proc_root=folder) def test_binding_exact_nested_server_and_portal_refusal(self): self.assertEqual(self.binding_fixture(),self.bound) with self.assertRaisesRegex(ValueError,'portal'):self.binding_fixture(lambda t:t[41]['args'].append('-enable-ei-portal')) with self.assertRaisesRegex(ValueError,'exactly one'):self.binding_fixture(lambda t:t[41].update(ppid=1)) with self.assertRaisesRegex(ValueError,'exactly one'):self.binding_fixture(lambda t:t.update({42:proc(42,40,'/usr/bin/Xwayland',['Xwayland',':2'])})) def test_failed_binding_reports_sanitized_names_and_ancestry_without_relaxing(self): def change(table): table[41]['exe']='/private/runtime/Xwayland-versioned' table[41]['args']=['Xwayland-versioned',':2','-auth','/private/secret-cookie','--unknown-secret'] with self.assertRaises(ValueError) as raised: self.binding_fixture(change) text=str(raised.exception) details=json.loads(text.split('preflight=',1)[1]) self.assertFalse(details['display_opened']);self.assertFalse(details['input_sent']) self.assertEqual(details['gamescope']['pid'],40) item=details['same_user_xwayland_candidates'][0] self.assertEqual(item['executable_basename'],'Xwayland-versioned') self.assertTrue(item['descendant']);self.assertFalse(item['exact_executable_name']) self.assertEqual(item['display_args'],[':2']) self.assertNotIn('/private',text);self.assertNotIn('secret',text) with self.assertRaises(ValueError) as raised: self.binding_fixture(lambda t:t[41].update(ppid=1)) self.assertFalse(json.loads(str(raised.exception).split('preflight=',1)[1])['same_user_xwayland_candidates'][0]['descendant']) def test_failed_binding_diagnostic_candidates_are_bounded(self): def change(table): table[41]['ppid']=1 for pid in range(100,140): table[pid]=proc(pid,1,'/usr/bin/Xwayland',['Xwayland',':3','-auth','private']) with self.assertRaises(ValueError) as raised:self.binding_fixture(change) details=json.loads(str(raised.exception).split('preflight=',1)[1]) self.assertEqual(details['candidate_count'],41) self.assertEqual(len(details['same_user_xwayland_candidates']),16) self.assertNotIn('private',str(raised.exception)) def test_reparented_candidate_socket_nonce_diagnostics_require_every_check(self): self.binding_fixture() folder=self.root/'proc';(folder/'40/fd').mkdir();(folder/'40/fd/7').symlink_to('socket:[100]') scope=proc(40,1,'/usr/bin/gamescope',['gamescope','--backend','headless','--']) candidate=proc(41,1,'/usr/bin/Xwayland',['Xwayland',':2'],11) records={40:scope,41:candidate} nonce='application.id=org.escapistsrereintroduced.game.'+'a'*32 env={'WINEPREFIX':str(self.prefix),'PULSE_PROP':nonce} with patch.dict(os.environ,{'PULSE_PROP':nonce}): report=probe.candidate_relationship(candidate,scope,2,self.prefix,records.__getitem__,lambda _:env,folder) for key in ('start_not_older_than_gamescope','wrapper_run_nonce_valid','xwayland_prefix_matches','gamescope_prefix_matches', 'xwayland_run_nonce_matches','gamescope_run_nonce_matches','same_listener_inodes','process_identities_stable'): self.assertTrue(report[key],key) self.assertEqual(report['listener_inodes'],[100]) self.assertNotIn(nonce,json.dumps(report));self.assertNotIn(str(self.prefix),json.dumps(report)) (folder/'40/fd/7').unlink() with patch.dict(os.environ,{'PULSE_PROP':'wrong'}): report=probe.candidate_relationship(candidate,scope,2,self.prefix,records.__getitem__,lambda _:env,folder) self.assertFalse(report['same_listener_inodes']);self.assertFalse(report['wrapper_run_nonce_valid']) # Missing Gamescope socket/nonce evidence still cannot substitute for ancestry. with self.assertRaisesRegex(ValueError,'exactly one'): self.binding_fixture(lambda t:t[41].update(ppid=1)) def test_shared_relationship_gate_rejects_every_missing_false_nonboolean_and_inode_fault(self): good={key:True for key in probe.SHARED_CHECKS};good['listener_inodes']=[100,101] self.assertTrue(probe.shared_relationship_valid(good)) for key in probe.SHARED_CHECKS: for value in (False,None,1): bad={**good,key:value};self.assertFalse(probe.shared_relationship_valid(bad),(key,value)) bad=dict(good);del bad[key];self.assertFalse(probe.shared_relationship_valid(bad),key) for inodes in (None,[],[True],[0],[-1],[100,100],[1,2,3]): self.assertFalse(probe.shared_relationship_valid({**good,'listener_inodes':inodes})) def test_actual_reparented_binding_requires_shared_fds_environment_and_stable_checks(self): self.binding_fixture();folder=self.root/'proc';(folder/'40/fd').mkdir();(folder/'40/fd/7').symlink_to('socket:[100]') table={50:proc(50,40),40:proc(40,1,'/usr/bin/gamescope',['gamescope','--backend','headless','--']), 41:proc(41,1,'/usr/bin/Xwayland',['Xwayland',':2'],11)} nonce='application.id=org.escapistsrereintroduced.game.'+'b'*32 env={'WINEPREFIX':str(self.prefix),'PULSE_PROP':nonce} def bind(env_read=lambda _:env,read=table.__getitem__): with patch.dict(os.environ,{'DISPLAY':':2','WINEPREFIX':str(self.prefix),'PULSE_PROP':nonce}),patch.object(probe.os,'getppid',return_value=50): return probe.binding(':0',self.prefix,read=read,env_read=env_read,proc_root=folder) result=bind();self.assertEqual(result['relationship']['mode'],'shared-listeners') self.assertTrue(probe.shared_relationship_valid(result['relationship']['checks'])) for pid in (40,41): for key,value in [('WINEPREFIX','/foreign'),('PULSE_PROP',nonce+'mismatch')]: with self.assertRaises(ValueError):bind(lambda current: {**env,key:value} if current==pid else env) table[41]['start']=9 with self.assertRaises(ValueError):bind() table[41]['start']=11 table[41]['args'].append('-enable-ei-portal') with self.assertRaisesRegex(ValueError,'portal'):bind() table[41]['args'].pop() (folder/'40/fd/7').unlink() with self.assertRaisesRegex(ValueError,'exactly one'):bind() def test_shared_binding_refuses_changed_listener_evidence_before_connection(self): self.binding_fixture();folder=self.root/'proc' table={50:proc(50,40),40:proc(40,1,'/usr/bin/gamescope',['gamescope','--backend','headless','--']), 41:proc(41,1,'/usr/bin/Xwayland',['Xwayland',':2'],11)} good={key:True for key in probe.SHARED_CHECKS};good['listener_inodes']=[100] changed={**good,'listener_inodes':[101]} with patch.dict(os.environ,{'DISPLAY':':2','WINEPREFIX':str(self.prefix)}),patch.object(probe.os,'getppid',return_value=50),\ patch.object(probe,'candidate_relationship',side_effect=[good,changed]): with self.assertRaisesRegex(ValueError,'changed during binding'): probe.binding(':0',self.prefix,read=table.__getitem__,proc_root=folder) def test_relationship_unreadable_environment_and_reused_pid_fail_diagnostics(self): self.binding_fixture();folder=self.root/'proc' scope=proc(40);candidate=proc(41,start=1) def inaccessible(pid):raise PermissionError('sensitive-path-must-not-be-emitted') result=probe.candidate_relationship(candidate,scope,2,self.prefix,lambda pid:proc(pid,start=99),inaccessible,folder) self.assertIsNone(result['xwayland_prefix_matches']) self.assertIsNone(result['gamescope_run_nonce_matches']) self.assertFalse(result['process_identities_stable']);self.assertFalse(result['start_not_older_than_gamescope']) self.assertNotIn('sensitive',json.dumps(result)) def test_display_socket_ownership_prevents_accidental_host_query(self): self.binding_fixture() folder=self.root/'proc' self.assertEqual(probe.owned_sockets(2,41,folder),self.bound['sockets']) with (folder/'net/unix').open('a') as out: out.write('0000: 00000002 00000000 00010000 0001 01 101 /tmp/.X11-unix/X2\n') with self.assertRaisesRegex(ValueError,'not held'):probe.owned_sockets(2,41,folder) (folder/'41/fd/8').symlink_to('socket:[101]') self.assertEqual(len(probe.owned_sockets(2,41,folder)),2) (folder/'net/unix').write_text('header\n') with self.assertRaisesRegex(ValueError,'No unique'):probe.owned_sockets(2,41,folder) def test_exact_game_window_requires_server_pid_and_native_process_metadata(self): window={'window':17,'pid':90,'server_pid':90,'title':'Arma Reforger'} p=proc(90,args=[str(self.exe)]) env={'WINEPREFIX':str(self.prefix),'DISPLAY':':2'} def owns():return probe.owned_window(window,self.bound,self.prefix,self.exe,lambda _:p,lambda _:env) self.assertIsNotNone(owns()) for key,value in [('server_pid',91),('pid',None)]: old=window[key];window[key]=value;self.assertIsNone(owns());window[key]=old env['WINEPREFIX']='/foreign';self.assertIsNone(owns());env['WINEPREFIX']=str(self.prefix) env['DISPLAY']=':0';self.assertIsNone(owns());env['DISPLAY']=':2' p['args']=['ArmaReforgerSteam.exe'];self.assertIsNone(owns()) p['args']=['/usr/bin/foreign',str(self.exe)];self.assertIsNone(owns()) p['args']=['Z:'+str(self.exe).replace('/','\\')];self.assertIsNotNone(owns()) with patch.object(probe,'process'): self.assertIsNone(probe.owned_window(window,self.bound,self.prefix,self.exe,Mock(side_effect=[p,{**p,'start':99}]),lambda _:env)) def test_observe_closes_native_connection_and_rejects_process_reuse(self): native=Mock();native.windows.return_value=[] with patch.object(probe,'binding',side_effect=[self.bound,{**self.bound,'xwayland':{**self.bound['xwayland'],'start':999}}]): with self.assertRaisesRegex(ValueError,'changed'):probe.observe(self.args,lambda _:native) native.close.assert_called_once() native=Mock();native.windows.side_effect=ValueError('query failed') with patch.object(probe,'binding',return_value=self.bound): with self.assertRaises(ValueError):probe.observe(self.args,lambda _:native) native.close.assert_called_once() def test_new_direct_profile_report_required_before_child(self): probe.validate_paths(self.args) for report in (self.root/'elsewhere.json',self.profile/'sub'/'out.json',self.profile/'..'/'out.json'): self.args.report=report with self.assertRaises(ValueError):probe.validate_paths(self.args) self.args.report=self.report;self.report.write_text('preserve historical evidence') with patch.object(probe.subprocess,'Popen') as popen,patch.object(probe,'binding',return_value=self.bound): with self.assertRaises(ValueError):probe.main(self.cli()) popen.assert_not_called() self.assertEqual(self.report.read_text(),'preserve historical evidence') self.report.unlink();self.report.symlink_to(self.root/'absent') with self.assertRaises(ValueError):probe.validate_paths(self.args) def test_marker_alias_nonfinite_and_observation_budget(self): for seconds in (0,31,float('nan'),float('inf')): self.args.observe_seconds=seconds with self.assertRaises(ValueError):probe.validate_paths(self.args) self.args.observe_seconds=20 marker=self.profile/'.erri-game-profile.json';marker.write_text('{}') with self.assertRaises(ValueError):probe.validate_paths(self.args) def test_actual_xres_status_zero_is_success_and_pid_mask_required(self): native=object.__new__(probe.X11);native.d=123 value=probe.X11.Client(probe.X11.Spec(17,2),4,None) keep=(probe.X11.Client*1)(value) def query(display,n,spec,count,out): C.cast(count,C.POINTER(C.c_long))[0]=1 C.cast(out,C.POINTER(C.POINTER(probe.X11.Client)))[0]=C.cast(keep,C.POINTER(probe.X11.Client)) return 0 native.r=SimpleNamespace(XResQueryClientIds=query,XResGetClientPid=Mock(return_value=90),XResClientIdsDestroy=Mock()) self.assertEqual(native.server_pid(17),90) native.r.XResClientIdsDestroy.assert_called_once() keep[0].spec.mask=1;self.assertIsNone(native.server_pid(17)) native.r.XResQueryClientIds=lambda *args:1;self.assertIsNone(native.server_pid(17)) def test_read_only_wrapper_records_binding_preserves_child_exit(self): child=Mock();child.poll.return_value=None;child.wait.return_value=17 observed={'binding':self.bound,'owned_windows':[{'window':17,'pid':90,'server_pid':90}], 'owned_game_window_verified':True,'unverified_window_count':0,'window_count':1} worker=SimpleNamespace(returncode=0,stdout=json.dumps(observed)) with patch.object(probe,'binding',return_value=self.bound),patch.object(probe,'process',return_value=proc(55)),\ patch.object(probe.subprocess,'Popen',return_value=child) as spawn,patch.object(probe.subprocess,'run',return_value=worker) as query,\ patch.object(probe.signal,'signal'),patch.object(probe.time,'sleep'): self.assertEqual(probe.main(self.cli()),17) self.assertEqual(spawn.call_args.args[0],[str(self.proton),'run',str(self.exe),'-noBackend']) self.assertEqual(spawn.call_args.kwargs,{}) # inherit parent's owned process group self.assertLessEqual(query.call_args.kwargs['timeout'],3) report=json.loads(self.report.read_text()) self.assertTrue(report['owned_game_window_verified']);self.assertTrue(report['read_only']) self.assertFalse(report['input_sent']);self.assertFalse(report['release_qualified']) self.assertEqual(report['command'],spawn.call_args.args[0]);self.assertEqual(len(report['helper_sha256']),64) self.assertEqual(self.report.stat().st_mode&0o777,0o600) self.assertEqual(report['first_window_count'],1);self.assertEqual(report['last_window_count'],1) self.assertEqual(report['successful_queries'],1);self.assertEqual(report['last_query_status'],'ok') child.terminate.assert_not_called();child.kill.assert_not_called() def test_query_failure_does_not_become_ownership_or_change_child_exit(self): child=Mock();child.poll.side_effect=[None,0];child.wait.return_value=0 with patch.object(probe,'binding',return_value=self.bound),patch.object(probe,'process',return_value=proc(55)),\ patch.object(probe.subprocess,'Popen',return_value=child),patch.object(probe.subprocess,'run',return_value=SimpleNamespace(returncode=2)),\ patch.object(probe.signal,'signal'),patch.object(probe.time,'sleep'): self.assertEqual(probe.main(self.cli()),0) self.assertFalse(json.loads(self.report.read_text())['owned_game_window_verified']) def test_timeout_and_signal_forwarding_leave_native_lifetime_with_parent(self): child=Mock();child.poll.return_value=None;child.wait.return_value=23 handlers={} def handle(sig,callback): old=handlers.get(sig,signal.SIG_DFL);handlers[sig]=callback;return old def timed_out(*args,**kwargs): handlers[signal.SIGTERM](signal.SIGTERM,None) raise probe.subprocess.TimeoutExpired('read-only observer',3) with patch.object(probe,'binding',return_value=self.bound),patch.object(probe,'process',return_value=proc(55)),\ patch.object(probe.subprocess,'Popen',return_value=child),patch.object(probe.subprocess,'run',side_effect=timed_out),\ patch.object(probe.signal,'signal',side_effect=handle),patch.object(probe.time,'sleep'): self.assertEqual(probe.main(self.cli()),23) child.send_signal.assert_called_once_with(signal.SIGTERM) self.assertFalse(json.loads(self.report.read_text())['owned_game_window_verified']) self.assertEqual(handlers[signal.SIGTERM],signal.SIG_DFL) child.kill.assert_not_called() def test_foreign_or_malformed_command_refused_without_child_or_report(self): original=self.cli() for command in ([str(self.proton),'other',str(self.exe)],['/bin/echo',str(self.exe)],[str(self.proton),'run','foreign.exe']): args=original[:original.index('--')]+['--']+command with patch.object(probe,'binding',return_value=self.bound),patch.object(probe.subprocess,'Popen') as spawn: with self.assertRaises(ValueError):probe.main(args) spawn.assert_not_called();self.assertFalse(self.report.exists()) def test_window_diagnostics_pin_rejection_without_titles_or_private_values(self): secret_title='private unrelated window title' window={'window':17,'pid':90,'server_pid':91,'title':secret_title,'xres_query':{'status':0,'count':1,'mask':2}} records={90:proc(90,args=['/private/other.exe']),91:proc(91,args=[str(self.exe)])} environment={'WINEPREFIX':str(self.prefix),'DISPLAY':':2','SECRET':'never emit'} report=probe.window_diagnostic(window,self.bound,self.prefix,self.exe,records.__getitem__,lambda _:environment) self.assertEqual(report['rejection_stage'],'property-server-pid-disagreement') self.assertFalse(report['property_process']['game_executable_match']) self.assertTrue(report['server_process']['game_executable_match']) text=json.dumps(report) for secret in (secret_title,str(self.exe),str(self.prefix),'/private','never emit'): self.assertNotIn(secret,text) window['pid']=91 self.assertEqual(probe.window_diagnostic(window,self.bound,self.prefix,self.exe,records.__getitem__,lambda _:environment)['rejection_stage'],'checks-passed') records[91]['args']=['z:'+str(self.exe).replace('/','\\')] report=probe.window_diagnostic(window,self.bound,self.prefix,self.exe,records.__getitem__,lambda _:environment) self.assertEqual(report['rejection_stage'],'game-executable-mismatch') self.assertEqual(report['server_process']['argv0_class'],'game-basename-noncanonical-path') self.assertTrue(report['server_process']['argv0_case_insensitive_exact']) self.assertIsNone(probe.owned_window(window,self.bound,self.prefix,self.exe,records.__getitem__,lambda _:environment)) def test_window_diagnostics_unavailable_pid_environment_and_mutation(self): window={'window':17,'pid':90,'server_pid':None} def unreadable(_):raise PermissionError('private error detail') report=probe.window_diagnostic(window,self.bound,self.prefix,self.exe,unreadable,unreadable) self.assertEqual(report['rejection_stage'],'server-pid-unavailable') self.assertFalse(report['property_process']['readable']) self.assertNotIn('private error',json.dumps(report)) window['server_pid']=90 p=proc(90,args=[str(self.exe)]) env={'WINEPREFIX':str(self.prefix),'DISPLAY':':2'} for key,value,stage in [('WINEPREFIX','/foreign','private-prefix-unverified'),('DISPLAY',':0','private-display-unverified')]: report=probe.window_diagnostic(window,self.bound,self.prefix,self.exe,lambda _:p,lambda _:{**env,key:value}) self.assertEqual(report['rejection_stage'],stage) changed=Mock(side_effect=[p,{**p,'start':99}]) report=probe.window_diagnostic(window,self.bound,self.prefix,self.exe,changed,lambda _:env) self.assertEqual(report['rejection_stage'],'native-process-changed') def test_observe_diagnostics_bounded_and_no_unverified_titles_serialized(self): native=Mock();native.windows.return_value=[{'window':i,'pid':90,'server_pid':91,'title':'never serialize'} for i in range(20)] safe={'rejection_stage':'property-server-pid-disagreement'} with patch.object(probe,'binding',return_value=self.bound),patch.object(probe,'owned_window',return_value=None),\ patch.object(probe,'window_diagnostic',return_value=safe) as diagnostic: report=probe.observe(self.args,lambda _:native) self.assertEqual(diagnostic.call_count,16);self.assertEqual(report['window_count'],20) self.assertFalse(report['owned_game_window_verified']);self.assertNotIn('never serialize',json.dumps(report)) def test_prefix_equivalence_is_only_exact_spelling_plus_trailing_slashes(self): alias=self.root/'alias';alias.symlink_to(self.prefix,target_is_directory=True) for suffix in ('','/','////'): self.assertTrue(probe.prefix_matches(str(self.prefix)+suffix,self.prefix)) for value in (None,b'',str(self.prefix)+'-other/',str(self.prefix)+'/./',str(self.prefix)+'/../pfx/', str(alias)+'/',str(self.prefix).replace('/pfx','//pfx'),str(self.prefix)+'\\',str(self.prefix)+' /','pfx/'): with self.subTest(value=value):self.assertFalse(probe.prefix_matches(value,self.prefix)) def test_actual_window_and_relationship_accept_proton_trailing_slash_only(self): window={'window':17,'pid':90,'server_pid':90} p=proc(90,args=[str(self.exe)]) for suffix in ('','/','///'): env={'WINEPREFIX':str(self.prefix)+suffix,'DISPLAY':':2'} self.assertIsNotNone(probe.owned_window(window,self.bound,self.prefix,self.exe,lambda _:p,lambda _:env)) self.assertEqual(probe.window_diagnostic(window,self.bound,self.prefix,self.exe,lambda _:p,lambda _:env)['rejection_stage'],'checks-passed') self.binding_fixture();folder=self.root/'proc' scope=proc(40);candidate=proc(41,start=11) nonce='application.id=org.escapistsrereintroduced.game.'+'c'*32 records={40:scope,41:candidate} for value,expected in ((str(self.prefix)+'///',True),(str(self.prefix)+'-other/',False),(str(self.prefix)+'/../pfx/',False)): env={'WINEPREFIX':value,'PULSE_PROP':nonce} with patch.dict(os.environ,{'PULSE_PROP':nonce}): result=probe.candidate_relationship(candidate,scope,2,self.prefix,records.__getitem__,lambda _:env,folder) self.assertIs(result['xwayland_prefix_matches'],expected) self.assertIs(result['gamescope_prefix_matches'],expected) def test_native_xwindow_attributes_abi_and_signatures_match_installed_headers(self): compiler=shutil.which('cc') self.assertIsNotNone(compiler,'Installed C compiler required for native ABI check') fields=[name for name,_ in probe.WindowAttributes._fields_] lines=['#include ','#include ','#include ', '_Static_assert(__builtin_types_compatible_p(__typeof__(&XGetWindowAttributes), Status (*)(Display*,Window,XWindowAttributes*)), "attributes signature");', '_Static_assert(__builtin_types_compatible_p(__typeof__(&XGetInputFocus), int (*)(Display*,Window*,int*)), "focus signature");', '_Static_assert(__builtin_types_compatible_p(__typeof__(&XTranslateCoordinates), Bool (*)(Display*,Window,Window,int,int,int*,int*,Window*)), "translation signature");', 'int main(void) { printf("%zu\\n", sizeof(XWindowAttributes));'] for field in fields: native='class' if field=='window_class' else field lines.append('printf("%zu\\n", offsetof(XWindowAttributes, '+native+'));') lines.append('return 0;}') source=self.root/'abi.c';binary=self.root/'abi' source.write_text('\n'.join(lines)) subprocess.run([compiler,'-std=gnu11','-Werror',str(source),'-o',str(binary)],check=True,capture_output=True,timeout=15) # This tiny binary only prints sizeof/offsetof; it does not link X11 or open a display. values=[int(value) for value in subprocess.check_output([str(binary)],text=True,timeout=3).splitlines()] self.assertEqual(values[0],C.sizeof(probe.WindowAttributes)) self.assertEqual(values[1:],[getattr(probe.WindowAttributes,field).offset for field in fields]) def test_ctypes_geometry_focus_adapter_signatures_and_values_without_connection(self): x=Mock();r=Mock();x.XOpenDisplay.return_value=123 with patch.object(probe.C,'CDLL',side_effect=[x,r]):native=probe.X11(':2') self.assertEqual(x.XGetWindowAttributes.argtypes,[C.c_void_p,C.c_ulong,C.POINTER(probe.WindowAttributes)]) self.assertEqual(x.XGetInputFocus.argtypes,[C.c_void_p,C.POINTER(C.c_ulong),C.POINTER(C.c_int)]) def attrs(display,window,out): value=C.cast(out,C.POINTER(probe.WindowAttributes)).contents value.root=2;value.window_class=1;value.map_state=2;value.width=1280;value.height=720 return 1 def focus(display,out,revert): C.cast(out,C.POINTER(C.c_ulong))[0]=20;C.cast(revert,C.POINTER(C.c_int))[0]=2;return 1 def translate(display,window,root,sx,sy,dx,dy,child): C.cast(dx,C.POINTER(C.c_int))[0]=-20;C.cast(dy,C.POINTER(C.c_int))[0]=30;return 1 x.XGetWindowAttributes.side_effect=attrs;x.XGetInputFocus.side_effect=focus;x.XTranslateCoordinates.side_effect=translate self.assertEqual(native.attributes(20)['width'],1280);self.assertEqual(native.focus(),(20,2)) self.assertEqual(native.translate(20,2),(-20,30)) x.XGetWindowAttributes.side_effect=None;x.XGetWindowAttributes.return_value=0 with self.assertRaises(ValueError):native.attributes(20) native.query_budget=0 with self.assertRaisesRegex(ValueError,'budget'):native.focus() native.close();x.XCloseDisplay.assert_called_once_with(123) def test_exact_mapped_focus_and_owned_descendant_qualify_read_only(self): query=FakeWindowQuery();owner={'window':20,'pid':90} direct=probe.input_target(query,owner) self.assertTrue(direct['verified']);self.assertEqual(direct['focus_chain'],[20]) self.assertEqual(direct['geometry']['clipped_bounds'],[20,30,820,630]) query.focused=(21,2) child=probe.input_target(query,owner) self.assertTrue(child['verified']);self.assertEqual(child['focus_chain'],[21,20]) self.assertEqual(child['focused_clipped_bounds'],[40,50,240,150]) query.positions[20]=(-10,-20) self.assertTrue(probe.input_target(query,owner)['verified']) self.assertEqual(probe.input_target(query,owner)['geometry']['clipped_bounds'],[0,0,790,580]) def test_unmapped_input_only_invalid_root_geometry_and_foreign_focus_refused(self): for window,key,value in ((20,'map_state',0),(20,'window_class',2),(20,'width',0),(20,'height',65536), (20,'root',99),(20,'width',True),(2,'width',0),(2,'window_class',2)): query=FakeWindowQuery();query.attrs[window][key]=value with self.subTest(window=window,key=key,value=value):self.assertFalse(probe.input_target(query,{'window':20,'pid':90})['verified']) for focused in ((0,2),(1,2),(True,2),(20,9),(20,True)): query=FakeWindowQuery();query.focused=focused self.assertFalse(probe.input_target(query,{'window':20,'pid':90})['verified']) query=FakeWindowQuery();query.positions[20]=(2000,0) self.assertFalse(probe.input_target(query,{'window':20,'pid':90})['verified']) query=FakeWindowQuery();query.focused=(21,2);query.pids[21]=99 result=probe.input_target(query,{'window':20,'pid':90}) self.assertFalse(result['verified']);self.assertEqual(result['reason'],'focus-client-not-owned-game') query.pids[21]=90;query.parents[21]=2 self.assertFalse(probe.input_target(query,{'window':20,'pid':90})['verified']) def test_child_clipping_cycle_transient_focus_or_geometry_refused(self): owner={'window':20,'pid':90} query=FakeWindowQuery();query.focused=(21,2);query.positions[21]=(900,50) self.assertFalse(probe.input_target(query,owner)['verified']) query=FakeWindowQuery();query.focused=(21,2);query.parents[21]=21 self.assertFalse(probe.input_target(query,owner)['verified']) query=FakeWindowQuery();query.focus=Mock(side_effect=[(20,2),(21,2)]) self.assertFalse(probe.input_target(query,owner)['verified']) query=FakeWindowQuery();query.translate=Mock(side_effect=[(20,30),(20,30),(21,30)]) self.assertFalse(probe.input_target(query,owner)['verified']) query=FakeWindowQuery();query.attributes=Mock(side_effect=ValueError('native-window-attributes-unavailable')) self.assertFalse(probe.input_target(query,owner)['verified']) def test_focus_ancestry_has_fixed_depth_bound_and_stale_process_is_not_target(self): query=FakeWindowQuery();query.focused=(100,2) for window in range(100,133): query.attrs[window]=query.attribute(100,100);query.positions[window]=(40,50) query.parents[window]=window+1;query.pids[window]=90 query.parents[132]=20 result=probe.input_target(query,{'window':20,'pid':90}) self.assertFalse(result['verified']);self.assertEqual(result['reason'],'focus-ancestry-budget-exceeded') owner={'window':20,'pid':90,'server_pid':90,'process':{'pid':90,'start':100,'uid':os.getuid()}} query=FakeWindowQuery();query.windows=lambda:[owner];query.close=Mock() with patch.object(probe,'binding',return_value=self.bound),patch.object(probe,'owned_window',side_effect=[owner,None]),\ patch.object(probe,'window_diagnostic',return_value={}): result=probe.observe(self.args,lambda _:query) self.assertTrue(result['owned_game_window_verified']);self.assertFalse(result['input_target_verified']) self.assertEqual(result['input_target_observations'][0]['reason'],'game-process-changed-during-target-observation') def test_observe_preserves_ownership_when_input_target_is_unverified(self): window={'window':20,'pid':90,'server_pid':90,'process':{'pid':90,'start':100,'uid':os.getuid()}} query=FakeWindowQuery();query.windows=lambda:[window];query.close=Mock();query.focused=(0,0) with patch.object(probe,'binding',return_value=self.bound),patch.object(probe,'owned_window',return_value=window),\ patch.object(probe,'window_diagnostic',return_value={}): result=probe.observe(self.args,lambda _:query) self.assertTrue(result['owned_game_window_verified']);self.assertFalse(result['input_target_verified']) self.assertIsNone(result['input_target']);query.close.assert_called_once() query.focused=(20,2) with patch.object(probe,'binding',return_value=self.bound),patch.object(probe,'owned_window',return_value=window),\ patch.object(probe,'window_diagnostic',return_value={}): result=probe.observe(self.args,lambda _:query) self.assertTrue(result['input_target_verified']);self.assertEqual(result['input_target']['window'],20) def test_no_native_input_or_focus_functions_are_bound(self): source=Path(probe.__file__).read_text() for forbidden in ('XTestFake','XSendEvent','XSetInputFocus','XWarpPointer','XRaiseWindow','XGrabPointer','XGrabKeyboard'): self.assertNotIn(forbidden,source) self.assertIn('XOpenDisplay(display.encode',source) if __name__=='__main__':unittest.main()