184 lines
12 KiB
Python
184 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Synthetic packed Chat closure evidence only; never launches an engine."""
|
|
from pathlib import Path
|
|
import contextlib
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]/'tools'))
|
|
import headless_game as runner
|
|
|
|
|
|
class PackedChatTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(prefix='erri-packed-chat-test-')
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.candidate = self.root/'packages'; self.candidate.mkdir()
|
|
for key, (name, guid, dependencies) in runner.PACKED_PROJECTS.items():
|
|
folder = self.candidate/key; folder.mkdir()
|
|
deps = [runner.PACKED_BASE]+[runner.PACKED_PROJECTS[d][1] for d in dependencies]
|
|
project = f'GameProject {{ ID "{name}" GUID "{guid}" TITLE "Synthetic {key}" Dependencies {{ '+ ' '.join('"'+d+'"' for d in deps)+' } }\n'
|
|
(folder/'addon.gproj').write_text(project)
|
|
(folder/'data.pak').write_bytes(b'SYNTHETIC NOT NATIVE '+key.encode())
|
|
(folder/'resourceDatabase.rdb').write_bytes(b'SYNTHETIC DATABASE '+key.encode())
|
|
self.session_temp = tempfile.TemporaryDirectory(prefix='erri-packed-chat-session-')
|
|
self.addCleanup(self.session_temp.cleanup)
|
|
self.session = Path(self.session_temp.name); self.session.rmdir()
|
|
self.args = SimpleNamespace(packed_core=None, packed_addons=self.candidate, packed_session=self.session,
|
|
project='chat', profile=Path('/tmp/erri-game-profile'), settle=15)
|
|
self.packed = runner.packed_request(self.args)
|
|
self.steam = Path('/Steam'); self.profile = self.session/'profile'
|
|
|
|
def content(self):
|
|
base = self.packed['input']
|
|
text = f"""00 ENGINE : Addon dirs:
|
|
00 ENGINE : dir: '{base}'
|
|
00 ENGINE : dir: './addons'
|
|
00 ENGINE : dir: '{self.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 key, (_, guid, _) in runner.PACKED_PROJECTS.items():
|
|
text += f"00 ENGINE : gproj: '{base/key}/addon.gproj' guid: '{guid}'\n"
|
|
text += "00 ENGINE : FileSystem: Adding package './addons/core/' (pak count: 2) to filesystem under name core\n"
|
|
text += "00 ENGINE : FileSystem: Adding package './addons/data/' (pak count: 14) to filesystem under name ArmaReforger\n"
|
|
for i, (key, (name, _, _)) in enumerate(runner.PACKED_PROJECTS.items()):
|
|
text += f"00 ENGINE : FileSystem: Adding package '{base/key}/' (pak count: 1) to filesystem under name {name}\n"
|
|
text += f"00 RESOURCES : ResourceDB: loading cache (id={i} name={name} path={base/key}/resourceDatabase.rdb)\n"
|
|
return text+'''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
|
|
'''
|
|
|
|
def evidence(self, content=None):
|
|
return runner.packed_evidence(self.content() if content is None else content, self.packed, self.steam, self.profile)
|
|
|
|
def test_exact_fifteen_artifacts_and_all_addon_identities_dependencies(self):
|
|
manifest = runner.packed_payload_inventory(self.candidate, 'chat')
|
|
self.assertEqual(set(manifest), {key+'/'+name for key in runner.PACKED_PROJECTS for name in runner.PACKED_FILES})
|
|
for key in runner.PACKED_PROJECTS:
|
|
p = self.candidate/key/'addon.gproj'; original = p.read_text()
|
|
for bad in (original.replace('GUID "', 'GUID "FFFF'), original.replace('ID "ERR_', 'ID "Other_'),
|
|
original.replace('Dependencies {', 'Dependencies { "58D0FB3206B6F859"'),
|
|
original.replace('"58D0FB3206B6F859"', '"1111111111111111"')):
|
|
p.write_text(bad)
|
|
with self.subTest(key=key), self.assertRaises(ValueError): runner.packed_payload_inventory(self.candidate, 'chat')
|
|
p.write_text(original)
|
|
chat = self.candidate/'chat/addon.gproj'; chat.write_text(chat.read_text().replace('"5192138CB9D5B911"', ''))
|
|
with self.assertRaises(ValueError): runner.packed_payload_inventory(self.candidate, 'chat')
|
|
|
|
def test_exact_folder_set_symlinks_traversal_extra_missing_empty_refused(self):
|
|
for folder in ('extra', 'Scripts'):
|
|
(self.candidate/folder).mkdir()
|
|
with self.assertRaises(ValueError): runner.packed_payload_inventory(self.candidate, 'chat')
|
|
(self.candidate/folder).rmdir()
|
|
original = self.candidate/'map'; original.rename(self.root/'map')
|
|
with self.assertRaises(ValueError): runner.packed_payload_inventory(self.candidate, 'chat')
|
|
original.symlink_to(self.root/'map', target_is_directory=True)
|
|
with self.assertRaises(ValueError): runner.packed_payload_inventory(self.candidate, 'chat')
|
|
original.unlink(); (self.root/'map').rename(original)
|
|
with self.assertRaises(ValueError): runner.packed_payload_inventory(self.candidate/'..'/'packages', 'chat')
|
|
for key in runner.PACKED_PROJECTS:
|
|
for name in runner.PACKED_FILES:
|
|
p = self.candidate/key/name; original = p.read_bytes(); p.write_bytes(b'')
|
|
with self.subTest(key=key, name=name), self.assertRaises(ValueError): runner.packed_payload_inventory(self.candidate, 'chat')
|
|
p.unlink(); p.symlink_to(self.root/'missing')
|
|
with self.assertRaises(ValueError): runner.packed_payload_inventory(self.candidate, 'chat')
|
|
p.unlink(); p.write_bytes(original)
|
|
|
|
def test_request_requires_mutual_exclusion_chat_session_and_fixed_settle(self):
|
|
for key, value in [('packed_core', self.candidate/'core'), ('project', 'core'), ('packed_addons', None),
|
|
('packed_session', None), ('settle', 14), ('profile', Path('/different')),
|
|
('packed_session', Path('/tmp/erri-packed-core-wrong')),
|
|
('packed_session', self.root/'erri-packed-chat-nested')]:
|
|
args = SimpleNamespace(**vars(self.args)); setattr(args, key, value)
|
|
with self.subTest(key=key), self.assertRaises(ValueError): runner.packed_request(args)
|
|
self.session.mkdir()
|
|
with self.assertRaises(ValueError): runner.packed_request(self.args)
|
|
|
|
def test_private_copy_v2_provenance_and_each_of_fifteen_artifacts_sealed(self):
|
|
runner.prepare_packed(self.packed)
|
|
raw = runner.packed_snapshot(self.packed, ['synthetic'], self.profile); metadata = json.loads(raw)
|
|
self.assertEqual((metadata['schema_version'], metadata['kind'], metadata['project']), (2, 'packed-chat-runtime-input', 'chat'))
|
|
self.assertEqual(metadata['addon_order'], ['core', 'ai', 'intel', 'map', 'chat'])
|
|
self.assertEqual(metadata['artifacts'], self.packed['manifest'])
|
|
for name in metadata['artifacts']:
|
|
source = self.candidate/name; target = self.packed['input']/name
|
|
self.assertNotEqual(source.stat().st_ino, target.stat().st_ino)
|
|
self.assertEqual(hashlib.sha256(target.read_bytes()).hexdigest(), metadata['artifacts'][name]['sha256'])
|
|
original = target.read_bytes(); target.write_bytes(b'changed')
|
|
self.assertFalse(runner.packed_seal(self.packed, raw, None)['packed_inputs_unchanged_after_run'])
|
|
target.write_bytes(original)
|
|
self.assertTrue(runner.packed_seal(self.packed, raw, None)['packed_inputs_unchanged_after_run'])
|
|
(self.candidate/'intel/resourceDatabase.rdb').unlink()
|
|
self.assertFalse(runner.packed_seal(self.packed, raw, None)['packed_inputs_unchanged_after_run'])
|
|
|
|
def test_candidate_changes_before_copy_and_before_launch_refused(self):
|
|
p = self.candidate/'ai/data.pak'; original = p.read_bytes(); p.write_bytes(b'changed')
|
|
with self.assertRaises(ValueError): runner.prepare_packed(self.packed)
|
|
p.write_bytes(original)
|
|
# Failed prepare owns its session; it is deliberately not reusable.
|
|
self.assertTrue(self.session.exists())
|
|
with self.assertRaises(FileExistsError): runner.prepare_packed(self.packed)
|
|
|
|
def test_command_selects_chat_from_private_closure_never_loose_source(self):
|
|
command = runner.command_for(Path('/forbidden-source'), self.steam, Path('/compat'), self.profile, 'chat', self.packed)
|
|
self.assertEqual(command[command.index('-addonsDir')+1], 'Z:'+str(self.packed['input']))
|
|
self.assertEqual(command[command.index('-addons')+1], runner.PACKED_PROJECTS['chat'][1])
|
|
self.assertEqual(command[command.index('-errOrdinaryScenario')+1], 'chat')
|
|
self.assertFalse(any('/forbidden-source' in arg or str(self.candidate) in arg for arg in command))
|
|
with self.assertRaises(ValueError): runner.command_for(Path('/unused'), self.steam, Path('/compat'), self.profile, 'core', self.packed)
|
|
|
|
def test_exact_seven_loaded_addons_and_five_mounts_databases_required(self):
|
|
self.assertTrue(self.evidence()['ready'])
|
|
self.assertFalse(self.evidence()['release_qualified'])
|
|
self.assertFalse(self.evidence()['build_qualified'])
|
|
source = self.content()
|
|
for key, (name, guid, _) in runner.PACKED_PROJECTS.items():
|
|
for fragment in (f"gproj: '{self.packed['input']/key}/addon.gproj'", f'to filesystem under name {name}', f'name={name} path='):
|
|
broken = '\n'.join(line for line in source.splitlines() if fragment not in line)+'\n'
|
|
with self.subTest(key=key, fragment=fragment): self.assertFalse(self.evidence(broken)['ready'])
|
|
redirected = source.replace(str(self.packed['input']/key), '/foreign/'+key)
|
|
self.assertTrue(self.evidence(redirected)['failed'])
|
|
self.assertTrue(self.evidence(source+f"00 ENGINE : FileSystem: Adding relative directory '/loose/{key}' to filesystem under name {name}\n")['failed'])
|
|
self.assertTrue(self.evidence(source.replace(f'name={name} path=', f'name=ERR_Foreign path='))['failed'])
|
|
self.assertTrue(self.evidence(source.replace('(pak count: 1)', '(pak count: 2)'))['failed'])
|
|
self.assertFalse(self.evidence(source.replace('Loaded addons:', 'Available addons:'))['ready'])
|
|
self.assertFalse(self.evidence(source.replace('project=chat', 'project=core'))['ready'])
|
|
|
|
def test_foreign_duplicate_errors_and_workbench_never_masked_by_later_good_block(self):
|
|
source = self.content(); guid = runner.PACKED_PROJECTS['ai'][1]
|
|
entry = next(line for line in source.splitlines() if "guid: '"+guid+"'" in line)+'\n'
|
|
for bad in (source.replace(entry, entry+entry), source.replace(entry, entry.replace(guid, '1111111111111111')),
|
|
source+'00 SCRIPT (E): synthetic error\n', source+'ERR SMOKE PASS: forbidden fixture\n',
|
|
source.replace('PLATFORM_WINDOWS', 'PLATFORM_WINDOWS,WORKBENCH')):
|
|
with self.subTest(bad=bad):
|
|
self.assertTrue(self.evidence(bad+source)['failed'])
|
|
self.assertFalse(self.evidence(bad+source)['ready'])
|
|
|
|
def test_mocked_failed_launch_keeps_chat_artifacts_not_source_provenance(self):
|
|
steam = self.root/'steam'; compat = self.root/'compat'
|
|
(compat/'pfx').mkdir(parents=True); (compat/'pfx/system.reg').write_text('synthetic')
|
|
for relative in ['steamapps/common/Proton - Experimental/proton', 'steamapps/common/Arma Reforger/ArmaReforgerSteam.exe']:
|
|
p = steam/relative; p.parent.mkdir(parents=True, exist_ok=True); p.write_bytes(b'not executed')
|
|
args = SimpleNamespace(**vars(self.args), steam=steam, compat=compat, timeout=60, failure_grace=0)
|
|
with patch.object(runner, 'assert_no_game'), patch.object(runner, 'assert_prefix_idle'), patch.object(runner.shutil, 'which', return_value='/mock'), patch.object(runner.subprocess, 'check_output', return_value='11'), patch.object(runner.subprocess, 'run'), patch.object(runner.subprocess, 'Popen', side_effect=OSError('synthetic failed launch')), patch.object(runner, 'snapshot', side_effect=AssertionError('source snapshot forbidden')), patch.object(runner, 'seal_run', side_effect=AssertionError('source seal forbidden')), contextlib.redirect_stdout(io.StringIO()):
|
|
self.assertEqual(runner.run(args), 1)
|
|
result = json.loads((self.session/'packed-runner.json').read_text())
|
|
self.assertEqual(result['project'], 'chat'); self.assertEqual(result['result'], 'runner-failed')
|
|
self.assertFalse(result['packed_consumption_verified']); self.assertNotIn('launch_source_sha256', result)
|
|
metadata = json.loads((self.session/runner.PACKED_MANIFEST).read_text())
|
|
self.assertEqual(metadata['schema_version'], 2); self.assertEqual(len(metadata['artifacts']), 15)
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|