truf-server/app/runtime_document.py

from collections.abc import Hashable
from dataclasses import dataclass
import hashlib
import ipaddress
import json
import math
import posixpath
import re
from urllib.parse import urlsplit

import yaml

from capacity_model import MAX_RESULT_BUNDLE_BYTES, validate_remote_assignment_capacity
from managed_files import (
    ManagedFileConfigurationError,
    managed_file_root_registry_from_config,
    normalize_managed_file_roots,
)
from query_policy import QueryPolicyError, validate_rejected_query_policy


_MESSAGES = {
    'invalid_input': 'YAML document input is invalid',
    'size': 'YAML document exceeds its byte bound',
    'encoding': 'YAML document is not valid UTF-8',
    'duplicate_key': 'YAML document contains a duplicate mapping key',
    'syntax': 'YAML document is invalid',
    'mapping_root': 'Runtime document root must be a mapping',
    'unknown_key': 'Runtime document contains an unsupported field',
    'type': 'Runtime document field has an invalid type',
    'bounds': 'Runtime document field exceeds its bound',
    'schema': 'Runtime document is incomplete',
    'core_profile': 'Runtime core profile is invalid',
    'auth_pool': 'Runtime credential pool is invalid',
    'reference': 'Runtime document reference is invalid',
    'capability': 'Runtime package capability is invalid',
    'deployment_path': 'Runtime deployment path is invalid',
}
_MERGE_KEY = object()
_MISSING = object()

MAX_DOCUMENT_DEPTH = 64
MAX_DOCUMENT_NODES = 100000
MAX_CONFIG_DOCUMENT_BYTES = 4 * 1024 * 1024
MAX_SECRETS_DOCUMENT_BYTES = 1024 * 1024
MAX_MAPPING_ENTRIES = 10000
MAX_SEQUENCE_ITEMS = 100000
MAX_MAPPING_KEY_BYTES = 512
MAX_SCALAR_BYTES = 1024 * 1024
MAX_PATH_BYTES = 4096
MAX_NAME_BYTES = 128
MAX_AUTH_POOLS = 64
MAX_AUTH_ENTRIES = 256
MAX_COMPATIBILITY_PROFILES = 16
MAX_PACKAGE_CAPABILITIES = 16
_MIN_INTEGER = -(1 << 63)
_MAX_INTEGER = (1 << 63) - 1

_SAFE_DOCUMENTS = frozenset({'config', 'secrets', 'package_capabilities', 'schema'})
_SAFE_FIELD_PATH = re.compile(r'^[a-z0-9_.\[\]-]{1,512}$')
_ENV_NAME = re.compile(r'^[A-Z_][A-Z0-9_]{0,127}$')
_PLACEHOLDER = re.compile(r'{([A-Za-z_][A-Za-z0-9_]*)}')
_CORE_SOURCES = ('gitlab', 'dockerhub', 'huggingface')
_KNOWN_CAPABILITIES = frozenset({
    ('github', 'github', 'exact_git_v1'),
    ('gitlab', 'gitlab', 'exact_git_v1'),
    ('dockerhub', 'docker', 'docker_direct_v1'),
    ('huggingface', 'huggingface', 'huggingface_space_v1'),
})
_SOURCE_CAPABILITIES = {
    'gitlab': ('gitlab', 'gitlab', 'exact_git_v1'),
    'dockerhub': ('dockerhub', 'docker', 'docker_direct_v1'),
    'huggingface': ('huggingface', 'huggingface', 'huggingface_space_v1'),
}
_REQUIRED_CONFIG_PATHS = (
    ('global',),
    ('supervisor',),
    ('sources',),
    ('global', 'root_dir'),
    ('global', 'project_dir'),
    ('global', 'runtime_dir'),
    ('global', 'postgres_data_dir'),
    ('global', 'postgres_bin_dir'),
    ('global', 'result_bundle_dir'),
    ('global', 'work_dir'),
    ('global', 'control_dir'),
    ('global', 'secrets_file'),
    ('supervisor', 'enabled_sources'),
    ('supervisor', 'control_dir'),
    ('supervisor', 'worker_api'),
    ('supervisor', 'worker_api', 'enabled'),
    ('supervisor', 'worker_api', 'sources'),
    ('supervisor', 'worker_api', 'auth_entries'),
    ('supervisor', 'worker_api', 'compatibility_profiles'),
    ('sources', 'gitlab'),
    ('sources', 'gitlab', 'enabled'),
    ('sources', 'gitlab', 'auth_pool'),
    ('sources', 'dockerhub'),
    ('sources', 'dockerhub', 'enabled'),
    ('sources', 'dockerhub', 'auth_pool'),
    ('sources', 'huggingface'),
    ('sources', 'huggingface', 'enabled'),
    ('sources', 'huggingface', 'auth_pool'),
)
_FIXED_PATHS = {
    'root_dir': '/opt/truf',
    'project_dir': '/opt/truf/app',
    'runtime_dir': '/data/runtime-linux',
    'postgres_data_dir': '/data/postgres-linux',
    'postgres_bin_dir': '/usr/lib/postgresql/16/bin',
    'result_bundle_dir': '/data/scanner-result-bundles',
    'work_dir': '/data/scanner-work',
    'control_dir': '/run/truf/control',
    'secrets_file': '/data/config/secrets.yaml',
}
_GLOBAL_PATH_FIELDS = frozenset({
    'result_bundle_dir', 'work_dir', 'root_dir', 'project_dir', 'runtime_dir',
    'postgres_data_dir', 'postgres_bin_dir', 'control_dir', 'secrets_file',
    'result_spool_dir', 'legacy_result_spool_dir', 'results_dir', 'queue_dir',
    'state_dir', 'log_dir', 'keycheck_dir', 'postman_cache_dir',
    'gharchive_cache_dir', 'database_path', 'state_file', 'api_proxy_file',
    'download_proxy_file', 'dashboard_db_path', 'scan_limiter_db',
    'dockerhub_tag_cache_path', 'proxy_file', 'trufflehog_path',
    'trufflehog_config',
})
_SUPERVISOR_PATH_FIELDS = frozenset({
    'log_dir', 'control_dir', 'instance_file', 'lock_file', 'supervisor_log',
    'status_file', 'dashboard_log', 'state_dir',
})
_KEYCHECK_PATH_FIELDS = frozenset({
    'input', 'proxy_file', 'keycheck_dir', 'summary_tsv', 'summary_json',
    'alive_summary_tsv',
})
_SOURCE_PATH_FIELDS = frozenset({
    'target_file', 'trufflehog_config', 'postman_cache_dir',
    'gharchive_cache_dir',
})
_CONFIG_TEMPLATE_SCHEMA_SHA256 = (
    '45d957d8168c796a9d4081ab0936d83d797c4675c8d477baf63195117363a9d8'
)


