import ast import json import os from pathlib import Path import shutil import subprocess import sys import tempfile import unittest sys.dont_write_bytecode = True ROOT = Path(__file__).resolve().parents[1] APP_DIR = ROOT / 'app' sys.path.insert(0, str(APP_DIR)) from runtime_security import ensure_private_directory, harden_private_file DIRECT_ENTRYPOINTS = { 'audit_github_tokens.py', 'child_bootstrap.py', 'console_runner.py', 'container_runtime.py', 'dashboard.py', 'docker_depth_operator.py', 'docker_depth_report.py', 'docker_shadow.py', 'keycheck_accounting_smoke.py', 'keycheck_runner.py', 'janitor.py', 'jsonl_projector.py', 'result_ingester.py', 'keycheckers/anthropic/anthropicKeycheck.py', 'keycheckers/aws/awsKeycheck.py', 'keycheckers/azure/azureKeycheck.py', 'keycheckers/deepseek/deepseekKeycheck.py', 'keycheckers/dockerhub/dockerhubKeycheck.py', 'keycheckers/gcp/gcpKeycheck.py', 'keycheckers/gemini/geminiKeycheck.py', 'keycheckers/github/githubKeycheck.py', 'keycheckers/gitlab/gitlabKeycheck.py', 'keycheckers/groq/groqKeycheck.py', 'keycheckers/huggingface/huggingfaceKeycheck.py', 'keycheckers/kimi/kimiKeycheck.py', 'keycheckers/openai/Keycheck.py', 'keycheckers/openrouter/OpenrouterKeycheck.py', 'keycheckers/provider_resolver/providerResolverKeycheck.py', 'keycheckers/qwen/qwenKeycheck.py', 'keycheckers/replicate/replicateKeycheck.py', 'keycheckers/xai/xaiKeycheck.py', 'keycheckers/zai/zaiKeycheck.py', 'migrate_layout.py', 'migrate_observability_db.py', 'migrate_runtime_safety.py', 'optimize_dashboard_db.py', 'owned_process.py', 'postgres_runtime.py', 'remote_worker_bootstrap.py', 'remote_worker_client.py', 'runtime_bootstrap.py', 'scanner_error_policy_smoke.py', 'supervisor.py', 'sync_alive_github_tokens.py', 'worker_api.py', 'worker_cli.py', 'worker_package_builder.py', } CORE_ENTRYPOINTS = { 'child_bootstrap.py', 'console_runner.py', 'container_runtime.py', 'dashboard.py', 'docker_shadow.py', 'keycheck_runner.py', 'janitor.py', 'jsonl_projector.py', 'result_ingester.py', 'owned_process.py', 'postgres_runtime.py', 'remote_worker_bootstrap.py', 'remote_worker_client.py', 'runtime_bootstrap.py', 'supervisor.py', 'worker_api.py', 'worker_cli.py', } def _is_main_name(node): return isinstance(node, ast.Name) and node.id == '__name__' def _is_main_value(node): return isinstance(node, ast.Constant) and node.value == '__main__' def _has_main_guard(tree): for node in ast.walk(tree): if not isinstance(node, ast.Compare) or len(node.ops) != 1 or not isinstance(node.ops[0], ast.Eq): continue if len(node.comparators) != 1: continue right = node.comparators[0] if (_is_main_name(node.left) and _is_main_value(right)) or ( _is_main_value(node.left) and _is_main_name(right) ): return True return False def _bytecode_assignment_line(tree): for node in tree.body: if not isinstance(node, ast.Assign) or not isinstance(node.value, ast.Constant) or node.value.value is not True: continue for target in node.targets: if ( isinstance(target, ast.Attribute) and target.attr == 'dont_write_bytecode' and isinstance(target.value, ast.Name) and target.value.id == 'sys' ): return node.lineno return None def _local_import_lines(tree, local_names): lines = [] for node in tree.body: if isinstance(node, ast.Import): if any(alias.name.partition('.')[0] in local_names for alias in node.names): lines.append(node.lineno) elif isinstance(node, ast.ImportFrom): root = (node.module or '').partition('.')[0] if node.level or root in local_names: lines.append(node.lineno) return lines def _has_fail_closed_check(tree, assignment_line): for node in tree.body: if not isinstance(node, ast.If) or node.lineno <= assignment_line: continue checks_policy = any( isinstance(item, ast.Attribute) and item.attr == 'dont_write_bytecode' and isinstance(item.value, ast.Name) and item.value.id == 'sys' for item in ast.walk(node.test) ) if checks_policy and any(isinstance(item, ast.Raise) for item in ast.walk(node)): return True return False class DirectEntrypointPolicyTests(unittest.TestCase): def test_inventory_sets_bytecode_policy_before_local_imports(self): parsed = {} discovered = set() for path in APP_DIR.rglob('*.py'): tree = ast.parse(path.read_text(encoding='utf-8')) relative = path.relative_to(APP_DIR).as_posix() parsed[relative] = tree if _has_main_guard(tree): discovered.add(relative) self.assertEqual(discovered, DIRECT_ENTRYPOINTS) local_names = {path.stem for path in APP_DIR.rglob('*.py')} local_names.update(path.name for path in APP_DIR.iterdir() if path.is_dir()) for relative in sorted(discovered): with self.subTest(entrypoint=relative): tree = parsed[relative] assignment_line = _bytecode_assignment_line(tree) self.assertIsNotNone(assignment_line) local_imports = _local_import_lines(tree, local_names) if local_imports: self.assertLess(assignment_line, min(local_imports)) if relative in CORE_ENTRYPOINTS: self.assertTrue(_has_fail_closed_check(tree, assignment_line)) def test_supported_direct_commands_create_no_application_bytecode_without_dash_b(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) app_dir = root / 'app' shutil.copytree( APP_DIR, app_dir, ignore=shutil.ignore_patterns('__pycache__', '*.pyc'), ) authority_runtime = root / 'runtime' ensure_private_directory(str(authority_runtime), reject_reparse=True) authority_files = [authority_runtime / 'check-openrouter-keys.ps1'] shutil.copy2(ROOT / 'runtime' / 'check-openrouter-keys.ps1', authority_files[0]) git_dir = authority_runtime / 'git' / 'cmd' ensure_private_directory(str(git_dir), reject_reparse=True) git_executable = git_dir / 'git.exe' git_executable.write_bytes(b'fixture Git executable') authority_files.append(git_executable) for name in ('start_runtime.ps1', 'stop_runtime.ps1'): target = root / name shutil.copy2(ROOT / name, target) authority_files.append(target) for path in authority_files: harden_private_file(str(path)) data_dir = root / 'fixture-data' postgres_dir = data_dir / 'postgres' ensure_private_directory(str(data_dir), reject_reparse=True) ensure_private_directory(str(postgres_dir), reject_reparse=True) executable = data_dir / ('trufflehog.exe' if os.name == 'nt' else 'trufflehog') executable.write_bytes(b'fixture') harden_private_file(str(executable)) state_file = data_dir / 'runner-state.json' state_file.write_text('{"version": 1, "sources": {}}\n', encoding='ascii') harden_private_file(str(state_file)) common_directory = str(data_dir) bundle_directory = data_dir / 'bundles' for path in ( bundle_directory, bundle_directory / 'tmp', bundle_directory / 'ready', bundle_directory / 'quarantine', ): ensure_private_directory(str(path), reject_reparse=True) config = { 'global': { 'root_dir': common_directory, 'project_dir': common_directory, 'runtime_dir': common_directory, 'result_spool_dir': common_directory, 'result_bundle_dir': str(bundle_directory), 'results_dir': common_directory, 'queue_dir': common_directory, 'state_dir': common_directory, 'log_dir': common_directory, 'control_dir': common_directory, 'keycheck_dir': common_directory, 'postman_cache_dir': common_directory, 'gharchive_cache_dir': common_directory, 'work_dir': common_directory, 'state_file': str(state_file), 'trufflehog_path': str(executable), }, 'sources': {}, } config_path = root / 'read-only-config.yaml' config_path.write_text(json.dumps(config), encoding='ascii') harden_private_file(str(config_path)) environment = os.environ.copy() for key in list(environment): normalized = key.upper() if normalized.startswith('TRUF_SUPERVISOR_') or normalized in { 'SCANNER_SUPERVISED', 'TRUF_MANAGED_POSTGRES_DSN', 'SCANNER_DB_URL', 'DATABASE_URL', 'SCANNER_DASHBOARD_DB_URL', 'KEYCHECK_DB_URL', 'PYTHONDONTWRITEBYTECODE', 'PYTHONPYCACHEPREFIX', 'PYTHONPATH', }: environment.pop(key, None) commands = ( ( 'migration help', [sys.executable, str(app_dir / 'migrate_runtime_safety.py'), '--help'], 0, 'usage:', ), ( 'keycheck print plan', [ sys.executable, str(app_dir / 'keycheck_runner.py'), '--config', str(config_path), '--service', 'github', '--no-summary', '--print-plan', ], 0, 'mode: run-keychecks', ), ( 'console show state', [ sys.executable, str(app_dir / 'console_runner.py'), '--config', str(config_path), '--show-state', ], 0, 'State file:', ), ( 'provider authority rejection', [ sys.executable, str(app_dir / 'keycheckers' / 'github' / 'githubKeycheck.py'), '--input', str(root / 'missing.jsonl'), ], 1, 'direct mutation is retired', ), ) for name, command, expected_code, expected_output in commands: with self.subTest(command=name): self.assertNotIn('-B', command) completed = subprocess.run( command, cwd=root, env=environment, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=30, check=False, ) if expected_code == 0: self.assertEqual(completed.returncode, 0, completed.stdout) else: self.assertNotEqual(completed.returncode, 0, completed.stdout) self.assertIn(expected_output, completed.stdout) self.assertEqual(list(app_dir.rglob('*.pyc')), []) self.assertEqual( [path for path in app_dir.rglob('__pycache__') if path.is_dir()], [], ) if __name__ == '__main__': unittest.main()