truf-server/tests/test_docker_recovery_diagnostics.py

import json
import logging
from pathlib import Path
import shutil
import sys
from types import SimpleNamespace

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'app'))
import scanner
import scanner_db


TARGET = 'offline/diagnostics@sha256:' + 'a' * 64
PRIVATE = 'private-diagnostic-sentinel'
CONFIG_MEDIA = 'application/vnd.oci.image.config.v1+json'
LAYER_MEDIA = 'application/vnd.oci.image.layer.v1.tar'
FINISHED = json.dumps({'level': 'info-0', 'logger': 'trufflehog', 'msg': 'finished scanning'})
GZIP_WARNING = json.dumps({'level': 'error', 'msg': 'error processing layer', 'error': 'gzip: invalid header'})
EXACT_EOF = {'level': 'error', 'msg': 'error processing layer', 'error': 'unexpected EOF'}
DEFAULT_LIMITS = {
    'config_max_bytes': 1 << 20, 'layer_max_bytes': 256 << 20,
    'image_max_bytes': 1 << 30, 'max_layers': 8,
    'archive_max_size_bytes': 256 << 20, 'archive_max_depth': 4,
    'archive_timeout_sec': 30, 'blob_timeout_sec': 600,
    'filesystem_concurrency': 2, 'blob_max_attempts': 3,
}


@pytest.fixture(autouse=True)
def docker_diagnostics_offline_only(monkeypatch):
    def forbidden(*args, **kwargs):
        pytest.fail('Docker diagnostic unit tests cannot use network, databases, runtime, or child processes')

    for owner, name in (
        (scanner.socket.socket, 'connect'), (scanner.socket.socket, 'connect_ex'),
        (scanner.requests.sessions.Session, 'request'), (scanner.subprocess, 'Popen'),
        (scanner.sqlite3, 'connect'), (scanner_db, 'ScannerDB'),
        (scanner, 'initialize_scanner_runtime'), (scanner, 'run_command_streamed'),
        (scanner, 'resolve_docker_content_manifest'), (scanner, 'docker_registry_bearer_token'),
        (scanner, 'stream_docker_registry_blob'), (scanner, '_scan_docker_content_file'),
    ):
        monkeypatch.setattr(owner, name, forbidden)


@pytest.fixture
def docker_diagnostics_case(tmp_path, monkeypatch, docker_diagnostics_offline_only):
    config = {'digest': 'sha256:' + 'b' * 64, 'size': 1024, 'media_type': CONFIG_MEDIA}
    layer = {'digest': 'sha256:' + 'c' * 64, 'size': 1024, 'media_type': LAYER_MEDIA}
    state = SimpleNamespace(
        resolved={'image': TARGET, 'repository': 'offline/diagnostics',
                  'manifest_digest': TARGET.split('@')[1], 'config': config, 'layers': [layer]},
        stderr=GZIP_WARNING + '\n' + FINISHED, returncode=0, blob_stderr=FINISHED,
        resolutions=[], downloads=[], scans=[], commands=[], now=100.0,
        resolve_error=None, transfer_error=None, scan_error=None, after_native=lambda: None,
    )
    monkeypatch.delenv('DOCKER_CONFIG', raising=False)
    monkeypatch.setattr(scanner, '_docker_implicit_auth_present', lambda: False)
    monkeypatch.setattr(scanner, 'time', SimpleNamespace(monotonic=lambda: state.now))
    monkeypatch.setattr(scanner, 'get_work_dir', lambda: str(tmp_path))
    monkeypatch.setattr(scanner, 'harden_private_directory', lambda *a, **k: None)
    monkeypatch.setattr(scanner, 'write_temp_owner', lambda *a, **k: None)
    monkeypatch.setattr(scanner, 'cleanup_command_work_dir', shutil.rmtree)
    monkeypatch.setattr(scanner, 'apply_finding_filters', lambda result, *a, **k: result)
    monkeypatch.setattr(scanner, 'attach_docker_content_provenance', lambda findings, *a: findings)

    def resolve(*args, **kwargs):
        state.resolutions.append(kwargs)
        if state.resolve_error:
            raise state.resolve_error
        return state.resolved, None

    def download(repository, descriptor, destination, bearer_auth, **kwargs):
        state.downloads.append((descriptor, kwargs))
        if state.transfer_error:
            raise state.transfer_error
        return SimpleNamespace(bearer_auth=bearer_auth)

    def scan(destination, descriptor, limits, deadline, *args):
        state.scans.append((descriptor, limits, deadline))
        if state.scan_error:
            raise state.scan_error
        return scanner.apply_trufflehog_diagnostics(
            {'findings': [], 'errors': []}, state.blob_stderr, 0, 'filesystem', require_completion=True,
        )

    def command(args, timeout, env, **kwargs):
        state.commands.append((timeout, kwargs))
        state.after_native()
        return scanner.streamed_output_from_text(
            stdout=json.dumps({'DetectorName': 'OfflineRetained', 'Raw': 'synthetic-finding'}),
            stderr=state.stderr, returncode=state.returncode,
        )

    monkeypatch.setattr(scanner, 'resolve_docker_content_manifest', resolve)
    monkeypatch.setattr(scanner, 'stream_docker_registry_blob', download)
    monkeypatch.setattr(scanner, '_scan_docker_content_file', scan)
    monkeypatch.setattr(scanner, 'run_command_streamed', command)
    state.run = lambda **kwargs: scanner.scan_docker_image(
        TARGET, timeout_sec=600, no_verification=True, log_target=False, **kwargs,
    )
    return state