class RuntimeDocumentError(ValueError):
    def __init__(self, category, line=None, column=None, *, document=None, path=None):
        self.category = str(category)
        self.line = int(line) if line is not None else None
        self.column = int(column) if column is not None else None
        self.document = document if document in _SAFE_DOCUMENTS else None
        self.path = (
            path if type(path) is str and _SAFE_FIELD_PATH.fullmatch(path) else None
        )
        message = _MESSAGES[self.category]
        if self.line is not None and self.column is not None:
            message += f' at line {self.line}, column {self.column}'
        if self.document is not None:
            message += f' in {self.document}'
        if self.path is not None:
            message += f' at {self.path}'
        super().__init__(message)


@dataclass(frozen=True)
class ValidatedRuntimeDocuments:
    config: dict
    secrets: dict


class _MarkedDocumentError(Exception):
    def __init__(self, mark):
        self.mark = mark


class _DuplicateKeyError(_MarkedDocumentError):
    pass


class _DocumentSyntaxError(_MarkedDocumentError):
    pass


class _StrictSafeLoader(yaml.SafeLoader):
    def _mapping_key(self, key_node, deep):
        key = None
        try:
            if key_node.tag == 'tag:yaml.org,2002:merge':
                return _MERGE_KEY
            key = self.construct_object(key_node, deep=deep)
            if not isinstance(key, Hashable):
                raise _DocumentSyntaxError(key_node.start_mark)
            return key
        finally:
            key_node = deep = key = None

    def _reject_duplicate_keys(self, pairs, deep):
        seen = key_node = _value_node = key = duplicate = None
        try:
            seen = {}
            for key_node, _value_node in pairs:
                key = self._mapping_key(key_node, deep)
                try:
                    duplicate = key in seen
                except Exception:
                    raise _DocumentSyntaxError(key_node.start_mark) from None
                if duplicate:
                    raise _DuplicateKeyError(key_node.start_mark)
                seen[key] = None
        finally:
            pairs = deep = seen = key_node = _value_node = key = duplicate = None

    def flatten_mapping(self, node):
        try:
            # SafeLoader recursively calls this method for anonymous merge sources.
            self._reject_duplicate_keys(node.value, False)
            return super().flatten_mapping(node)
        finally:
            node = None

    def construct_mapping(self, node, deep=False):
        mapping = key_node = value_node = key = None
        try:
            if not isinstance(node, yaml.MappingNode):
                raise _DocumentSyntaxError(node.start_mark)

            self.flatten_mapping(node)
            self._reject_duplicate_keys(node.value, deep)

            mapping = {}
            for key_node, value_node in node.value:
                key = self._mapping_key(key_node, deep)
                mapping[key] = self.construct_object(value_node, deep=deep)
            return mapping
        finally:
            node = deep = mapping = key_node = value_node = key = None


def _location(mark):
    if mark is None:
        return None, None
    line = getattr(mark, 'line', None)
    column = getattr(mark, 'column', None)
    if type(line) is not int or type(column) is not int:
        return None, None
    return line + 1, column + 1


def _parse_yaml_document(payload):
    text = None
    try:
        text = payload.decode('utf-8', errors='strict')
    except UnicodeDecodeError:
        pass
    if text is None:
        return None, ('encoding', None, None)

    value = None
    failure = None
    loader = None
    try:
        loader = _StrictSafeLoader(text)
        value = loader.get_single_data()
    except _DuplicateKeyError as exc:
        failure = ('duplicate_key', *_location(exc.mark))
    except _DocumentSyntaxError as exc:
        failure = ('syntax', *_location(exc.mark))
    except Exception as exc:
        mark = getattr(exc, 'problem_mark', None) or getattr(exc, 'context_mark', None)
        failure = ('syntax', *_location(mark))
    except BaseException:
        if loader is not None:
            try:
                loader.dispose()
            except Exception:
                pass
        loader = None
        value = None
        text = None
        payload = None
        raise
    finally:
        if loader is not None:
            try:
                loader.dispose()
            except Exception:
                failure = failure or ('syntax', None, None)

    if failure is not None:
        value = None
    return value, failure


def load_yaml_document(payload, *, max_bytes):
    if type(payload) is not bytes or type(max_bytes) is not int or max_bytes <= 0:
        payload = None
        max_bytes = None
        raise RuntimeDocumentError('invalid_input')
    if len(payload) > max_bytes:
        payload = None
        raise RuntimeDocumentError('size')

    value = failure = None
    try:
        value, failure = _parse_yaml_document(payload)
    finally:
        payload = None
        max_bytes = None

    if failure is not None:
        raise RuntimeDocumentError(*failure)
    return value


def _preview_document(payload, max_bytes, document):
    try:
        value = load_yaml_document(payload, max_bytes=max_bytes)
    except RuntimeDocumentError as exc:
        return None, (
            exc.category, exc.line, exc.column, document, exc.path,
        )
    finally:
        payload = None
    return value, None


def _field_path(path):
    return '.'.join(path) if path else 'root'


def _failure(category, document, path):
    return category, document, _field_path(path)


def _consume(state, count=1):
    state['nodes'] += count
    return state['nodes'] <= MAX_DOCUMENT_NODES


def _bounded_text(value, limit=MAX_SCALAR_BYTES, *, nonempty=False):
    try:
        if type(value) is not str or '\x00' in value:
            return False
        try:
            size = len(value.encode('utf-8', errors='strict'))
        except UnicodeError:
            return False
        return size <= limit and (not nonempty or bool(value))
    finally:
        value = None


def _bounded_name(value):
    try:
        return (
            _bounded_text(value, MAX_NAME_BYTES, nonempty=True)
            and value == value.strip()
        )
    finally:
        value = None


def _normalize_string_list(value, path, state, *, max_items=MAX_SEQUENCE_ITEMS):
    normalized = item = None
    try:
        if type(value) is not list:
            return None, _failure('type', 'config', path)
        if len(value) > max_items:
            return None, _failure('bounds', 'config', path)
        normalized = []
        for item in value:
            if not _consume(state) or not _bounded_text(item, nonempty=True):
                return None, _failure('bounds', 'config', path + ('item',))
            normalized.append(item)
        return normalized, None
    finally:
        value = path = state = normalized = item = None


