#!/usr/bin/env python3 """Bounded ordinary-game startup check on a private hidden and muted display. No Workbench plugin, scripted smoke fixture, Steam profile, or existing process is used. """ import argparse import fcntl import json import os from pathlib import Path import re import shutil import signal import subprocess import time import uuid from native_provenance import snapshot, persist, bindings, seal_run from native_report import analyze ROOT = Path(__file__).resolve().parents[1] PROFILE_MARKER = '.erri-game-profile.json' PROFILE_ID = {'kind': 'erri-ordinary-game-disposable-profile', 'version': 1} GAME = re.compile(r'^armareforgersteam(?:diag)?(?:\.exe)?$', re.I) def refuse(message): raise ValueError(message) def assert_no_game(proc_root=Path('/proc')): """Read-only and fail closed; also recognizes Wine's truncated comm name.""" for proc in proc_root.iterdir(): if not proc.name.isdigit(): continue try: args = (proc/'cmdline').read_bytes().split(b'\0') comm = (proc/'comm').read_text().strip() except (FileNotFoundError, ProcessLookupError): continue # process exited during enumeration except OSError as error: refuse(f'Cannot exclude an existing game process {proc.name}: {error}') names = [re.split(r'[\\/]', arg.decode(errors='replace'))[-1] for arg in args] if any(GAME.fullmatch(name) for name in names) or comm.lower().startswith('armareforgerste'): refuse(f'An Arma Reforger game process already exists (PID {proc.name}); leave it untouched') def prefix_process_candidate(proc): """Classify readable process metadata before accessing protected environ. Linux services such as systemd can be non-dumpable even to their UID. Their environment is irrelevant. Windows/Wine/runtime candidates still fail closed. """ args = [value.decode(errors='replace') for value in (proc/'cmdline').read_bytes().split(b'\0') if value] comm = (proc/'comm').read_text().strip() try: executable = os.readlink(proc/'exe') except (FileNotFoundError, PermissionError): executable = '' values = args + [comm, executable] for value in values: name = re.split(r'[\\/]', value)[-1].lower() if (name.endswith('.exe') or name.startswith(('wine', 'proton', 'gamescope')) or name in {'reaper', 'pressure-vessel-wrap', 'pv-bwrap'} or name.startswith('armareforger')): return True # systemd's PAM child overwrites both names, and its non-dumpable exe is # inaccessible. This exact native helper is positively identified by both. if comm == '(sd-pam)' and args == ['(sd-pam)']: return False # Observed host Apache workers are also non-dumpable; accept only their # exact native foreground-service signature, after all runtime checks. if comm == 'apache2' and args == ['apache2', '-DFOREGROUND']: return False # A readable native executable is conclusive; otherwise require an absolute # Linux argv[0] corroborated by the kernel's (possibly truncated) comm name. if executable.startswith('/'): return False if args and args[0].startswith('/'): basename = Path(args[0]).name if comm and (comm == basename or comm == basename[:15]): return False argv0 = args[0] if args else '' refuse(f'Cannot classify possible Wine/runtime process {proc.name} (comm={comm!r}, argv0={argv0!r})') def assert_prefix_idle(compat, proc_root=Path('/proc')): target = ('WINEPREFIX='+str(compat/'pfx')).encode() failures = [] for proc in proc_root.iterdir(): if not proc.name.isdigit() or int(proc.name) == os.getpid(): continue try: if proc.stat().st_uid != os.getuid(): continue if not prefix_process_candidate(proc): continue environment = (proc/'environ').read_bytes().split(b'\0') except (FileNotFoundError, ProcessLookupError): continue except (OSError, ValueError) as error: failures.append(f'Cannot verify disposable prefix is idle (PID {proc.name}): {error}') continue if target in environment: failures.append(f'Disposable prefix already has a process (PID {proc.name}); leave it untouched') if failures: refuse('\n'.join(failures)) def validate_paths(root, steam, compat, profile): if compat.is_relative_to(steam) or not (compat/'pfx/system.reg').is_file(): refuse('--compat must be an existing disposable cloned prefix outside Steam') if any(profile.is_relative_to(path) or path.is_relative_to(profile) for path in (root, steam, compat)): refuse('--profile must be separate from source, Steam and disposable prefix trees') if profile.exists(): marker = profile/PROFILE_MARKER if marker.is_symlink() or not marker.is_file() or json.loads(marker.read_text()) != PROFILE_ID: refuse('Existing profile is not marked as this runner\'s disposable game profile') for path in profile.rglob('*'): if path.is_symlink() and path != profile/'addons/ArmaReforger': refuse(f'Unexpected profile symlink: {path}') def command_for(root, steam, compat, profile, project): project_text = (root/'addons'/project/'addon.gproj').read_text() guid = re.search(r'\bGUID "([A-F0-9]{16})"', project_text)[1] return ['gamescope', '--backend', 'headless', '-W', '1280', '-H', '720', '-w', '1280', '-h', '720', '-r', '30', '--', str(steam/'steamapps/common/Proton - Experimental/proton'), 'run', str(steam/'steamapps/common/Arma Reforger/ArmaReforgerSteam.exe'), '-addonsDir', 'Z:'+str(root/'addons'), '-addons', guid, '-profile', 'Z:'+str(profile), '-errOrdinaryScenario', project, '-noBackend', '-noSound', '-noSplash', '-noThrow', '-disableCrashReporter', '-logconsole'] def game_evidence(content, project=None): report = analyze(content, 'console.log') defines = re.findall(r'Module: Game;[^\n]*defines:\s*"([^"]*)"', content) non_workbench = bool(defines) and all(not {'WORKBENCH', 'ENF_WB'}.intersection(value.split(',')) for value in defines) headers = re.findall(r'ERR ORDINARY: header verified project=(core|chat) systems=MissionSystems', content) header_verified = bool(headers) and (project is None or project in headers) participants = re.findall(r'ERR ORDINARY: local participant verified project=(core|chat) phase=prison', content) participant_verified = bool(participants) and (project is None or project in participants) ready = header_verified and participant_verified and 'Rereintroduced [mission]: phase 0 -> 1' in content unexpected_fixture = 'ERR SMOKE ' in content failed = (bool(report['engine_errors'] or report['script_or_runtime_failures']) or 'Rereintroduced: scenario preparation failed' in content or 'Can\'t compile "Game"' in content or unexpected_fixture or any({'WORKBENCH', 'ENF_WB'}.intersection(value.split(',')) for value in defines)) return {**report, 'local_participant_verified': participant_verified, 'scenario_header_verified': header_verified, 'verified_header_projects': headers, 'game_module_defines': defines, 'workbench_undefined': non_workbench, 'normal_mission_ready': ready, 'unexpected_smoke_fixture': unexpected_fixture, 'failed': failed, 'ready': non_workbench and ready and not failed, 'scope': 'ordinary local game startup only; no player input, full escape, multiplayer or FPS proof'} def cleanup(child, proton, env): # Caller owns the prefix lock and verified no pre-existing prefix processes. try: subprocess.run([str(proton.parent/'files/bin/wineserver'), '-k'], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10) except (OSError, subprocess.TimeoutExpired): pass try: os.killpg(child.pid, signal.SIGTERM) except ProcessLookupError: pass try: child.wait(timeout=10) except subprocess.TimeoutExpired: pass finally: try: os.killpg(child.pid, signal.SIGKILL) except ProcessLookupError: pass child.wait(timeout=10) class StartupMonitor: """Latch failures while allowing a bounded diagnostic drain; no native I/O.""" def __init__(self, started, timeout, settle, failure_grace=0): if not 0 <= failure_grace <= 30: refuse('Use failure-grace 0..30 seconds') self.started = started self.deadline = started + timeout self.settle = settle self.failure_grace = failure_grace self.first_failure_at = None self.ready_at = None def observe(self, evidence, now, process_exited=False): if evidence.get('failed') and self.first_failure_at is None: self.first_failure_at = now if self.first_failure_at is not None: # Repeated errors and subsequent readiness cannot renew this lease. cutoff = min(self.deadline, self.first_failure_at + self.failure_grace) if process_exited or now >= cutoff: return 'game-failed' return None if now >= self.deadline: return 'timeout' if process_exited: return 'process-exited' if evidence.get('ready'): if self.ready_at is None: self.ready_at = now if now - self.ready_at >= self.settle: return 'game-started' else: self.ready_at = None return None def timeout_result(self): return 'game-failed' if self.first_failure_at is not None else 'timeout' def annotate(self, final): # Keep raw final error counts intact, but never lose an earlier failure # if the process truncated/replaced the log during the grace interval. final['failure_observed_during_polling'] = self.first_failure_at is not None if self.first_failure_at is not None: final['failed'] = True final['ready'] = False return final def run(a): steam, compat, profile = a.steam.resolve(), a.compat.resolve(), a.profile.resolve() if not 60 <= a.timeout <= 1800 or not 5 <= a.settle <= 120: refuse('Use timeout 60..1800 seconds and settle 5..120 seconds') if not 0 <= a.failure_grace <= 30: refuse('Use failure-grace 0..30 seconds') assert_no_game() validate_paths(ROOT, steam, compat, profile) # Use the SAME prefix lock as headless_workbench.py: never run them together. with (compat/'.erri-workbench.lock').open('a') as prefix_lock: try: fcntl.flock(prefix_lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: refuse('Another test runner owns this disposable prefix') assert_prefix_idle(compat) return run_locked(a, steam, compat, profile) def run_locked(a, steam, compat, profile): command = command_for(ROOT, steam, compat, profile, a.project) separator = command.index("--") proton, executable = Path(command[separator+1]), Path(command[separator+3]) if not proton.is_file() or not executable.is_file(): refuse('Installed ordinary ArmaReforgerSteam.exe and Proton Experimental are required') if not shutil.which('gamescope') or not shutil.which('pactl'): refuse('gamescope and pactl are required') # Refuse launcher/profile reuse; repeat immediately before any launch setup. assert_no_game() profile.mkdir(parents=True, exist_ok=True) marker = profile/PROFILE_MARKER if not marker.exists(): marker.write_text(json.dumps(PROFILE_ID)+'\n') addons = profile/'addons' addons.mkdir(exist_ok=True) vanilla = steam/'steamapps/common/Arma Reforger/addons/data' link = addons/'ArmaReforger' if not link.exists() and not link.is_symlink(): link.symlink_to(vanilla, target_is_directory=True) if link.resolve() != vanilla.resolve() or any(x.name not in {'ArmaReforger', 'saves'} for x in addons.iterdir()) or any((addons/'saves').rglob('*.gproj')): refuse('Private profile addons must contain only the installed vanilla link') run_id = uuid.uuid4().hex silent_sink = 'erri_game_'+run_id application_id = 'org.escapistsrereintroduced.game.'+run_id env = os.environ.copy() env.update(STEAM_COMPAT_DATA_PATH=str(compat), STEAM_COMPAT_CLIENT_INSTALL_PATH=str(steam), WINEPREFIX=str(compat/'pfx'), PULSE_SINK=silent_sink, PULSE_PROP='application.id='+application_id) sink_module = None child = None session = None captured = snapshot(ROOT, command) started = time.time() monitor = StartupMonitor(time.monotonic(), a.timeout, a.settle, a.failure_grace) result = 'timeout' runner_error = None cleanup_ok = True log_root = profile/'logs' previous = set(log_root.glob('logs_*')) launch_log = profile/('game-launch-'+run_id+'.log') try: sink_module = subprocess.check_output(['pactl', 'load-module', 'module-null-sink', 'sink_name='+silent_sink], text=True).strip() subprocess.run(['pactl', 'set-sink-mute', silent_sink, '1'], check=True) subprocess.run(['pactl', 'set-sink-volume', silent_sink, '0%'], check=True) assert_no_game() with launch_log.open('x') as output: child = subprocess.Popen(command, cwd=executable.parent, env=env, stdout=output, stderr=subprocess.STDOUT, start_new_session=True) while time.monotonic() < monitor.deadline: # -noSound prevents game audio; this catches a Proton startup stream # or a saved explicit Wine device. Only our fresh application ID. streams = json.loads(subprocess.check_output(['pactl', '-f', 'json', 'list', 'sink-inputs'], text=True)) for stream in streams: if stream.get('properties', {}).get('application.id') != application_id: continue index = str(stream['index']) subprocess.run(['pactl', 'move-sink-input', index, silent_sink], check=True) subprocess.run(['pactl', 'set-sink-input-mute', index, '1'], check=True) candidates = [x for x in log_root.glob('logs_*') if x not in previous] if len(candidates) > 1: result = 'unexpected-multiple-sessions'; break evidence = {} if candidates: if session is None: session = candidates[0] persist(session, captured) console = session/'console.log' content = console.read_text(errors='replace') if console.is_file() else '' evidence = game_evidence(content, a.project) outcome = monitor.observe(evidence, time.monotonic(), child.poll() is not None) if outcome is not None: result = outcome; break # Grace never extends the original absolute run deadline. cutoff = monitor.deadline if monitor.first_failure_at is not None: cutoff = min(cutoff, monitor.first_failure_at + a.failure_grace) time.sleep(max(0, min(1, cutoff - time.monotonic()))) else: result = monitor.timeout_result() except KeyboardInterrupt: result = 'interrupted' except (OSError, ValueError, subprocess.SubprocessError) as error: result = 'runner-failed' runner_error = str(error) finally: if child is not None: try: cleanup(child, proton, env) assert_prefix_idle(compat) except (OSError, ValueError, subprocess.SubprocessError) as error: cleanup_ok = False result = 'cleanup-failed' runner_error = str(error) if sink_module is not None: try: subprocess.run(['pactl', 'unload-module', sink_module], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5) except (OSError, subprocess.SubprocessError): pass # Never replace the user's audio defaults to compensate. sealed = seal_run(ROOT, session, captured) if cleanup_ok and session and (session/'console.log').is_file() else {} final = game_evidence((session/'console.log').read_text(errors='replace'), a.project) if session and (session/'console.log').is_file() else {} final = monitor.annotate(final) if final.get('failed'): result = 'game-failed' if sealed and not sealed['source_unchanged_after_run']: result = 'source-changed' summary = {**sealed, **bindings(captured), 'started_unix': started, 'result': result, 'session': str(session) if session else None, 'project': a.project, 'cleanup_confirmed': cleanup_ok, 'runner_error': runner_error, 'failure_grace_seconds': a.failure_grace, 'failure_observed_during_polling': monitor.first_failure_at is not None, 'first_failure_elapsed_seconds': (monitor.first_failure_at - monitor.started) if monitor.first_failure_at is not None else None, 'executable': str(executable), 'dedicated_server': False, 'display': 'headless Gamescope', 'audio': '-noSound plus own silent sink', 'scope': 'ordinary game compile/startup; no scripted fixture, full escape or multiplayer acceptance'} if session: (session/'runner.json').write_text(json.dumps(summary, indent=2)+'\n') (session/'game-startup-report.json').write_text(json.dumps(final, indent=2)+'\n') shutil.copyfile(launch_log, session/'headless-launch.log') print(json.dumps(summary, indent=2), flush=True) return 0 if result == 'game-started' else 1 def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--steam', type=Path, default=Path.home()/'.local/share/Steam') parser.add_argument('--compat', type=Path, required=True) parser.add_argument('--profile', type=Path, default=Path('/tmp/erri-game-profile')) parser.add_argument('--project', choices=['core', 'chat'], default='core') parser.add_argument('--inspect-processes', action='store_true', help='Read-only process preflight report; never launch, lock, write a profile or configure audio') parser.add_argument('--timeout', type=int, default=180) parser.add_argument('--settle', type=int, default=15) parser.add_argument('--failure-grace', type=int, default=0, help='Keep a failed run alive for diagnostic logs only, at most 0..30 seconds within the original timeout') args = parser.parse_args() try: if args.inspect_processes: report = {'read_only': True, 'game_process_check': 'clear', 'prefix_process_check': 'clear'} for key, check in [('game_process_check', assert_no_game), ('prefix_process_check', lambda: assert_prefix_idle(args.compat.resolve()))]: try: check() except (OSError, ValueError) as error: report[key] = str(error) print(json.dumps(report, indent=2)) return 0 if report['game_process_check'] == report['prefix_process_check'] == 'clear' else 1 return run(args) except (OSError, ValueError, subprocess.SubprocessError) as error: parser.error(str(error)) if __name__ == '__main__': raise SystemExit(main())