def assert_preflight_failure(case, result, error_class, diagnostic):
    assert result['errors'] and result['error_class'] == error_class
    assert result['retryable'] is False and not result.get('source_failure')
    assert len(result['findings']) == 1
    assert not case.downloads and not case.scans
    scope = result['scan_meta']['docker_full_recovery']
    assert scope['coverage_complete'] is False and scope['blob_transfer_attempted'] is False
    assert scope['scanned_descriptors'] == 0
    assert scope['diagnostic'] == {'phase': 'preflight', **diagnostic}
    for value in (PRIVATE, TARGET, TARGET.split('@')[1]):
        assert value not in json.dumps(result)
    return scope


@pytest.mark.parametrize('source', ['docker', 'filesystem', 'git'])
@pytest.mark.parametrize('returncode', [0, 1, -1])
def test_exact_eof_retains_public_error_policy(source, returncode):
    assert scanner._trufflehog_diagnostic_policy(json.dumps(EXACT_EOF), source, returncode) == (
        'error', 'network', True,
    )


@pytest.mark.parametrize('codec_warning,finished,returncode', [
    (True, True, 0), (False, True, 0), (True, False, 0), (True, True, 1), (False, False, -1),
])
def test_exact_eof_has_only_fixed_ambiguous_metadata(
    docker_diagnostics_case, caplog, codec_warning, finished, returncode,
):
    case = docker_diagnostics_case
    caplog.set_level(logging.INFO, logger='scanner')
    payload = dict(EXACT_EOF, target=TARGET, digest=TARGET.split('@')[1],
                   url='https://example.invalid/' + PRIVATE)
    case.stderr = '\n'.join([
        *([GZIP_WARNING] * 32 if codec_warning else []), json.dumps(payload),
        *([FINISHED] if finished else []),
    ])
    case.returncode = returncode
    result = case.run()
    expected_class = 'mixed' if codec_warning and returncode != 0 else 'network'
    assert result['errors'] and result['error_class'] == expected_class
    assert result['retryable'] is True and result['source_failure'] is False
    meta = result['scan_meta']
    assert meta['trufflehog_finished'] is finished and meta['trufflehog_returncode'] == returncode
    assert meta['docker_native_diagnostic'] == {
        'phase': 'native_layer_processing', 'subcause': 'unexpected_eof_ambiguous',
        'coverage_complete': False,
    }
    assert not case.resolutions and not case.downloads
    if codec_warning and returncode == 0:
        assert meta['docker_full_recovery'] == {
            'coverage_complete': False, 'blob_transfer_attempted': False,
            'diagnostic': {'phase': 'native_diagnostics', 'reason': 'unrelated_diagnostics'},
        }
    else:
        assert 'docker_full_recovery' not in meta
    for value in (PRIVATE, TARGET, TARGET.split('@')[1], 'https://'):
        assert value not in json.dumps(meta) and value not in caplog.text