def _normalize_dynamic_config(value, path, state):
    normalized = key = entry_name = profile_name = raw_profile = None
    item_path = package_manifest = profile = sources = failure = item = None
    query = raw_override = override = name = maximum = None
    try:
        if path == ('supervisor', 'worker_api', 'admin', 'managed_file_roots'):
            try:
                normalized, _registry = normalize_managed_file_roots(value)
            except ManagedFileConfigurationError as exc:
                return None, _failure(exc.category, 'config', path + exc.field)
            return normalized, None

        if path in (
            ('supervisor', 'worker_api', 'sources'),
            ('supervisor', 'defaults', 'extra_args'),
        ):
            return _normalize_string_list(value, path, state)

        if path == ('supervisor', 'worker_api', 'assignment_ttl_seconds_by_source'):
            if type(value) is not dict:
                return None, _failure('type', 'config', path)
            if len(value) > len(_CORE_SOURCES):
                return None, _failure('bounds', 'config', path)
            normalized = {}
            for key, item in value.items():
                item_path = path + (key,) if type(key) is str else path + ('field',)
                if not _consume(state):
                    return None, _failure('bounds', 'config', item_path)
                if key not in _CORE_SOURCES:
                    return None, _failure('unknown_key', 'config', item_path)
                if type(item) is not int or not 60 <= item <= 7 * 24 * 60 * 60:
                    return None, _failure('bounds', 'config', item_path)
                normalized[key] = item
            return normalized, None

        if path == ('supervisor', 'worker_api', 'auth_entries'):
            if type(value) is not dict:
                return None, _failure('type', 'config', path)
            if len(value) > len(_CORE_SOURCES) + 1:
                return None, _failure('bounds', 'config', path)
            normalized = {}
            for key, entry_name in value.items():
                if not _consume(state) or not _bounded_name(key) or not _bounded_name(entry_name):
                    return None, _failure('bounds', 'config', path + ('item',))
                normalized[key] = entry_name
            return normalized, None

        if path == ('supervisor', 'worker_api', 'compatibility_profiles'):
            if type(value) is not dict:
                return None, _failure('type', 'config', path)
            if len(value) > MAX_COMPATIBILITY_PROFILES:
                return None, _failure('bounds', 'config', path)
            normalized = {}
            for profile_name, raw_profile in value.items():
                item_path = path + ('profile',)
                if not _consume(state) or not _bounded_name(profile_name):
                    return None, _failure('bounds', 'config', item_path)
                if type(raw_profile) is not dict:
                    return None, _failure('type', 'config', item_path)
                if set(raw_profile) - {'package_manifest', 'sources'}:
                    return None, _failure('unknown_key', 'config', item_path)
                package_manifest = raw_profile.get('package_manifest', _MISSING)
                if package_manifest is _MISSING:
                    return None, _failure('schema', 'config', item_path + ('package_manifest',))
                if not _consume(state) or not _bounded_text(package_manifest, MAX_PATH_BYTES, nonempty=True):
                    return None, _failure('bounds', 'config', item_path + ('package_manifest',))
                profile = {'package_manifest': package_manifest}
                if 'sources' in raw_profile:
                    sources, failure = _normalize_string_list(
                        raw_profile['sources'], item_path + ('sources',), state,
                        max_items=len(_CORE_SOURCES),
                    )
                    if failure is not None:
                        return None, failure
                    profile['sources'] = sources
                normalized[profile_name] = profile
            return normalized, None

        if (
            path == ('keychecks', 'env')
            or len(path) == 4
            and path[0] == 'supervisor'
            and path[1] == 'sources'
            and path[3] == 'env'
        ):
            if type(value) is not dict:
                return None, _failure('type', 'config', path)
            if len(value) > 128:
                return None, _failure('bounds', 'config', path)
            normalized = {}
            for key, item in value.items():
                if (
                    not _consume(state)
                    or type(key) is not str
                    or _ENV_NAME.fullmatch(key) is None
                    or not _bounded_text(item)
                ):
                    return None, _failure('bounds', 'config', path + ('item',))
                normalized[key] = item
            return normalized, None

        if (
            len(path) == 3
            and path[0] == 'sources'
            and path[2] == 'query_overrides'
        ):
            if type(value) is not dict:
                return None, _failure('type', 'config', path)
            if len(value) > MAX_MAPPING_ENTRIES:
                return None, _failure('bounds', 'config', path)
            normalized = {}
            for query, raw_override in value.items():
                item_path = path + ('item',)
                if not _consume(state) or not _bounded_text(query, 512, nonempty=True):
                    return None, _failure('bounds', 'config', item_path)
                if type(raw_override) is not dict:
                    return None, _failure('type', 'config', item_path)
                if set(raw_override) - {'pages', 'per_page', 'max_targets'}:
                    return None, _failure('unknown_key', 'config', item_path)
                override = {}
                for name, item in raw_override.items():
                    maximum = 100000 if name == 'max_targets' else 1000
                    if (
                        not _consume(state)
                        or type(item) is not int
                        or not 1 <= item <= maximum
                    ):
                        return None, _failure('bounds', 'config', item_path + (name,))
                    override[name] = item
                normalized[query] = override
            return normalized, None

        return _MISSING
    finally:
        value = path = state = normalized = key = entry_name = None
        profile_name = raw_profile = item_path = package_manifest = profile = None
        sources = failure = item = query = raw_override = override = None
        name = maximum = None


def _normalize_scalar(value, template, path):
    try:
        if type(value) is not type(template):
            return None, _failure('type', 'config', path)
        if type(value) is int and not 0 <= value <= _MAX_INTEGER:
            return None, _failure('bounds', 'config', path)
        if type(value) is float and (not math.isfinite(value) or value < 0):
            return None, _failure('bounds', 'config', path)
        if type(value) is str and not _bounded_text(value):
            return None, _failure('bounds', 'config', path)
        if type(value) not in {type(None), bool, int, float, str}:
            return None, _failure('type', 'config', path)
        return value, None
    finally:
        value = template = path = None


