truf-server/app/docker_shadow.py

import sys

sys.dont_write_bytecode = True
if not sys.dont_write_bytecode:
    raise RuntimeError('Docker shadow runner could not disable bytecode writes')

import argparse
import copy
import hashlib
import math
import os
import re
import secrets
import socket
import time

import scanner
import console_runner
from keycheck_candidates import extract_candidates
from lifecycle_authority import require_active_supervisor_child
from scanner_db import (
    DOCKER_ADAPTIVE_GATE_MAX_CONTROLS,
    DOCKER_ADAPTIVE_GATE_MIN_CONTROLS,
    DOCKER_ADAPTIVE_LAYER_CLASS_ORDER,
    DOCKER_ADAPTIVE_SELECTOR_VERSION,
    DOCKER_ADAPTIVE_SHADOW_SELECTION_METRIC_KEYS,
    ScannerDB,
    canonical_docker_layer_plan_bytes,
    docker_content_media_class,
    docker_layer_execution_policy_sha256,
    docker_layer_selection_policy_sha256,
    finding_identity,
    select_docker_adaptive_payload,
    validate_docker_adaptive_checkpoint,
    validate_docker_adaptive_shadow_selection_metrics,
    validate_docker_layer_execution,
    validate_docker_layer_limits,
    validate_docker_layer_plan,
    validate_docker_layer_resolution,
)


_SHA256_RE = re.compile(r'[a-f0-9]{64}')
_ROUTED_SERVICE_RE = re.compile(r'[a-z0-9][a-z0-9_.-]{0,63}')
SHADOW_FAILURE_METRIC_KEYS = (
    'diagnostic_full_incomplete',
    'diagnostic_adaptive_incomplete',
    'diagnostic_blob_chunk_processing',
    'diagnostic_blob_detector_timeout',
    'diagnostic_blob_network',
    'diagnostic_blob_timeout',
    'diagnostic_blob_mixed',
    'diagnostic_blob_other',
)
_SHADOW_BLOB_FAILURE_METRICS = {
    'chunk_processing': 'diagnostic_blob_chunk_processing',
    'detector_timeout': 'diagnostic_blob_detector_timeout',
    'network': 'diagnostic_blob_network',
    'transfer_timeout': 'diagnostic_blob_timeout',
    'timeout': 'diagnostic_blob_timeout',
    'mixed': 'diagnostic_blob_mixed',
}


class DockerShadowPrivacyError(RuntimeError):
    pass


def empty_selection_metrics():
    return {name: 0 for name in DOCKER_ADAPTIVE_SHADOW_SELECTION_METRIC_KEYS}


def empty_failure_metrics():
    return {name: 0 for name in SHADOW_FAILURE_METRIC_KEYS}


def _blob_failure_metric(error_code):
    return _SHADOW_BLOB_FAILURE_METRICS.get(
        str(error_code or ''), 'diagnostic_blob_other',
    )


def _canonical_private_plan(plan):
    plan = validate_docker_layer_plan(plan)
    payload = canonical_docker_layer_plan_bytes(plan)
    return plan, hashlib.sha256(payload).hexdigest()


def _validate_duplicate_descriptors(descriptors):
    identities = {}
    for descriptor in descriptors:
        identity = (
            descriptor['kind'], descriptor['size'],
            docker_content_media_class(descriptor['kind'], descriptor['media_type']),
        )
        previous = identities.setdefault(descriptor['digest'], identity)
        if previous != identity:
            raise ValueError('Docker shadow duplicate digest metadata conflicts')