@pytest.mark.parametrize('payload,expected_class', [
    (dict(EXACT_EOF, error='unexpected EOF trailing detail'), 'network'),
    (dict(EXACT_EOF, error='unexpected eof'), 'network'),
    (dict(EXACT_EOF, error=' unexpected EOF'), 'network'),
    (dict(EXACT_EOF, msg='error reading chunk'), 'network'),
    (dict(EXACT_EOF, msg='Error processing layer'), 'network'),
    ({'msg': 'error processing layer', 'message': 'unexpected EOF'}, 'network'),
    ({'msg': 'error processing layer', 'errors': ['unexpected EOF']}, 'network'),
    (dict(EXACT_EOF, error=['unexpected EOF']), 'network'),
    (dict(EXACT_EOF, error='connection reset'), 'network'),
    (dict(EXACT_EOF, error='EOF'), 'trufflehog'),
    ('error processing layer: unexpected EOF', 'network'),
])
def test_other_eof_and_network_errors_do_not_gain_layer_subcause(docker_diagnostics_case, payload, expected_class):
    case = docker_diagnostics_case
    line = payload if isinstance(payload, str) else json.dumps(payload)
    case.stderr = '\n'.join([GZIP_WARNING, line, FINISHED])
    result = case.run()
    assert result['error_class'] == expected_class and result['retryable'] is True
    assert 'docker_native_diagnostic' not in result['scan_meta']
    assert not result['scan_meta']['docker_full_recovery']['coverage_complete']
    assert not case.resolutions and not case.downloads


@pytest.mark.parametrize('embedded', [False, True])
def test_independent_fatal_error_still_blocks_eof_recovery(docker_diagnostics_case, embedded):
    case = docker_diagnostics_case
    lines = [json.dumps(dict(EXACT_EOF, errors=['unknown flag']))] if embedded else [
        json.dumps(EXACT_EOF), json.dumps({'level': 'error', 'error': 'unknown flag'}),
    ]
    case.stderr = '\n'.join([GZIP_WARNING, *lines, FINISHED])
    result = case.run()
    assert result['error_class'] == ('source_configuration' if embedded else 'mixed')
    assert result['retryable'] is False and result['source_failure'] is True
    assert result['scan_meta']['docker_native_diagnostic']['coverage_complete'] is False
    assert result['scan_meta']['docker_full_recovery']['diagnostic']['reason'] == 'unrelated_diagnostics'
    assert not case.resolutions and not case.downloads


@pytest.mark.parametrize('details,expected_class,retryable', [
    ({'message': 'unexpected EOF'}, 'network', True),
    ({'message': 'unexpected EOF', 'errors': ['gzip: invalid header']}, 'network', True),
    ({'message': 'unknown flag'}, 'source_configuration', False),
    ({'message': 'unauthorized'}, 'docker_registry_access', False),
    ({'message': 'permission denied'}, 'auth_or_permission', True),
    ({'message': PRIVATE}, 'trufflehog', True),
    ({'message': {'detail': PRIVATE}}, 'trufflehog', True),
    ({'message': 'GZIP: INVALID HEADER'}, 'trufflehog', True),
    ({'message': ' '}, 'trufflehog', True),
    ({'error': 'unexpected EOF', 'message': 'gzip: invalid header'}, 'network', True),
    ({'error': 'unexpected EOF', 'message': 'unknown flag'}, 'source_configuration', False),
    ({'message': 'unexpected EOF', 'errors': ['unknown flag']}, 'source_configuration', False),
    ({'message': 'unknown flag', 'errors': ['unauthorized']}, 'source_configuration', False),
    ({'message': 'gzip: invalid header', 'errors': ['unexpected EOF']}, 'network', True),
], ids=['review-dual-eof', 'review-plural-dual-eof', 'review-dual-configuration', 'registry-auth',
        'permission', 'unknown', 'non-string', 'case-variant', 'blank', 'reversed-details',
        'two-fatal-details', 'plural-configuration', 'plural-auth-priority', 'duplicate-with-fatal-plural'])
