sashatrask
267 lines
10 KiB
Python
267 lines
10 KiB
Python
import hashlib
|
|
import importlib.machinery
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import py_compile
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
APP_DIR = ROOT / 'app'
|
|
sys.path.insert(0, str(APP_DIR))
|
|
|
|
import child_bootstrap
|
|
import lifecycle_authority
|
|
|
|
|
|
def sha256_file(path):
|
|
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
|
|
|
|
|
class ImportSuffixManifestTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name)
|
|
self.app_dir = self.root / 'app'
|
|
self.app_dir.mkdir()
|
|
for name in lifecycle_authority.CODE_AUTHORITY_FILES:
|
|
path = self.app_dir.joinpath(*name.split('/'))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(f'# fixture for {name}\n', encoding='ascii')
|
|
shutil.copyfile(APP_DIR / 'child_bootstrap.py', self.app_dir / 'child_bootstrap.py')
|
|
|
|
external = self.root / 'runtime' / 'check-openrouter-keys.ps1'
|
|
external.parent.mkdir()
|
|
external.write_text('# fixture\n', encoding='ascii')
|
|
for name in ('start_core_runtime.ps1', 'start_runtime.ps1', 'stop_runtime.ps1'):
|
|
(self.root / name).write_text(f'# fixture for {name}\n', encoding='ascii')
|
|
self.executable = self.root / 'trufflehog.exe'
|
|
self.executable.write_bytes(b'fixture executable')
|
|
self.config = self.app_dir / 'config.yaml'
|
|
self.config.write_text('{}\n', encoding='ascii')
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def build_manifest(self):
|
|
return lifecycle_authority.build_code_manifest(
|
|
app_dir=self.app_dir,
|
|
trufflehog_path=self.executable,
|
|
)
|
|
|
|
def test_manifest_constructs_and_hashes_exact_import_suffix_surface(self):
|
|
package = self.app_dir / 'package'
|
|
package.mkdir()
|
|
pyw = self.app_dir / 'window.PYW'
|
|
pyw.write_text('raise SystemExit\n', encoding='ascii')
|
|
source = self.root / 'bytecode_source.py'
|
|
source.write_text('VALUE = 1\n', encoding='ascii')
|
|
pyc = package / 'sourceless.pyc'
|
|
py_compile.compile(str(source), cfile=str(pyc), doraise=True)
|
|
source.unlink()
|
|
pyd = package / 'native.PYD'
|
|
pyd.write_bytes(b'not loaded during manifest construction')
|
|
arbitrary = package / 'notes.txt'
|
|
arbitrary.write_text('not importable\n', encoding='ascii')
|
|
cache = package / '__pycache__' / 'generated.pyc'
|
|
cache.parent.mkdir()
|
|
cache.write_bytes(b'generated cache')
|
|
|
|
manifest = self.build_manifest()
|
|
|
|
self.assertEqual(manifest['schema'], 5)
|
|
self.assertEqual(child_bootstrap.MANIFEST_SCHEMA, lifecycle_authority.CODE_MANIFEST_SCHEMA)
|
|
for path in (pyw, pyc, pyd, cache):
|
|
name = path.relative_to(self.app_dir).as_posix()
|
|
self.assertEqual(manifest['files'][name]['sha256'], sha256_file(path))
|
|
self.assertNotIn(arbitrary.relative_to(self.app_dir).as_posix(), manifest['files'])
|
|
|
|
def test_normal_pycache_generation_is_manifest_file_set_drift(self):
|
|
package = self.app_dir / 'package'
|
|
package.mkdir()
|
|
source = package / 'module.py'
|
|
source.write_text('VALUE = 1\n', encoding='ascii')
|
|
before = self.build_manifest()
|
|
|
|
cache = Path(py_compile.compile(str(source), doraise=True))
|
|
self.assertEqual(cache.parent.name, '__pycache__')
|
|
|
|
after = self.build_manifest()
|
|
cache_name = cache.relative_to(self.app_dir).as_posix()
|
|
self.assertNotEqual(after, before)
|
|
self.assertEqual(after['files'][cache_name]['sha256'], sha256_file(cache))
|
|
with self.assertRaisesRegex(
|
|
lifecycle_authority.LifecycleAuthorityError,
|
|
f'file set drifted: {cache_name}',
|
|
):
|
|
lifecycle_authority.verify_code_manifest(before)
|
|
|
|
def test_added_pyw_is_rejected_as_file_set_drift(self):
|
|
manifest = self.build_manifest()
|
|
added = self.app_dir / 'late.PyW'
|
|
added.write_text('raise SystemExit\n', encoding='ascii')
|
|
|
|
with self.assertRaises(lifecycle_authority.LifecycleAuthorityError) as raised:
|
|
lifecycle_authority.verify_code_manifest(manifest)
|
|
self.assertIn('file set drifted: late.PyW', str(raised.exception))
|
|
|
|
def test_added_pyd_is_rejected_without_loading_it(self):
|
|
manifest = self.build_manifest()
|
|
added = self.app_dir / 'native_shadow.PYD'
|
|
added.write_bytes(b'not a loadable extension')
|
|
|
|
with mock.patch.object(
|
|
importlib.machinery.ExtensionFileLoader,
|
|
'create_module',
|
|
side_effect=AssertionError('extension was loaded'),
|
|
) as create_module:
|
|
with self.assertRaises(lifecycle_authority.LifecycleAuthorityError) as raised:
|
|
lifecycle_authority.verify_code_manifest(manifest)
|
|
create_module.assert_not_called()
|
|
self.assertIn('file set drifted: native_shadow.PYD', str(raised.exception))
|
|
|
|
def test_each_root_launcher_is_hashed_and_detects_drift(self):
|
|
for name in ('start_core_runtime.ps1', 'start_runtime.ps1', 'stop_runtime.ps1'):
|
|
with self.subTest(name=name):
|
|
manifest = self.build_manifest()
|
|
path = self.root / name
|
|
manifest_name = f'../{name}'
|
|
self.assertEqual(manifest['files'][manifest_name]['sha256'], sha256_file(path))
|
|
original = path.read_bytes()
|
|
try:
|
|
path.write_bytes(original + b'# drift\n')
|
|
with self.assertRaisesRegex(
|
|
lifecycle_authority.LifecycleAuthorityError,
|
|
f'code authority drifted: {manifest_name}',
|
|
):
|
|
lifecycle_authority.verify_code_manifest(manifest)
|
|
finally:
|
|
path.write_bytes(original)
|
|
|
|
def test_required_launcher_removal_and_reparse_are_rejected(self):
|
|
path = self.root / 'start_runtime.ps1'
|
|
original = path.read_bytes()
|
|
manifest = self.build_manifest()
|
|
path.unlink()
|
|
try:
|
|
with self.assertRaisesRegex(
|
|
lifecycle_authority.LifecycleAuthorityError,
|
|
'required external code authority file is absent',
|
|
):
|
|
lifecycle_authority.verify_code_manifest(manifest)
|
|
finally:
|
|
path.write_bytes(original)
|
|
|
|
replacement = self.root / 'launcher-replacement.ps1'
|
|
replacement.write_bytes(original)
|
|
path.unlink()
|
|
try:
|
|
try:
|
|
os.symlink(replacement, path)
|
|
except (NotImplementedError, OSError) as exc:
|
|
self.skipTest(f'file symlink creation is unavailable: {exc}')
|
|
with self.assertRaisesRegex(
|
|
lifecycle_authority.LifecycleAuthorityError,
|
|
'reparse point is forbidden',
|
|
):
|
|
lifecycle_authority.verify_code_manifest(manifest)
|
|
finally:
|
|
if os.path.lexists(path):
|
|
path.unlink()
|
|
path.write_bytes(original)
|
|
|
|
def test_manifest_rejects_unlisted_external_authority_path(self):
|
|
manifest = self.build_manifest()
|
|
unlisted = self.root / 'unlisted.ps1'
|
|
unlisted.write_text('# unlisted\n', encoding='ascii')
|
|
manifest['files']['../unlisted.ps1'] = {
|
|
'path': lifecycle_authority.canonical_path(unlisted),
|
|
'sha256': sha256_file(unlisted),
|
|
}
|
|
|
|
with self.assertRaisesRegex(
|
|
lifecycle_authority.LifecycleAuthorityError,
|
|
'not an allowed external',
|
|
):
|
|
lifecycle_authority.verify_code_manifest(manifest)
|
|
|
|
def test_child_rejects_sourceless_shadow_pyc_before_payload_executes(self):
|
|
manifest = self.build_manifest()
|
|
manifest_digest = lifecycle_authority.code_manifest_sha256(manifest)
|
|
marker = self.root / 'injected-marker.txt'
|
|
payload_source = self.root / 'shadow_payload.py'
|
|
payload_source.write_text(
|
|
'import os\n'
|
|
'with open(os.environ["INJECTED_MARKER"], "w", encoding="utf-8") as handle:\n'
|
|
' handle.write(os.environ.get("TRUF_SUPERVISOR_TOKEN", ""))\n',
|
|
encoding='ascii',
|
|
)
|
|
py_compile.compile(
|
|
str(payload_source),
|
|
cfile=str(self.app_dir / 'requests.pyc'),
|
|
doraise=True,
|
|
)
|
|
payload_source.unlink()
|
|
|
|
instance_file = self.root / 'instance.json'
|
|
instance_id = 'import-suffix-test'
|
|
token = 't' * 48
|
|
dsn = 'postgresql://truf:fixture@127.0.0.1:5432/truf'
|
|
dsn_digest = lifecycle_authority.dsn_sha256(dsn)
|
|
metadata = {
|
|
'schema': 2,
|
|
'instance_file': str(instance_file),
|
|
'instance_id': instance_id,
|
|
'token': token,
|
|
'activation_state': 'ACTIVE',
|
|
'config_path': str(self.config),
|
|
'config_sha256': sha256_file(self.config),
|
|
'supervisor_path': str(self.app_dir / 'supervisor.py'),
|
|
'supervisor_sha256': sha256_file(self.app_dir / 'supervisor.py'),
|
|
'code_manifest': manifest,
|
|
'code_manifest_sha256': manifest_digest,
|
|
'canonical_dsn_sha256': dsn_digest,
|
|
'control': {'host': '127.0.0.1', 'port': 1},
|
|
}
|
|
instance_file.write_text(json.dumps(metadata), encoding='ascii')
|
|
environment = os.environ.copy()
|
|
environment.update(lifecycle_authority.supervised_child_environment(metadata, dsn, 'scanner'))
|
|
environment.update({
|
|
'SCANNER_DB_URL': dsn,
|
|
'DATABASE_URL': dsn,
|
|
'INJECTED_MARKER': str(marker),
|
|
})
|
|
|
|
completed = subprocess.run(
|
|
[
|
|
sys.executable,
|
|
'-I',
|
|
'-S',
|
|
'-B',
|
|
str(self.app_dir / 'child_bootstrap.py'),
|
|
'scanner',
|
|
],
|
|
cwd=self.app_dir,
|
|
env=environment,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
timeout=20,
|
|
check=False,
|
|
)
|
|
|
|
self.assertNotEqual(completed.returncode, 0)
|
|
self.assertIn('file set drifted: requests.pyc', completed.stdout)
|
|
self.assertFalse(marker.exists())
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|