truf-server/app/lifecycle_authority.py

import hashlib
import hmac
import json
import os
import shutil
import socket
import stat

from db_backend import parse_postgres_url
from process_identity import verify_retained_process
from runtime_security import (
    canonical_path,
    private_file_ready,
    read_private_json,
    reject_reparse_components,
    require_trusted_native_executable,
    sha256_file,
)


CODE_MANIFEST_SCHEMA = 5
APPLICATION_IMPORT_SUFFIXES = ('.py', '.pyw', '.pyc', '.pyd')
CONTROL_SCHEMA = 1
DISCOVERY_PRODUCER_ROLE = 'discovery-producer'
DISCOVERY_PRODUCER_SOURCES = ('gitlab', 'dockerhub', 'huggingface')
PHASE_INACTIVE = 'INACTIVE'
PHASE_ACTIVATING = 'ACTIVATING'
PHASE_ACTIVE = 'ACTIVE'
PHASE_STOPPING = 'STOPPING'
PHASE_FAILED_HOLD = 'FAILED_HOLD'
LIFECYCLE_PHASES = {
    PHASE_INACTIVE,
    PHASE_ACTIVATING,
    PHASE_ACTIVE,
    PHASE_STOPPING,
    PHASE_FAILED_HOLD,
}

# These files collectively decide process ownership, database authority, and
# what data may be launched or persisted by the supervisor.
CODE_AUTHORITY_FILES = (
    'owned_process.py',
    'supervisor.py',
    'supervisor_instance.py',
    'console_runner.py',
    'scanner.py',
    'docker_shadow.py',
    'keycheck_runner.py',
    'dashboard.py',
    'postgres_runtime.py',
    'process_identity.py',
    'runtime_security.py',
    'scanner_db.py',
    'db_backend.py',
    'result_spool.py',
    'janitor.py',
    'result_bundle.py',
    'result_ingester.py',
    'jsonl_projector.py',
    'admin_api.py',
    'worker_api.py',
    'worker_assignment.py',
    'worker_package.py',
    'scan_execution.py',
    'keycheck_candidates.py',
    'paths.py',
    'target_identity.py',
    'lifecycle_authority.py',
    'audit_github_tokens.py',
    'sync_alive_github_tokens.py',
    'child_bootstrap.py',
    'runtime_bootstrap.py',
    'runtime_document.py',
    'capacity_model.py',
    'runtime_document_io.py',
    'managed_files.py',
    'host_agent_client.py',
    'host_agent_protocol.py',
    'host_agent_reconcile.py',
    'host_agent_server.py',
    'host_agent_apply.py',
    'host_agent_lifecycle.py',
    'host_agent_runtime.py',
    'host_agent_state.py',
    'worker_contracts.py',
    'worker_assignment_runner.py',
)
REMOTE_WORKER_CODE_AUTHORITY_FILES = (
    'db_backend.py',
    'docker_depth_experiment.py',
    'janitor.py',
    'keycheck_candidates.py',
    'lifecycle_authority.py',
    'owned_process.py',
    'paths.py',
    'process_identity.py',
    'query_policy.py',
    'remote_worker_bootstrap.py',
    'remote_worker_client.py',
    'result_bundle.py',
    'result_spool.py',
    'runtime_security.py',
    'scan_execution.py',
    'scanner.py',
    'scanner_db.py',
    'supervisor_instance.py',
    'target_identity.py',
    'worker_contracts.py',
    'worker_assignment_runner.py',
    'worker_cli.py',
    'worker_local_state.py',
    'worker_supervisor.py',
    'worker_package.py',
)
EXTERNAL_CODE_AUTHORITY_FILES = (
    '../runtime/check-openrouter-keys.ps1',
    '../start_core_runtime.ps1',
    '../start_runtime.ps1',
    '../stop_runtime.ps1',
) if os.name == 'nt' else ()
# The launchers execute before a manifest can be captured. First-launch trust
# therefore requires offline ACL hardening; manifests only detect later drift.
TRUFFLEHOG_MANIFEST_NAME = 'trufflehog'
GIT_MANIFEST_NAME = 'git'