def build_private_adaptive_plan(
    resolved, payload_classes, limits, checkpoint, scan_policy_sha256,
):
    resolved = validate_docker_layer_resolution(resolved)
    limits = validate_docker_layer_limits(limits)
    checkpoint = validate_docker_adaptive_checkpoint(checkpoint)
    scan_policy_sha256 = str(scan_policy_sha256 or '').lower()
    if not _SHA256_RE.fullmatch(scan_policy_sha256):
        raise ValueError('Docker shadow scan policy must be a lowercase SHA-256')

    descriptors = [resolved['config'], *resolved['layers']]
    _validate_duplicate_descriptors(descriptors)
    decisions = select_docker_adaptive_payload(
        descriptors, payload_classes, limits, covered_digests=(),
    )
    metrics = empty_selection_metrics()
    planned = []
    omitted = 0
    for descriptor, selected, reason, payload_class in decisions:
        if reason == 'config_selected':
            metrics['selected_config'] += 1
        elif reason.startswith('selected_'):
            metrics[reason] += 1
        elif reason == 'already_covered':
            metrics['reuse_already_covered'] += 1
        elif reason == 'duplicate_digest':
            metrics['reuse_duplicate_digest'] += 1
        elif not selected:
            metric_name = f'omitted_{reason}'
            if metric_name not in metrics:
                raise ValueError('Docker shadow selection reason is not aggregate-safe')
            metrics[metric_name] += 1
            omitted += 1
        planned.append({
            **descriptor,
            'payload_class': payload_class,
            'selected': bool(selected),
            'selection_reason': reason,
            'coverage_state': 'selected' if selected else 'skipped',
            'lease_token': None,
            'attempt': 0,
            'max_attempts': limits['blob_max_attempts'],
        })

    plan, plan_sha256 = _canonical_private_plan({
        'version': 2,
        'image': resolved['image'],
        'repository': resolved['repository'],
        'manifest_digest': resolved['manifest_digest'],
        'platform_os': resolved['platform_os'],
        'platform_arch': resolved['platform_arch'],
        'manifest_media_type': resolved['manifest_media_type'],
        'limits': limits,
        'selector_version': DOCKER_ADAPTIVE_SELECTOR_VERSION,
        'selection_policy_sha256': docker_layer_selection_policy_sha256(limits),
        'scan_policy_sha256': scan_policy_sha256,
        'execution_policy_sha256': docker_layer_execution_policy_sha256(
            scan_policy_sha256, limits,
        ),
        'checkpoint': checkpoint,
        'descriptors': planned,
    })
    return {
        'plan': plan,
        'plan_sha256': plan_sha256,
        'selection_metrics': validate_docker_adaptive_shadow_selection_metrics(metrics),
        'omitted_descriptor_count': omitted,
    }


def _checkpoint_order(descriptor):
    if descriptor['kind'] == 'config':
        return (0, 0, -descriptor['position'], descriptor['size'], descriptor['digest'])
    try:
        class_rank = DOCKER_ADAPTIVE_LAYER_CLASS_ORDER.index(descriptor['payload_class'])
    except ValueError as exc:
        raise ValueError('Docker shadow descriptor class is not schedulable') from exc
    return (1, class_rank, -descriptor['position'], descriptor['size'], descriptor['digest'])


def lease_private_adaptive_checkpoint(plan, token_factory=None):
    plan = validate_docker_layer_plan(plan)
    if plan['version'] != 2:
        raise ValueError('Docker shadow checkpoints require a version-two plan')
    if any(
        descriptor['coverage_state'] in ('leased', 'shared_pending')
        for descriptor in plan['descriptors']
    ):
        raise ValueError('Docker shadow plan already contains active leases')

    pending = {}
    for descriptor in plan['descriptors']:
        if descriptor['coverage_state'] == 'selected':
            pending.setdefault(descriptor['digest'], []).append(descriptor)
    if not pending:
        return None

    groups = []
    for digest, descriptors in pending.items():
        attempts = {item['attempt'] for item in descriptors}
        maximums = {item['max_attempts'] for item in descriptors}
        if len(attempts) != 1 or len(maximums) != 1:
            raise ValueError('Docker shadow duplicate attempts conflict')
        if next(iter(attempts)) >= next(iter(maximums)):
            raise ValueError('Docker shadow plan exceeded its private attempt budget')
        representative = min(descriptors, key=_checkpoint_order)
        groups.append((representative, digest))
    groups.sort(key=lambda item: _checkpoint_order(item[0]))

    leased_digests = set()
    leased_bytes = 0
    max_blobs = plan['checkpoint']['max_blobs']
    max_bytes = plan['checkpoint']['max_bytes']
    for descriptor, digest in groups:
        if len(leased_digests) >= max_blobs:
            continue
        if leased_digests and leased_bytes + descriptor['size'] > max_bytes:
            continue
        leased_digests.add(digest)
        leased_bytes += descriptor['size']
    if not leased_digests:
        raise ValueError('Docker shadow checkpoint made no bounded progress')

    token_factory = token_factory or (lambda: secrets.token_urlsafe(32))
    tokens = {digest: str(token_factory()) for digest in leased_digests}
    leased_plan = copy.deepcopy(plan)
    for descriptor in leased_plan['descriptors']:
        if descriptor['digest'] not in leased_digests:
            continue
        descriptor['coverage_state'] = 'leased'
        descriptor['lease_token'] = tokens[descriptor['digest']]
        descriptor['attempt'] += 1
    leased_plan, plan_sha256 = _canonical_private_plan(leased_plan)
    return {'plan': leased_plan, 'plan_sha256': plan_sha256}


