truf-server/app/runtime_bootstrap.py

"""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 <supervisor|postgres-runtime|migrate-runtime-safety> -- <args>')
    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