CHILD_INSTANCE_FILE_ENV = 'TRUF_SUPERVISOR_INSTANCE_FILE'
CHILD_INSTANCE_ID_ENV = 'TRUF_SUPERVISOR_INSTANCE_ID'
CHILD_TOKEN_ENV = 'TRUF_SUPERVISOR_TOKEN'
CHILD_CONFIG_HASH_ENV = 'TRUF_SUPERVISOR_CONFIG_SHA256'
CHILD_SCRIPT_HASH_ENV = 'TRUF_SUPERVISOR_SHA256'
CHILD_MANIFEST_HASH_ENV = 'TRUF_SUPERVISOR_CODE_MANIFEST_SHA256'
CHILD_DSN_HASH_ENV = 'TRUF_SUPERVISOR_DSN_SHA256'
CHILD_KIND_ENV = 'TRUF_SUPERVISOR_CHILD_KIND'
PRIVATE_CHILD_ENV_KEYS = (
    CHILD_INSTANCE_FILE_ENV,
    CHILD_INSTANCE_ID_ENV,
    CHILD_TOKEN_ENV,
    CHILD_CONFIG_HASH_ENV,
    CHILD_SCRIPT_HASH_ENV,
    CHILD_MANIFEST_HASH_ENV,
    CHILD_DSN_HASH_ENV,
    CHILD_KIND_ENV,
    'SCANNER_SUPERVISED',
    'TRUF_MANAGED_POSTGRES_DSN',
    'SCANNER_DB_URL',
    'DATABASE_URL',
    'SCANNER_DASHBOARD_DB_URL',
    'KEYCHECK_DB_URL',
    'TRUF_DASHBOARD_CANONICAL_LAUNCH',
    'TRUF_DASHBOARD_HOST',
)
_LIBPQ_PRIVATE_ENV_KEYS = frozenset({
    'PGPASSWORD',
    'PGUSER',
    'PGDATABASE',
    'PGHOST',
    'PGHOSTADDR',
    'PGPORT',
    'PGSERVICE',
    'PGSERVICEFILE',
    'PGPASSFILE',
    'PGOPTIONS',
    'PGSSLMODE',
    'PGSSLKEY',
    'PGSSLCERT',
    'PGSSLROOTCERT',
})
_PRIVATE_EXTERNAL_ENV_KEYS = frozenset(PRIVATE_CHILD_ENV_KEYS) | _LIBPQ_PRIVATE_ENV_KEYS


class LifecycleAuthorityError(ValueError):
    pass


def _manifest_payload(manifest):
    return json.dumps(
        manifest,
        ensure_ascii=True,
        sort_keys=True,
        separators=(',', ':'),
    ).encode('utf-8')


def code_manifest_sha256(manifest):
    return hashlib.sha256(_manifest_payload(manifest)).hexdigest()


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 _application_root(app_dir=None):
    candidate = os.path.abspath(os.fspath(app_dir or os.path.dirname(os.path.abspath(__file__))))
    try:
        details = os.lstat(candidate)
        if _is_reparse_point(candidate):
            raise LifecycleAuthorityError(f'application root reparse point is forbidden: {candidate}')
    except OSError as exc:
        raise LifecycleAuthorityError(f'application root is unavailable: {candidate}') from exc
    if not stat.S_ISDIR(details.st_mode):
        raise LifecycleAuthorityError(f'application root is not a directory: {candidate}')
    return canonical_path(candidate)


def _external_authority_expected_path(root, name):
    if name not in EXTERNAL_CODE_AUTHORITY_FILES:
        raise LifecycleAuthorityError(f'code authority path is not an allowed external: {name}')
    project_root = os.path.normcase(os.path.abspath(os.path.dirname(root)))
    candidate = os.path.normcase(os.path.abspath(os.path.join(root, *name.split('/'))))
    try:
        contained = candidate != project_root and os.path.commonpath((project_root, candidate)) == project_root
    except ValueError:
        contained = False
    if not contained:
        raise LifecycleAuthorityError(f'external code authority path escapes the project root: {name}')
    return candidate


