192 lines
14 KiB
Python
192 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Runner integration policy tests: fake native processes and no display connections."""
|
|
import contextlib
|
|
import copy
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
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_game as runner
|
|
import headless_display_probe as helper
|
|
|
|
DEFAULT_REPORT = object()
|
|
|
|
|
|
class DisplayRunnerTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory(prefix='erri-display-runner-test-')
|
|
self.base=Path(self.tmp.name);self.root=self.base/'root';(self.root/'tools').mkdir(parents=True)
|
|
(self.root/'tools/headless_display_probe.py').write_text('synthetic helper bytes for provenance test')
|
|
self.compat=self.base/'compat';self.profile=self.base/'profile';self.profile.mkdir()
|
|
self.steam=self.base/'Steam';self.exe=self.steam/'steamapps/common/Arma Reforger/ArmaReforgerSteam.exe'
|
|
self.proton=self.steam/'steamapps/common/Proton - Experimental/proton'
|
|
for path in (self.exe,self.proton):path.parent.mkdir(parents=True,exist_ok=True);path.write_bytes(b'synthetic file, never launched')
|
|
self.command=['gamescope','--backend','headless','-W','1280','--',str(self.proton),'run',str(self.exe),'-noBackend']
|
|
self.wrapped,self.context=runner.display_probe_command(self.command,self.root,self.compat,self.profile,self.exe,'fixed-run-id',':0')
|
|
self.session=self.base/'session';self.session.mkdir()
|
|
def ident(pid):return {'pid':pid,'start':100,'uid':os.getuid()}
|
|
self.report={**{k:self.context[k] for k in ('helper_sha256','prefix','profile','game_executable','report','command')},
|
|
'schema':1,'kind':'headless-display-read-only','owned_game_window_verified':True,
|
|
'read_only':True,'input_sent':False,'observation_finished':True,'wrapper':ident(10),
|
|
'binding':{'display':':2','gamescope':ident(20),'xwayland':ident(30),
|
|
'sockets':[{'path':'@/tmp/.X11-unix/X2','inode':1000}],
|
|
'relationship':{'mode':'descendant','descendant':True}},
|
|
'owned_windows':[{'window':12345,'pid':40,'server_pid':40,'process':ident(40),'title':'fixture game'}]}
|
|
def tearDown(self):self.tmp.cleanup()
|
|
def save(self,value=DEFAULT_REPORT):
|
|
raw=(json.dumps(self.report if value is DEFAULT_REPORT else value,indent=3)+'\n').encode()
|
|
Path(self.context['report']).write_bytes(raw);return raw
|
|
def test_exact_gamescope_boundary_and_unmodified_proton_child(self):
|
|
separator=self.command.index('--')
|
|
self.assertEqual(self.wrapped[:separator+1],self.command[:separator+1])
|
|
self.assertEqual(self.wrapped[-len(self.context['command']):],self.command[separator+1:])
|
|
self.assertEqual(self.wrapped[separator+1],sys.executable)
|
|
self.assertEqual(self.wrapped[separator+2],str(self.root/'tools/headless_display_probe.py'))
|
|
for flag,value in (('--report',self.context['report']),('--host-display',':0'),('--prefix',str(self.compat/'pfx'))):
|
|
self.assertEqual(self.wrapped[self.wrapped.index(flag)+1],value)
|
|
self.assertEqual(self.command[separator+1],str(self.proton)) # caller array untouched
|
|
def test_default_command_has_no_wrapper_and_remains_ordinary(self):
|
|
command=runner.command_for(runner.ROOT,self.steam,self.compat,self.profile,'core')
|
|
separator=command.index('--')
|
|
self.assertEqual(command[separator+1:separator+4],[str(self.proton),'run',str(self.exe)])
|
|
self.assertNotIn('--host-display',command);self.assertNotIn('--query-worker',command)
|
|
self.assertIn('-noSound',command);self.assertIn('-noBackend',command)
|
|
def test_bad_boundary_missing_helper_and_existing_report_refused(self):
|
|
wrong=self.command.copy();wrong[2]='wayland'
|
|
with self.assertRaises(ValueError):runner.display_probe_command(wrong,self.root,self.compat,self.profile,self.exe,'fixed-run-id',':0')
|
|
self.save()
|
|
with self.assertRaises(ValueError):runner.display_probe_command(self.command,self.root,self.compat,self.profile,self.exe,'fixed-run-id',':0')
|
|
Path(self.context['report']).unlink()
|
|
helper=self.root/'tools/headless_display_probe.py';helper.unlink();helper.symlink_to(self.base/'absent')
|
|
with self.assertRaises(ValueError):runner.display_probe_command(self.command,self.root,self.compat,self.profile,self.exe,'fixed-run-id',':0')
|
|
def test_packed_and_vanilla_reject_before_resolution_or_launch(self):
|
|
for project,packed in [('core',{'session':self.base/'packed'}),(runner.CONTROL,None)]:
|
|
args=SimpleNamespace(project=project,inspect_headless_display=True)
|
|
with patch.object(runner,'packed_request',return_value=packed),patch.object(runner.subprocess,'Popen') as spawn,patch.object(runner,'assert_no_game') as process_check:
|
|
with self.assertRaisesRegex(ValueError,'loose Core or Chat'):runner.run(args)
|
|
spawn.assert_not_called();process_check.assert_not_called()
|
|
def test_exact_bytes_archived_with_hash_without_promoting_gameplay(self):
|
|
raw=self.save();result=runner.display_probe_result(self.context,self.session)
|
|
self.assertTrue(result['verified']);self.assertTrue(result['read_only']);self.assertFalse(result['input_sent'])
|
|
self.assertEqual((self.session/'display-ownership.json').read_bytes(),raw)
|
|
self.assertEqual(result['sha256'],hashlib.sha256(raw).hexdigest())
|
|
self.assertEqual(result['archive_file'],'display-ownership.json')
|
|
self.assertFalse(runner.display_probe_result(self.context,self.session)['verified']) # exclusive archive copy
|
|
self.assertEqual((self.session/'display-ownership.json').read_bytes(),raw)
|
|
def test_each_launch_binding_mismatch_fails(self):
|
|
for key in ('helper_sha256','prefix','profile','game_executable','report','command'):
|
|
altered=copy.deepcopy(self.report);altered[key]=['foreign'] if key=='command' else 'foreign'
|
|
self.save(altered)
|
|
with self.subTest(key=key):self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
def test_absent_malformed_unverified_and_input_claims_fail(self):
|
|
self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
for value in ([],None,{}, {**self.report,'owned_game_window_verified':False}, {**self.report,'read_only':False},
|
|
{**self.report,'input_sent':True}, {**self.report,'observation_finished':False}, {**self.report,'schema':2}):
|
|
self.save(value);self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
Path(self.context['report']).write_text('{truncated')
|
|
self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
Path(self.context['report']).unlink();Path(self.context['report']).symlink_to(self.base/'absent')
|
|
self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
def test_host_display_uid_pid_start_or_window_disagreement_fails(self):
|
|
faults=[]
|
|
for target in ('wrapper','gamescope','xwayland','window'):
|
|
for key,value in [('uid',os.getuid()+1),('pid',0),('start',0),('pid',True)]:
|
|
bad=copy.deepcopy(self.report)
|
|
item=(bad['wrapper'] if target=='wrapper' else bad['owned_windows'][0]['process'] if target=='window' else bad['binding'][target])
|
|
item[key]=value;faults.append(bad)
|
|
for display in (':0',':0.1','unix:0','remote:2',''):
|
|
bad=copy.deepcopy(self.report);bad['binding']['display']=display;faults.append(bad)
|
|
for key,value in [('pid',41),('server_pid',41),('window',0),('window',None),('window',True)]:
|
|
bad=copy.deepcopy(self.report);bad['owned_windows'][0][key]=value;faults.append(bad)
|
|
bad=copy.deepcopy(self.report);bad['owned_windows']=[];faults.append(bad)
|
|
for bad in faults:
|
|
self.save(bad)
|
|
with self.subTest(report=bad):self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
def test_missing_foreign_malformed_duplicate_socket_evidence_fails(self):
|
|
for sockets in (None,[],[{'path':'@/tmp/.X11-unix/X0','inode':1000}],
|
|
[{'path':'@/tmp/.X11-unix/X2','inode':0}], [{'path':'@/tmp/.X11-unix/X2','inode':True}],
|
|
[{'path':'@/tmp/.X11-unix/X2','inode':1000}]*2):
|
|
bad=copy.deepcopy(self.report);bad['binding']['sockets']=sockets;self.save(bad)
|
|
with self.subTest(sockets=sockets):self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
def test_missing_unknown_or_unproven_relationship_refused(self):
|
|
for relationship in (None,{}, {'mode':'unknown'}, {'mode':'descendant'},
|
|
{'mode':'descendant','descendant':False},{'mode':'descendant','descendant':1},
|
|
{'mode':'shared-listeners'}, {'mode':'shared-listeners','checks':{}},'descendant'):
|
|
bad=copy.deepcopy(self.report);bad['binding']['relationship']=relationship;self.save(bad)
|
|
with self.subTest(relationship=relationship):self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
bad=copy.deepcopy(self.report);del bad['binding']['relationship'];self.save(bad)
|
|
self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
def test_shared_relationship_requires_every_check_and_exact_socket_inodes(self):
|
|
checks={key:True for key in helper.SHARED_CHECKS};checks['listener_inodes']=[1000]
|
|
good=copy.deepcopy(self.report);good['binding']['relationship']={'mode':'shared-listeners','checks':checks}
|
|
self.save(good);self.assertTrue(runner.display_probe_result(self.context,None)['verified'])
|
|
for key in helper.SHARED_CHECKS:
|
|
for value in (False,None,1):
|
|
bad=copy.deepcopy(good);bad['binding']['relationship']['checks'][key]=value;self.save(bad)
|
|
with self.subTest(key=key,value=value):self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
bad=copy.deepcopy(good);del bad['binding']['relationship']['checks'][key];self.save(bad)
|
|
self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
for inodes in (None,[],[1001],[True],[1000,1000],[1000,1001]):
|
|
bad=copy.deepcopy(good);bad['binding']['relationship']['checks']['listener_inodes']=inodes;self.save(bad)
|
|
with self.subTest(inodes=inodes):self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
bad=copy.deepcopy(good);bad['binding']['xwayland']['start']=99;self.save(bad)
|
|
self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
good['binding']['sockets'].append({'path':'/tmp/.X11-unix/X2','inode':1001})
|
|
good['binding']['relationship']['checks']['listener_inodes']=[1000,1001]
|
|
self.save(good);self.assertTrue(runner.display_probe_result(self.context,None)['verified'])
|
|
def test_boolean_server_pid_is_not_an_integer_process_identity(self):
|
|
bad=copy.deepcopy(self.report);window=bad['owned_windows'][0]
|
|
window['pid']=1;window['process']['pid']=1;window['server_pid']=True
|
|
self.save(bad);self.assertFalse(runner.display_probe_result(self.context,None)['verified'])
|
|
def simulate_run_locked(self,display_ok,ordinary_failed=False):
|
|
(self.exe.parent/'addons/data').mkdir(parents=True,exist_ok=True)
|
|
args=SimpleNamespace(project='core',capture_prison_view=False,inspect_headless_display=True,timeout=120,settle=15,failure_grace=0)
|
|
session=self.profile/'logs/logs_fixture'
|
|
child=Mock();child.poll.return_value=None
|
|
def spawn(*args,**kwargs):
|
|
session.mkdir(parents=True);(session/'console.log').write_text('synthetic ordinary startup fixture')
|
|
return child
|
|
def output(args,**kwargs):return '[]' if '-f' in args else '11'
|
|
monitor=SimpleNamespace(deadline=time.monotonic()+60,started=0,first_failure_at=None,
|
|
observe=lambda *args:'game-started',annotate=lambda value:value)
|
|
evidence={'ready':not ordinary_failed,'failed':ordinary_failed,'scope':'synthetic test only'}
|
|
with contextlib.ExitStack() as stack:
|
|
for name in ('assert_no_game','assert_prefix_idle','persist','cleanup'):
|
|
stack.enter_context(patch.object(runner,name))
|
|
stack.enter_context(patch.object(runner,'snapshot',return_value={}))
|
|
stack.enter_context(patch.object(runner,'bindings',return_value={}))
|
|
stack.enter_context(patch.object(runner,'seal_run',return_value={'source_unchanged_after_run':True}))
|
|
stack.enter_context(patch.object(runner,'StartupMonitor',return_value=monitor))
|
|
stack.enter_context(patch.object(runner,'game_evidence',return_value=evidence))
|
|
stack.enter_context(patch.object(runner,'display_probe_result',return_value={'verified':display_ok}))
|
|
stack.enter_context(patch.object(runner.shutil,'which',return_value='/usr/bin/synthetic'))
|
|
stack.enter_context(patch.object(runner.subprocess,'Popen',side_effect=spawn))
|
|
stack.enter_context(patch.object(runner.subprocess,'check_output',side_effect=output))
|
|
stack.enter_context(patch.object(runner.subprocess,'run'))
|
|
stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
|
|
status=runner.run_locked(args,self.steam,self.compat,self.profile)
|
|
return status,json.loads((session/'runner.json').read_text())
|
|
def test_missing_display_proof_turns_actual_runner_path_into_failure(self):
|
|
status,report=self.simulate_run_locked(False)
|
|
self.assertEqual(status,1);self.assertEqual(report['result'],'display-probe-incomplete')
|
|
self.assertEqual(report['startup_result_before_display_check'],'game-started')
|
|
def test_valid_display_does_not_clear_native_failure(self):
|
|
status,report=self.simulate_run_locked(True,ordinary_failed=True)
|
|
self.assertEqual(status,1);self.assertEqual(report['result'],'game-failed')
|
|
def test_valid_ownership_preserves_clean_startup_only(self):
|
|
status,report=self.simulate_run_locked(True)
|
|
self.assertEqual(status,0);self.assertEqual(report['result'],'game-started')
|
|
self.assertTrue(report['display_ownership']['verified'])
|
|
self.assertNotIn('multiplayer_verified',report)
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|