Files
EscapistsReintroduced/tools/headless_game.py
T

448 lines
23 KiB
Python

#!/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}
CONTROL = 'vanilla-control'
CONTROL_MARKER = '.erri-vanilla-control.json'
CONTROL_ID = {'kind': 'erri-vanilla-only-startup-control', '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):
if project not in {'core', 'chat', CONTROL}:
refuse('Unsupported ordinary game project/control')
command = ['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')]
if project != CONTROL:
project_text = (root/'addons'/project/'addon.gproj').read_text()
guid = re.search(r'\bGUID "([A-F0-9]{16})"', project_text)[1]
command += ['-addonsDir', 'Z:'+str(root/'addons'), '-addons', guid]
command += ['-profile', 'Z:'+str(profile)]
if project != CONTROL:
command += ['-errOrdinaryScenario', project, '-noSplash']
return command + ['-noBackend', '-noSound', '-noThrow', '-disableCrashReporter', '-logconsole']
def validate_control_profile(profile):
"""Never reuse the successor or user's profile as a vanilla control."""
if profile.exists():
marker = profile/CONTROL_MARKER
if marker.is_symlink() or not marker.is_file() or json.loads(marker.read_text()) != CONTROL_ID:
refuse('Vanilla control requires a new or explicitly control-owned disposable profile')
def vanilla_evidence(content):
report = analyze(content, 'console.log')
modules = re.findall(r'Module: (GameLib|Game);[^\n]*defines:\s*"([^"]*)"', content)
ordinary = {name for name, flags in modules} == {'GameLib', 'Game'}
bad_defines = any({'WORKBENCH', 'ENF_WB'}.intersection(flags.split(',')) for _, flags in modules)
# Only the engine's Loaded-addons blocks count, not its available inventory.
blocks = re.findall(r"Loaded addons:\r?\n((?:[^\n]*ENGINE[^\n]*gproj:[^\n]*\n)+)", content)
guids = [guid.upper() for block in blocks for guid in re.findall(r"guid: '([0-9A-Fa-f]{16})'", block)]
allowed = {'5614BBCCBB55ED1C', '58D0FB3206B6F859'}
wrong_addons = bool(set(guids) - allowed)
base_only = bool(blocks) and set(guids) == allowed and not wrong_addons
successor = any(token in content for token in ('ERR ORDINARY:', 'ERR SMOKE ', 'Rereintroduced [mission]'))
menu = "Entities load 'worlds/MainMenuWorld/MainMenuWorld.ent'" in content and 'Entered main menu.' in content
failed = bool(report['engine_errors'] or report['script_or_runtime_failures']) or bad_defines or wrong_addons or successor
return {**report, 'control_only': True, 'loaded_addon_guids': guids, 'vanilla_addons_only': base_only,
'game_module_defines': modules, 'workbench_undefined': ordinary and not bad_defines,
'main_menu_initialized': menu, 'unexpected_successor_activity': successor,
'normal_mission_ready': False, 'local_participant_verified': False,
'scenario_header_verified': False, 'failed': failed,
'ready': ordinary and base_only and menu and not failed,
'scope': 'vanilla-only ordinary main-menu startup control; no successor mission, input, multiplayer or release qualification'}
def game_evidence(content, project=None):
if project == CONTROL:
return vanilla_evidence(content)
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()
if a.project == CONTROL:
if a.timeout > 120 or a.settle > 30:
refuse('Vanilla control is bounded to timeout 60..120 and settle 5..30 seconds')
validate_control_profile(profile)
elif (profile/CONTROL_MARKER).exists():
refuse('Successor runs must not reuse the vanilla-control profile')
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')
if a.project == CONTROL:
(profile/CONTROL_MARKER).write_text(json.dumps(CONTROL_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'
if a.project == CONTROL and result == 'game-started':
result = 'vanilla-control-started'
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',
'control_only': a.project == CONTROL,
'source_manifest_scope': 'successor addon snapshot retained for provenance bookkeeping only; none loaded by vanilla control' if a.project == CONTROL else 'successor addon sources',
'scope': final.get('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 in {'game-started', 'vanilla-control-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', CONTROL], 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())