def _require_external_authority_file(root, name):
    path = _external_authority_expected_path(root, name)
    try:
        reject_reparse_components(path)
    except OSError as exc:
        raise LifecycleAuthorityError(f'external code authority reparse point is forbidden: {name}') from exc
    try:
        details = os.stat(path, follow_symlinks=False)
    except OSError as exc:
        raise LifecycleAuthorityError(f'required external code authority file is absent: {name}') from exc
    if not stat.S_ISREG(details.st_mode):
        raise LifecycleAuthorityError(f'external code authority file is not regular: {name}')
    if canonical_path(path) != path:
        raise LifecycleAuthorityError(f'external code authority path is not exact: {name}')
    return path


def _application_code_files(root):
    """Return the exact importable application code surface."""
    names = []

    def raise_walk_error(exc):
        raise LifecycleAuthorityError(f'unable to inspect the application root: {exc}') from exc

    for current, directories, files in os.walk(root, followlinks=False, onerror=raise_walk_error):
        for name in directories:
            candidate = os.path.join(current, name)
            try:
                linked = _is_reparse_point(candidate)
            except OSError as exc:
                raise LifecycleAuthorityError(f'unable to inspect application directory: {candidate}') from exc
            if linked:
                relative = os.path.relpath(candidate, root).replace(os.sep, '/')
                raise LifecycleAuthorityError(f'application directory reparse point is forbidden: {relative}')
        relative_current = os.path.relpath(current, root)
        in_cache = any(part.lower() == '__pycache__' for part in relative_current.split(os.sep))
        suffixes = ('.pyc',) if in_cache else APPLICATION_IMPORT_SUFFIXES
        for name in files:
            source_path = os.path.abspath(os.path.join(current, name))
            try:
                linked = _is_reparse_point(source_path)
            except OSError as exc:
                raise LifecycleAuthorityError(f'unable to inspect application file: {source_path}') from exc
            if linked:
                relative = os.path.relpath(source_path, root).replace(os.sep, '/')
                raise LifecycleAuthorityError(f'application file reparse point is forbidden: {relative}')
            if not name.lower().endswith(suffixes):
                continue
            path = canonical_path(source_path)
            try:
                if os.path.commonpath((root, path)) != root:
                    raise LifecycleAuthorityError(f'code authority path escapes the application root: {path}')
            except ValueError as exc:
                raise LifecycleAuthorityError(f'code authority path escapes the application root: {path}') from exc
            names.append(os.path.relpath(source_path, root).replace(os.sep, '/'))
    return sorted(names)


def code_authority_file_names(app_dir=None, existing_only=False):
    root = _application_root(app_dir)
    names = list(_application_code_files(root))
    for name in (*CODE_AUTHORITY_FILES, *EXTERNAL_CODE_AUTHORITY_FILES):
        path = canonical_path(os.path.join(root, name))
        if not existing_only or os.path.isfile(path):
            names.append(name)
    return tuple(dict.fromkeys(names))


def resolve_manifest_executable(value, *, name=TRUFFLEHOG_MANIFEST_NAME, app_dir=None):
    if name not in {TRUFFLEHOG_MANIFEST_NAME, GIT_MANIFEST_NAME}:
        raise LifecycleAuthorityError('unsupported manifested executable')
    label = 'TruffleHog' if name == TRUFFLEHOG_MANIFEST_NAME else 'Git'
    text = str(value or '').strip()
    if not text:
        if name == GIT_MANIFEST_NAME:
            text = 'git'
            if os.name == 'nt':
                private_git = os.path.join(os.path.dirname(_application_root(app_dir)), 'runtime', 'git', 'cmd', 'git.exe')
                try:
                    reject_reparse_components(private_git)
                except OSError as exc:
                    raise LifecycleAuthorityError('project-private Git path contains a reparse point') from exc
                text = private_git if os.path.lexists(private_git) else text
        else:
            from paths import default_trufflehog_path

            text = default_trufflehog_path()
    candidate = shutil.which(text) if not os.path.isabs(text) and not any(sep in text for sep in ('/', '\\')) else text
    if not candidate:
        raise LifecycleAuthorityError(f'configured {label} executable is unavailable: {text}')
    if os.name != 'nt':
        if not os.path.isabs(candidate) or candidate != os.path.normpath(candidate):
            raise LifecycleAuthorityError(f'configured {label} executable path must be exact and absolute: {candidate}')
        try:
            reject_reparse_components(candidate)
        except (OSError, ValueError) as exc:
            raise LifecycleAuthorityError(f'configured {label} executable path is unsafe: {candidate}') from exc
    path = canonical_path(candidate)
    if not os.path.isfile(path):
        raise LifecycleAuthorityError(f'configured {label} executable is not a regular file: {path}')
    return path


