sashatrask
358 lines
13 KiB
Python
358 lines
13 KiB
Python
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()
|