#!/usr/bin/env python3 """Exercise process refusal, command scope and evidence gates without native launch.""" from pathlib import Path import sys import tempfile import unittest from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1]/'tools')) import headless_game as runner class GameRunnerTests(unittest.TestCase): def test_exact_native_smartcard_helper_does_not_own_a_wine_prefix(self): with tempfile.TemporaryDirectory() as directory: proc = self.process(Path(directory), b'scdaemon\0--multi-server\0', comm='scdaemon') self.assertFalse(runner.prefix_process_candidate(proc)) (proc/'cmdline').write_bytes(b'scdaemon\0--multi-server\0proton\0') self.assertTrue(runner.prefix_process_candidate(proc)) (proc/'cmdline').write_bytes(b'scdaemon\0') with self.assertRaisesRegex(ValueError, 'Cannot classify'): runner.prefix_process_candidate(proc) def test_only_confirmed_exited_zombie_can_have_empty_process_identity(self): with tempfile.TemporaryDirectory() as directory: proc = self.process(Path(directory), b'', comm='codex') (proc/'status').write_text('Name:\tcodex\nState:\tZ (zombie)\n') self.assertFalse(runner.prefix_process_candidate(proc)) for status in ('State:\tS (sleeping)\n', '', 'State:\tR (running)\n'): (proc/'status').write_text(status) with self.assertRaisesRegex(ValueError, 'Cannot classify'): runner.prefix_process_candidate(proc) def process(self, root, command, comm='wine', pid='123'): proc = root/pid proc.mkdir() (proc/'cmdline').write_bytes(command) (proc/'comm').write_text(comm) (proc/'environ').write_bytes(b'') return proc def test_refuses_existing_game_variants_without_mutation(self): for command in (b'Z:/Games/ArmaReforgerSteamDiag.exe\0-world\0x', b'C:\\Games\\ArmaReforgerSteam.exe\0', b'proton\0run\0/Z/ArmaReforgerSteamDiag.exe\0'): with self.subTest(command=command), tempfile.TemporaryDirectory() as directory: root = Path(directory) proc = self.process(root, command) with self.assertRaisesRegex(ValueError, 'game process already exists'): runner.assert_no_game(root) self.assertEqual((proc/'cmdline').read_bytes(), command) def test_truncated_comm_still_refuses_empty_cmdline(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) self.process(root, b'', comm='ArmaReforgerStea') with self.assertRaisesRegex(ValueError, 'game process already exists'): runner.assert_no_game(root) def test_unrelated_workbench_is_not_a_game_process(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) self.process(root, b'Z:/Workbench/ArmaReforgerWorkbenchSteamDiag.exe\0', comm='ArmaReforgerWor') runner.assert_no_game(root) def test_unreadable_process_fails_closed(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) self.process(root, b'') with patch.object(Path, 'read_bytes', side_effect=PermissionError('blocked')): with self.assertRaisesRegex(ValueError, 'Cannot exclude an existing game'): runner.assert_no_game(root) def test_same_prefix_process_refused_other_prefix_preserved(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) compat = root/'compat' proc = self.process(root, b'wineserver\0') (proc/'environ').write_bytes(('WINEPREFIX='+str(compat/'pfx')).encode()+b'\0') with self.assertRaisesRegex(ValueError, 'prefix already has a process'): runner.assert_prefix_idle(compat, root) runner.assert_prefix_idle(root/'different', root) def test_proton_trailing_prefix_slashes_refuse_existing_workbench_and_wine(self): for command in (b'wineserver\0', b'Z:/Workbench/ArmaReforgerWorkbenchSteamDiag.exe\0'): for suffix in ('','/','////'): with self.subTest(command=command,suffix=suffix),tempfile.TemporaryDirectory() as directory: root=Path(directory);compat=root/'compat';proc=self.process(root,command) data=('WINEPREFIX='+str(compat/'pfx')+suffix).encode()+b'\0OTHER=unchanged\0' (proc/'environ').write_bytes(data) with self.assertRaisesRegex(ValueError,'prefix already has a process'): runner.assert_prefix_idle(compat,root) self.assertEqual((proc/'environ').read_bytes(),data) def test_prefix_idle_does_not_resolve_neighbours_dot_components_or_symlink_spellings(self): with tempfile.TemporaryDirectory() as directory: root=Path(directory);compat=root/'compat';prefix=compat/'pfx';prefix.mkdir(parents=True) alias=root/'alias';alias.symlink_to(prefix,target_is_directory=True) proc=self.process(root,b'wineserver\0') for value in (str(prefix)+'-other/',str(prefix)+'/./',str(prefix)+'/../pfx/',str(alias)+'/', str(prefix).replace('/pfx','//pfx'),str(prefix)+'\\','pfx/',str(prefix)+' /'): data=('WINEPREFIX='+value).encode()+b'\0';(proc/'environ').write_bytes(data) with self.subTest(value=value):runner.assert_prefix_idle(compat,root) self.assertEqual((proc/'environ').read_bytes(),data) (proc/'environ').write_bytes(('OTHER_WINEPREFIX='+str(prefix)+'/').encode()+b'\0') runner.assert_prefix_idle(compat,root) def test_nondumpable_native_service_does_not_require_environment(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) proc = self.process(root, b'/usr/lib/systemd/systemd\0--user\0', comm='systemd') original = Path.read_bytes def protected_read(path): if path.name == 'environ': self.fail('Unrelated native service environment must not be read') return original(path) with patch.object(Path, 'read_bytes', protected_read), patch.object(runner.os, 'readlink', side_effect=PermissionError('non-dumpable')): runner.assert_prefix_idle(root/'compat', root) self.assertTrue(proc.exists()) def test_runtime_candidates_require_environment_even_if_exe_is_protected(self): for command, comm in [(b'/usr/bin/wineserver\0', 'wineserver'), (b'/usr/bin/python3\0/Steam/Proton - Experimental/proton\0run\0', 'python3'), (b'/usr/bin/gamescope\0--backend\0headless\0', 'gamescope'), (b'C:\\windows\\system32\\services.exe\0', 'services.exe')]: with self.subTest(comm=comm), tempfile.TemporaryDirectory() as directory: root = Path(directory); self.process(root, command, comm=comm) original = Path.read_bytes def protected_read(path): if path.name == 'environ': raise PermissionError('protected candidate') return original(path) with patch.object(Path, 'read_bytes', protected_read), patch.object(runner.os, 'readlink', side_effect=PermissionError('protected exe')): with self.assertRaisesRegex(ValueError, 'Cannot verify disposable prefix'): runner.assert_prefix_idle(root/'compat', root) def test_unclassifiable_process_remains_fail_closed(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory); self.process(root, b'', comm='unknown') with patch.object(runner.os, 'readlink', side_effect=PermissionError('protected exe')): with self.assertRaisesRegex(ValueError, 'Cannot classify possible Wine'): runner.assert_prefix_idle(root/'compat', root) def test_exact_systemd_pam_helper_is_native_without_environment(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory); self.process(root, b'(sd-pam)\0', comm='(sd-pam)') original = Path.read_bytes def protected_read(path): if path.name == 'environ': self.fail('Native PAM helper environment must not be read') return original(path) with patch.object(Path, 'read_bytes', protected_read), patch.object(runner.os, 'readlink', side_effect=PermissionError('non-dumpable')): runner.assert_prefix_idle(root/'compat', root) def test_observed_native_apache_signature_is_narrowly_classified(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) proc = self.process(root, b'apache2\0-DFOREGROUND\0', comm='apache2') with patch.object(runner.os, 'readlink', side_effect=PermissionError('protected exe')): self.assertFalse(runner.prefix_process_candidate(proc)) (proc/'cmdline').write_bytes(b'apache2\0-DFOREGROUND\0/Steam/proton\0') self.assertTrue(runner.prefix_process_candidate(proc)) (proc/'cmdline').write_bytes(b'apache2\0') with self.assertRaisesRegex(ValueError, 'Cannot classify possible Wine'): runner.prefix_process_candidate(proc) def test_bare_matching_names_are_not_generically_trusted(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) self.process(root, b'unknown\0', comm='unknown', pid='101') self.process(root, b'another\0', comm='another', pid='102') with patch.object(runner.os, 'readlink', side_effect=PermissionError('protected exe')): with self.assertRaises(ValueError) as raised: runner.assert_prefix_idle(root/'compat', root) self.assertIn('101', str(raised.exception)) self.assertIn('102', str(raised.exception)) # report all blockers in one pass def test_pam_name_does_not_hide_runtime_arguments(self): with tempfile.TemporaryDirectory() as directory: root = Path(directory) proc = self.process(root, b'(sd-pam)\0/Steam/proton\0', comm='(sd-pam)') with patch.object(runner.os, 'readlink', side_effect=PermissionError('protected exe')): self.assertTrue(runner.prefix_process_candidate(proc)) def test_readonly_inspection_does_not_enter_launcher_or_audio(self): with patch.object(sys, 'argv', ['headless_game.py', '--compat', '/unused-prefix', '--inspect-processes']): with patch.object(runner, 'assert_no_game'), patch.object(runner, 'assert_prefix_idle', side_effect=ValueError('ambiguous process')): with patch.object(runner, 'run') as launch, patch.object(runner.subprocess, 'Popen') as spawn, patch.object(runner.subprocess, 'run') as commands, patch('builtins.print'): self.assertEqual(runner.main(), 1) launch.assert_not_called(); spawn.assert_not_called(); commands.assert_not_called() def test_existing_unmarked_profile_and_steam_prefix_refused(self): with tempfile.TemporaryDirectory() as directory: base = Path(directory) compat = base/'compat'; (compat/'pfx').mkdir(parents=True) (compat/'pfx/system.reg').touch() profile = base/'profile'; profile.mkdir(); (profile/'sentinel').write_text('keep') with self.assertRaisesRegex(ValueError, 'not marked'): runner.validate_paths(base/'source', base/'steam', compat, profile) self.assertEqual((profile/'sentinel').read_text(), 'keep') with self.assertRaisesRegex(ValueError, 'outside Steam'): runner.validate_paths(base/'source', base, compat, base/'fresh') def test_profile_cannot_overlap_owned_prefix_or_source(self): with tempfile.TemporaryDirectory() as directory: base = Path(directory); compat = base/'compat' (compat/'pfx').mkdir(parents=True); (compat/'pfx/system.reg').touch() for profile in (base/'source/profile', compat/'profile', base): with self.assertRaisesRegex(ValueError, 'separate from source'): runner.validate_paths(base/'source', base/'steam', compat, profile) def test_command_is_ordinary_game_with_explicit_successor_identity(self): for project, guid in [('core', '2528466DE220D91C'), ('chat', '2528466DE220D91C')]: command = runner.command_for(runner.ROOT, Path('/Steam'), Path('/compat'), Path('/profile'), project) self.assertEqual(command[command.index('--')+3], '/Steam/steamapps/common/Arma Reforger/ArmaReforgerSteam.exe') self.assertFalse(any('ArmaReforger_BE.exe' in arg or 'ArmaReforgerSteamDiag.exe' in arg for arg in command)) self.assertEqual(command[command.index('-addons')+1], guid) self.assertNotIn('-world', command) self.assertEqual(command[command.index('-errOrdinaryScenario')+1], project) for flag in ['-noBackend', '-noSound', '-noSplash', '-noThrow', '-disableCrashReporter']: self.assertIn(flag, command) for unwanted in ['-server', '-gproj', '-errPlay', '-errSmoke', '-plugin', 'Workbench', '-silentCrashReport', '-noCrashDialog']: self.assertFalse(any(unwanted in arg for arg in command)) def test_prison_view_capture_is_explicit_and_excludes_controls_and_packed_inputs(self): args = (runner.ROOT, Path('/Steam'), Path('/compat'), Path('/profile')) for project in ('core', 'chat'): ordinary = runner.command_for(*args, project) self.assertNotIn('-errOrdinaryScreenshot', ordinary) captured = runner.command_for(*args, project, capture_prison_view=True) self.assertEqual(captured[captured.index('-errOrdinaryScreenshot')+1], '1') with self.assertRaisesRegex(ValueError, 'loose successor'): runner.command_for(*args, runner.CONTROL, capture_prison_view=True) with self.assertRaisesRegex(ValueError, 'loose successor'): runner.command_for(*args, 'core', packed={}, capture_prison_view=True) def test_ready_requires_game_module_without_workbench_and_normal_phase(self): module = 'SCRIPT : Module: Game; loaded 12x files; defines: "PLATFORM_WINDOWS,ENABLE_DIAG"; CRC32: a\n' phase = 'SCRIPT : Reintroduced [mission]: phase 0 -> 1\nSCRIPT : ERR ORDINARY: header verified project=core systems=MissionSystems\nSCRIPT : ERR ORDINARY: local participant verified project=core phase=prison\n' self.assertTrue(runner.game_evidence(module+phase)['ready']) self.assertFalse(runner.game_evidence(phase)['ready']) self.assertFalse(runner.game_evidence(module)['ready']) self.assertFalse(runner.game_evidence(module.replace('ENABLE_DIAG', 'WORKBENCH')+phase)['ready']) for error in ['SCRIPT (E): failure', 'RESOURCES (E): missing', 'ERR SMOKE PASS: fixture leaked', 'Reintroduced: scenario preparation failed']: self.assertTrue(runner.game_evidence(module+phase+error)['failed']) def test_startup_requires_selected_header_and_actual_systems_marker(self): module = 'SCRIPT : Module: Game; defines: "PLATFORM_WINDOWS"; CRC32: a\n' phase = 'SCRIPT : Reintroduced [mission]: phase 0 -> 1\n' core = 'SCRIPT : ERR ORDINARY: header verified project=core systems=MissionSystems\nSCRIPT : ERR ORDINARY: local participant verified project=core phase=prison\n' chat = 'SCRIPT : ERR ORDINARY: header verified project=chat systems=MissionSystems\nSCRIPT : ERR ORDINARY: local participant verified project=chat phase=prison\n' self.assertFalse(runner.game_evidence(module+phase, 'core')['ready']) self.assertFalse(runner.game_evidence(module+phase+core, 'chat')['ready']) self.assertTrue(runner.game_evidence(module+phase+chat, 'chat')['ready']) self.assertFalse(runner.game_evidence(module+phase+core.replace('MissionSystems', 'unknown'), 'core')['ready']) error = "GUI (E): Unknown class 'SCR_WidgetExportRuleRoot'\n" result = runner.game_evidence(error+module+phase+core, 'core') self.assertTrue(result['scenario_header_verified']) self.assertTrue(result['failed']) self.assertFalse(result['ready']) def test_world_without_normal_local_admission_is_not_ready(self): world = ('SCRIPT : Module: Game; defines: "PLATFORM_WINDOWS"; CRC32: a\n' 'SCRIPT : Reintroduced [mission]: phase 0 -> 1\n' 'SCRIPT : ERR ORDINARY: header verified project=core systems=MissionSystems\n') result = runner.game_evidence(world, 'core') self.assertTrue(result['scenario_header_verified']) self.assertFalse(result['local_participant_verified']) self.assertFalse(result['ready']) self.assertFalse(result['failed']) # incomplete admission may still arrive self.assertFalse(runner.game_evidence(world+'ERR ORDINARY: local participant verified project=chat phase=prison\n', 'core')['ready']) self.assertTrue(runner.game_evidence(world+'ERR ORDINARY: local participant verified project=core phase=prison\n', 'core')['ready']) def test_no_engine_or_audio_spawn_during_import_and_process_refusal(self): with patch.object(runner, 'assert_no_game', side_effect=ValueError('existing game')): with patch.object(runner.subprocess, 'Popen') as spawn, patch.object(runner.subprocess, 'run') as commands: class Args: steam=Path('/steam'); compat=Path('/compat'); profile=Path('/profile') timeout=180; settle=15; failure_grace=0 with self.assertRaisesRegex(ValueError, 'existing game'): runner.run(Args()) spawn.assert_not_called(); commands.assert_not_called() class FailureGraceTests(unittest.TestCase): def test_exit_at_settle_boundary_is_not_success(self): monitor = runner.StartupMonitor(100, 180, 5) self.assertIsNone(monitor.observe({'ready': True}, 101)) self.assertEqual(monitor.observe({'ready': True}, 106, True), 'process-exited') def test_missing_readiness_resets_continuous_settle(self): monitor = runner.StartupMonitor(100, 180, 5) self.assertIsNone(monitor.observe({'ready': True}, 101)) self.assertIsNone(monitor.observe({}, 104)) self.assertIsNone(monitor.observe({'ready': True}, 106)) self.assertEqual(monitor.observe({'ready': True}, 111), 'game-started') def test_default_error_stops_immediately(self): monitor = runner.StartupMonitor(100, 180, 15) self.assertEqual(monitor.observe({'failed': True}, 101), 'game-failed') self.assertEqual(monitor.first_failure_at, 101) def test_repeated_errors_do_not_renew_grace(self): monitor = runner.StartupMonitor(100, 180, 15, 30) self.assertIsNone(monitor.observe({'failed': True}, 105)) self.assertIsNone(monitor.observe({'failed': True}, 134.99)) self.assertEqual(monitor.first_failure_at, 105) self.assertEqual(monitor.observe({'failed': True}, 135), 'game-failed') def test_readiness_cannot_turn_latched_error_into_success(self): monitor = runner.StartupMonitor(100, 180, 5, 30) self.assertIsNone(monitor.observe({'failed': True}, 101)) for now in (102, 110, 130): self.assertIsNone(monitor.observe({'ready': True}, now)) self.assertEqual(monitor.observe({'ready': True}, 131), 'game-failed') self.assertIsNone(monitor.ready_at) def test_failure_after_readiness_aborts_settle_success(self): monitor = runner.StartupMonitor(100, 180, 5, 30) self.assertIsNone(monitor.observe({'ready': True}, 101)) self.assertIsNone(monitor.observe({'failed': True}, 105)) self.assertIsNone(monitor.observe({'ready': True}, 106)) self.assertEqual(monitor.observe({}, 135), 'game-failed') def test_absolute_timeout_cuts_short_grace(self): monitor = runner.StartupMonitor(100, 60, 5, 30) self.assertIsNone(monitor.observe({'failed': True}, 159)) self.assertEqual(monitor.observe({'ready': True}, 160), 'game-failed') self.assertEqual(monitor.timeout_result(), 'game-failed') self.assertEqual(monitor.deadline, 160) def test_process_exit_during_grace_returns_failed_immediately(self): monitor = runner.StartupMonitor(100, 180, 15, 30) monitor.observe({'failed': True}, 101) self.assertEqual(monitor.observe({}, 102, process_exited=True), 'game-failed') def test_final_raw_analysis_cannot_erase_observed_failure(self): monitor = runner.StartupMonitor(100, 180, 15, 30) monitor.observe({'failed': True}, 101) final = monitor.annotate({'ready': True, 'failed': False, 'engine_errors': [], 'game_module_observed': True}) self.assertTrue(final['failure_observed_during_polling']) self.assertTrue(final['failed']) self.assertFalse(final['ready']) self.assertEqual(final['engine_errors'], []) # no invented final-log errors self.assertTrue(final['game_module_observed']) def test_grace_does_not_delay_normal_success_or_exit(self): monitor = runner.StartupMonitor(100, 180, 15, 30) self.assertIsNone(monitor.observe({'ready': True}, 101)) self.assertEqual(monitor.observe({'ready': True}, 116), 'game-started') monitor = runner.StartupMonitor(100, 180, 15, 30) self.assertEqual(monitor.observe({}, 101, process_exited=True), 'process-exited') self.assertEqual(monitor.observe({}, 280), 'timeout') self.assertEqual(monitor.timeout_result(), 'timeout') def test_grace_bounds_rejected_before_any_native_mutation(self): for grace in (-1, 121): with self.subTest(grace=grace): with self.assertRaisesRegex(ValueError, 'failure-grace 0..120'): runner.StartupMonitor(100, 180, 15, grace) class Args: steam=Path('/steam'); compat=Path('/compat'); profile=Path('/profile') timeout=180; settle=15; failure_grace=grace with patch.object(runner, 'assert_no_game') as processes, patch.object(runner.subprocess, 'Popen') as spawn: with self.assertRaisesRegex(ValueError, 'failure-grace 0..120'): runner.run(Args()) processes.assert_not_called(); spawn.assert_not_called() def test_longer_diagnostic_grace_remains_failed_and_absolute_bounded(self): monitor = runner.StartupMonitor(100, 180, 15, 120) self.assertIsNone(monitor.observe({'failed': True}, 105)) self.assertIsNone(monitor.observe({'ready': True}, 224)) self.assertEqual(monitor.observe({'ready': True}, 225), 'game-failed') self.assertFalse(monitor.annotate({'ready': True})['ready']) monitor = runner.StartupMonitor(100, 60, 15, 120) self.assertIsNone(monitor.observe({'failed': True}, 159)) self.assertEqual(monitor.observe({'ready': True}, 160), 'game-failed') def test_cli_default_and_explicit_grace(self): for extra, expected in (([], 0), (['--failure-grace', '30'], 30), (['--failure-grace', '120'], 120)): with self.subTest(extra=extra), patch.object(sys, 'argv', ['headless_game.py', '--compat', '/unused', *extra]): with patch.object(runner, 'run', return_value=1) as run: self.assertEqual(runner.main(), 1) self.assertEqual(run.call_args.args[0].failure_grace, expected) if __name__ == '__main__': unittest.main()