def manifest_authority_paths(
    app_dir=None, trufflehog_path=None, policy_paths=None, existing_only=False, *,
    git_path=None, include_executables=True,
):
    """List authority paths; exclude executables when applying private-file policy."""
    root = _application_root(app_dir)
    paths = []
    names = list(code_authority_file_names(root, existing_only=existing_only))
    # Required externals may never disappear from read-only preflight or an
    # offline hardening plan, even when optional paths use existing_only.
    names.extend(name for name in EXTERNAL_CODE_AUTHORITY_FILES if name not in names)
    for name in names:
        if name in EXTERNAL_CODE_AUTHORITY_FILES:
            paths.append(_require_external_authority_file(root, name))
            continue
        path = canonical_path(os.path.join(root, *name.split('/')))
        if os.path.isfile(path):
            paths.append(path)
        elif not existing_only:
            raise LifecycleAuthorityError(f'code authority file is absent or outside the application root: {name}')
    if include_executables:
        if trufflehog_path or not existing_only:
            try:
                paths.append(resolve_manifest_executable(trufflehog_path))
            except LifecycleAuthorityError:
                if not existing_only or trufflehog_path:
                    raise
        # Git is required even in preflight/offline hardening's existing-only mode.
        paths.append(resolve_manifest_executable(git_path, name=GIT_MANIFEST_NAME, app_dir=root))
    for value in policy_paths or ():
        if not value:
            continue
        path = canonical_path(value)
        if os.path.isfile(path):
            paths.append(path)
        elif not existing_only:
            raise LifecycleAuthorityError(f'configured policy authority file is unavailable: {path}')
    return tuple(dict.fromkeys(paths))


def build_code_manifest(
    app_dir=None, trufflehog_path=None, policy_paths=None, *, git_path=None,
    include_trufflehog=True,
):
    root = _application_root(app_dir)
    files = {}
    for name in code_authority_file_names(root):
        path = (
            _require_external_authority_file(root, name)
            if name in EXTERNAL_CODE_AUTHORITY_FILES
            else canonical_path(os.path.join(root, *name.split('/')))
        )
        try:
            contained = os.path.commonpath((root, path)) == root
        except ValueError:
            contained = False
        explicitly_external = (
            name in EXTERNAL_CODE_AUTHORITY_FILES
            and path == _external_authority_expected_path(root, name)
        )
        if (not contained and not explicitly_external) or not os.path.isfile(path):
            raise LifecycleAuthorityError(f'code authority file is absent or outside the application root: {name}')
        files[name] = {'path': path, 'sha256': sha256_file(path)}
    git_executable = resolve_manifest_executable(git_path, name=GIT_MANIFEST_NAME, app_dir=root)
    executables = {
        GIT_MANIFEST_NAME: {'path': git_executable, 'sha256': sha256_file(git_executable)},
    }
    if include_trufflehog:
        executable = resolve_manifest_executable(trufflehog_path)
        executables[TRUFFLEHOG_MANIFEST_NAME] = {
            'path': executable,
            'sha256': sha256_file(executable),
        }
    assets = {}
    for value in policy_paths or ():
        if not value:
            continue
        path = canonical_path(value)
        if not os.path.isfile(path):
            raise LifecycleAuthorityError(f'configured policy authority file is unavailable: {path}')
        assets[path] = {'path': path, 'sha256': sha256_file(path)}
    return {
        'schema': CODE_MANIFEST_SCHEMA,
        'root': root,
        'files': files,
        'executables': executables,
        'assets': assets,
    }