def test_distinct_docker_detail_channels_cannot_be_cleared(
    docker_diagnostics_case, caplog, details, expected_class, retryable,
):
    case = docker_diagnostics_case
    caplog.set_level(logging.INFO, logger='scanner')
    line = json.dumps(dict(json.loads(GZIP_WARNING), **details))
    case.stderr = line + '\n' + FINISHED
    result = case.run()
    assert result['errors'] and not case.resolutions and not case.downloads and not case.scans
    assert result['error_class'] == expected_class and result['retryable'] is retryable
    assert result['source_failure'] is (expected_class == 'source_configuration')
    assert scanner._trufflehog_diagnostic_policy(line, 'docker', 0) == ('error', expected_class, retryable)
    meta = result['scan_meta']
    assert meta['trufflehog_finished'] and meta['trufflehog_returncode'] == 0
    assert not meta.get('docker_full_recovery', {}).get('coverage_complete')
    assert PRIVATE not in json.dumps(meta) and PRIVATE not in caplog.text


@pytest.mark.parametrize('details', [
    {}, {'message': None}, {'message': ''}, {'message': 'gzip: invalid header'},
    {'message': 'gzip: invalid header', 'errors': ['gzip: invalid header']},
], ids=['single-channel', 'null', 'empty', 'exact-duplicate', 'duplicate-with-plural'])
def test_clean_or_duplicate_codec_detail_keeps_full_recovery(docker_diagnostics_case, details):
    case = docker_diagnostics_case
    line = json.dumps(dict(json.loads(GZIP_WARNING), **details))
    case.stderr = line + '\n' + FINISHED
    assert scanner._trufflehog_diagnostic_policy(line, 'docker', 0) == ('warning', 'docker_layer_gzip', False)
    result = case.run()
    scope = result['scan_meta']['docker_full_recovery']
    assert not result['errors'] and not result.get('warnings') and not result.get('degraded')
    assert scope['coverage_complete'] and scope['recovered_codec']
    assert scope['scanned_descriptors'] == scope['descriptor_count'] == 2
    assert len(case.resolutions) == 1 and len(case.downloads) == len(case.scans) == 2


@pytest.mark.parametrize('causes,settings,expected_policy', [
    ('unexpected EOF', {}, ('error', 'trufflehog', True)),
    (['gzip: invalid header'] * 3, {'trufflehog_diagnostic_max_errors': 2}, ('error', 'output_limit', False)),
    (['x' * 129], {'trufflehog_diagnostic_max_line_chars': 128}, ('error', 'output_limit', False)),
    (['\u00e9' * 65], {'trufflehog_diagnostic_max_line_bytes': 128}, ('error', 'output_limit', False)),
], ids=['invalid-plural-shape', 'plural-count-bound', 'plural-character-bound', 'plural-byte-bound'])
def test_docker_detail_channels_do_not_bypass_original_plural_validation(monkeypatch, causes, settings, expected_policy):
    for key, value in settings.items():
        monkeypatch.setattr(scanner.scan_config, key, value)
    line = json.dumps(dict(json.loads(GZIP_WARNING), message='unexpected EOF', errors=causes))
    assert scanner._trufflehog_diagnostic_policy(line, 'docker', 0) == expected_policy


@pytest.mark.parametrize('source', ['git', 'huggingface', 'filesystem'])
def test_detail_channel_reduction_does_not_change_other_sources(source):
    line = json.dumps(dict(EXACT_EOF, message='unknown flag'))
    assert scanner._trufflehog_diagnostic_policy(line, source, 0) == ('error', 'network', True)


def test_missing_native_completion_cannot_be_cleared_by_recovery(docker_diagnostics_case):
    case = docker_diagnostics_case
    case.stderr = GZIP_WARNING
    result = case.run()
    assert result['error_class'] == 'command_incomplete' and result['retryable'] is True
    assert not result['scan_meta']['trufflehog_finished']
    assert result['scan_meta']['docker_full_recovery']['diagnostic']['reason'] == 'unrelated_diagnostics'
    assert not case.resolutions and not case.downloads


def test_count_bound_is_first_and_does_not_claim_byte_observations(docker_diagnostics_case):
    case = docker_diagnostics_case
    case.resolved['config']['media_type'] = PRIVATE
    case.resolved['layers'] *= 26
    case.resolved['layers'][0]['size'] = (256 << 20) + 1
    scope = assert_preflight_failure(case, case.run(), 'recovery_budget', {
        'reason': 'count_bound', 'observed': 26, 'limit': 8,
    })
    assert scope['descriptor_count'] == 27
    assert type(scope['diagnostic']['observed']) is int and type(scope['diagnostic']['limit']) is int