def _normalize_config_shape(value, template, path, state, depth=0):
    identity = dynamic = normalized = key = item = copied = failure = None
    try:
        if depth > MAX_DOCUMENT_DEPTH or not _consume(state):
            return None, _failure('bounds', 'config', path)

        if type(value) in (dict, list):
            identity = id(value)
            if identity in state['active']:
                return None, _failure('bounds', 'config', path)
            state['active'].add(identity)
            try:
                dynamic = _normalize_dynamic_config(value, path, state)
                if dynamic is not _MISSING:
                    return dynamic

                if type(template) is not type(value):
                    return None, _failure('type', 'config', path)
                if type(value) is dict:
                    if len(value) > MAX_MAPPING_ENTRIES:
                        return None, _failure('bounds', 'config', path)
                    normalized = {}
                    for key, item in value.items():
                        if (
                            type(key) is not str
                            or not _bounded_text(key, MAX_MAPPING_KEY_BYTES, nonempty=True)
                        ):
                            return None, _failure('type', 'config', path + ('field',))
                        if key not in template:
                            if (
                                len(path) == 2
                                and path[0] == 'sources'
                                and key in _SOURCE_PATH_FIELDS
                            ):
                                copied, failure = _normalize_scalar(
                                    item, '', path + (key,),
                                )
                                if failure is not None:
                                    return None, failure
                                normalized[key] = copied
                                continue
                            if (
                                len(path) == 3
                                and path[:2] == ('supervisor', 'sources')
                                and key == 'enabled'
                            ):
                                copied, failure = _normalize_scalar(
                                    item, False, path + (key,),
                                )
                                if failure is not None:
                                    return None, failure
                                normalized[key] = copied
                                continue
                            if (
                                len(path) == 3
                                and path[:2] == ('supervisor', 'sources')
                                and key == 'env'
                            ):
                                copied, failure = _normalize_dynamic_config(
                                    item, path + (key,), state,
                                )
                                if failure is not None:
                                    return None, failure
                                normalized[key] = copied
                                continue
                            return None, _failure('unknown_key', 'config', path + ('unknown',))
                        copied, failure = _normalize_config_shape(
                            item, template[key], path + (key,), state, depth + 1,
                        )
                        if failure is not None:
                            return None, failure
                        normalized[key] = copied
                    return normalized, None

                if len(value) > MAX_SEQUENCE_ITEMS:
                    return None, _failure('bounds', 'config', path)
                if not template:
                    if value:
                        return None, _failure('type', 'config', path)
                    return [], None
                normalized = []
                for item in value:
                    copied, failure = _normalize_config_shape(
                        item, template[0], path + ('item',), state, depth + 1,
                    )
                    if failure is not None:
                        return None, failure
                    normalized.append(copied)
                return normalized, None
            finally:
                state['active'].remove(identity)

        return _normalize_scalar(value, template, path)
    finally:
        value = template = path = state = identity = dynamic = None
        normalized = key = item = copied = failure = None


def _dynamic_schema_kind(path):
    if path == ('supervisor', 'worker_api', 'admin', 'managed_file_roots'):
        return 'managed_file_roots'
    if path in (
        ('supervisor', 'worker_api', 'sources'),
        ('supervisor', 'defaults', 'extra_args'),
    ):
        return 'string_list'
    if path == ('supervisor', 'worker_api', 'assignment_ttl_seconds_by_source'):
        return 'assignment_ttl_seconds_by_source'
    if path == ('supervisor', 'worker_api', 'auth_entries'):
        return 'auth_entries'
    if path == ('supervisor', 'worker_api', 'compatibility_profiles'):
        return 'compatibility_profiles'
    if (
        path == ('keychecks', 'env')
        or len(path) == 4
        and path[0] == 'supervisor'
        and path[1] == 'sources'
        and path[3] == 'env'
    ):
        return 'environment'
    if (
        len(path) == 3
        and path[0] == 'sources'
        and path[2] == 'query_overrides'
    ):
        return 'query_overrides'
    return None


def _config_template_schema_identity(value, path=()):
    dynamic_kind = scalar_names = None
    try:
        dynamic_kind = _dynamic_schema_kind(path)
        if dynamic_kind is not None:
            return {'dynamic': dynamic_kind}
        if type(value) is dict:
            return {
                'mapping': [
                    [key, _config_template_schema_identity(value[key], path + (key,))]
                    for key in sorted(value)
                ],
            }
        if type(value) is list:
            return {
                'sequence': (
                    _config_template_schema_identity(value[0], path + ('item',))
                    if value else None
                ),
            }
        scalar_names = {
            type(None): 'null', bool: 'bool', int: 'int', float: 'float', str: 'str',
        }
        return {'scalar': scalar_names.get(type(value), 'unsupported')}
    finally:
        value = path = dynamic_kind = scalar_names = None


def _config_template_schema_hash(value):
    encoded = None
    try:
        encoded = json.dumps(
            _config_template_schema_identity(value),
            ensure_ascii=True,
            separators=(',', ':'),
            sort_keys=True,
        ).encode('ascii')
        return hashlib.sha256(encoded).hexdigest()
    finally:
        value = encoded = None


def _normalize_secrets(value):
    state = {'nodes': 0, 'active': set()}
    pools = normalized = active = None
    pool_name = raw_entries = raw_entry = entries = names = None
    name = token = entry = username = None
    try:
        if type(value) is not dict:
            return None, _failure('mapping_root', 'secrets', ())
        if set(value) - {'auth_pools'}:
            return None, _failure('unknown_key', 'secrets', ('unknown',))
        pools = value.get('auth_pools', {})
        if type(pools) is not dict:
            return None, _failure('type', 'secrets', ('auth_pools',))
        if len(pools) > MAX_AUTH_POOLS:
            return None, _failure('bounds', 'secrets', ('auth_pools',))

        normalized = {}
        active = {id(value), id(pools)}
        state['active'].update(active)
        if not _consume(state, 2):
            return None, _failure('bounds', 'secrets', ())
        for pool_name, raw_entries in pools.items():
            pool_path = ('auth_pools', 'pool')
            if not _bounded_name(pool_name) or type(raw_entries) is not list:
                return None, _failure('auth_pool', 'secrets', pool_path)
            if id(raw_entries) in state['active'] or len(raw_entries) > MAX_AUTH_ENTRIES:
                return None, _failure('bounds', 'secrets', pool_path)
            state['active'].add(id(raw_entries))
            raw_entries_id = id(raw_entries)
            entries = []
            names = set()
            try:
                for raw_entry in raw_entries:
                    entry_path = pool_path + ('entry',)
                    if not _consume(state) or type(raw_entry) is not dict:
                        return None, _failure('auth_pool', 'secrets', entry_path)
                    if id(raw_entry) in state['active']:
                        return None, _failure('bounds', 'secrets', entry_path)
                    state['active'].add(id(raw_entry))
                    try:
                        if set(raw_entry) not in ({'name', 'token'}, {'name', 'username', 'token'}):
                            return None, _failure('auth_pool', 'secrets', entry_path)
                        name = raw_entry.get('name')
                        token = raw_entry.get('token')
                        if not _bounded_name(name) or not _bounded_text(token, nonempty=True):
                            return None, _failure('auth_pool', 'secrets', entry_path)
                        if name in names:
                            return None, _failure('auth_pool', 'secrets', entry_path + ('name',))
                        names.add(name)
                        entry = {'name': name}
                        if 'username' in raw_entry:
                            username = raw_entry['username']
                            if not _bounded_text(username, MAX_NAME_BYTES, nonempty=True):
                                return None, _failure('auth_pool', 'secrets', entry_path + ('username',))
                            entry['username'] = username
                        entry['token'] = token
                        entries.append(entry)
                    finally:
                        state['active'].remove(id(raw_entry))
            finally:
                state['active'].remove(raw_entries_id)
            normalized[pool_name] = entries
        return {'auth_pools': normalized}, None
    finally:
        value = pools = normalized = active = None
        pool_name = raw_entries = raw_entry = entries = names = None
        name = token = entry = username = None
        state = None