def normalize_code_manifest(manifest, *, required_names=None, external_names=None):
    if not isinstance(manifest, dict) or manifest.get('schema') != CODE_MANIFEST_SCHEMA:
        raise LifecycleAuthorityError('unsupported code authority manifest schema')
    root_value = manifest.get('root') or ''
    root = _application_root(root_value) if root_value else ''
    values = manifest.get('files')
    expected_names = set(values) if isinstance(values, dict) else set()
    external_names = set(
        EXTERNAL_CODE_AUTHORITY_FILES if external_names is None else external_names
    )
    if not external_names <= set(EXTERNAL_CODE_AUTHORITY_FILES):
        raise LifecycleAuthorityError('code authority manifest has unsupported external files')
    required_names = set(CODE_AUTHORITY_FILES if required_names is None else required_names)
    required_names.update(external_names)
    if not root or not isinstance(values, dict) or not required_names.issubset(expected_names):
        raise LifecycleAuthorityError('code authority manifest has an incomplete file set')
    files = {}
    for name in sorted(expected_names):
        value = values.get(name)
        if not isinstance(value, dict):
            raise LifecycleAuthorityError(f'invalid code authority entry: {name}')
        if name in external_names:
            path = os.path.normcase(os.path.abspath(os.fspath(value.get('path') or '')))
            expected_path = _external_authority_expected_path(root, name)
        else:
            path = canonical_path(value.get('path') or '')
            expected_path = canonical_path(os.path.join(root, *name.split('/')))
            try:
                contained = os.path.commonpath((root, expected_path)) == root
            except ValueError:
                contained = False
            if not contained:
                raise LifecycleAuthorityError(f'code authority path is not an allowed external: {name}')
        digest = str(value.get('sha256') or '')
        if path != expected_path:
            raise LifecycleAuthorityError(f'code authority path mismatch: {name}')
        if len(digest) != 64 or any(ch not in '0123456789abcdef' for ch in digest):
            raise LifecycleAuthorityError(f'invalid code authority digest: {name}')
        files[name] = {'path': path, 'sha256': digest}
    executables = manifest.get('executables')
    executable_names = set(executables) if isinstance(executables, dict) else set()
    if executable_names not in (
        {GIT_MANIFEST_NAME},
        {TRUFFLEHOG_MANIFEST_NAME, GIT_MANIFEST_NAME},
    ):
        raise LifecycleAuthorityError('code authority manifest has an incomplete executable set')
    normalized_executables = {}
    for name, label in ((TRUFFLEHOG_MANIFEST_NAME, 'TruffleHog'), (GIT_MANIFEST_NAME, 'Git')):
        if name not in executables:
            continue
        executable = executables[name]
        if not isinstance(executable, dict):
            raise LifecycleAuthorityError(f'invalid {label} authority entry')
        raw_path = executable.get('path')
        executable_digest = str(executable.get('sha256') or '')
        if not isinstance(raw_path, str) or not os.path.isabs(raw_path) or '\x00' in raw_path or len(executable_digest) != 64 or any(ch not in '0123456789abcdef' for ch in executable_digest):
            raise LifecycleAuthorityError(f'invalid {label} authority identity')
        executable_path = canonical_path(raw_path)
        if os.name != 'nt' and executable_path != raw_path:
            raise LifecycleAuthorityError(f'{label} authority path is not exact')
        normalized_executables[name] = {'path': executable_path, 'sha256': executable_digest}
    assets_value = manifest.get('assets')
    if not isinstance(assets_value, dict):
        raise LifecycleAuthorityError('code authority manifest has an invalid asset set')
    assets = {}
    for name in sorted(assets_value):
        value = assets_value[name]
        if not isinstance(value, dict):
            raise LifecycleAuthorityError(f'invalid policy authority entry: {name}')
        path = canonical_path(value.get('path') or '')
        digest = str(value.get('sha256') or '')
        if name != path or not os.path.isabs(path) or len(digest) != 64 or any(ch not in '0123456789abcdef' for ch in digest):
            raise LifecycleAuthorityError(f'invalid policy authority identity: {name}')
        assets[name] = {'path': path, 'sha256': digest}
    return {
        'schema': CODE_MANIFEST_SCHEMA,
        'root': root,
        'files': files,
        'executables': normalized_executables,
        'assets': assets,
    }