@pytest.mark.parametrize('kind,index,limit', [('config', 0, 1 << 20), ('layer', 1, 256 << 20)])
def test_descriptor_byte_bounds_distinguish_config_and_layer(docker_diagnostics_case, kind, index, limit):
    case = docker_diagnostics_case
    descriptor = case.resolved['config'] if kind == 'config' else case.resolved['layers'][0]
    descriptor['size'] = limit + 1
    assert_preflight_failure(case, case.run(), 'recovery_budget', {
        'reason': 'descriptor_byte_bound', 'observed': limit + 1, 'limit': limit,
        'descriptor_kind': kind, 'descriptor_index': index,
    })


@pytest.mark.parametrize('fits', [False, True])
def test_image_byte_bound_uses_unique_descriptors_and_accepts_exact_boundary(docker_diagnostics_case, fits):
    case = docker_diagnostics_case
    layers = [dict(case.resolved['layers'][0], digest='sha256:' + f'{index:064x}', size=256 << 20)
              for index in range(1, 5)]
    if fits:
        layers[-1]['size'] -= case.resolved['config']['size']
    case.resolved['layers'] = [*layers, dict(layers[0])]
    result = case.run()
    if not fits:
        scope = assert_preflight_failure(case, result, 'recovery_budget', {
            'reason': 'image_byte_bound', 'observed': (1 << 30) + 1024, 'limit': 1 << 30,
        })
        assert scope['descriptor_count'] == 6
    else:
        scope = result['scan_meta']['docker_full_recovery']
        assert not result['errors'] and scope['coverage_complete']
        assert scope['descriptor_count'] == scope['scanned_descriptors'] == 6
        assert len(case.downloads) == len(case.scans) == 5
        assert 'diagnostic' not in scope


def test_default_limits_and_deadline_unchanged_with_deduplicated_success(docker_diagnostics_case):
    case = docker_diagnostics_case
    case.resolved['config']['size'] = 1 << 20
    case.resolved['layers'][0]['size'] = 256 << 20
    case.resolved['layers'] *= 8
    result = case.run()
    scope = result['scan_meta']['docker_full_recovery']
    assert not result['errors'] and not result.get('warnings') and not result.get('degraded')
    assert scope['coverage_complete'] and scope['recovered_codec']
    assert scope['scanned_descriptors'] == scope['descriptor_count'] == 9
    assert len(case.downloads) == len(case.scans) == 2
    assert case.commands[0][0] == 600 and case.commands[0][1]['deadline'] == 700
    assert case.resolutions[0]['deadline'] == 700
    assert all(limits == DEFAULT_LIMITS and deadline == 700 for _, limits, deadline in case.scans)
    assert all(options['deadline'] == 700 and options['min_free_bytes'] == 20 << 30
               for _, options in case.downloads)
    assert scanner.DOCKER_CONFIG_MEDIA_TYPES == {CONFIG_MEDIA, 'application/vnd.docker.container.image.v1+json'}
    assert scanner.DOCKER_LAYER_MEDIA_TYPES == {
        LAYER_MEDIA, LAYER_MEDIA + '+gzip', LAYER_MEDIA + '+zstd',
        'application/vnd.docker.image.rootfs.diff.tar', 'application/vnd.docker.image.rootfs.diff.tar.gzip',
    }


@pytest.mark.parametrize('kind', ['config', 'layer'])
@pytest.mark.parametrize('media', [
    'application/vnd.oci.image.layer.nondistributable.v1.tar',
    'application/vnd.oci.image.layer.nondistributable.v1.tar+gzip',
    'application/vnd.oci.image.layer.nondistributable.v1.tar+zstd',
    'application/vnd.docker.image.rootfs.foreign.diff.tar',
    'application/vnd.docker.image.rootfs.foreign.diff.tar.gzip',
    'application/vnd.oci.image.manifest.v1+json',
    'application/vnd.oci.image.index.v1+json',
    'application/vnd.docker.distribution.manifest.v1+json',
    'application/vnd.docker.distribution.manifest.v1+prettyjws',
    'application/vnd.docker.distribution.manifest.v2+json',
    'application/vnd.docker.distribution.manifest.list.v2+json',
])
def test_known_media_is_reported_without_becoming_supported(docker_diagnostics_case, kind, media):
    case = docker_diagnostics_case
    descriptor = case.resolved['config'] if kind == 'config' else case.resolved['layers'][0]
    descriptor['media_type'] = media
    assert_preflight_failure(case, case.run(), 'unsupported_media_type', {
        'reason': 'media_type', 'media_type': media,
        'descriptor_kind': kind, 'descriptor_index': 0 if kind == 'config' else 1,
    })