def apply_private_adaptive_execution(plan, execution):
    plan, plan_sha256 = _canonical_private_plan(plan)
    execution = validate_docker_layer_execution(execution, plan, plan_sha256)
    records = {record['digest']: record for record in execution['blobs']}
    next_plan = copy.deepcopy(plan)
    for descriptor in next_plan['descriptors']:
        if descriptor['coverage_state'] != 'leased':
            continue
        status = records[descriptor['digest']]['status']
        if status == 'covered':
            next_state = 'covered'
        elif status == 'retryable_failed' and descriptor['attempt'] < descriptor['max_attempts']:
            next_state = 'selected'
        else:
            next_state = 'terminal_failed'
        descriptor['coverage_state'] = next_state
        descriptor['lease_token'] = None
    return _canonical_private_plan(next_plan)[0]


def private_result_identities(result, normalized_target):
    if not isinstance(result, dict):
        raise ValueError('Docker shadow private result must be an object')
    routed = set()
    detectors = set()
    try:
        scanner.strip_nearby_context_for_persistence(result)
        findings = result.get('findings') or []
        if not isinstance(findings, list):
            raise ValueError('Docker shadow private findings must be a list')
        for finding in findings:
            if not isinstance(finding, dict):
                raise ValueError('Docker shadow private finding must be an object')
            detector_hash = str(
                finding_identity('dockerhub', normalized_target, finding)[2] or ''
            ).lower()
            if not _SHA256_RE.fullmatch(detector_hash):
                raise DockerShadowPrivacyError('Docker shadow detector identity is invalid')
            detectors.add(detector_hash)
            for candidate in extract_candidates(finding):
                service = str(candidate.service or '')
                provider_key_hash = str(candidate.provider_key_hash or '').lower()
                if (
                    not _ROUTED_SERVICE_RE.fullmatch(service)
                    or not _SHA256_RE.fullmatch(provider_key_hash)
                ):
                    raise DockerShadowPrivacyError('Docker shadow routed identity is invalid')
                routed.add((service, provider_key_hash))
    finally:
        findings = result.get('findings') if isinstance(result, dict) else None
        if isinstance(findings, list):
            for finding in findings:
                if isinstance(finding, dict):
                    finding.clear()
            findings.clear()
        result.clear()
    return frozenset(routed), frozenset(detectors)


def run_timed_private_side(
    side, operation, durable_checkpoint, *, timeout_sec,
    monotonic_ns=time.monotonic_ns,
):
    if side not in ('full', 'adaptive'):
        raise ValueError('Docker shadow side is invalid')
    with scanner.scan_slot_scope(['docker-shadow', side], timeout_sec=timeout_sec):
        started_ns = monotonic_ns()
        metrics = operation()
        if not isinstance(metrics, dict) or any(
            not isinstance(name, str)
            or isinstance(value, bool)
            or not isinstance(value, int)
            or value < 0
            for name, value in metrics.items()
        ):
            raise ValueError('Docker shadow private sink accepts only non-negative aggregates')
        durable_checkpoint()
        elapsed_ns = max(1, monotonic_ns() - started_ns)
    return metrics, max(1, math.ceil(elapsed_ns / 1_000_000))


def _clear_private_result(result):
    if not isinstance(result, dict):
        return
    findings = result.get('findings')
    if isinstance(findings, list):
        for finding in findings:
            if isinstance(finding, dict):
                finding.clear()
        findings.clear()
    result.clear()