def verify_code_manifest(
    manifest, expected_sha256=None, require_private_acl=False, *,
    required_names=None, external_names=None,
):
    normalized = normalize_code_manifest(
        manifest, required_names=required_names, external_names=external_names,
    )
    digest = code_manifest_sha256(normalized)
    if expected_sha256 and not hmac.compare_digest(digest, str(expected_sha256)):
        raise LifecycleAuthorityError('code authority manifest digest mismatch')
    for name in (EXTERNAL_CODE_AUTHORITY_FILES if external_names is None else external_names):
        if normalized['files'][name]['path'] != _require_external_authority_file(normalized['root'], name):
            raise LifecycleAuthorityError(f'code authority path mismatch: {name}')
    current_code = set(_application_code_files(normalized['root']))
    manifested_code = {
        name for name, value in normalized['files'].items()
        if name.lower().endswith(APPLICATION_IMPORT_SUFFIXES)
        and os.path.commonpath((normalized['root'], value['path'])) == normalized['root']
    }
    if current_code != manifested_code:
        added = sorted(current_code - manifested_code)
        removed = sorted(manifested_code - current_code)
        detail = added[0] if added else removed[0] if removed else 'unknown'
        raise LifecycleAuthorityError(f'application code authority file set drifted: {detail}')
    entries = [(name, value, False) for name, value in normalized['files'].items()] + [
        (f'executable:{name}', value, True) for name, value in normalized['executables'].items()
    ] + [(f'asset:{name}', value, False) for name, value in normalized['assets'].items()]
    for name, value, native in entries:
        if require_private_acl:
            if native and os.name != 'nt':
                try:
                    require_trusted_native_executable(value['path'])
                except (OSError, ValueError) as exc:
                    raise LifecycleAuthorityError(f'code authority executable is not trusted: {name}: {exc}') from exc
            elif not private_file_ready(value['path']):
                raise LifecycleAuthorityError(f'code authority ACL is not exact-private: {name}')
        try:
            current = sha256_file(value['path'])
        except OSError as exc:
            raise LifecycleAuthorityError(f'unable to verify code authority file: {name}') from exc
        if not hmac.compare_digest(current, value['sha256']):
            raise LifecycleAuthorityError(f'code authority drifted: {name}')
    return normalized


def dsn_sha256(dsn):
    value = str(dsn or '')
    return hashlib.sha256(value.encode('utf-8')).hexdigest() if value else ''


def supervised_child_environment(metadata, canonical_dsn, child_kind):
    return {
        'SCANNER_SUPERVISED': '1',
        CHILD_INSTANCE_FILE_ENV: str(metadata['instance_file']),
        CHILD_INSTANCE_ID_ENV: str(metadata['instance_id']),
        CHILD_TOKEN_ENV: str(metadata['token']),
        CHILD_CONFIG_HASH_ENV: str(metadata['config_sha256']),
        CHILD_SCRIPT_HASH_ENV: str(metadata['supervisor_sha256']),
        CHILD_MANIFEST_HASH_ENV: str(metadata['code_manifest_sha256']),
        CHILD_DSN_HASH_ENV: str(metadata.get('canonical_dsn_sha256') or ''),
        CHILD_KIND_ENV: str(child_kind),
        'TRUF_MANAGED_POSTGRES_DSN': str(canonical_dsn or ''),
    }


def strip_supervisor_credentials(env):
    for key in list(env):
        normalized = str(key).upper()
        if normalized.startswith('TRUF_POSTGRES_') or normalized in _PRIVATE_EXTERNAL_ENV_KEYS:
            env.pop(key, None)
    return env


def _send_handshake(metadata, timeout=3):
    control = metadata.get('control') or {}
    request = {
        'schema': CONTROL_SCHEMA,
        'instance_id': metadata['instance_id'],
        'token': metadata['token'],
        'action': 'handshake',
    }
    payload = json.dumps(request, ensure_ascii=True, separators=(',', ':')).encode('utf-8') + b'\n'
    with socket.create_connection((control.get('host'), int(control.get('port') or 0)), timeout=timeout) as sock:
        sock.settimeout(timeout)
        sock.sendall(payload)
        sock.shutdown(socket.SHUT_WR)
        chunks = []
        total = 0
        while True:
            chunk = sock.recv(65536)
            if not chunk:
                break
            total += len(chunk)
            if total > 1024 * 1024:
                raise LifecycleAuthorityError('supervisor handshake response is too large')
            chunks.append(chunk)
    try:
        response = json.loads(b''.join(chunks).decode('utf-8'))
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise LifecycleAuthorityError('invalid supervisor handshake response') from exc
    if (
        not isinstance(response, dict)
        or response.get('schema') != CONTROL_SCHEMA
        or response.get('instance_id') != metadata['instance_id']
        or response.get('ok') is not True
        or not isinstance(response.get('result'), dict)
    ):
        raise LifecycleAuthorityError('authenticated supervisor handshake failed')
    return response['result']


