202 lines
13 KiB
Python
202 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Archive a stopped packed-only Core or Chat run without attaching loose-source identity."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import math
|
|
from pathlib import Path
|
|
import re
|
|
import sys
|
|
import uuid
|
|
|
|
import headless_game
|
|
from headless_game import PACKED_FILES, PACKED_MANIFEST, command_for, packed_evidence, packed_inventory
|
|
from native_provenance import encoded, strict_json
|
|
from native_report import analyze
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
ADDON_ORDER = ['core', 'ai', 'intel', 'map', 'chat']
|
|
RESULTS = {'packed-core-started', 'packed-evidence-incomplete', 'packed-input-changed',
|
|
'cleanup-failed', 'unexpected-multiple-sessions', 'process-exited',
|
|
'interrupted', 'runner-failed', 'game-failed', 'timeout'}
|
|
|
|
|
|
def require(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def read(path):
|
|
require(path == path.resolve() and path.is_file() and not path.is_symlink(), 'Missing, noncanonical or symlinked evidence: '+str(path))
|
|
return path.read_bytes()
|
|
|
|
|
|
def digest(raw):
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def timestamp(value):
|
|
return type(value) in (int, float) and math.isfinite(value) and value > 0
|
|
|
|
|
|
|
|
def schema(metadata):
|
|
fields = {'schema_version', 'kind', 'launch_id', 'captured_unix', 'candidate_directory', 'runtime_directory', 'profile', 'command', 'artifacts', 'build_qualification'}
|
|
require(isinstance(metadata, dict), 'Invalid packed launch metadata object')
|
|
version = metadata.get('schema_version')
|
|
require(type(version) is int, 'Invalid packed schema version')
|
|
if version == 1:
|
|
require(set(metadata) == fields and metadata.get('kind') == 'packed-core-runtime-input', 'Invalid packed Core metadata fields/kind')
|
|
return 'core'
|
|
require(version == 2 and set(metadata) == fields | {'project', 'addon_order'}
|
|
and metadata.get('kind') == 'packed-chat-runtime-input' and metadata.get('project') == 'chat'
|
|
and metadata.get('addon_order') == ADDON_ORDER, 'Invalid packed Chat metadata fields/kind/dependency order')
|
|
return 'chat'
|
|
|
|
|
|
def payload_inventory(folder, project):
|
|
# Preserve the original Core schema and exact-three-file validation. Chat
|
|
# uses the runner's exact-five-addon closure validator, not an arbitrary tree.
|
|
if project == 'core':
|
|
return packed_inventory(folder)
|
|
return headless_game.packed_payload_inventory(folder, project)
|
|
|
|
def inspect(session):
|
|
"""Read and verify first; no writes and no engine/process operation."""
|
|
session = Path(session)
|
|
require(session.is_absolute() and session == session.resolve() and session.is_dir(), 'Use the canonical stopped packed session directory')
|
|
raw = read(session/PACKED_MANIFEST)
|
|
metadata = strict_json(raw)
|
|
project = schema(metadata)
|
|
expected_result = 'packed-'+project+'-started'
|
|
require(metadata['build_qualification'] == 'unqualified candidate; packaging verdict is unchanged', 'Packed metadata cannot assert build qualification')
|
|
require(isinstance(metadata['launch_id'], str), 'Invalid packed launch ID')
|
|
try:
|
|
uuid.UUID(metadata['launch_id'])
|
|
except ValueError as error:
|
|
raise ValueError('Invalid packed launch UUID') from error
|
|
runtime = session/'input/addons/core' if project == 'core' else session/'input/addons'
|
|
profile = session/'profile'
|
|
require(metadata['runtime_directory'] == str(runtime) and metadata['profile'] == str(profile), 'Packed runtime/profile paths do not belong to this session')
|
|
candidate = metadata['candidate_directory']
|
|
require(isinstance(candidate, str) and Path(candidate).is_absolute() and '..' not in Path(candidate).parts, 'Invalid recorded candidate path')
|
|
artifacts = metadata['artifacts']
|
|
expected_files = PACKED_FILES if project == 'core' else {addon+'/'+name for addon in ADDON_ORDER for name in PACKED_FILES}
|
|
require(isinstance(artifacts, dict) and set(artifacts) == expected_files, 'Packed provenance must hash the exact complete runtime artifact set')
|
|
for name, info in artifacts.items():
|
|
require(isinstance(info, dict) and set(info) == {'bytes', 'sha256'} and type(info['bytes']) is int and info['bytes'] > 0
|
|
and isinstance(info['sha256'], str) and re.fullmatch('[0-9a-f]{64}', info['sha256']), 'Invalid packed artifact hash/size: '+name)
|
|
require(payload_inventory(runtime, project) == artifacts, 'Runtime package bytes differ from launch manifest; refusing replacement payload')
|
|
runner_raw = read(session/'packed-runner.json')
|
|
runner = strict_json(runner_raw)
|
|
require(isinstance(runner, dict), 'Invalid stopped runner report')
|
|
require(runner.get('packed_only') is True and runner.get('project') == project and runner.get('dedicated_server') is False and runner.get('control_only') is False, 'Runner is not the recorded packed-only ordinary project')
|
|
require(not any(key in runner for key in ('source_provenance', 'launch_source_sha256', 'launch_source_manifest', 'launch_source_metadata', 'source_unchanged_after_run', 'end_source_sha256')), 'Packed run cannot carry loose-source qualification')
|
|
require(runner.get('build_qualified') is False and runner.get('release_qualified') is False, 'Packed consumption cannot qualify a build or release')
|
|
require(runner.get('packed_input_provenance') == PACKED_MANIFEST and runner.get('packed_input_provenance_sha256') == digest(raw) and runner.get('packed_launch_id') == metadata['launch_id'], 'Runner packed manifest/launch binding differs')
|
|
require(runner.get('cleanup_confirmed') is True, 'Archive only after owned cleanup is confirmed')
|
|
require('packed_input_error' in runner and (runner['packed_input_error'] is None or isinstance(runner['packed_input_error'], str)), 'Missing/malformed packed input diagnostic')
|
|
require(type(runner.get('packed_inputs_unchanged_after_run')) is bool and type(runner.get('packed_consumption_verified')) is bool, 'Missing packed input/consumption verdicts')
|
|
require(type(runner.get('failure_observed_during_polling')) is bool, 'Missing polling failure latch')
|
|
allowed_results = (RESULTS - {'packed-core-started'}) | {expected_result}
|
|
require(isinstance(runner.get('result'), str) and runner['result'] in allowed_results, 'Unknown packed runner result')
|
|
require(timestamp(metadata['captured_unix']) and timestamp(runner.get('started_unix')) and timestamp(runner.get('stopped_unix'))
|
|
and metadata['captured_unix'] <= runner['started_unix'] <= runner['stopped_unix'], 'Invalid snapshot/start/stop chronology')
|
|
command = metadata['command']
|
|
require(isinstance(command, list) and len(command) > 16 and all(isinstance(arg, str) for arg in command), 'Invalid recorded native command')
|
|
executable = runner.get('executable')
|
|
require(isinstance(executable, str) and Path(executable).is_absolute(), 'Invalid recorded executable')
|
|
exe = Path(executable)
|
|
require(exe.parts[-4:] == ('steamapps', 'common', 'Arma Reforger', 'ArmaReforgerSteam.exe'), 'Recorded executable is not installed ordinary Arma Reforger')
|
|
steam = exe.parents[3]
|
|
packed_context = {'input': runtime, 'project': project}
|
|
expected = command_for(Path('/unused-source'), steam, Path('/unused-prefix'), profile, project, packed_context)
|
|
require(command == expected, 'Recorded command differs from exact packed-only ordinary launch; unexpected flags/search roots')
|
|
native = runner.get('session')
|
|
require(isinstance(native, str), 'No sealed native log session was produced')
|
|
native = Path(native)
|
|
require(native.parent == profile/'logs' and re.fullmatch(r'logs_[A-Za-z0-9_-]+', native.name) and native == native.resolve(), 'Native log session escapes the recorded private profile')
|
|
require(read(native/PACKED_MANIFEST) == raw and read(native/'runner.json') == runner_raw, 'Native and root packed manifest/runner copies disagree')
|
|
startup_raw = read(session/'packed-startup-report.json')
|
|
require(read(native/'game-startup-report.json') == startup_raw, 'Native and root raw startup reports disagree')
|
|
require(isinstance(strict_json(startup_raw), dict), 'Invalid raw startup report')
|
|
console = read(native/'console.log')
|
|
require(runner.get('console_sha256') == digest(console), 'Raw console differs from stopped runner seal')
|
|
files = {PACKED_MANIFEST: raw, 'runner.json.txt': runner_raw, 'raw-startup-report.json': startup_raw, 'console.log.txt': console}
|
|
auxiliary = {}
|
|
for name in ('script.log', 'error.log', 'crash.log', 'headless-launch.log', 'rereintroduced-hq.log'):
|
|
if (native/name).exists() or (native/name).is_symlink():
|
|
data = read(native/name)
|
|
files[name+'.txt'] = data
|
|
auxiliary[name] = analyze(data.decode(errors='replace'), name)
|
|
for name in sorted(expected_files):
|
|
data = read(runtime/name)
|
|
require(len(data) == artifacts[name]['bytes'] and digest(data) == artifacts[name]['sha256'], 'Packed artifact changed during archive read: '+name)
|
|
files['package/'+name] = data
|
|
# Recompute from the console, not the saved ready flag; auxiliary errors also
|
|
# remain failures. Warnings in error.log are not mislabeled as native errors.
|
|
evidence = packed_evidence(console.decode(errors='replace'), packed_context, steam, profile)
|
|
auxiliary_failed = any(report['engine_errors'] or report['script_or_runtime_failures'] or report['missing_end_screen'] for report in auxiliary.values())
|
|
settled_claim = (runner['result'] == expected_result and runner['stopped_unix'] - runner['started_unix'] >= 15)
|
|
passed = (evidence['ready'] is True and not auxiliary_failed and settled_claim
|
|
and runner['packed_inputs_unchanged_after_run'] is True and runner.get('packed_input_error') is None
|
|
and runner['packed_consumption_verified'] is True and runner['failure_observed_during_polling'] is False and runner.get('runner_error') is None)
|
|
report = {**evidence, 'packed_consumption_verified': passed, 'archived_packed_input_verified': True,
|
|
'runner_result': runner['result'], 'runner_failure_latched': runner['failure_observed_during_polling'],
|
|
'packed_inputs_unchanged_after_run': runner['packed_inputs_unchanged_after_run'],
|
|
'auxiliary_native_reports': auxiliary, 'auxiliary_native_failure': auxiliary_failed,
|
|
'runner_settle_claim_consistent': settled_claim,
|
|
'settle_evidence': 'Native readiness recomputed; 15-second continuous settling is the sealed runner verdict, not an independently recorded polling timeline.',
|
|
'build_qualified': False, 'release_qualified': False,
|
|
'qualification': 'packed-only ordinary startup consumption only; original package build verdict and current source acceptance remain separate'}
|
|
report['failed'] = not passed
|
|
report['ready'] = passed
|
|
report['verifier_sha256'] = {name: digest(read(Path(__file__).resolve().parent/name)) for name in ('archive_packed.py', 'headless_game.py', 'native_report.py', 'project_schema.py')}
|
|
files['report.json'] = encoded(report)
|
|
archive_kind = {'schema_version': metadata['schema_version'], 'kind': 'packed-'+project+'-consumption', 'packed_launch_id': metadata['launch_id'],
|
|
'build_qualified': False, 'release_qualified': False}
|
|
if project == 'chat':
|
|
archive_kind.update(project=project, addon_order=ADDON_ORDER)
|
|
files['archive-kind.json'] = encoded(archive_kind)
|
|
return files, report
|
|
|
|
|
|
def archive(session, name, root=ROOT):
|
|
require(isinstance(name, str) and re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]{0,127}', name) is not None, 'Archive name must be one safe ASCII component')
|
|
root = Path(root)
|
|
evidence = root/'docs/evidence'
|
|
require(root == root.resolve() and evidence == evidence.resolve(), 'Evidence destination must not redirect through symlinks')
|
|
destination = evidence/name
|
|
require(not destination.exists() and not destination.is_symlink(), 'Evidence already exists; refusing overwrite')
|
|
files, report = inspect(session)
|
|
# No mutation until every provenance binding and artifact byte has passed.
|
|
destination.mkdir(parents=True, exist_ok=False)
|
|
for relative, data in files.items():
|
|
target = destination/relative
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
with target.open('xb') as output:
|
|
output.write(data)
|
|
hashes = {name: digest(raw) for name, raw in sorted(files.items())}
|
|
with (destination/'archive-sha256.json').open('xb') as output:
|
|
output.write(encoded(hashes))
|
|
metadata = strict_json(files[PACKED_MANIFEST])
|
|
require(payload_inventory(destination/'package', schema(metadata)) == metadata['artifacts'], 'Archived packed payload verification failed')
|
|
return destination, 0 if report['packed_consumption_verified'] else 1
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('session', type=Path, help='Stopped /tmp/erri-packed-core-* or erri-packed-chat-* session root, not its logs subdirectory')
|
|
parser.add_argument('name', help='New docs/evidence directory name; never overwritten')
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
destination, status = archive(args.session, args.name)
|
|
except (OSError, ValueError) as error:
|
|
parser.error(str(error))
|
|
print(f'Archived {destination}; packed consumption exit={status} (nonzero retains failed evidence)')
|
|
return status
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|