def _full_scan_incomplete_reason(result):
    if result.get('source_failure'):
        return 'full_scan_source_failure'
    if result.get('skipped'):
        return 'full_scan_skipped'
    return 'full_scan_error'


def _adaptive_scan_incomplete_reason(exc):
    if isinstance(exc, scanner.DockerLayerInfrastructureError):
        return 'adaptive_scan_infrastructure'
    if isinstance(exc, scanner.DockerContentTransferError):
        return 'adaptive_scan_transfer'
    if isinstance(exc, TimeoutError):
        return 'adaptive_scan_timeout'
    if isinstance(exc, scanner.DockerRemoteAccessError):
        return 'adaptive_scan_remote_access'
    if isinstance(exc, ValueError):
        return 'adaptive_scan_contract'
    return 'adaptive_scan_error'


def shadow_failure_reason_code(exc):
    if isinstance(exc, DockerShadowPrivacyError):
        return 'privacy_violation'
    if isinstance(exc, (KeyboardInterrupt, SystemExit)):
        return 'operator_interrupt'
    if exc.__class__.__name__ == 'ScanSlotFatalError':
        return 'scan_slot_fatal'
    if isinstance(exc, TimeoutError):
        return 'timeout'
    if isinstance(exc, ValueError):
        return 'invalid_contract'
    if isinstance(exc, RuntimeError):
        message = str(exc).lower()
        if 'checkpoint' in message:
            return 'checkpoint_failure'
        if 'fence' in message or 'lease' in message:
            return 'report_fence_failure'
        if 'cohort' in message or 'control' in message:
            return 'control_failure'
        return 'runtime_failure'
    return 'unexpected_failure'


def execute_private_adaptive_scan(
    normalized_target, *, limits, checkpoint, scan_policy_sha256, timeout_sec,
    platform_os='linux', platform_arch='amd64', min_free_bytes=0,
    scan_kwargs=None,
):
    timeout_sec = max(1, int(timeout_sec))
    deadline = time.monotonic() + timeout_sec
    scan_kwargs = dict(scan_kwargs or {})
    resolved, bearer_auth = scanner.resolve_docker_content_manifest(
        normalized_target,
        platform_os=str(platform_os or 'linux'),
        platform_arch=str(platform_arch or 'amd64'),
        deadline=deadline,
    )
    payload_classes, bearer_auth = scanner.fetch_docker_config_payload_classes(
        resolved, bearer_auth, deadline=deadline,
        min_free_bytes=max(0, int(min_free_bytes or 0)),
    )
    built = build_private_adaptive_plan(
        resolved, payload_classes, limits, checkpoint, scan_policy_sha256,
    )
    plan = built['plan']
    selection_metrics = dict(built['selection_metrics'])
    failure_metrics = empty_failure_metrics()
    routed = set()
    detectors = set()
    selected_unique = {
        item['digest']: item['max_attempts']
        for item in plan['descriptors'] if item['selected']
    }
    max_checkpoints = max(1, sum(selected_unique.values()))
    checkpoint_count = 0
    while True:
        leased = lease_private_adaptive_checkpoint(plan)
        if leased is None:
            break
        checkpoint_count += 1
        if checkpoint_count > max_checkpoints or time.monotonic() >= deadline:
            raise RuntimeError('Docker shadow adaptive checkpoint budget was exhausted')
        work = {
            'plan': leased['plan'],
            'plan_sha256': leased['plan_sha256'],
            'bearer_auth': bearer_auth,
            'min_free_bytes': max(0, int(min_free_bytes or 0)),
            'deadline': deadline,
        }
        result = scanner.scan_docker_layer_plan(
            normalized_target, work,
            timeout_sec=max(1, math.ceil(deadline - time.monotonic())),
            detectors=scan_kwargs.get('detectors'),
            exclude_detectors=scan_kwargs.get('exclude_detectors'),
            no_verification=bool(scan_kwargs.get('no_verification', False)),
            trufflehog_config=scan_kwargs.get('trufflehog_config'),
            log_target=False,
        )
        try:
            execution = result.get('docker_layer_execution')
            next_plan = apply_private_adaptive_execution(
                leased['plan'], result.get('docker_layer_execution'),
            )
            records = {
                record['digest']: record for record in execution['blobs']
            }
            newly_terminal = {
                item['digest'] for item in next_plan['descriptors']
                if item['coverage_state'] == 'terminal_failed'
                and item['digest'] in records
            }
            for digest in newly_terminal:
                failure_metrics[
                    _blob_failure_metric(records[digest]['error_code'])
                ] += 1
            plan = next_plan
            checkpoint_routed, checkpoint_detectors = private_result_identities(
                result, normalized_target,
            )
        except Exception:
            _clear_private_result(result)
            raise
        routed.update(checkpoint_routed)
        detectors.update(checkpoint_detectors)
        del checkpoint_routed, checkpoint_detectors

    if any(
        item['coverage_state'] in ('selected', 'leased', 'shared_pending')
        for item in plan['descriptors']
    ):
        raise RuntimeError('Docker shadow adaptive plan did not reach a terminal state')
    selection_metrics['adaptive_checkpoints'] += checkpoint_count
    terminal_digests = {
        item['digest'] for item in plan['descriptors']
        if item['coverage_state'] == 'terminal_failed'
    }
    if sum(failure_metrics.values()) != len(terminal_digests):
        raise RuntimeError('Docker shadow terminal failure accounting is inconsistent')
    return {
        'routed': frozenset(routed),
        'detectors': frozenset(detectors),
        'selection_metrics': validate_docker_adaptive_shadow_selection_metrics(
            selection_metrics,
        ),
        'omitted_descriptor_count': int(built['omitted_descriptor_count']),
        'failure_count': len(terminal_digests),
        'failure_metrics': failure_metrics,
    }