def _send_handshake_with_timeout_retry(metadata, timeout_retries=0):
    retries = min(1, max(0, int(timeout_retries or 0)))
    for attempt in range(retries + 1):
        try:
            return _send_handshake(metadata)
        except TimeoutError:
            if attempt >= retries:
                raise


def verify_supervisor_command_line(arguments, supervisor_path, config_path):
    """Verify the unique script binding and config value retained by the OS."""
    arguments = [str(argument) for argument in arguments]
    options = {
        '--runtime-bootstrap-entrypoint': [],
        '--config': [],
    }
    option_value_indices = set()
    for index, argument in enumerate(arguments):
        for option in options:
            if argument == option:
                value = arguments[index + 1] if index + 1 < len(arguments) else ''
                options[option].append(value)
                if index + 1 < len(arguments):
                    option_value_indices.add(index + 1)
            elif argument.startswith(option + '='):
                options[option].append(argument.split('=', 1)[1])

    expected_supervisor = canonical_path(supervisor_path)
    direct_bindings = []
    for index, argument in enumerate(arguments):
        if index in option_value_indices or not argument or argument.startswith('-'):
            continue
        try:
            if canonical_path(argument) == expected_supervisor:
                direct_bindings.append(argument)
        except (OSError, TypeError, ValueError):
            continue

    bindings = options['--runtime-bootstrap-entrypoint'] + direct_bindings
    if len(bindings) != 1:
        raise LifecycleAuthorityError('supervisor command line must contain exactly one bound supervisor script')
    try:
        binding_matches = canonical_path(bindings[0]) == expected_supervisor
    except (OSError, TypeError, ValueError):
        binding_matches = False
    if not binding_matches:
        raise LifecycleAuthorityError('supervisor command line bound supervisor script mismatch')

    configs = options['--config']
    if len(configs) != 1:
        raise LifecycleAuthorityError('supervisor command line must contain exactly one bound config argument')
    try:
        config_matches = canonical_path(configs[0]) == canonical_path(config_path)
    except (OSError, TypeError, ValueError):
        config_matches = False
    if not config_matches:
        raise LifecycleAuthorityError('supervisor command line bound config argument mismatch')


def _verify_supervisor_process(metadata):
    process = verify_retained_process(
        metadata['pid'],
        metadata['process_creation_time'],
        metadata['executable'],
    )
    try:
        arguments = process.command_line()
        verify_supervisor_command_line(
            arguments,
            metadata['supervisor_path'],
            metadata['config_path'],
        )
    finally:
        process.close()