@pytest.mark.parametrize('kind,media', [('config', LAYER_MEDIA), ('layer', CONFIG_MEDIA)])
def test_supported_media_in_wrong_descriptor_kind_is_reported_and_rejected(docker_diagnostics_case, kind, media):
    case = docker_diagnostics_case
    descriptor = case.resolved['config'] if kind == 'config' else case.resolved['layers'][0]
    descriptor['media_type'] = media
    assert_preflight_failure(case, case.run(), 'unsupported_media_type', {
        'reason': 'media_type', 'media_type': media,
        'descriptor_kind': kind, 'descriptor_index': 0 if kind == 'config' else 1,
    })


@pytest.mark.parametrize('media', [
    None, True, 7, [], {'url': PRIVATE}, b'private-diagnostic-sentinel',
    'application/octet-stream', 'https://example.invalid/' + PRIVATE,
    LAYER_MEDIA + ';token=' + PRIVATE, LAYER_MEDIA + '\n' + PRIVATE,
    PRIVATE + TARGET, PRIVATE * 10000,
], ids=['none', 'bool', 'integer', 'list', 'object', 'bytes', 'unknown', 'url',
        'parameters', 'newline', 'identifier', 'oversized'])
def test_unknown_and_non_string_media_fail_closed_without_disclosure(docker_diagnostics_case, caplog, media):
    case = docker_diagnostics_case
    caplog.set_level(logging.INFO, logger='scanner')
    case.resolved['layers'].append(dict(case.resolved['layers'][0], media_type=media,
                                        kind=PRIVATE, position=PRIVATE))
    assert_preflight_failure(case, case.run(), 'unsupported_media_type', {
        'reason': 'media_type', 'media_type': 'other', 'descriptor_kind': 'layer', 'descriptor_index': 2,
    })
    assert PRIVATE not in caplog.text and TARGET not in caplog.text


def test_missing_media_is_not_an_infrastructure_exception(docker_diagnostics_case):
    case = docker_diagnostics_case
    del case.resolved['config']['media_type']
    assert_preflight_failure(case, case.run(), 'unsupported_media_type', {
        'reason': 'media_type', 'media_type': 'other', 'descriptor_kind': 'config', 'descriptor_index': 0,
    })


@pytest.mark.parametrize('size', [None, True, -1, PRIVATE, {}])
def test_invalid_sizes_are_not_byte_budget_observations(docker_diagnostics_case, size):
    case = docker_diagnostics_case
    case.resolved['config']['size'] = size
    assert_preflight_failure(case, case.run(), 'invalid_descriptor', {
        'reason': 'integrity', 'descriptor_kind': 'config', 'descriptor_index': 0,
    })


def test_invalid_digest_preserves_class_but_does_not_claim_media_failure(docker_diagnostics_case):
    case = docker_diagnostics_case
    case.resolved['config']['digest'] = PRIVATE
    assert_preflight_failure(case, case.run(), 'unsupported_media_type', {
        'reason': 'integrity', 'descriptor_kind': 'config', 'descriptor_index': 0,
    })


def test_conflicting_duplicate_has_bounded_integrity_context(docker_diagnostics_case):
    case = docker_diagnostics_case
    case.resolved['layers'].append(dict(case.resolved['layers'][0], size=2048))
    assert_preflight_failure(case, case.run(), 'conflicting_descriptors', {
        'reason': 'integrity', 'descriptor_kind': 'layer', 'descriptor_index': 2,
    })