def private_full_side_metrics(db, control, scan_kwargs):
    reference_routed, reference_detectors = db.docker_adaptive_shadow_control_identities(
        control['target_scan_id'],
    )
    reference_routed = set(reference_routed)
    reference_detectors = set(reference_detectors)
    rerun_routed = set()
    rerun_detectors = set()
    result = None
    try:
        max_attempts = min(10, max(1, int(scan_kwargs.get('max_attempts', 3) or 3)))
        incomplete_reason = 'full_scan_error'
        attempts = 0
        for attempts in range(1, max_attempts + 1):
            result = scanner.scan_docker_image(
                control['normalized_target'],
                timeout_sec=int(scan_kwargs['timeout_sec']),
                detectors=scan_kwargs.get('detectors'),
                exclude_detectors=scan_kwargs.get('exclude_detectors'),
                no_verification=bool(scan_kwargs.get('no_verification', False)),
                trufflehog_config=scan_kwargs.get('trufflehog_config'),
                config_dir=scanner.docker_token_manager.get_next_config(),
                trufflehog_concurrency=int(scan_kwargs.get('trufflehog_concurrency', 0) or 0),
                docker_recovery_limits=scan_kwargs.get('docker_recovery_limits'),
                docker_recovery_min_free_bytes=int(scan_kwargs.get('docker_recovery_min_free_bytes', 20 << 30)),
                log_target=False,
            )
            if not isinstance(result, dict):
                raise ValueError('Docker shadow full result must be an object')
            if not (
                result.get('errors') or result.get('skipped')
                or result.get('source_failure')
                or result.get('warnings') or result.get('degraded')
                or ((result.get('scan_meta') or {}).get('docker_layer_scope') or {}).get('coverage_complete') is False
                or ((result.get('scan_meta') or {}).get('docker_full_recovery') or {}).get('coverage_complete') is False
            ):
                rerun_routed, rerun_detectors = map(
                    set, private_result_identities(result, control['normalized_target']),
                )
                result = None
                return {
                    'full_routed_count': len(reference_routed),
                    'full_detector_count': len(reference_detectors),
                    'failure_count': 0,
                    'safety_regression_count': int(
                        rerun_routed != reference_routed
                        or rerun_detectors != reference_detectors
                    ),
                    **empty_failure_metrics(),
                }
            incomplete_reason = _full_scan_incomplete_reason(result)
            retryable = bool(result.get('retryable', True))
            _clear_private_result(result)
            result = None
            if not retryable:
                break
        print(
            'Docker adaptive shadow full side incomplete: '
            f'reason_code={incomplete_reason} attempts={attempts}'
        )
        metrics = empty_failure_metrics()
        metrics['diagnostic_full_incomplete'] = 1
        metrics.update({
            'full_routed_count': len(reference_routed),
            'full_detector_count': len(reference_detectors),
            'failure_count': 1,
            'safety_regression_count': 0,
        })
        return metrics
    finally:
        _clear_private_result(result)
        reference_routed.clear()
        reference_detectors.clear()
        rerun_routed.clear()
        rerun_detectors.clear()


