EscapistsReintroduced/docs/evidence/dedicated-packed-2026-10-02-objective-map/dedicated_native.py

#!/usr/bin/env python3
"""Bounded Linux server diagnostic: isolated network, frozen sources, no clients."""
import argparse
import fcntl
import hashlib
import json
import os
import re
import stat
from pathlib import Path
import shutil
import signal
import subprocess
import time

PROBE = '''import json,os,socket,sys
state={"netns":os.readlink("/proc/self/ns/net"),"interfaces":socket.if_nameindex(),"routes":open("/proc/net/route").read(),"ipv6Routes":open("/proc/net/ipv6_route").read()}
print("ERR ISOLATION "+json.dumps(state),flush=True)
assert state["netns"] != sys.argv[1], "network namespace was not isolated"
assert [name for _,name in state["interfaces"]] == ["lo"], "non-loopback interface present"
assert len(state["routes"].splitlines()) == 1, "IPv4 route present"
assert all(line.split()[-1] == "lo" for line in state["ipv6Routes"].splitlines()), "non-loopback IPv6 route present"
if len(sys.argv)>2: os.execv(sys.argv[2],sys.argv[2:])
'''


def manifest(source):
    result = {}
    for path in sorted(source.rglob('*')):
        if path.is_symlink():
            raise ValueError('Linked source entry is not allowed: ' + str(path))
        if path.is_file():
            result[str(path.relative_to(source))] = hashlib.sha256(path.read_bytes()).hexdigest()
    return result


def sandbox(runtime, source, run):
    return ['bwrap', '--unshare-net', '--unshare-pid', '--unshare-ipc', '--unshare-uts',
            '--die-with-parent', '--cap-drop', 'ALL', '--clearenv',
            '--ro-bind', '/', '/', '--tmpfs', '/home', '--tmpfs', '/tmp', '--tmpfs', '/run',
            '--proc', '/proc', '--dev', '/dev', '--ro-bind', str(runtime), '/tmp/runtime',
            '--bind', str(source), '/tmp/source', '--bind', str(run), '/tmp/work',
            '--setenv', 'HOME', '/tmp/work/home', '--setenv', 'TMPDIR', '/tmp/work/tmp',
            '--setenv', 'XDG_CONFIG_HOME', '/tmp/work/home/.config',
            '--setenv', 'XDG_DATA_HOME', '/tmp/work/home/.local/share',
            '--setenv', 'XDG_CACHE_HOME', '/tmp/work/home/.cache',
            '--setenv', 'PATH', '/usr/bin:/bin', '--setenv', 'LANG', 'C.UTF-8',
            '--chdir', '/tmp/runtime']


PROJECTS = {
    'core': ('2528466DE220D91C', '{AD62322D5213C440}Missions/ERR_Everon_Base.conf'),
    'chat': ('2528466DE220D91C', '{C1E20667DCE324ED}Missions/ERR_Everon.conf'),
}
ADDON_GUIDS = {'core': PROJECTS['core'][0]}

WORLD = '{B6C516F86302C679}Worlds/ERR_Everon.ent'
SYSTEMS = '{C50579A2EC48E8EF}Configs/Systems/MissionSystems.conf'


def engine_args(route, project='core'):
    if route not in ('header', 'diag-world', 'vanilla-everon') or project not in PROJECTS:
        raise ValueError('Unsupported diagnostic route or project')
    if route == 'vanilla-everon' and project != 'core':
        raise ValueError('vanilla-everon cannot be combined with --project chat')
    args = ['/tmp/runtime/ArmaReforgerServer' + ('Diag' if route == 'diag-world' else ''),
            '-profile', '/tmp/work/profile', '-noBackend', '-noSound', '-noSplash', '-noThrow',
            '-bindIP', '127.0.0.1', '-bindPort', '2001', '-maxFPS', '30', '-logStats', '5000']
    if route == 'vanilla-everon':
        return args + ['-server', 'Missions/21_GM_Eden.conf']
    addon, header = PROJECTS[project]
    args += ['-addonsDir', '/tmp/source/addons', '-addons', addon]
    if route == 'header':
        return args + ['-server', header]
    return args + ['-server', WORLD, '-worldSystemsConfig', SYSTEMS]


