jakub
cab998fb64
Merge owned systems into one addon while preserving resource identities. Gate player admission on the replicated prison scene, restore the original framed door, and hold enemy AI until escape. Correct scenario branding and author metadata, include publishing assets, and retain regression and native verification evidence.
216 lines
17 KiB
Python
216 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""Packed archival verification without native execution or production writes."""
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]/'tools'))
|
|
import archive_packed as archiver
|
|
import headless_game as runner
|
|
import test_packed_core_game as fixtures
|
|
|
|
|
|
class PackedArchiveTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.fixture=fixtures.PackedCoreTests();self.fixture.setUp()
|
|
self.session=self.fixture.session;self.root=self.fixture.root/'archive-repository';self.root.mkdir()
|
|
runner.prepare_packed(self.fixture.packed)
|
|
command=runner.command_for(Path('/unused'),self.fixture.steam,Path('/compat'),self.fixture.profile,'core',self.fixture.packed)
|
|
self.raw=runner.packed_snapshot(self.fixture.packed,command,self.fixture.profile)
|
|
self.meta=json.loads(self.raw)
|
|
self.native=self.fixture.profile/'logs/logs_2026-09-19_11-02-00';self.native.mkdir(parents=True)
|
|
(self.native/'console.log').write_text(self.fixture.content())
|
|
self.report=runner.packed_seal(self.fixture.packed,self.raw,self.native)
|
|
self.report.update(started_unix=self.meta['captured_unix']+0.001,stopped_unix=self.meta['captured_unix']+30,result='packed-core-started',session=str(self.native),project='core',cleanup_confirmed=True,runner_error=None,failure_observed_during_polling=False,executable=command[16],dedicated_server=False,control_only=False,packed_consumption_verified=True)
|
|
# The command executable follows Proton/run; avoid an assumed path in fixtures.
|
|
self.report['executable']=command[command.index('--')+3]
|
|
self.startup={'ready':True,'failed':False,'packed_only':True}
|
|
self.write_reports()
|
|
def tearDown(self):self.fixture.tearDown()
|
|
def write_reports(self):
|
|
raw=archiver.encoded(self.report)
|
|
(self.session/'packed-runner.json').write_bytes(raw);(self.native/'runner.json').write_bytes(raw)
|
|
startup=archiver.encoded(self.startup)
|
|
(self.session/'packed-startup-report.json').write_bytes(startup);(self.native/'game-startup-report.json').write_bytes(startup)
|
|
(self.native/runner.PACKED_MANIFEST).write_bytes(self.raw)
|
|
def update_metadata(self):
|
|
self.raw=archiver.encoded(self.meta);(self.session/runner.PACKED_MANIFEST).write_bytes(self.raw)
|
|
self.report['packed_input_provenance_sha256']=archiver.digest(self.raw);self.write_reports()
|
|
def reseal_console(self,text):
|
|
raw=text.encode();(self.native/'console.log').write_bytes(raw);self.report['console_sha256']=archiver.digest(raw);self.write_reports()
|
|
def test_exact_artifacts_reports_and_hashes_retained_without_source_claim(self):
|
|
dest,status=archiver.archive(self.session,'packed-test',self.root);self.assertEqual(status,0)
|
|
self.assertEqual(runner.packed_inventory(dest/'package'),self.meta['artifacts'])
|
|
for name in runner.PACKED_FILES:self.assertEqual((dest/'package'/name).read_bytes(),(self.fixture.packed['input']/name).read_bytes())
|
|
self.assertEqual((dest/runner.PACKED_MANIFEST).read_bytes(),self.raw)
|
|
hashes=json.loads((dest/'archive-sha256.json').read_text())
|
|
for name,digest in hashes.items():self.assertEqual(archiver.digest((dest/name).read_bytes()),digest)
|
|
self.assertFalse((dest/'source-sha256.json').exists());self.assertFalse((dest/'source-provenance.json').exists())
|
|
report=json.loads((dest/'report.json').read_text());self.assertTrue(report['packed_consumption_verified']);self.assertFalse(report['build_qualified']);self.assertFalse(report['release_qualified'])
|
|
def test_valid_failed_run_archives_nonzero_and_retains_raw_failure(self):
|
|
self.report.update(result='game-failed',packed_consumption_verified=False,failure_observed_during_polling=True)
|
|
self.startup={'ready':False,'packed_input_violations':['historical alias parser refusal']};self.write_reports()
|
|
dest,status=archiver.archive(self.session,'failed-run',self.root)
|
|
self.assertEqual(status,1);self.assertIn('historical alias parser refusal',(dest/'raw-startup-report.json').read_text())
|
|
self.assertTrue(json.loads((dest/'report.json').read_text())['runner_failure_latched'])
|
|
def test_corrected_ready_parser_cannot_erase_latched_run_failure(self):
|
|
self.report.update(result='game-failed',packed_consumption_verified=False,failure_observed_during_polling=True);self.write_reports()
|
|
files,report=archiver.inspect(self.session)
|
|
self.assertFalse(report['packed_consumption_verified']);self.assertTrue(report['packed_loaded_paths_verified'])
|
|
def test_native_raw_error_overrides_claimed_success_and_is_retained(self):
|
|
self.reseal_console(self.fixture.content()+'00 SCRIPT (E): actual failure\n')
|
|
dest,status=archiver.archive(self.session,'error-run',self.root);self.assertEqual(status,1)
|
|
self.assertIn('actual failure',(dest/'console.log.txt').read_text())
|
|
def test_auxiliary_errors_fail_but_warnings_do_not(self):
|
|
(self.native/'error.log').write_text('00 SCRIPT (W): warning\n')
|
|
self.assertTrue(archiver.inspect(self.session)[1]['packed_consumption_verified'])
|
|
(self.native/'error.log').write_text('00 SCRIPT (E): failure absent from console\n')
|
|
files,report=archiver.inspect(self.session);self.assertFalse(report['packed_consumption_verified']);self.assertTrue(report['auxiliary_native_failure']);self.assertIn('error.log.txt',files)
|
|
def test_forged_saved_ready_report_cannot_substitute_native_readiness(self):
|
|
self.reseal_console(self.fixture.content().replace('local participant verified','not ready'))
|
|
self.assertFalse(archiver.inspect(self.session)[1]['packed_consumption_verified'])
|
|
def test_unchanged_flag_and_minimum_duration_required(self):
|
|
self.report['packed_inputs_unchanged_after_run']=False;self.write_reports()
|
|
self.assertFalse(archiver.inspect(self.session)[1]['packed_consumption_verified'])
|
|
self.report['packed_inputs_unchanged_after_run']=True;self.report['stopped_unix']=self.report['started_unix']+14.9;self.write_reports()
|
|
self.assertFalse(archiver.inspect(self.session)[1]['packed_consumption_verified'])
|
|
def test_tampered_payload_console_or_copy_is_refused_before_destination(self):
|
|
payload=self.fixture.packed['input']/'data.pak';original=payload.read_bytes();payload.write_bytes(b'changed')
|
|
with self.assertRaisesRegex(ValueError,'package bytes differ'):archiver.archive(self.session,'tampered',self.root)
|
|
self.assertFalse((self.root/'docs/evidence/tampered').exists());payload.write_bytes(original)
|
|
(self.native/'console.log').write_text('changed')
|
|
with self.assertRaisesRegex(ValueError,'console differs'):archiver.inspect(self.session)
|
|
def test_root_native_report_disagreement_refused(self):
|
|
(self.native/'runner.json').write_text('{}')
|
|
with self.assertRaisesRegex(ValueError,'copies disagree'):archiver.inspect(self.session)
|
|
def test_manifest_seal_mismatch_and_duplicate_keys_refused(self):
|
|
(self.session/runner.PACKED_MANIFEST).write_bytes(self.raw+b' ')
|
|
with self.assertRaisesRegex(ValueError,'binding differs'):archiver.inspect(self.session)
|
|
(self.session/runner.PACKED_MANIFEST).write_text('{"schema_version":1,"schema_version":1}')
|
|
with self.assertRaisesRegex(ValueError,'duplicate JSON key'):archiver.inspect(self.session)
|
|
def test_unexpected_command_loose_search_or_server_flag_refused(self):
|
|
self.meta['command'].extend(['-addonsDir','Z:/loose/successor/addons']);self.update_metadata()
|
|
with self.assertRaisesRegex(ValueError,'Recorded command differs'):archiver.inspect(self.session)
|
|
def test_recorded_native_session_escape_refused(self):
|
|
self.report['session']='/tmp/foreign/logs_foo';self.write_reports()
|
|
with self.assertRaisesRegex(ValueError,'escapes'):archiver.inspect(self.session)
|
|
def test_runtime_or_profile_redirection_refused(self):
|
|
self.meta['runtime_directory']='/tmp/foreign/core';self.update_metadata()
|
|
with self.assertRaisesRegex(ValueError,'do not belong'):archiver.inspect(self.session)
|
|
def test_cleanup_unknown_result_and_source_relabel_refused(self):
|
|
original=dict(self.report)
|
|
for key,value in [('cleanup_confirmed',False),('result','all good'),('source_provenance','launch_snapshot'),('build_qualified',True),('packed_only',False),('packed_inputs_unchanged_after_run',1)]:
|
|
self.report=dict(original);self.report[key]=value;self.write_reports()
|
|
with self.subTest(key=key),self.assertRaises(ValueError):archiver.inspect(self.session)
|
|
def test_canonical_names_no_overwrite_and_symlink_destination(self):
|
|
for name in ('../outside','nested/name','.','..','/absolute','bad name','_hidden'):
|
|
with self.subTest(name=name),self.assertRaises(ValueError):archiver.archive(self.session,name,self.root)
|
|
dest,status=archiver.archive(self.session,'once',self.root)
|
|
with self.assertRaisesRegex(ValueError,'overwrite'):archiver.archive(self.session,'once',self.root)
|
|
(self.root/'docs/evidence/redirect').symlink_to(dest,target_is_directory=True)
|
|
with self.assertRaises(ValueError):archiver.archive(self.session,'redirect',self.root)
|
|
def test_symlinked_evidence_refused(self):
|
|
path=self.native/'error.log';path.symlink_to(self.native/'console.log')
|
|
with self.assertRaisesRegex(ValueError,'symlinked'):archiver.inspect(self.session)
|
|
def test_timestamp_and_missing_hash_validation(self):
|
|
self.report['started_unix']=self.meta['captured_unix']-1;self.write_reports()
|
|
with self.assertRaisesRegex(ValueError,'chronology'):archiver.inspect(self.session)
|
|
|
|
class PackedChatArchiveTests(unittest.TestCase):
|
|
# Reuse the existing Core fixture construction without inheriting/rerunning its
|
|
# tests. Convert the private payload to the runner's actual schema-2 closure.
|
|
write_reports=PackedArchiveTests.write_reports
|
|
update_metadata=PackedArchiveTests.update_metadata
|
|
reseal_console=PackedArchiveTests.reseal_console
|
|
def setUp(self):
|
|
base=PackedArchiveTests();base.setUp();self.__dict__.update(base.__dict__)
|
|
runtime=self.session/'input/addons';candidate=self.fixture.root/'chat-candidate';candidate.mkdir()
|
|
for project in archiver.ADDON_ORDER:
|
|
folder=runtime/project;folder.mkdir(exist_ok=True)
|
|
# This schema-2 archive represents the frozen vanilla-only closure. Current
|
|
# source projects may add external dependencies that this payload does not carry.
|
|
name,guid,dependencies=runner.PACKED_PROJECTS[project]
|
|
deps=[runner.PACKED_BASE]+[runner.PACKED_PROJECTS[key][1] for key in dependencies]
|
|
declaration=f'GameProject {{ ID "{name}" GUID "{guid}" TITLE "Synthetic {project}" Dependencies {{ '
|
|
declaration+=' '.join('"'+guid+'"' for guid in deps)+' } }\n'
|
|
(folder/'addon.gproj').write_text(declaration)
|
|
(folder/'data.pak').write_bytes(('fixture-pak-'+project).encode())
|
|
(folder/'resourceDatabase.rdb').write_bytes(('fixture-db-'+project).encode())
|
|
shutil.copytree(folder,candidate/project)
|
|
self.packed={'project':'chat','input':runtime,'candidate':candidate,'session':self.session,'manifest':runner.packed_payload_inventory(runtime,'chat')}
|
|
command=runner.command_for(Path('/unused'),self.fixture.steam,Path('/compat'),self.fixture.profile,'chat',self.packed)
|
|
self.raw=runner.packed_snapshot(self.packed,command,self.fixture.profile);self.meta=json.loads(self.raw)
|
|
(self.native/'console.log').write_text(self.content())
|
|
self.report=runner.packed_seal(self.packed,self.raw,self.native)
|
|
self.report.update(started_unix=self.meta['captured_unix']+0.001,stopped_unix=self.meta['captured_unix']+30,result='packed-chat-started',session=str(self.native),project='chat',cleanup_confirmed=True,runner_error=None,failure_observed_during_polling=False,executable=command[command.index('--')+3],dedicated_server=False,control_only=False,packed_consumption_verified=True)
|
|
self.startup={'ready':True,'failed':False,'packed_only':True};self.write_reports()
|
|
def tearDown(self):self.fixture.tearDown()
|
|
def content(self):
|
|
folder=self.session/'input/addons'
|
|
lines=[f"00 ENGINE : dir: '{folder}'", "00 ENGINE : dir: './addons'", "00 ENGINE : dir: '$profile:../addons/'", '00 ENGINE : Loaded addons:', "00 ENGINE : gproj: './addons/core/core.gproj' guid: '5614BBCCBB55ED1C'", "00 ENGINE : gproj: './addons/data/ArmaReforger.gproj' guid: '58D0FB3206B6F859'"]
|
|
for project in archiver.ADDON_ORDER:
|
|
name,guid,deps=runner.PACKED_PROJECTS[project]
|
|
lines.append(f"00 ENGINE : gproj: '{folder/project}/addon.gproj' guid: '{guid}'")
|
|
lines.extend(["00 ENGINE : FileSystem: Adding package './addons/core/' (pak count: 2) to filesystem under name core", "00 ENGINE : FileSystem: Adding package './addons/data/' (pak count: 14) to filesystem under name ArmaReforger"])
|
|
for i,project in enumerate(archiver.ADDON_ORDER):
|
|
name,guid,deps=runner.PACKED_PROJECTS[project]
|
|
lines.extend([f"00 ENGINE : FileSystem: Adding package '{folder/project}/' (pak count: 1) to filesystem under name {name}", f"00 RESOURCES : ResourceDB: loading cache (id={i} name={name} path={folder/project}/resourceDatabase.rdb)"])
|
|
lines.extend(['00 SCRIPT : Module: GameLib; defines: "PLATFORM_WINDOWS"','00 SCRIPT : Module: Game; defines: "PLATFORM_WINDOWS"','00 SCRIPT : ERR ORDINARY: header verified project=chat systems=MissionSystems','00 SCRIPT : ERR ORDINARY: local participant verified project=chat phase=prison','00 SCRIPT : Reintroduced [mission]: phase 0 -> 1'])
|
|
return '\n'.join(lines)+'\n'
|
|
def test_chat_preserves_all_fifteen_artifacts_and_schema(self):
|
|
destination,status=archiver.archive(self.session,'packed-chat',self.root);self.assertEqual(status,0)
|
|
self.assertEqual(runner.packed_payload_inventory(destination/'package','chat'),self.meta['artifacts'])
|
|
self.assertEqual(len(self.meta['artifacts']),15)
|
|
for name,info in self.meta['artifacts'].items():self.assertEqual(archiver.digest((destination/'package'/name).read_bytes()),info['sha256'])
|
|
kind=json.loads((destination/'archive-kind.json').read_text());self.assertEqual(kind['schema_version'],2);self.assertEqual(kind['kind'],'packed-chat-consumption');self.assertEqual(kind['addon_order'],archiver.ADDON_ORDER)
|
|
report=json.loads((destination/'report.json').read_text());self.assertTrue(report['packed_consumption_verified']);self.assertFalse(report['release_qualified'])
|
|
self.assertFalse((destination/'source-sha256.json').exists())
|
|
def test_missing_extra_or_traversal_manifest_key_refused(self):
|
|
original=dict(self.meta['artifacts'])
|
|
for operation in ('missing','extra','traversal'):
|
|
self.meta['artifacts']=dict(original)
|
|
if operation=='missing':del self.meta['artifacts']['map/data.pak']
|
|
elif operation=='extra':self.meta['artifacts']['foreign/data.pak']=original['map/data.pak']
|
|
else:self.meta['artifacts']['../data.pak']=self.meta['artifacts'].pop('map/data.pak')
|
|
self.update_metadata()
|
|
with self.subTest(operation=operation),self.assertRaisesRegex(ValueError,'artifact set'):archiver.inspect(self.session)
|
|
def test_schema_kind_project_order_and_core_result_cannot_mix(self):
|
|
original=dict(self.meta)
|
|
for key,value in [('schema_version',1),('kind','packed-core-runtime-input'),('project','core'),('addon_order',['core','chat','ai','intel','map'])]:
|
|
self.meta=dict(original);self.meta[key]=value;self.update_metadata()
|
|
with self.subTest(key=key),self.assertRaises(ValueError):archiver.inspect(self.session)
|
|
self.meta=original;self.update_metadata();self.report['result']='packed-core-started';self.write_reports()
|
|
with self.assertRaisesRegex(ValueError,'result'):archiver.inspect(self.session)
|
|
def test_wrong_companion_identity_is_not_accepted_as_hash_manifest(self):
|
|
path=self.packed['input']/'ai/addon.gproj';path.write_bytes((self.packed['input']/'core/addon.gproj').read_bytes())
|
|
data=path.read_bytes();self.meta['artifacts']['ai/addon.gproj']={'bytes':len(data),'sha256':archiver.digest(data)};self.update_metadata()
|
|
with self.assertRaises(ValueError):archiver.inspect(self.session)
|
|
def test_external_dependency_cannot_expand_frozen_packed_closure(self):
|
|
path=self.packed['input']/'ai/addon.gproj'
|
|
path.write_text(path.read_text().replace('Dependencies {','Dependencies { "595F2BF2F44836FB"'))
|
|
data=path.read_bytes();self.meta['artifacts']['ai/addon.gproj']={'bytes':len(data),'sha256':archiver.digest(data)}
|
|
self.update_metadata()
|
|
with self.assertRaisesRegex(ValueError,'identity or exact declared dependencies'):archiver.inspect(self.session)
|
|
def test_each_companion_loaded_mount_and_database_evidence_required(self):
|
|
for fragment in ("under name ERR_Ai", "name=ERR_Intel path=", "guid: '9AAEE9041D79809E'", 'header verified project=chat', 'local participant verified project=chat'):
|
|
self.reseal_console(self.content().replace(fragment,'missing'))
|
|
with self.subTest(fragment=fragment):self.assertFalse(archiver.inspect(self.session)[1]['packed_consumption_verified'])
|
|
def test_redirected_companion_and_loose_extra_search_fail_native_review(self):
|
|
text=self.content().replace(str(self.packed['input']/'ai'),'Z:/loose/ai')
|
|
self.reseal_console(text)
|
|
self.assertFalse(archiver.inspect(self.session)[1]['packed_consumption_verified'])
|
|
self.reseal_console(self.content()+"00 ENGINE : dir: '/loose/source/addons'\n")
|
|
self.assertFalse(archiver.inspect(self.session)[1]['packed_consumption_verified'])
|
|
def test_valid_failed_chat_run_archives_nonzero(self):
|
|
self.report.update(result='timeout',packed_consumption_verified=False);self.write_reports()
|
|
destination,status=archiver.archive(self.session,'failed-chat',self.root);self.assertEqual(status,1)
|
|
self.assertTrue((destination/'package/ai/data.pak').is_file())
|
|
def test_recorded_command_cannot_select_core_or_an_external_addon_parent(self):
|
|
original=list(self.meta['command'])
|
|
for flag,value in [('-addons','2528466DE220D91C'),('-addonsDir','Z:/loose/addons')]:
|
|
self.meta['command']=list(original);self.meta['command'][self.meta['command'].index(flag)+1]=value;self.update_metadata()
|
|
with self.subTest(flag=flag),self.assertRaisesRegex(ValueError,'Recorded command differs'):archiver.inspect(self.session)
|
|
|
|
if __name__=='__main__':unittest.main()
|