def private_adaptive_side_metrics(
    db, control, *, limits, checkpoint, scan_policy_sha256, scan_kwargs,
    platform_os='linux', platform_arch='amd64', min_free_bytes=0,
):
    reference_routed, reference_detectors = db.docker_adaptive_shadow_control_identities(
        control['target_scan_id'],
    )
    reference_routed = set(reference_routed)
    reference_detectors = set(reference_detectors)
    adaptive_routed = set()
    adaptive_detectors = set()
    outcome = None
    try:
        max_attempts = min(10, max(1, int(scan_kwargs.get('max_attempts', 3) or 3)))
        incomplete_reason = 'adaptive_scan_error'
        attempts = 0
        for attempts in range(1, max_attempts + 1):
            try:
                outcome = execute_private_adaptive_scan(
                    control['normalized_target'], limits=limits, checkpoint=checkpoint,
                    scan_policy_sha256=scan_policy_sha256,
                    timeout_sec=int(scan_kwargs['timeout_sec']),
                    platform_os=platform_os, platform_arch=platform_arch,
                    min_free_bytes=min_free_bytes, scan_kwargs=scan_kwargs,
                )
                adaptive_routed = set(outcome.pop('routed'))
                adaptive_detectors = set(outcome.pop('detectors'))
                metrics = {
                    'adaptive_routed_count': len(adaptive_routed),
                    'routed_intersection_count': len(reference_routed & adaptive_routed),
                    'adaptive_detector_count': len(adaptive_detectors),
                    'detector_intersection_count': len(reference_detectors & adaptive_detectors),
                    'omitted_descriptor_count': int(outcome['omitted_descriptor_count']),
                    'failure_count': int(outcome['failure_count']),
                }
                metrics.update(outcome['selection_metrics'])
                metrics.update(outcome['failure_metrics'])
                return metrics
            except DockerShadowPrivacyError:
                raise
            except scanner.ScanSlotFatalError:
                raise
            except MemoryError:
                raise
            except Exception as exc:
                incomplete_reason = _adaptive_scan_incomplete_reason(exc)
                retryable = bool(getattr(exc, 'retryable', not isinstance(exc, ValueError)))
                if not retryable:
                    break
            finally:
                if isinstance(outcome, dict):
                    outcome.clear()
                outcome = None
                adaptive_routed.clear()
                adaptive_detectors.clear()
        print(
            'Docker adaptive shadow adaptive side incomplete: '
            f'reason_code={incomplete_reason} attempts={attempts}'
        )
        metrics = empty_selection_metrics()
        metrics.update(empty_failure_metrics())
        metrics['diagnostic_adaptive_incomplete'] = 1
        metrics.update({
            'adaptive_routed_count': 0,
            'routed_intersection_count': 0,
            'adaptive_detector_count': 0,
            'detector_intersection_count': 0,
            'omitted_descriptor_count': 0,
            'failure_count': 1,
        })
        return metrics
    finally:
        if isinstance(outcome, dict):
            outcome.clear()
        reference_routed.clear()
        reference_detectors.clear()
        adaptive_routed.clear()
        adaptive_detectors.clear()