@pytest.mark.parametrize('failure,reason,phase,error_class,retryable', [
    ('identity', 'integrity', 'preflight', 'recovery_identity_mismatch', False),
    ('manifest', 'manifest_invalid', 'manifest_resolution', 'recovery_manifest_invalid', False),
    ('opaque', 'other', 'manifest_resolution', 'recovery_infrastructure', True),
    ('limits', 'configuration', 'configuration', 'recovery_infrastructure', True),
])
def test_early_failure_omits_unobserved_descriptor_counts(
    docker_diagnostics_case, failure, reason, phase, error_class, retryable,
):
    case = docker_diagnostics_case
    options = {}
    if failure == 'identity':
        case.resolved['manifest_digest'] = PRIVATE
    elif failure == 'manifest':
        case.resolve_error = scanner.DockerRegistryResolutionError(PRIVATE + TARGET)
    elif failure == 'opaque':
        case.resolve_error = RuntimeError(PRIVATE + TARGET)
    else:
        options['docker_recovery_limits'] = {}
    result = case.run(**options)
    scope = result['scan_meta']['docker_full_recovery']
    assert result['error_class'] == error_class and result['retryable'] is retryable
    assert scope['diagnostic'] == {'phase': phase, 'reason': reason}
    assert 'descriptor_count' not in scope and scope['scanned_descriptors'] == 0
    assert not scope['coverage_complete'] and not scope['blob_transfer_attempted']
    assert not case.downloads and PRIVATE not in json.dumps(result)


@pytest.mark.parametrize('code,retryable,reason', [
    ('digest_mismatch', False, 'integrity'), ('size_mismatch', False, 'integrity'),
    ('transfer_timeout', True, 'timeout'), ('remote_transient', True, 'other'),
])
def test_transfer_failure_keeps_retry_semantics_without_exception_text(docker_diagnostics_case, code, retryable, reason):
    case = docker_diagnostics_case
    case.transfer_error = scanner.DockerContentTransferError(code, PRIVATE + TARGET, retryable)
    result = case.run()
    scope = result['scan_meta']['docker_full_recovery']
    assert result['errors'] and result['error_class'] == code and result['retryable'] is retryable
    assert not result.get('source_failure') and not scope['coverage_complete']
    assert scope['blob_transfer_attempted'] and scope['scanned_descriptors'] == 0
    assert scope['diagnostic'] == {
        'phase': 'blob_transfer', 'reason': reason, 'descriptor_kind': 'config', 'descriptor_index': 0,
    }
    assert len(case.downloads) == 1 and not case.scans
    assert PRIVATE not in json.dumps(result) and TARGET not in json.dumps(result)


def test_transfer_failure_uses_original_index_after_deduplication(docker_diagnostics_case, monkeypatch):
    case = docker_diagnostics_case
    first = case.resolved['layers'][0]
    case.resolved['layers'] = [first, dict(first), dict(first, digest='sha256:' + 'd' * 64)]
    original = scanner.stream_docker_registry_blob

    def download(*args, **kwargs):
        if args[1]['position'] == 3:
            case.transfer_error = scanner.DockerContentTransferError('digest_mismatch', PRIVATE, False)
        return original(*args, **kwargs)

    monkeypatch.setattr(scanner, 'stream_docker_registry_blob', download)
    result = case.run()
    scope = result['scan_meta']['docker_full_recovery']
    assert result['error_class'] == 'digest_mismatch' and result['retryable'] is False
    assert not scope['coverage_complete'] and scope['scanned_descriptors'] == 3
    assert scope['descriptor_count'] == 4 and len(case.downloads) == 3
    assert scope['diagnostic'] == {
        'phase': 'blob_transfer', 'reason': 'integrity', 'descriptor_kind': 'layer', 'descriptor_index': 3,
    }


@pytest.mark.parametrize('code', ['invalid_config_json', 'invalid_layer_archive'])
def test_content_integrity_error_is_not_mislabeled_as_transport(docker_diagnostics_case, code):
    case = docker_diagnostics_case
    case.scan_error = scanner.DockerContentScanError(code, PRIVATE + TARGET)
    result = case.run()
    assert result['error_class'] == code and result['retryable'] is False
    assert result['scan_meta']['docker_full_recovery']['diagnostic'] == {
        'phase': 'blob_scan', 'reason': 'integrity', 'descriptor_kind': 'config', 'descriptor_index': 0,
    }
    assert not result['scan_meta']['docker_full_recovery']['coverage_complete']
    assert PRIVATE not in json.dumps(result)