def _get_path(value, path):
    current = name = None
    try:
        current = value
        for name in path:
            if type(current) is not dict or name not in current:
                return _MISSING
            current = current[name]
        return current
    finally:
        value = path = current = name = None


def _validate_required_fields(config):
    path = None
    try:
        for path in _REQUIRED_CONFIG_PATHS:
            if _get_path(config, path) is _MISSING:
                return _failure('schema', 'config', path)
        return None
    finally:
        config = path = None


def _validate_auth_references(config, secrets):
    pools = sources = worker_config = raw_worker_sources = None
    worker_sources = auth_entries = source_config = matches = None
    docker_pool = source_name = pool_name = entry_name = None
    try:
        pools = secrets['auth_pools']
        sources = config['sources']
        docker_pool = sources['dockerhub'].get('auth_pool') or ''

        for source_name, source_config in sources.items():
            pool_name = source_config.get('auth_pool')
            if pool_name and pool_name not in pools:
                return _failure('reference', 'config', ('sources', source_name, 'auth_pool'))

        if docker_pool:
            if any('username' not in entry for entry in pools[docker_pool]):
                return _failure('auth_pool', 'secrets', ('auth_pools', 'pool', 'entry'))

        worker_config = config['supervisor']['worker_api']
        raw_worker_sources = worker_config['sources']
        if len(raw_worker_sources) != len(set(raw_worker_sources)):
            return _failure('core_profile', 'config', ('supervisor', 'worker_api', 'sources'))
        if raw_worker_sources:
            if any(source not in _CORE_SOURCES for source in raw_worker_sources):
                return _failure('core_profile', 'config', ('supervisor', 'worker_api', 'sources'))
            worker_sources = tuple(source for source in _CORE_SOURCES if source in raw_worker_sources)
        else:
            worker_sources = _CORE_SOURCES
        worker_config['sources'] = list(worker_sources)

        auth_entries = worker_config['auth_entries']
        if set(auth_entries) - (set(worker_sources) | {'github'}):
            return _failure('reference', 'config', ('supervisor', 'worker_api', 'auth_entries'))
        if set(auth_entries) - {'github', 'gitlab'}:
            return _failure('reference', 'config', ('supervisor', 'worker_api', 'auth_entries'))
        for source_name, entry_name in auth_entries.items():
            source_config = sources.get(source_name)
            if type(source_config) is not dict or source_config.get('enabled') is not True:
                return _failure('reference', 'config', ('supervisor', 'worker_api', 'auth_entries'))
            pool_name = source_config.get('auth_pool')
            if not pool_name or pool_name not in pools:
                return _failure('reference', 'config', ('supervisor', 'worker_api', 'auth_entries'))
            matches = [entry for entry in pools[pool_name] if entry['name'] == entry_name]
            if len(matches) != 1:
                return _failure('reference', 'config', ('supervisor', 'worker_api', 'auth_entries'))
        return None
    finally:
        config = secrets = pools = sources = None
        worker_config = raw_worker_sources = worker_sources = None
        auth_entries = source_config = matches = None
        docker_pool = source_name = pool_name = entry_name = None


def _normalize_capabilities(package_capabilities, profiles, worker_sources, enabled):
    if package_capabilities is None:
        package_capabilities = {}
    if type(package_capabilities) is not dict:
        return None, _failure('capability', 'package_capabilities', ())
    if set(package_capabilities) != set(profiles):
        return None, _failure('capability', 'package_capabilities', ('profiles',))
    if enabled and not profiles:
        return None, _failure(
            'capability', 'config',
            ('supervisor', 'worker_api', 'compatibility_profiles'),
        )

    normalized_capabilities = {}
    for profile_name, evidence in package_capabilities.items():
        path = ('profiles', 'profile')
        if (
            not _bounded_name(profile_name)
            or type(evidence) is not dict
            or set(evidence) != {'package_manifest', 'capabilities'}
            or evidence.get('package_manifest') != profiles[profile_name]['package_manifest']
        ):
            return None, _failure('capability', 'package_capabilities', path)
        raw_capabilities = evidence['capabilities']
        if type(raw_capabilities) is not list:
            return None, _failure('capability', 'package_capabilities', path)
        if not 1 <= len(raw_capabilities) <= MAX_PACKAGE_CAPABILITIES:
            return None, _failure('capability', 'package_capabilities', path)
        seen = set()
        capabilities = []
        for raw_item in raw_capabilities:
            item_path = path + ('capability',)
            if type(raw_item) is not dict or set(raw_item) != {
                'source', 'platform', 'planning_kind',
            }:
                return None, _failure('capability', 'package_capabilities', item_path)
            if any(type(raw_item[name]) is not str for name in raw_item):
                return None, _failure('capability', 'package_capabilities', item_path)
            capability = (
                raw_item['source'], raw_item['platform'], raw_item['planning_kind'],
            )
            if capability not in _KNOWN_CAPABILITIES or capability in seen:
                return None, _failure('capability', 'package_capabilities', item_path)
            seen.add(capability)
            capabilities.append(capability)
        normalized_capabilities[profile_name] = frozenset(capabilities)

    covered = set()
    for profile_name, profile in profiles.items():
        capabilities = normalized_capabilities[profile_name]
        raw_sources = profile.get('sources') or [item[0] for item in capabilities]
        if len(raw_sources) != len(set(raw_sources)):
            return None, _failure(
                'capability', 'config',
                ('supervisor', 'worker_api', 'compatibility_profiles', 'profile', 'sources'),
            )
        profile_sources = tuple(source for source in _CORE_SOURCES if source in raw_sources)
        if (
            not profile_sources
            or set(raw_sources) != set(profile_sources)
            or not set(profile_sources) <= set(worker_sources)
            or any(_SOURCE_CAPABILITIES[source] not in capabilities for source in profile_sources)
        ):
            return None, _failure(
                'capability', 'config',
                ('supervisor', 'worker_api', 'compatibility_profiles', 'profile', 'sources'),
            )
        profile['sources'] = list(profile_sources)
        covered.update(profile_sources)
    if enabled and covered != set(worker_sources):
        return None, _failure(
            'capability', 'config',
            ('supervisor', 'worker_api', 'compatibility_profiles'),
        )
    return normalized_capabilities, None