def preflight_addons(source, route, project='core'):
    """Presence/identity checks only; native loading must validate RDB contents."""
    engine_args(route, project)  # also reject incompatible direct helper calls
    if route == 'vanilla-everon':
        return
    addons = source / 'addons'
    if addons.is_symlink() or not addons.is_dir():
        raise ValueError('Expected a regular addons directory in diagnostic source')
    closure = ('core',) if project == 'core' else tuple(ADDON_GUIDS)
    for addon in closure:
        folder = addons / addon
        if folder.is_symlink() or not folder.is_dir():
            raise ValueError('Missing or linked required addon directory: ' + addon)
        for name in ('addon.gproj', 'resourceDatabase.rdb'):
            path = folder / name
            try:
                info = path.lstat()
            except FileNotFoundError as error:
                raise ValueError('Missing required addon input: ' + str(path)) from error
            if not stat.S_ISREG(info.st_mode) or info.st_size <= 0:
                raise ValueError('Required addon input must be a nonempty regular file: ' + str(path))
        project_file = folder / 'addon.gproj'
        if project_file.stat().st_size > 65536:
            raise ValueError('Unsupported oversized addon project: ' + str(project_file))
        try:
            definition = project_file.read_text(encoding='utf-8')
        except UnicodeDecodeError as error:
            raise ValueError('Addon project is not UTF-8: ' + str(project_file)) from error
        if re.findall(r'\bGUID\s+"([A-Fa-f0-9]{16})"', definition) != [ADDON_GUIDS[addon]]:
            raise ValueError('Required addon identity mismatch: ' + str(project_file))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--source', type=Path, required=True)
    parser.add_argument('--runtime', type=Path, default=Path('/tmp/erri-dedicated-runtime/server'))
    parser.add_argument('--archive', type=Path, required=True)
    parser.add_argument('--route', choices=['header', 'diag-world', 'vanilla-everon'], default='header')
    parser.add_argument('--project', choices=['core', 'chat'], default='core',
                        help='Diagnostic scenario header; both select the combined addon')
    parser.add_argument('--seconds', type=int, default=90)
    args = parser.parse_args()
    if args.route == 'vanilla-everon' and args.project != 'core':
        parser.error('vanilla-everon cannot be combined with --project chat')
    if not 1 <= args.seconds <= 110:
        parser.error('Runtime bound must be 1..110 seconds (plus at most 10 seconds owned cleanup)')
    source, runtime, archive = args.source.resolve(), args.runtime.resolve(), args.archive.resolve()
    if not source.is_relative_to('/tmp') or not runtime.is_relative_to('/tmp') or not archive.is_relative_to('/tmp'):
        parser.error('Sources, runtime and archive must be isolated /tmp paths')
    if archive == source or source in archive.parents or archive in source.parents or archive == runtime or runtime in archive.parents or archive in runtime.parents:
        parser.error('Source, archive and runtime must not overlap')
    if archive.exists() or archive.is_symlink():
        parser.error('Archive must be new')
    try:
        preflight_addons(source, args.route, args.project)
    except (ValueError, OSError) as error:
        parser.error(str(error))
    origin = runtime.parent / 'runtime-origin.json'
    if not origin.is_file() or json.loads(origin.read_text()).get('app_id') != 1874900:
        parser.error('Stable private runtime origin missing')
    binary = runtime / Path(engine_args(args.route, args.project)[0]).name
    with binary.open('rb') as stream:
        elf = stream.read(20)
    if binary.is_symlink() or len(elf) != 20 or elf[:6] != b'\x7fELF\x02\x01' or int.from_bytes(elf[18:20], 'little') != 62:
        parser.error('Expected real native x86-64 ELF server')
    archive.mkdir(mode=0o700, parents=True)
    (archive / 'runner.py').write_bytes(Path(__file__).read_bytes())
    for child in ['home', 'profile', 'tmp']:
        (archive / child).mkdir()
    # Reject links before copytree follows them into the isolated source snapshot.
    manifest(source)
    frozen = archive / 'source'
    shutil.copytree(source, frozen, ignore=shutil.ignore_patterns('.git', '__pycache__', 'evidence', 'generated'))
    preflight_addons(frozen, args.route, args.project)
    initial = manifest(frozen)
    (archive / 'source-manifest.json').write_text(json.dumps(initial, indent=2)+'\n')
    base = sandbox(runtime, frozen, archive)
    netns = os.readlink('/proc/self/ns/net')
    check = base + ['/usr/bin/python3', '-c', PROBE, netns]
    preflight = subprocess.run(check, capture_output=True, text=True, timeout=15)
    (archive / 'isolation-preflight.json').write_text(json.dumps({'command':check,'exit':preflight.returncode,'stdout':preflight.stdout,'stderr':preflight.stderr}, indent=2)+'\n')
    if preflight.returncode:
        print('Isolation refused; no server launched:', archive)
        return 2
    command = base + ['/usr/bin/python3', '-c', PROBE, netns] + engine_args(args.route, args.project)
    record = {'command': command, 'engine': engine_args(args.route, args.project), 'seconds':args.seconds,
              'route': args.route, 'project': None if args.route == 'vanilla-everon' else args.project,
              'binarySHA256':hashlib.sha256(binary.read_bytes()).hexdigest(),
              'serverLaunched':False,'realPlayers':0,'multiplayerProven':False,
              'scope':'vanilla Everon control; no successor loaded' if args.route == 'vanilla-everon' else f'successor {args.project} empty-server startup'}
    (archive / 'command.json').write_text(json.dumps(record, indent=2)+'\n')
    lock_fd = os.open('/tmp/erri-dedicated-native-diagnostic.lock', os.O_WRONLY | os.O_CREAT | os.O_NOFOLLOW, 0o600)
    with os.fdopen(lock_fd, 'a') as lock:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        with (archive / 'stdout.log').open('wb') as log:
            process = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
            record['serverLaunched'] = True
            record['supervisorPid'] = process.pid
            record['startedUTC'] = time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())
            try:
                record['exit'] = process.wait(timeout=args.seconds)
                record['timeout'] = False
            except subprocess.TimeoutExpired:
                record['timeout'] = True
            finally:
                if process.poll() is None:
                    os.killpg(process.pid, signal.SIGTERM)
                    try: process.wait(timeout=5)
                    except subprocess.TimeoutExpired:
                        os.killpg(process.pid, signal.SIGKILL)
                        process.wait(timeout=5)
                record['exit'] = process.returncode
                record['finishedUTC'] = time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())
                (archive / 'result.json').write_text(json.dumps(record, indent=2)+'\n')
                (archive / 'source-manifest-after.json').write_text(json.dumps(manifest(frozen), indent=2)+'\n')
    print('Diagnostic archived; inspect actual compilation/header/systems/errors:',archive)
    return 0

if __name__ == '__main__':
    raise SystemExit(main())