import hashlib import itertools import json import os from pathlib import Path import runpy 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 lifecycle_authority import supervisor import supervisor_instance def canonical(path): return os.path.normcase(os.path.realpath(os.path.abspath(os.fspath(path)))) class RetainedProcess: def __init__(self, command): self.command = list(command) self.closed = False def command_line(self): return list(self.command) def close(self): self.closed = True class ApplicationTreeReparseTests(unittest.TestCase): def _create_link(self, kind, target, link): if kind == 'symlink': try: os.symlink(target, link, target_is_directory=True) except (NotImplementedError, OSError) as exc: self.skipTest(f'directory symlink creation is unavailable: {exc}') return if os.name != 'nt': self.skipTest('directory junctions are Windows-only') completed = subprocess.run( ['cmd.exe', '/d', '/c', 'mklink', '/J', str(link), str(target)], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=10, check=False, ) if completed.returncode != 0 or not getattr(os.path, 'isjunction', lambda _path: False)(link): self.skipTest(f'directory junction creation is unavailable: {completed.stdout.strip()}') def _assert_package_link_rejected_before_payload(self, kind): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) app_dir = root / 'app' app_dir.mkdir() for name in ('child_bootstrap.py', 'runtime_bootstrap.py'): shutil.copy2(APP_DIR / name, app_dir / name) payload_marker = root / 'package.marker' target_marker = root / 'target.marker' external_package = root / 'external_package' external_package.mkdir() (external_package / '__init__.py').write_text( 'import os\nopen(os.environ["PACKAGE_MARKER"], "w", encoding="ascii").write("executed")\n', encoding='ascii', ) supervisor_path = app_dir / 'supervisor.py' supervisor_path.write_text( 'import os\nimport shadow_package\n' 'open(os.environ["TARGET_MARKER"], "w", encoding="ascii").write("executed")\n', encoding='ascii', ) for name in lifecycle_authority.CODE_AUTHORITY_FILES: path = app_dir.joinpath(*name.split('/')) path.parent.mkdir(parents=True, exist_ok=True) if not path.exists(): path.write_text(f'# fixture for {name}\n', encoding='ascii') external_authority = root / 'runtime' / 'check-openrouter-keys.ps1' external_authority.parent.mkdir() external_authority.write_text('# fixture\n', encoding='ascii') for name in ('start_runtime.ps1', 'stop_runtime.ps1'): (root / name).write_text(f'# fixture for {name}\n', encoding='ascii') executable = root / 'trufflehog.exe' executable.write_bytes(b'fixture') clean_manifest = lifecycle_authority.build_code_manifest( app_dir=app_dir, trufflehog_path=executable, ) linked_package = app_dir / 'shadow_package' self._create_link(kind, external_package, linked_package) with self.assertRaisesRegex( lifecycle_authority.LifecycleAuthorityError, 'application directory reparse point is forbidden', ): lifecycle_authority.build_code_manifest(app_dir=app_dir, trufflehog_path=executable) with self.assertRaisesRegex( lifecycle_authority.LifecycleAuthorityError, 'application directory reparse point is forbidden', ): lifecycle_authority.verify_code_manifest(clean_manifest) manifest = { 'schema': lifecycle_authority.CODE_MANIFEST_SCHEMA, 'root': canonical(app_dir), 'files': {}, 'executables': {}, 'assets': {}, } digest = hashlib.sha256(json.dumps( manifest, ensure_ascii=True, sort_keys=True, separators=(',', ':'), ).encode('utf-8')).hexdigest() child_namespace = runpy.run_path(str(app_dir / 'child_bootstrap.py')) with self.assertRaisesRegex(RuntimeError, 'application directory reparse point is forbidden'): child_namespace['_verify_manifest']( {'code_manifest': manifest, 'code_manifest_sha256': digest}, {'code_manifest_sha256': digest}, ) completed = subprocess.run( [ sys.executable, '-I', '-S', '-B', str(app_dir / 'runtime_bootstrap.py'), 'supervisor', '--', '--runtime-bootstrap-entrypoint', str(supervisor_path.resolve()), ], cwd=root, env={ **os.environ, 'PACKAGE_MARKER': str(payload_marker), 'TARGET_MARKER': str(target_marker), }, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=20, check=False, ) self.assertNotEqual(completed.returncode, 0) self.assertIn('application directory reparse point is forbidden', completed.stdout) self.assertFalse(payload_marker.exists()) self.assertFalse(target_marker.exists()) def test_general_package_symlink_is_rejected_before_payload(self): self._assert_package_link_rejected_before_payload('symlink') def test_general_package_junction_is_rejected_before_payload_when_supported(self): self._assert_package_link_rejected_before_payload('junction') class SupervisorEntrypointBindingTests(unittest.TestCase): def setUp(self): self.supervisor_path = canonical(APP_DIR / 'supervisor.py') self.config_path = canonical(APP_DIR / 'config.yaml') self.bootstrap_path = canonical(APP_DIR / 'runtime_bootstrap.py') self.bound_command = [ sys.executable, '-I', '-S', '-B', self.bootstrap_path, 'supervisor', '--', '--runtime-bootstrap-entrypoint', self.supervisor_path, '--config', self.config_path, '--with-postgres', ] self.unbound_command = [ sys.executable, '-I', '-S', '-B', self.bootstrap_path, 'supervisor', '--', '--config', self.config_path, '--with-postgres', ] def _lifecycle_metadata(self): return { 'pid': 123, 'process_creation_time': 'fixture', 'executable': canonical(sys.executable), 'supervisor_path': self.supervisor_path, 'config_path': self.config_path, } def _instance_metadata(self): return { **self._lifecycle_metadata(), 'lifecycle_mode': 'foreground', 'code_manifest': {}, 'code_manifest_sha256': '1' * 64, 'supervisor_sha256': '2' * 64, 'config_sha256': '3' * 64, } def _verify_with_supervisor_instance(self, command): retained = RetainedProcess(command) def digest(path): return '2' * 64 if canonical(path) == self.supervisor_path else '3' * 64 with mock.patch.object( supervisor_instance, 'validate_instance_metadata', return_value=self._instance_metadata(), ), mock.patch.object(supervisor_instance, 'verify_code_manifest'), mock.patch.object( supervisor_instance, 'sha256_file', side_effect=digest, ), mock.patch.object( supervisor_instance, 'verify_retained_process', return_value=retained, ): result = supervisor_instance.verify_instance_process(self._instance_metadata()) return result def test_realistic_foreground_command_passes_both_retained_process_verifiers(self): lifecycle_process = RetainedProcess(self.bound_command) with mock.patch.object( lifecycle_authority, 'verify_retained_process', return_value=lifecycle_process, ): lifecycle_authority._verify_supervisor_process(self._lifecycle_metadata()) self.assertTrue(lifecycle_process.closed) instance_process = self._verify_with_supervisor_instance(self.bound_command) self.assertFalse(instance_process.closed) instance_process.close() def test_missing_foreground_binding_is_rejected_by_bootstrap_and_both_verifiers(self): lifecycle_process = RetainedProcess(self.unbound_command) with mock.patch.object( lifecycle_authority, 'verify_retained_process', return_value=lifecycle_process, ): with self.assertRaisesRegex( lifecycle_authority.LifecycleAuthorityError, 'bound supervisor script', ): lifecycle_authority._verify_supervisor_process(self._lifecycle_metadata()) self.assertTrue(lifecycle_process.closed) with self.assertRaisesRegex( supervisor_instance.InstanceMetadataError, 'bound supervisor script', ): self._verify_with_supervisor_instance(self.unbound_command) namespace = runpy.run_path(str(APP_DIR / 'runtime_bootstrap.py')) with self.assertRaisesRegex(RuntimeError, 'explicit bootstrap entrypoint binding'): namespace['_require_supervisor_entrypoint_binding']([], self.supervisor_path) with tempfile.TemporaryDirectory() as temp_dir: app_dir = Path(temp_dir) / 'app' app_dir.mkdir() for name in ('child_bootstrap.py', 'runtime_bootstrap.py'): shutil.copy2(APP_DIR / name, app_dir / name) marker = Path(temp_dir) / 'payload.marker' (app_dir / 'supervisor.py').write_text( 'import os\nopen(os.environ["PAYLOAD_MARKER"], "w", encoding="ascii").write("executed")\n', encoding='ascii', ) completed = subprocess.run( [ sys.executable, '-I', '-S', '-B', str(app_dir / 'runtime_bootstrap.py'), 'supervisor', '--', ], cwd=temp_dir, env={**os.environ, 'PAYLOAD_MARKER': str(marker)}, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=20, check=False, ) self.assertNotEqual(completed.returncode, 0) self.assertIn('explicit bootstrap entrypoint binding', completed.stdout) self.assertFalse(marker.exists()) def test_canonical_powershell_invocations_pass_the_explicit_binding(self): for path in (ROOT / 'start_runtime.ps1', ROOT / 'stop_runtime.ps1'): content = path.read_text(encoding='ascii') self.assertIn('--runtime-bootstrap-entrypoint $Supervisor', content) class LifecycleModeExclusionTests(unittest.TestCase): def test_all_lifecycle_action_modes_are_pairwise_mutually_exclusive(self): modes = ( ('--dry-run',), ('--background',), ('--stop-background',), ('--background-status',), ('--attach',), ('--cmd', 'status'), ) lifecycle_names = ( 'start_background', 'stop_background', 'background_status', 'attach_background', 'send_background_command', ) for first, second in itertools.combinations(modes, 2): with self.subTest(first=first[0], second=second[0]), mock.patch.object( sys, 'argv', ['supervisor.py', *first, *second], ), mock.patch.multiple( supervisor, **{name: mock.DEFAULT for name in lifecycle_names}, ) as actions: with self.assertRaises(SystemExit) as raised: supervisor.main() self.assertEqual(raised.exception.code, 2) for action in actions.values(): action.assert_not_called() self.assertTrue(supervisor._preimport_runtime_launch_requested(['--dry-run', '--background'])) if __name__ == '__main__': unittest.main()