204 lines
16 KiB
Python
204 lines
16 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)
|
|
(folder/'addon.gproj').write_bytes((runner.ROOT/'addons'/project/'addon.gproj').read_bytes())
|
|
(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 : Rereintroduced [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_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()
|