"""Stdlib-only pre-import boundary for canonical runtime entrypoints.""" import sys import os if __name__ == '__main__': if sys.platform != 'linux' or not os.path.isfile('/.dockerenv') or os.path.abspath(__file__) != '/opt/truf/app/runtime_bootstrap.py': raise SystemExit('Docker development copy: runtime control is disabled outside the prepared container. See DOCKER_MIGRATION.md.') import runpy runpy.run_path('/opt/truf/app/container_runtime.py')['require_container']() sys.dont_write_bytecode = True if not sys.dont_write_bytecode: raise RuntimeError('runtime bootstrap could not disable bytecode writes') import runpy import stat RUNTIME_BOOTSTRAP_ENV = 'TRUF_RUNTIME_BOOTSTRAP' RUNTIME_BOOTSTRAP_VALUE = '1' APPLICATION_IMPORT_SUFFIXES = ('.py', '.pyw', '.pyc', '.pyd') SUPERVISOR_ENTRYPOINT_FLAG = '--runtime-bootstrap-entrypoint' TARGETS = { 'supervisor': 'supervisor.py', 'postgres-runtime': 'postgres_runtime.py', 'migrate-runtime-safety': 'migrate_runtime_safety.py', } def _require_isolated_startup(): if not ( sys.flags.isolated and sys.flags.no_site and sys.flags.dont_write_bytecode and sys.dont_write_bytecode ): raise RuntimeError('runtime bootstrap requires isolated no-site bytecode-free startup (-I -S -B)') def _canonical(path): return os.path.normcase(os.path.realpath(os.path.abspath(os.fspath(path)))) def _is_reparse_point(path): details = os.lstat(path) if stat.S_ISLNK(details.st_mode): return True attributes = getattr(details, 'st_file_attributes', 0) reparse_attribute = getattr(stat, 'FILE_ATTRIBUTE_REPARSE_POINT', 0) return bool(attributes & reparse_attribute) or getattr(os.path, 'isjunction', lambda _path: False)(path) def _reject_cached_bytecode(app_dir): def raise_walk_error(exc): raise RuntimeError(f'unable to inspect the application root: {exc}') from exc try: root_details = os.lstat(app_dir) except OSError as exc: raise RuntimeError(f'application root is unavailable: {app_dir}') from exc if _is_reparse_point(app_dir): raise RuntimeError(f'application root reparse point is forbidden: {app_dir}') if not stat.S_ISDIR(root_details.st_mode): raise RuntimeError(f'application root is not a directory: {app_dir}') canonical_root = _canonical(app_dir) for current, directories, files in os.walk(app_dir, followlinks=False, onerror=raise_walk_error): for name in directories: candidate = os.path.join(current, name) if _is_reparse_point(candidate): relative = os.path.relpath(candidate, app_dir).replace(os.sep, '/') if name.lower() == '__pycache__': raise RuntimeError(f'application __pycache__ link is forbidden: {relative}') raise RuntimeError(f'application directory reparse point is forbidden: {relative}') relative_current = os.path.relpath(current, app_dir) in_cache = any(part.lower() == '__pycache__' for part in relative_current.split(os.sep)) for name in files: candidate = os.path.join(current, name) relative = os.path.relpath(candidate, app_dir).replace(os.sep, '/') if _is_reparse_point(candidate): raise RuntimeError(f'application file reparse point is forbidden: {relative}') if name.lower().endswith(APPLICATION_IMPORT_SUFFIXES): try: contained = os.path.commonpath((canonical_root, _canonical(candidate))) == canonical_root except ValueError: contained = False if not contained: raise RuntimeError(f'application Python authority escapes its root: {relative}') if in_cache and name.lower().endswith('.pyc'): raise RuntimeError(f'application __pycache__ bytecode is forbidden: {relative}') def _require_supervisor_entrypoint_binding(arguments, entrypoint): bindings = [] for index, argument in enumerate(arguments): text = str(argument) if text == SUPERVISOR_ENTRYPOINT_FLAG: if index + 1 >= len(arguments): raise RuntimeError('supervisor runtime bootstrap entrypoint binding has no path') bindings.append(str(arguments[index + 1])) elif text.startswith(SUPERVISOR_ENTRYPOINT_FLAG + '='): bindings.append(text.split('=', 1)[1]) if len(bindings) != 1: raise RuntimeError('supervisor runtime requires exactly one explicit bootstrap entrypoint binding') binding = bindings[0] if not os.path.isabs(binding) or _canonical(binding) != _canonical(entrypoint): raise RuntimeError('supervisor runtime bootstrap entrypoint binding is not canonical supervisor.py') def main(): _require_isolated_startup() if len(sys.argv) < 3 or sys.argv[2] != '--': raise RuntimeError('usage: runtime_bootstrap.py -- ') target_name = str(sys.argv[1]).strip().lower() target_file = TARGETS.get(target_name) if not target_file: raise RuntimeError(f'unsupported canonical runtime target: {target_name}') app_dir = os.path.dirname(os.path.abspath(__file__)) _reject_cached_bytecode(app_dir) entrypoint = _canonical(os.path.join(app_dir, target_file)) arguments = list(sys.argv[3:]) if target_name == 'supervisor': _require_supervisor_entrypoint_binding(arguments, entrypoint) child_namespace = runpy.run_path(os.path.join(app_dir, 'child_bootstrap.py')) enable_dependencies = child_namespace.get('_enable_dependency_paths') if not callable(enable_dependencies): raise RuntimeError('authenticated dependency path bootstrap is unavailable') enable_dependencies(target_name) os.environ[RUNTIME_BOOTSTRAP_ENV] = RUNTIME_BOOTSTRAP_VALUE sys.path.insert(0, app_dir) sys.argv = [entrypoint, *arguments] runpy.run_path(entrypoint, run_name='__main__') if __name__ == '__main__': try: main() except Exception as exc: raise SystemExit(f'canonical runtime bootstrap rejected launch: {exc}') from exc