def _validate_worker_config(config, package_capabilities):
    worker_config = enabled = address_text = address = integer_bounds = None
    name = minimum = maximum = value = source = effective_ttl = minimum_ttl = None
    ttl_overrides = None
    admin_config = default = origin = marker = parsed = parsed_port = None
    _normalized = failure = managed_file_roots = None
    try:
        worker_config = config['supervisor']['worker_api']
        enabled = worker_config['enabled']
        if type(enabled) is not bool:
            return _failure('type', 'config', ('supervisor', 'worker_api', 'enabled'))

        address_text = worker_config.get('address', '127.0.0.1')
        try:
            address = ipaddress.ip_address(address_text)
        except (TypeError, ValueError):
            return _failure('bounds', 'config', ('supervisor', 'worker_api', 'address'))
        if address.is_unspecified or address.is_multicast or not (
            address.is_loopback or address.is_private
        ):
            return _failure('bounds', 'config', ('supervisor', 'worker_api', 'address'))

        integer_bounds = {
            'port': (1024, 65535),
            'assignment_ttl_seconds': (60, 7 * 24 * 60 * 60),
            'max_bundle_bytes': (
                1024 * 1024,
                min(MAX_RESULT_BUNDLE_BYTES, config['global'].get(
                    'result_bundle_max_event_bytes', MAX_RESULT_BUNDLE_BYTES,
                )),
            ),
            'reaper_interval_seconds': (5, 3600),
            'reaper_batch_size': (1, 1000),
            'limit_concurrency': (1, 1024),
            'body_idle_timeout_seconds': (1, 120),
            'json_body_timeout_seconds': (1, 300),
            'bundle_body_timeout_seconds': (30, 86400),
        }
        for name, (minimum, maximum) in integer_bounds.items():
            value = worker_config.get(name)
            if type(value) is not int or not minimum <= value <= maximum:
                return _failure('bounds', 'config', ('supervisor', 'worker_api', name))

        ttl_overrides = worker_config.get('assignment_ttl_seconds_by_source', {})
        for source in worker_config['sources']:
            effective_ttl = ttl_overrides.get(
                source, worker_config['assignment_ttl_seconds'],
            )
            minimum_ttl = (
                config['sources'][source]['timeout']
                + worker_config['bundle_body_timeout_seconds']
                + 60
            )
            if effective_ttl < minimum_ttl:
                return _failure(
                    'bounds', 'config',
                    (
                        'supervisor', 'worker_api',
                        'assignment_ttl_seconds_by_source', source,
                    ) if source in ttl_overrides else (
                        'supervisor', 'worker_api', 'assignment_ttl_seconds',
                    ),
                )

        admin_config = worker_config.get('admin') or {}
        if type(admin_config) is not dict or type(admin_config.get('enabled', False)) is not bool:
            return _failure('type', 'config', ('supervisor', 'worker_api', 'admin'))
        for name, default, minimum, maximum in (
            ('max_body_bytes', 8192, 1024, 65536),
            ('snapshot_limit', 200, 1, 500),
            ('requeue_limit', 100, 1, 500),
        ):
            value = admin_config.get(name, default)
            if type(value) is not int or not minimum <= value <= maximum:
                return _failure(
                    'bounds', 'config', ('supervisor', 'worker_api', 'admin', name),
                )
        try:
            managed_file_roots = managed_file_root_registry_from_config(config)
        except ManagedFileConfigurationError as exc:
            return _failure(
                exc.category, 'config',
                ('supervisor', 'worker_api', 'admin', 'managed_file_roots')
                + exc.field,
            )
        if admin_config.get('enabled') is True:
            origin = admin_config.get('origin')
            marker = admin_config.get('edge_marker')
            try:
                parsed = urlsplit(origin)
                parsed_port = parsed.port
            except (TypeError, ValueError):
                parsed = None
                parsed_port = None
            if (
                type(origin) is not str
                or not 1 <= len(origin) <= 512
                or any(character.isspace() for character in origin)
                or parsed is None
                or parsed.scheme != 'https'
                or not parsed.hostname
                or parsed.username is not None
                or parsed.password is not None
                or parsed.path
                or parsed.query
                or parsed.fragment
                or parsed_port is not None and not 1 <= parsed_port <= 65535
            ):
                return _failure(
                    'bounds', 'config', ('supervisor', 'worker_api', 'admin', 'origin'),
                )
            if (
                type(marker) is not str
                or not 32 <= len(marker) <= 512
                or any(character.isspace() for character in marker)
            ):
                return _failure(
                    'bounds', 'config', ('supervisor', 'worker_api', 'admin', 'edge_marker'),
                )

        _normalized, failure = _normalize_capabilities(
            package_capabilities,
            worker_config['compatibility_profiles'],
            worker_config['sources'],
            enabled,
        )
        return failure
    finally:
        config = package_capabilities = worker_config = enabled = address_text = None
        address = integer_bounds = name = minimum = maximum = value = None
        source = effective_ttl = minimum_ttl = ttl_overrides = None
        admin_config = default = origin = marker = None
        parsed = parsed_port = _normalized = failure = managed_file_roots = None