@pytest.mark.parametrize('cleanup_fails', [False, True])
def test_blob_cleanup_phase_only_replaces_a_failure_if_cleanup_fails(docker_diagnostics_case, monkeypatch, cleanup_fails):
    case = docker_diagnostics_case
    case.scan_error = scanner.DockerContentScanError('invalid_config_json', PRIVATE)
    monkeypatch.setattr(scanner.os.path, 'lexists', lambda path: True)

    def unlink(path):
        if cleanup_fails:
            raise RuntimeError(PRIVATE)

    monkeypatch.setattr(scanner, 'durable_unlink', unlink)
    result = case.run()
    scope = result['scan_meta']['docker_full_recovery']
    assert result['error_class'] == ('recovery_infrastructure' if cleanup_fails else 'invalid_config_json')
    assert result['retryable'] is cleanup_fails and not scope['coverage_complete']
    assert scope['diagnostic'] == {
        'phase': 'cleanup' if cleanup_fails else 'blob_scan',
        'reason': 'other' if cleanup_fails else 'integrity',
        'descriptor_kind': 'config', 'descriptor_index': 0,
    }
    assert PRIVATE not in json.dumps(result)


@pytest.mark.parametrize('finished', [False, True])
def test_incomplete_filesystem_scan_does_not_clear_native_warnings(docker_diagnostics_case, finished):
    case = docker_diagnostics_case
    case.blob_stderr = '\n'.join([
        json.dumps({'level': 'info-2', 'msg': 'skipping file: size exceeds max allowed'}),
        *([FINISHED] if finished else []),
    ])
    result = case.run()
    assert result['error_class'] == 'recovery_scan_incomplete' and result['retryable'] is not finished
    assert result['warnings'] and result['degraded'] and len(result['findings']) == 1
    scope = result['scan_meta']['docker_full_recovery']
    assert not scope['coverage_complete'] and scope['scanned_descriptors'] == 0
    assert scope['diagnostic'] == {
        'phase': 'blob_scan', 'reason': 'scan_incomplete', 'descriptor_kind': 'config', 'descriptor_index': 0,
    }


def test_no_fresh_recovery_deadline_or_unobserved_byte_fields(docker_diagnostics_case):
    case = docker_diagnostics_case
    case.after_native = lambda: setattr(case, 'now', 700.0)
    result = case.run()
    scope = result['scan_meta']['docker_full_recovery']
    assert result['error_class'] == 'timeout' and result['retryable'] is True
    assert scope['diagnostic'] == {'phase': 'preflight', 'reason': 'timeout'}
    assert not scope['coverage_complete'] and not scope['blob_transfer_attempted']
    assert 'descriptor_count' not in scope and not case.resolutions and not case.downloads


def test_later_deadline_preserves_first_budget_reason_and_nonretryability(docker_diagnostics_case, monkeypatch):
    case = docker_diagnostics_case
    case.resolved['layers'] *= 26

    def filtering(result, *args, **kwargs):
        case.now = 700.0
        return result

    monkeypatch.setattr(scanner, 'apply_finding_filters', filtering)
    result = case.run()
    assert result['scan_meta']['docker_deadline_exceeded']
    assert_preflight_failure(case, result, 'recovery_budget', {
        'reason': 'count_bound', 'observed': 26, 'limit': 8,
    })


@pytest.mark.parametrize('stage', ['cleanup', 'filter'])
def test_post_scan_failure_keeps_coverage_incomplete(docker_diagnostics_case, monkeypatch, stage):
    case = docker_diagnostics_case
    if stage == 'cleanup':
        def cleanup(path):
            shutil.rmtree(path)
            raise RuntimeError(PRIVATE)
        monkeypatch.setattr(scanner, 'cleanup_command_work_dir', cleanup)
    else:
        def filtering(result, *args, **kwargs):
            case.now = 700.0
            return result
        monkeypatch.setattr(scanner, 'apply_finding_filters', filtering)
    result = case.run()
    scope = result['scan_meta']['docker_full_recovery']
    assert result['error_class'] == ('private_cleanup' if stage == 'cleanup' else 'timeout')
    assert result['retryable'] is True and not scope['coverage_complete']
    assert scope['scanned_descriptors'] == scope['descriptor_count'] == 2
    assert scope['diagnostic'] == {
        'phase': 'cleanup' if stage == 'cleanup' else 'completion',
        'reason': 'cleanup' if stage == 'cleanup' else 'timeout',
    }
    assert PRIVATE not in json.dumps(result)