def _shadow_config(config):
    supervisor = config.get('supervisor') if isinstance(config, dict) else None
    supervisor = supervisor if isinstance(supervisor, dict) else {}
    value = supervisor.get('docker_shadow')
    if not isinstance(value, dict):
        raise ValueError('Docker shadow supervisor configuration is required')
    allowed = {'enabled', 'cohort_size', 'lease_seconds'}
    if set(value) - allowed:
        raise ValueError('Docker shadow supervisor configuration has unsupported fields')
    if not console_runner.bool_config(value.get('enabled'), False):
        raise ValueError('Docker shadow operator command is disabled')
    cohort_size = int(value.get('cohort_size', DOCKER_ADAPTIVE_GATE_MIN_CONTROLS))
    lease_seconds = int(value.get('lease_seconds', 3600))
    if not DOCKER_ADAPTIVE_GATE_MIN_CONTROLS <= cohort_size <= DOCKER_ADAPTIVE_GATE_MAX_CONTROLS:
        raise ValueError('Docker shadow cohort size must be between 50 and 100')
    if not 60 <= lease_seconds <= 86400:
        raise ValueError('Docker shadow lease duration is outside the supported range')
    return {'cohort_size': cohort_size, 'lease_seconds': lease_seconds}


def _scan_kwargs(source_args):
    return {
        'timeout_sec': max(1, int(source_args.timeout)),
        'detectors': source_args.detectors,
        'exclude_detectors': source_args.exclude_detectors,
        'no_verification': bool(source_args.no_verification),
        'trufflehog_config': source_args.trufflehog_config,
        'trufflehog_concurrency': max(0, int(source_args.trufflehog_concurrency or 0)),
        'max_attempts': max(
            1, int(getattr(source_args, 'target_retry_max_attempts', 3) or 3),
        ),
    }