def _expand_path(raw_value, resolved_globals):
    if not _bounded_text(raw_value, MAX_PATH_BYTES) or any(
        marker in raw_value for marker in ('\\', '$', '~')
    ):
        return None
    references = _PLACEHOLDER.findall(raw_value)
    if len(references) > 32:
        return None
    if any(reference not in resolved_globals for reference in references):
        return None
    expanded = _PLACEHOLDER.sub(lambda match: resolved_globals[match.group(1)], raw_value)
    if '{' in expanded or '}' in expanded or not expanded:
        return None
    if not _bounded_text(expanded, MAX_PATH_BYTES) or not expanded.startswith('/'):
        return None
    if posixpath.normpath(expanded) != expanded:
        return None
    return expanded


def _resolve_global_deployment_paths(config):
    global_config = resolved = active = name = raw_value = None
    try:
        global_config = config['global']
        resolved = {}
        active = set()

        def resolve(name):
            raw_value = references = reference = value = None
            try:
                if name in resolved:
                    return resolved[name]
                if name in active or name not in global_config or name not in _GLOBAL_PATH_FIELDS:
                    return None
                raw_value = global_config[name]
                if type(raw_value) is not str:
                    return None
                active.add(name)
                references = _PLACEHOLDER.findall(raw_value)
                for reference in references:
                    if resolve(reference) is None:
                        active.remove(name)
                        return None
                value = _expand_path(raw_value, resolved)
                active.remove(name)
                if value is not None:
                    resolved[name] = value
                return value
            finally:
                name = raw_value = references = reference = value = None

        for name in _GLOBAL_PATH_FIELDS & set(global_config):
            raw_value = global_config[name]
            if raw_value == '':
                resolved[name] = ''
                continue
            if resolve(name) is None:
                return None, _failure('deployment_path', 'config', ('global', name))

        return resolved, None
    finally:
        config = global_config = resolved = active = name = raw_value = None


def _resolve_package_manifest_path(config, raw_value):
    resolved = failure = path = None
    try:
        try:
            resolved, failure = _resolve_global_deployment_paths(config)
        except Exception:
            return None
        if failure is not None:
            return None
        path = _expand_path(raw_value, resolved)
        if path is None or not path.startswith('/data/worker-packages/'):
            return None
        return path
    finally:
        config = raw_value = resolved = failure = path = None


def _validate_deployment_paths(config):
    global_config = resolved = failure = name = expected = runtime_root = None
    value = trufflehog_config = supervisor = supervisor_paths = None
    supervisor_roots = root = keychecks = keycheck_paths = None
    source_name = source_config = source_paths = source_roots = None
    profiles = profile = path = None
    try:
        global_config = config['global']
        resolved, failure = _resolve_global_deployment_paths(config)
        if failure is not None:
            return failure

        for name, expected in _FIXED_PATHS.items():
            if resolved.get(name) != expected:
                return _failure('deployment_path', 'config', ('global', name))

        def contained(path, root):
            return path == root or path.startswith(root.rstrip('/') + '/')

        runtime_root = resolved['runtime_dir']
        for name in (
            'result_spool_dir', 'legacy_result_spool_dir', 'results_dir', 'queue_dir',
            'state_dir', 'log_dir', 'keycheck_dir', 'postman_cache_dir',
            'gharchive_cache_dir', 'database_path', 'state_file', 'api_proxy_file',
            'download_proxy_file', 'dashboard_db_path', 'scan_limiter_db',
            'dockerhub_tag_cache_path', 'proxy_file',
        ):
            value = resolved.get(name)
            if value and not contained(value, runtime_root):
                return _failure('deployment_path', 'config', ('global', name))
        trufflehog_config = resolved.get('trufflehog_config')
        if trufflehog_config and not contained(trufflehog_config, resolved['project_dir']):
            return _failure('deployment_path', 'config', ('global', 'trufflehog_config'))

        def validate_section(section, fields, prefix):
            values = name = raw_value = value = None
            try:
                values = {}
                for name in fields & set(section):
                    raw_value = section[name]
                    if raw_value == '':
                        values[name] = ''
                        continue
                    value = _expand_path(raw_value, resolved)
                    if value is None:
                        return _failure('deployment_path', 'config', prefix + (name,))
                    values[name] = value
                return values
            finally:
                section = fields = prefix = values = name = raw_value = value = None

        supervisor = config['supervisor']
        supervisor_paths = validate_section(
            supervisor, _SUPERVISOR_PATH_FIELDS, ('supervisor',),
        )
        if type(supervisor_paths) is tuple:
            return supervisor_paths
        if supervisor_paths.get('control_dir') != _FIXED_PATHS['control_dir']:
            return _failure('deployment_path', 'config', ('supervisor', 'control_dir'))
        supervisor_roots = {
            'log_dir': resolved.get('log_dir'),
            'instance_file': resolved['control_dir'],
            'lock_file': resolved['control_dir'],
            'supervisor_log': resolved.get('log_dir'),
            'status_file': resolved.get('log_dir'),
            'dashboard_log': resolved.get('log_dir'),
            'state_dir': resolved.get('state_dir'),
        }
        for name, root in supervisor_roots.items():
            value = supervisor_paths.get(name)
            if value and (not root or not contained(value, root)):
                return _failure('deployment_path', 'config', ('supervisor', name))

        keychecks = config.get('keychecks')
        if type(keychecks) is dict:
            keycheck_paths = validate_section(keychecks, _KEYCHECK_PATH_FIELDS, ('keychecks',))
            if type(keycheck_paths) is tuple:
                return keycheck_paths
            for name, value in keycheck_paths.items():
                root = resolved.get('proxy_file') if name == 'proxy_file' else resolved.get('keycheck_dir')
                if value and (not root or not contained(value, root)):
                    return _failure('deployment_path', 'config', ('keychecks', name))
        for source_name, source_config in config['sources'].items():
            source_paths = validate_section(
                source_config, _SOURCE_PATH_FIELDS, ('sources', source_name),
            )
            if type(source_paths) is tuple:
                return source_paths
            source_roots = {
                'target_file': resolved.get('queue_dir'),
                'trufflehog_config': resolved.get('trufflehog_config'),
                'postman_cache_dir': resolved.get('postman_cache_dir'),
                'gharchive_cache_dir': resolved.get('gharchive_cache_dir'),
            }
            for name, value in source_paths.items():
                root = source_roots[name]
                if value and (not root or not contained(value, root)):
                    return _failure(
                        'deployment_path', 'config', ('sources', source_name, name),
                    )

        profiles = supervisor['worker_api']['compatibility_profiles']
        for profile in profiles.values():
            path = _resolve_package_manifest_path(config, profile['package_manifest'])
            if path is None:
                return _failure(
                    'deployment_path', 'config',
                    ('supervisor', 'worker_api', 'compatibility_profiles', 'profile', 'package_manifest'),
                )
        return None
    finally:
        config = global_config = resolved = failure = name = expected = None
        runtime_root = value = trufflehog_config = supervisor = supervisor_paths = None
        supervisor_roots = root = keychecks = keycheck_paths = None
        source_name = source_config = source_paths = source_roots = None
        profiles = profile = path = None