def require_active_supervisor_child(
    config_path=None, child_kind=None, require_dsn=True, handshake_timeout_retries=0,
):
    """Authenticate a mutating child before it reads application inputs."""
    instance_file = os.getenv(CHILD_INSTANCE_FILE_ENV) or ''
    inherited_id = os.getenv(CHILD_INSTANCE_ID_ENV) or ''
    inherited_token = os.getenv(CHILD_TOKEN_ENV) or ''
    inherited_config_hash = os.getenv(CHILD_CONFIG_HASH_ENV) or ''
    inherited_script_hash = os.getenv(CHILD_SCRIPT_HASH_ENV) or ''
    inherited_manifest_hash = os.getenv(CHILD_MANIFEST_HASH_ENV) or ''
    inherited_dsn_hash = os.getenv(CHILD_DSN_HASH_ENV) or ''
    inherited_kind = os.getenv(CHILD_KIND_ENV) or ''
    if not all((instance_file, inherited_id, inherited_token, inherited_config_hash, inherited_script_hash, inherited_manifest_hash)):
        raise LifecycleAuthorityError('direct mutation is retired; use an authenticated active supervisor command')
    if child_kind and inherited_kind != str(child_kind):
        raise LifecycleAuthorityError('supervised child kind does not match the requested mutation entrypoint')

    # Imported lazily to avoid a module cycle while supervisor metadata support
    # itself imports the manifest helpers above.
    from supervisor_instance import load_instance_metadata

    try:
        metadata = load_instance_metadata(instance_file)
    except (OSError, ValueError) as exc:
        raise LifecycleAuthorityError('private supervisor instance metadata is unavailable') from exc
    if not hmac.compare_digest(metadata['instance_id'], inherited_id):
        raise LifecycleAuthorityError('supervisor child instance identity mismatch')
    if not hmac.compare_digest(metadata['token'], inherited_token):
        raise LifecycleAuthorityError('supervisor child credential mismatch')
    if metadata.get('activation_state') != PHASE_ACTIVE:
        raise LifecycleAuthorityError('supervisor is not ACTIVE; mutation is refused')
    if canonical_path(instance_file) != canonical_path(metadata.get('instance_file') or instance_file):
        raise LifecycleAuthorityError('supervisor child instance path mismatch')
    if config_path and canonical_path(config_path) != metadata['config_path']:
        raise LifecycleAuthorityError('supervisor child config path mismatch')

    expected_pairs = (
        ('config_sha256', inherited_config_hash),
        ('supervisor_sha256', inherited_script_hash),
        ('code_manifest_sha256', inherited_manifest_hash),
    )
    for key, inherited in expected_pairs:
        if not hmac.compare_digest(str(metadata.get(key) or ''), inherited):
            raise LifecycleAuthorityError(f'supervisor child {key} authority mismatch')
    if not hmac.compare_digest(sha256_file(metadata['config_path']), metadata['config_sha256']):
        raise LifecycleAuthorityError('supervisor config authority drifted')
    verify_code_manifest(
        metadata['code_manifest'],
        metadata['code_manifest_sha256'],
        require_private_acl=True,
    )
    _verify_supervisor_process(metadata)

    dsn = os.getenv('TRUF_MANAGED_POSTGRES_DSN') or ''
    if require_dsn:
        try:
            parsed = parse_postgres_url(dsn)
        except ValueError as exc:
            raise LifecycleAuthorityError('a canonical managed PostgreSQL DSN is required') from exc
        if parsed['host'] != '127.0.0.1':
            raise LifecycleAuthorityError('managed PostgreSQL DSN is not loopback-bound')
        actual_dsn_hash = dsn_sha256(dsn)
        if not inherited_dsn_hash or not hmac.compare_digest(actual_dsn_hash, inherited_dsn_hash):
            raise LifecycleAuthorityError('managed PostgreSQL DSN authority mismatch')
        if not hmac.compare_digest(str(metadata.get('canonical_dsn_sha256') or ''), inherited_dsn_hash):
            raise LifecycleAuthorityError('private metadata PostgreSQL DSN authority mismatch')
        for key in ('SCANNER_DB_URL', 'DATABASE_URL'):
            if not hmac.compare_digest(str(os.getenv(key) or ''), dsn):
                raise LifecycleAuthorityError(f'{key} does not match the managed PostgreSQL DSN')
        if any(key.upper().startswith('PG') for key in os.environ):
            raise LifecycleAuthorityError('libpq PG* environment overrides are forbidden for managed children')

    handshake = _send_handshake_with_timeout_retry(metadata, handshake_timeout_retries)
    if handshake.get('instance_id') != metadata['instance_id'] or handshake.get('activation_state') != PHASE_ACTIVE:
        raise LifecycleAuthorityError('supervisor handshake did not confirm ACTIVE authority')
    for key, value in expected_pairs:
        if not hmac.compare_digest(str(handshake.get(key) or ''), value):
            raise LifecycleAuthorityError(f'supervisor handshake {key} mismatch')
    if require_dsn and not hmac.compare_digest(str(handshake.get('canonical_dsn_sha256') or ''), inherited_dsn_hash):
        raise LifecycleAuthorityError('supervisor handshake PostgreSQL DSN authority mismatch')
    return metadata