485 lines
36 KiB
Python
485 lines
36 KiB
Python
#!/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 <X11/Xlib.h>','#include <stddef.h>','#include <stdio.h>',
|
|
'_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()
|