def _validate_runtime_semantics(config, secrets, package_capabilities):
    failure = enabled_sources = supervisor_sources = None
    source_name = source_config = queries = overrides = None
    try:
        failure = _validate_required_fields(config)
        if failure is not None:
            return failure

        enabled_sources = config['supervisor']['enabled_sources']
        if (
            type(enabled_sources) is not list
            or len(enabled_sources) != len(_CORE_SOURCES)
            or set(enabled_sources) != set(_CORE_SOURCES)
        ):
            return _failure('core_profile', 'config', ('supervisor', 'enabled_sources'))
        config['supervisor']['enabled_sources'] = list(_CORE_SOURCES)
        if any(config['sources'][source]['enabled'] is not True for source in _CORE_SOURCES):
            return _failure('core_profile', 'config', ('sources',))

        try:
            validate_remote_assignment_capacity(config['global'])
        except ValueError as exc:
            return _failure('bounds', 'config', ('global', str(exc)))

        supervisor_sources = config['supervisor'].get('sources', {})
        if any(source not in config['sources'] for source in supervisor_sources):
            return _failure('reference', 'config', ('supervisor', 'sources'))
        for source_name, source_config in config['sources'].items():
            queries = source_config.get('queries', [])
            overrides = source_config.get('query_overrides', {})
            if any(query not in queries for query in overrides):
                return _failure('reference', 'config', ('sources', source_name, 'query_overrides'))
            if source_config.get('target_claim_order', 'oldest') not in (
                'oldest', 'newest', 'balanced',
            ):
                return _failure(
                    'core_profile', 'config',
                    ('sources', source_name, 'target_claim_order'),
                )

        failure = _validate_auth_references(config, secrets)
        if failure is not None:
            return failure
        failure = _validate_worker_config(config, package_capabilities)
        if failure is not None:
            return failure
        failure = _validate_deployment_paths(config)
        if failure is not None:
            return failure

        try:
            validate_rejected_query_policy(config)
        except QueryPolicyError:
            return _failure('reference', 'config', ('query_policy',))
        return None
    finally:
        config = secrets = package_capabilities = None
        failure = enabled_sources = supervisor_sources = None
        source_name = source_config = queries = overrides = None


def _validate_runtime_documents_inner(config, secrets, config_template, package_capabilities):
    template = normalized_config = normalized_secrets = None
    template_state = state = failure = None
    try:
        if type(config) is not dict:
            return None, _failure('mapping_root', 'config', ())
        if type(config_template) is not dict:
            return None, _failure('mapping_root', 'schema', ())

        template_state = {'nodes': 0, 'active': set()}
        template, failure = _normalize_config_shape(
            config_template, config_template, (), template_state,
        )
        if failure is not None:
            return None, _failure('schema', 'schema', ())
        if _config_template_schema_hash(template) != _CONFIG_TEMPLATE_SCHEMA_SHA256:
            return None, _failure('schema', 'schema', ())
        state = {'nodes': 0, 'active': set()}
        normalized_config, failure = _normalize_config_shape(
            config, template, (), state,
        )
        if failure is not None:
            return None, failure
        normalized_secrets, failure = _normalize_secrets(secrets)
        if failure is not None:
            return None, failure
        failure = _validate_runtime_semantics(
            normalized_config, normalized_secrets, package_capabilities,
        )
        if failure is not None:
            return None, failure
        return ValidatedRuntimeDocuments(
            config=normalized_config,
            secrets=normalized_secrets,
        ), None
    except Exception:
        return None, _failure('schema', 'schema', ())
    except BaseException:
        config = secrets = config_template = package_capabilities = None
        template = normalized_config = normalized_secrets = None
        template_state = state = failure = None
        raise


def validate_runtime_documents(
    config,
    secrets,
    *,
    config_template,
    package_capabilities=None,
):
    validated = failure = None
    try:
        validated, failure = _validate_runtime_documents_inner(
            config, secrets, config_template, package_capabilities,
        )
    finally:
        config = None
        secrets = None
        config_template = None
        package_capabilities = None

    if failure is not None:
        raise RuntimeDocumentError(
            failure[0], document=failure[1], path=failure[2],
        )
    return validated


def _preview_runtime_documents_inner(
    config_payload,
    secrets_payload,
    config_template_payload,
    package_capabilities,
):
    config = secrets = config_template = validated = None
    try:
        config, failure = _preview_document(
            config_payload, MAX_CONFIG_DOCUMENT_BYTES, 'config',
        )
        if failure is not None:
            return None, failure
        secrets, failure = _preview_document(
            secrets_payload, MAX_SECRETS_DOCUMENT_BYTES, 'secrets',
        )
        if failure is not None:
            return None, failure
        config_template, failure = _preview_document(
            config_template_payload, MAX_CONFIG_DOCUMENT_BYTES, 'schema',
        )
        if failure is not None:
            return None, failure
        validated, failure = _validate_runtime_documents_inner(
            config, secrets, config_template, package_capabilities,
        )
        if failure is not None:
            return None, (failure[0], None, None, failure[1], failure[2])
        return validated, None
    except Exception:
        return None, ('schema', None, None, 'schema', 'root')
    finally:
        config_payload = None
        secrets_payload = None
        config_template_payload = None
        package_capabilities = None
        config = None
        secrets = None
        config_template = None
        validated = None


def preview_runtime_documents(
    config_payload,
    secrets_payload,
    *,
    config_template_payload,
    package_capabilities=None,
):
    validated = failure = None
    try:
        validated, failure = _preview_runtime_documents_inner(
            config_payload,
            secrets_payload,
            config_template_payload,
            package_capabilities,
        )
    finally:
        config_payload = None
        secrets_payload = None
        config_template_payload = None
        package_capabilities = None

    if failure is not None:
        raise RuntimeDocumentError(
            failure[0], failure[1], failure[2],
            document=failure[3], path=failure[4],
        )
    return validated