def run_shadow(config_path):
    require_active_supervisor_child(
        config_path, child_kind='docker-shadow', require_dsn=True,
    )
    config = console_runner.load_config(config_path)
    settings = _shadow_config(config)
    source_config = (config.get('sources') or {}).get('dockerhub')
    if not isinstance(source_config, dict):
        raise ValueError('Docker shadow requires the DockerHub source configuration')
    global_config = config.get('global') or {}
    if not isinstance(global_config, dict):
        raise ValueError('Docker shadow global configuration must be a mapping')

    console_runner.apply_global_config(global_config)
    scanner.initialize_scanner_runtime(preflight_complete=True)
    secrets_config = console_runner.load_secrets(config, config_path)
    state = {
        'version': 1,
        'sources': {'dockerhub': console_runner.default_source_state()},
    }
    console_runner.configure_source_auth(
        'dockerhub', source_config, state=state, secrets=secrets_config,
    )
    source_args = console_runner.build_args_from_source_config(
        'dockerhub', source_config, global_config, '', auth_entry=None,
    )
    scanner.scan_config.drop_detectors = scanner.csv_items(source_args.drop_detectors)
    scanner.scan_config.trufflehog_job_memory_limit_bytes = int(
        source_args.trufflehog_job_memory_limit_bytes
    )
    scan_kwargs = _scan_kwargs(source_args)
    limits = console_runner.docker_layer_limits(source_args)
    scan_kwargs['docker_recovery_limits'] = limits
    scan_kwargs['docker_recovery_min_free_bytes'] = int(getattr(source_args, 'docker_layer_min_free_bytes', 20 << 30))
    checkpoint = console_runner.docker_adaptive_checkpoint(source_args)
    scan_policy_sha256 = console_runner.docker_layer_scan_policy_sha256(
        source_args, scan_kwargs,
    )
    execution_policy_sha256 = docker_layer_execution_policy_sha256(
        scan_policy_sha256, limits,
    )
    selection_policy_sha256 = docker_layer_selection_policy_sha256(limits)
    selection_salt = hashlib.sha256(
        ('docker-shadow-controls-v1:' + scan_policy_sha256 + ':'
         + execution_policy_sha256 + ':' + selection_policy_sha256).encode('ascii')
    ).hexdigest()

    db_url = str(os.getenv('TRUF_MANAGED_POSTGRES_DSN') or '')
    if not db_url:
        raise RuntimeError('Docker shadow canonical PostgreSQL authority is unavailable')
    db = ScannerDB(db_url=db_url, initialize=False)
    controls = []
    report = None
    owner = 'docker-shadow:{}:{}'.format(
        os.getpid(), hashlib.sha256(socket.gethostname().encode('utf-8')).hexdigest()[:16],
    )
    try:
        if not db.enabled:
            raise RuntimeError('Docker shadow PostgreSQL connection is unavailable')
        db.set_application_name('truf-docker-adaptive-shadow')
        controls = db.docker_adaptive_shadow_controls(
            scan_policy_sha256, settings['cohort_size'], selection_salt,
        )
        report = db.start_docker_adaptive_shadow_report(
            scan_policy_sha256, execution_policy_sha256, selection_policy_sha256,
            settings['cohort_size'], owner, lease_seconds=settings['lease_seconds'],
        )
        aggregate = {
            'completed_pairs': 0,
            'full_routed_count': 0,
            'adaptive_routed_count': 0,
            'routed_intersection_count': 0,
            'full_detector_count': 0,
            'adaptive_detector_count': 0,
            'detector_intersection_count': 0,
            'full_slot_ms': 0,
            'adaptive_slot_ms': 0,
            'omitted_descriptor_count': 0,
            'failure_count': 0,
            'privacy_violation_count': 0,
            'safety_regression_count': 0,
        }
        selection_metrics = empty_selection_metrics()
        failure_metrics = empty_failure_metrics()

        def durable_checkpoint():
            db.checkpoint_docker_adaptive_shadow_report(
                report['report_token'], owner, report['lease_token'],
                lease_seconds=settings['lease_seconds'],
            )

        for index, control in enumerate(controls):
            sides = ('full', 'adaptive') if index % 2 == 0 else ('adaptive', 'full')
            for side in sides:
                if side == 'full':
                    operation = lambda control=control: private_full_side_metrics(
                        db, control, scan_kwargs,
                    )
                else:
                    operation = lambda control=control: private_adaptive_side_metrics(
                        db, control, limits=limits, checkpoint=checkpoint,
                        scan_policy_sha256=scan_policy_sha256,
                        scan_kwargs=scan_kwargs,
                        platform_os=source_args.docker_platform_os,
                        platform_arch=source_args.docker_platform_arch,
                        min_free_bytes=source_args.docker_layer_min_free_bytes,
                    )
                side_metrics, elapsed_ms = run_timed_private_side(
                    side, operation, durable_checkpoint,
                    timeout_sec=scan_kwargs['timeout_sec'],
                )
                aggregate[f'{side}_slot_ms'] += elapsed_ms
                for name, value in side_metrics.items():
                    if name in selection_metrics:
                        selection_metrics[name] += value
                    elif name in failure_metrics:
                        failure_metrics[name] += value
                    else:
                        aggregate[name] += value
                side_metrics.clear()
            aggregate['completed_pairs'] += 1
            control.clear()

        completed = db.finish_docker_adaptive_shadow_report(
            report['report_token'], owner, report['lease_token'],
            selection_metrics=selection_metrics, **aggregate,
        )
        print(
            'Docker adaptive shadow report: '
            f'id={completed["report_id"]} passed={str(completed["passed"]).lower()} '
            f'controls={completed["completed_pairs"]} '
            f'routed_recall_ppm={completed["routed_recall_ppm"]} '
            f'slot_ratio_ppm={completed["slot_ratio_ppm"]} '
            'failure_categories=' + ','.join(
                f'{name.removeprefix("diagnostic_")}:{failure_metrics[name]}'
                for name in SHADOW_FAILURE_METRIC_KEYS
                if failure_metrics[name]
            )
        )
        return 0 if completed['passed'] else 2
    except BaseException as exc:
        if report is not None:
            try:
                db.fail_docker_adaptive_shadow_report(
                    report['report_token'], owner, report['lease_token'],
                    privacy_violation_count=int(isinstance(exc, DockerShadowPrivacyError)),
                    safety_regression_count=0,
                )
            except Exception:
                pass
        print(
            'Docker adaptive shadow report failed safely: '
            f'reason_code={shadow_failure_reason_code(exc)}'
        )
        return 1
    finally:
        for control in controls:
            if isinstance(control, dict):
                control.clear()
        controls.clear()
        db.close()
        scanner.docker_token_manager.cleanup()


def parse_args(argv=None):
    parser = argparse.ArgumentParser(description='Run one private Docker adaptive shadow report')
    parser.add_argument('--config', required=True)
    return parser.parse_args(argv)


def main(argv=None):
    args = parse_args(argv)
    return run_shadow(args.config)


if __name__ == '__main__':
    raise SystemExit(main())