import sys sys.dont_write_bytecode = True import argparse import json import math import os from collections import Counter, defaultdict RANK_BUCKETS = ('ranks_1_3', 'ranks_4_10') INVALID_OR_UNKNOWN = 'invalid_or_unknown' DOCKER_DEPTH_SOURCES = frozenset(('dockerhub',)) EXPERIMENT_STATES = frozenset(( 'collecting', 'planned', 'holding', 'resolving', 'active', 'draining', 'completed', 'released', 'held', )) REPOSITORY_STATES = frozenset(( 'pending', 'resolving', 'resolved', 'held', 'failed', 'skipped', )) TARGET_STATES = frozenset(( 'pending', 'reserved', 'scanning', 'done', 'failed', 'held', 'quarantined', 'skipped', )) BINDING_STATES = frozenset(( 'reserved', 'scanning', 'completed', 'failed', 'quarantined', 'released', )) RESERVATION_STATES = frozenset(( 'scanning', 'ready', 'ingesting', 'db_committed', 'acknowledged', 'refunded', 'quarantined', )) SCAN_STATUSES = frozenset(('clean', 'found', 'skipped', 'error', 'degraded')) CANDIDATE_STATES = frozenset(( 'pending', 'leased', 'deferred', 'completed', 'quarantined', )) KEYCHECK_STATUSES = frozenset(( 'ALIVE', 'VALID', 'VALID_2FA', 'VALID_RATE_LIMITED', 'VERTEX', 'BEDROCK', 'FOUNDRY', 'ADMIN', 'CANARY', 'NO_BALANCE', 'NO_QUOTA', 'LIMITED_OR_NO_BALANCE', 'LIMITED_OR_QUOTA', 'LIMITED', 'RATE_LIMITED', 'NO_CONTEXT', 'NO_USERNAME', 'NO_TARGET', 'NO_TARGET_MODELS', 'NO_GENERATION_MODEL', 'RESTRICTED', 'API_DISABLED', 'ACCESS_DENIED', 'QUARANTINED', 'DISABLED', 'DEAD', 'INVALID', 'EXPIRED', 'LEAKED_REVOKED', 'INVALID_OR_REVOKED', 'NETWORK', 'NETWORK_ERROR', 'FOUNDRY_UNRESOLVED', 'FOUNDRY_BAD_ENDPOINT', 'GENERATION_OK', 'OK', 'OPENAI_BAD_ENDPOINT', 'OPENAI_UNRESOLVED', 'REFRESH_TOKEN', 'UNKNOWN', )) KEYCHECK_STATUS_GROUPS = frozenset(( 'alive', 'dead', 'restricted', 'no_balance', 'no_context', 'limited', 'network', 'unknown', )) ERROR_CATEGORIES = frozenset(( 'secondary_rate_limit', 'rate_limit', 'auth_invalid', 'auth_forbidden', 'query_invalid', 'not_found', 'server_error', 'disk_space', 'timeout', 'extract', 'download', 'network', 'api', 'trufflehog', 'unknown', )) REPORT_MAX_MATERIALIZED_ROWS = 250000 _RESERVATION_BOUND_BINDINGS_CTE = '''reservation_bound_bindings AS ( SELECT t.id AS target_id, t.target_queue_id, t.manifest_id, b.id AS binding_id, b.reservation_id, b.attempt, b.state AS binding_state, reservation.state AS reservation_state, ts.id AS scan_id, ts.status AS scan_status, ts.duration_sec, ts.error_count AS scan_error_count FROM bounded_targets t LEFT JOIN target_queue queue ON queue.id = t.target_queue_id LEFT JOIN docker_depth_experiment_scan_bindings b ON b.experiment_target_id = t.id AND EXISTS ( SELECT 1 FROM result_reservations exact_reservation WHERE exact_reservation.id = b.reservation_id AND exact_reservation.queue_id = t.target_queue_id AND exact_reservation.source = queue.source AND exact_reservation.platform = queue.platform AND COALESCE(exact_reservation.query, '') = COALESCE(queue.query, '') AND exact_reservation.target = queue.target AND exact_reservation.normalized_target = queue.normalized_target ) LEFT JOIN result_reservations reservation ON reservation.id = b.reservation_id AND reservation.queue_id = t.target_queue_id LEFT JOIN target_scans ts ON ts.id = b.target_scan_id AND ts.queue_id = t.target_queue_id AND ts.queue_id = reservation.queue_id AND ts.result_reservation_id = reservation.id AND ts.result_reservation_id = b.reservation_id AND ts.scan_event_id = reservation.scan_event_id AND ts.claim_lease_token = reservation.claim_lease_token AND ts.source = reservation.source AND COALESCE(ts.query, '') = COALESCE(reservation.query, '') AND ts.target = reservation.target AND ts.normalized_target = reservation.normalized_target AND ts.scan_type = reservation.platform )''' def _rows(connection, sql, params, *, max_rows=REPORT_MAX_MATERIALIZED_ROWS): if isinstance(max_rows, bool) or not isinstance(max_rows, int) or max_rows < 1: raise ValueError('report row bound must be a positive integer') cursor = connection.execute(sql, params) columns = [item[0] for item in cursor.description] result = [] while True: batch = cursor.fetchmany(512) if not batch: break for row in batch: if len(result) >= max_rows: raise RuntimeError('Docker depth report row bound is exceeded') if hasattr(row, 'keys'): result.append({key: row[key] for key in row.keys()}) else: result.append(dict(zip(columns, row))) return result def _row(connection, sql, params): rows = _rows(connection, sql, params) return rows[0] if rows else None def _integer(value, default=0): try: return int(value) except (TypeError, ValueError, OverflowError): return default def _duration(value): try: result = float(value) except (TypeError, ValueError, OverflowError): return None return result if math.isfinite(result) and result >= 0 else None def _category(value, allowed): return value if isinstance(value, str) and value in allowed else INVALID_OR_UNKNOWN def _experiment_key(value): if not isinstance(value, str) or not 1 <= len(value) <= 128: return INVALID_OR_UNKNOWN allowed = frozenset('abcdefghijklmnopqrstuvwxyz0123456789._-') if value[0].isalnum() and value[-1].isalnum() and all(char in allowed for char in value): return value return INVALID_OR_UNKNOWN def _sha256(value): if ( isinstance(value, str) and len(value) == 64 and all(char in '0123456789abcdef' for char in value)): return value return INVALID_OR_UNKNOWN def _counts(values, allowed): counts = Counter(_category(value, allowed) for value in values) return {key: counts[key] for key in sorted(counts)} def _rank_bucket(rank): rank = _integer(rank, -1) if 1 <= rank <= 3: return 'ranks_1_3' if 4 <= rank <= 10: return 'ranks_4_10' return None def _finding_identity(row): fingerprint = str(row.get('finding_fingerprint') or '') detector_hash = str(row.get('detector_secret_hash') or '') if fingerprint or detector_hash: return ('identity', fingerprint, detector_hash) return ('finding_id', _integer(row.get('finding_id'))) def _finding_metrics(rows, layer_rows): finding_ids = set() identities = set() identity_by_finding = {} detector_hashes = set() missing_attribution_findings = set() exact_attribution_count = 0 exact_findings = set() explicit_unattributed_count = 0 explicit_unattributed_findings = set() invalid_exact_count = 0 invalid_attribution_count = 0 for row in rows: finding_id = _integer(row.get('finding_id')) identity = _finding_identity(row) finding_ids.add(finding_id) identities.add(identity) identity_by_finding[finding_id] = identity detector_hash = str(row.get('detector_secret_hash') or '') if detector_hash: detector_hashes.add(detector_hash) attribution_count = max(0, _integer(row.get('attribution_count'))) if attribution_count == 0: missing_attribution_findings.add(finding_id) exact_count = max(0, _integer(row.get('exact_attribution_count'))) exact_attribution_count += exact_count if exact_count: exact_findings.add(finding_id) unattributed_count = max( 0, _integer(row.get('unattributed_attribution_count')), ) explicit_unattributed_count += unattributed_count if unattributed_count: explicit_unattributed_findings.add(finding_id) invalid_exact_count += max( 0, _integer(row.get('invalid_exact_attribution_count')), ) invalid_attribution_count += max( 0, _integer(row.get('invalid_attribution_count')), ) positions_from_base = Counter() positions_from_top = Counter() manifest_layers = set() for row in layer_rows: count = max(0, _integer(row.get('attribution_count'))) manifest_layers.add(_integer(row.get('matched_manifest_layer_id'))) positions_from_base[_integer(row.get('position_from_base'))] += count positions_from_top[_integer(row.get('position_from_top'))] += count unattributed_findings = finding_ids - exact_findings exact_identities = {identity_by_finding[item] for item in exact_findings} unattributed_identities = { identity_by_finding[item] for item in unattributed_findings } return { 'findings': { 'occurrence_count': len(finding_ids), 'deduplicated_count': len(identities), 'detector_secret_identity_count': len(detector_hashes), 'without_detector_secret_hash_count': sum( 1 for identity in identities if identity[0] == 'finding_id' or not identity[2] ), }, 'layers': { 'exact_attribution_count': exact_attribution_count, 'exact_finding_occurrence_count': len(exact_findings), 'exact_deduplicated_finding_count': len(exact_identities), 'exact_manifest_layer_count': len(manifest_layers), 'unattributed_attribution_count': explicit_unattributed_count, 'unattributed_finding_occurrence_count': len(unattributed_findings), 'unattributed_deduplicated_finding_count': len(unattributed_identities), 'missing_attribution_finding_count': len(missing_attribution_findings), 'invalid_exact_attribution_count': invalid_exact_count, 'invalid_or_unknown_attribution_count': invalid_attribution_count, 'positions_from_base': { str(key): positions_from_base[key] for key in sorted(positions_from_base) }, 'positions_from_top': { str(key): positions_from_top[key] for key in sorted(positions_from_top) }, }, } def _keycheck_metrics(rows): candidates = {} for row in rows: candidates.setdefault(_integer(row.get('candidate_id')), row) credentials = { _integer(row.get('credential_id')) for row in candidates.values() if row.get('credential_id') is not None } frozen_results = set() frozen_by_credential = {} current_by_credential = {} for row in candidates.values(): credential_id = _integer(row.get('credential_id')) frozen_result_id = row.get('frozen_result_id') if frozen_result_id is not None: frozen_result_id = _integer(frozen_result_id) frozen_results.add(frozen_result_id) previous = frozen_by_credential.get(credential_id) if previous is None or frozen_result_id > previous[0]: frozen_by_credential[credential_id] = ( frozen_result_id, row.get('frozen_status'), row.get('frozen_status_group'), ) if row.get('current_state_credential_id') is not None: current_by_credential[credential_id] = ( row.get('current_status'), row.get('current_status_group'), row.get('matched_current_result_id') is not None, ) candidate_attempts = sum( max(0, _integer(row.get('candidate_attempts'))) for row in candidates.values() ) return { 'credentials': { 'deduplicated_count': len(credentials), }, 'keychecks': { 'candidate_count': len(candidates), 'candidate_state_counts': _counts( (row.get('candidate_state') for row in candidates.values()), CANDIDATE_STATES, ), 'attempt_count': candidate_attempts, 'retry_count': sum( max(0, _integer(row.get('candidate_attempts')) - 1) for row in candidates.values() ), 'pending_candidate_count': sum( str(row.get('candidate_state') or '').lower() == 'pending' for row in candidates.values() ), 'missing_frozen_result_candidate_count': sum( row.get('frozen_result_id') is None for row in candidates.values() ), 'frozen': { 'result_count': len(frozen_results), 'credential_count': len(frozen_by_credential), 'missing_credential_count': len(credentials - set(frozen_by_credential)), 'status_counts': _counts( (value[1] for value in frozen_by_credential.values()), KEYCHECK_STATUSES, ), 'status_group_counts': _counts( (value[2] for value in frozen_by_credential.values()), KEYCHECK_STATUS_GROUPS, ), }, 'current': { 'credential_count': len(current_by_credential), 'missing_credential_count': len(credentials - set(current_by_credential)), 'missing_backing_result_count': sum( not value[2] for value in current_by_credential.values() ), 'status_counts': _counts( (value[0] for value in current_by_credential.values()), KEYCHECK_STATUSES, ), 'status_group_counts': _counts( (value[1] for value in current_by_credential.values()), KEYCHECK_STATUS_GROUPS, ), }, }, } def _aggregate( target_ids, physical_rows, finding_rows, layer_rows, candidate_rows, error_rows): target_ids = set(target_ids) target_records = {} bindings = {} scans = {} for row in physical_rows: target_id = _integer(row.get('target_id')) if target_id not in target_ids: continue target_records.setdefault(target_id, row) if row.get('binding_id') is not None: bindings.setdefault(_integer(row.get('binding_id')), row) if row.get('scan_id') is not None: scans.setdefault(_integer(row.get('scan_id')), row) selected_findings = [ row for row in finding_rows if _integer(row.get('target_id')) in target_ids ] selected_candidates = [ row for row in candidate_rows if _integer(row.get('target_id')) in target_ids ] selected_errors = [ row for row in error_rows if _integer(row.get('target_id')) in target_ids ] selected_layers = [ row for row in layer_rows if _integer(row.get('target_id')) in target_ids ] durations = { scan_id: value for scan_id, row in scans.items() if (value := _duration(row.get('duration_sec'))) is not None } errors = { _integer(row.get('error_id')): row for row in selected_errors if row.get('error_id') is not None } error_ids = set(errors) scans_with_error_rows = { _integer(row.get('scan_id')) for row in selected_errors if row.get('error_id') is not None } scans_with_reported_errors = { scan_id for scan_id, row in scans.items() if _integer(row.get('scan_error_count')) > 0 } finding_metrics = _finding_metrics(selected_findings, selected_layers) keycheck_metrics = _keycheck_metrics(selected_candidates) duration_values = list(durations.values()) duration_total = sum(duration_values) targets_with_bindings = { _integer(row.get('target_id')) for row in bindings.values() } targets_with_scans = { _integer(row.get('target_id')) for row in scans.values() } return { 'target_count': len(target_ids), 'manifest_count': len({ _integer(row.get('matched_manifest_id')) for row in target_records.values() if row.get('matched_manifest_id') is not None }), 'declared_manifest_layer_count': sum( max(0, _integer(row.get('manifest_layer_count'))) for row in target_records.values() if row.get('matched_manifest_id') is not None ), 'target_state_counts': _counts( (row.get('target_state') for row in target_records.values()), TARGET_STATES, ), 'targets_with_bindings': len(targets_with_bindings), 'targets_without_bindings': len(target_ids - targets_with_bindings), 'targets_with_scans': len(targets_with_scans), 'targets_without_scans': len(target_ids - targets_with_scans), 'scan_bindings': { 'attempt_count': len(bindings), 'retry_count': sum( _integer(row.get('attempt')) > 1 for row in bindings.values() ), 'refunded_attempt_count': sum( str(row.get('binding_state') or '') == 'released' for row in bindings.values() ), 'maximum_attempt': max( (_integer(row.get('attempt')) for row in bindings.values()), default=0, ), 'state_counts': _counts( (row.get('binding_state') for row in bindings.values()), BINDING_STATES, ), 'reservation_state_counts': _counts( (row.get('reservation_state') for row in bindings.values()), RESERVATION_STATES, ), }, 'scans': { 'count': len(scans), 'status_counts': _counts( (row.get('scan_status') for row in scans.values()), SCAN_STATUSES, ), 'duration': { 'count': len(duration_values), 'missing_count': len(scans) - len(duration_values), 'total_seconds': duration_total, 'average_seconds': ( duration_total / len(duration_values) if duration_values else 0.0 ), 'minimum_seconds': min(duration_values, default=0.0), 'maximum_seconds': max(duration_values, default=0.0), }, 'errors': { 'row_count': len(error_ids), 'category_counts': _counts( (row.get('error_category') for row in errors.values()), ERROR_CATEGORIES, ), 'reported_count': sum( max(0, _integer(row.get('scan_error_count'))) for row in scans.values() ), 'scan_count': len( scans_with_error_rows | scans_with_reported_errors ), }, }, **finding_metrics, **keycheck_metrics, } def _marginal_minimum_rank_yield(target_ranks, finding_rows, candidate_rows): minimum_finding_rank = {} minimum_detector_rank = {} minimum_credential_rank = {} def record_minimum(mapping, identity, rank): if rank is None: mapping.setdefault(identity, None) elif mapping.get(identity) is None or rank < mapping[identity]: mapping[identity] = rank for row in finding_rows: target_id = _integer(row.get('target_id')) if target_id not in target_ranks: continue rank = target_ranks[target_id] record_minimum(minimum_finding_rank, _finding_identity(row), rank) detector_hash = str(row.get('detector_secret_hash') or '') if detector_hash: record_minimum(minimum_detector_rank, detector_hash, rank) for row in candidate_rows: target_id = _integer(row.get('target_id')) if target_id not in target_ranks or row.get('credential_id') is None: continue record_minimum( minimum_credential_rank, _integer(row.get('credential_id')), target_ranks[target_id], ) def distribution(mapping): buckets = Counter(_rank_bucket(rank) or 'unranked' for rank in mapping.values()) return { 'deduplicated_total': len(mapping), 'ranks_1_3': buckets['ranks_1_3'], 'ranks_4_10': buckets['ranks_4_10'], 'unranked': buckets['unranked'], } return { 'findings': distribution(minimum_finding_rank), 'detector_secret_identities': distribution(minimum_detector_rank), 'credentials': distribution(minimum_credential_rank), } def _scope_report( target_ranks, physical_rows, finding_rows, layer_rows, candidate_rows, error_rows): target_ids = set(target_ranks) bucket_targets = { bucket: { target_id for target_id, rank in target_ranks.items() if _rank_bucket(rank) == bucket } for bucket in RANK_BUCKETS } return { 'totals': _aggregate( target_ids, physical_rows, finding_rows, layer_rows, candidate_rows, error_rows, ), 'rank_buckets': { bucket: _aggregate( bucket_targets[bucket], physical_rows, finding_rows, layer_rows, candidate_rows, error_rows, ) for bucket in RANK_BUCKETS }, 'unranked_target_count': sum( _rank_bucket(rank) is None for rank in target_ranks.values() ), 'marginal_minimum_rank_yield': _marginal_minimum_rank_yield( target_ranks, finding_rows, candidate_rows, ), } def _overlap_summary(item_queries): credits = sum(len(query_ids) for query_ids in item_queries.values()) attributed = sum(bool(query_ids) for query_ids in item_queries.values()) return { 'physical_count': len(item_queries), 'attributed_physical_count': attributed, 'unattributed_physical_count': len(item_queries) - attributed, 'attribution_credit_count': credits, 'overlap_credit_count': credits - attributed, 'shared_physical_count': sum( len(query_ids) > 1 for query_ids in item_queries.values() ), } def _repository_overlap_summary(repository_queries): summary = _overlap_summary(repository_queries) return { 'physical_count': summary['physical_count'], 'query_membership_credit_count': summary['attribution_credit_count'], 'overlap_credit_count': summary['overlap_credit_count'], 'shared_physical_count': summary['shared_physical_count'], } def _report_connection(connection): if not hasattr(connection, 'execute'): connection = getattr(connection, 'conn', None) if connection is None or not hasattr(connection, 'execute'): raise ValueError('a database connection is required') module = str(type(connection).__module__ or '').lower() if module.startswith(('psycopg', 'psycopg2')) and not hasattr( connection, 'is_postgres', ): raise ValueError( 'raw PostgreSQL connections are unsupported; use DatabaseConnection' ) return connection def _begin_report_snapshot(connection): is_postgres = bool(getattr(connection, 'is_postgres', False)) raw_connection = getattr(connection, '_conn', connection) if is_postgres: transaction_status = getattr( getattr(raw_connection, 'info', None), 'transaction_status', None, ) if transaction_status is not None and int(transaction_status) != 0: raise RuntimeError('Docker depth report requires an idle database connection') statement = ( 'BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY' ) else: if bool(getattr(raw_connection, 'in_transaction', False)): raise RuntimeError('Docker depth report requires an idle database connection') statement = 'BEGIN' try: connection.execute(statement) except Exception: connection.rollback() raise return 'repeatable_read' if is_postgres else 'sqlite_transaction' def _build_docker_depth_report_snapshot( connection, *, experiment_id=None, experiment_key=None): """Return secret-safe aggregates for one durable Docker depth experiment.""" connection = _report_connection(connection) if (experiment_id is None) == (experiment_key is None): raise ValueError('provide exactly one of experiment_id or experiment_key') if experiment_id is not None: if isinstance(experiment_id, bool) or not isinstance(experiment_id, int): raise ValueError('experiment_id must be a positive integer') if experiment_id < 1: raise ValueError('experiment_id must be a positive integer') where = 'id = ?' selector = experiment_id else: experiment_key = str(experiment_key or '').strip() if not experiment_key: raise ValueError('experiment_key must be non-empty') where = 'experiment_key = ?' selector = experiment_key experiment = _row( connection, f'''SELECT id, experiment_key, source, state, query_count, target_count, selection_count FROM docker_depth_experiments WHERE {where}''', (selector,), ) if experiment is None: raise LookupError('Docker depth experiment was not found') experiment_id = _integer(experiment['id']) query_rows = _rows( connection, '''WITH bounded_queries AS ( SELECT id, experiment_id, source, query_ordinal, query, query_sha256, required_repository_count, selected_repository_count FROM docker_depth_experiment_queries WHERE experiment_id = ? ) SELECT q.id AS query_id, q.query_ordinal, q.query_sha256, q.source, q.required_repository_count, q.selected_repository_count, r.id AS repository_id, COALESCE(r.replacement_repository_queue_id, r.repository_queue_id) AS repository_queue_id, r.repository_queue_id AS planned_repository_queue_id, r.repository_rank, r.work_state AS repository_state, r.resolver_attempts, s.id AS selection_id, s.image_rank, t.id AS target_id FROM bounded_queries q LEFT JOIN docker_depth_experiment_repositories r ON r.experiment_id = q.experiment_id AND r.query_ordinal = q.query_ordinal AND r.source = q.source AND r.query = q.query LEFT JOIN docker_depth_experiment_selections s ON s.experiment_id = q.experiment_id AND s.query_ordinal = q.query_ordinal AND s.experiment_repository_id = r.id LEFT JOIN docker_depth_experiment_targets t ON t.experiment_id = q.experiment_id AND t.id = s.experiment_target_id ORDER BY q.query_ordinal, q.id, r.id, s.id''', (experiment_id,), ) physical_rows = _rows( connection, f'''WITH bounded_targets AS ( SELECT id, experiment_id, target_queue_id, manifest_id, state FROM docker_depth_experiment_targets WHERE experiment_id = ? ), target_ranks AS ( SELECT experiment_target_id, MIN(image_rank) AS image_rank FROM docker_depth_experiment_selections WHERE experiment_id = ? GROUP BY experiment_target_id ), {_RESERVATION_BOUND_BINDINGS_CTE} SELECT t.id AS target_id, t.state AS target_state, tr.image_rank, m.id AS matched_manifest_id, m.layer_count AS manifest_layer_count, rb.binding_id, rb.attempt, rb.binding_state, rb.reservation_state, rb.scan_id, rb.scan_status, rb.duration_sec, rb.scan_error_count FROM bounded_targets t LEFT JOIN target_ranks tr ON tr.experiment_target_id = t.id LEFT JOIN docker_image_manifests m ON m.id = t.manifest_id AND m.target_queue_id = t.target_queue_id LEFT JOIN reservation_bound_bindings rb ON rb.target_id = t.id ORDER BY t.id, rb.binding_id''', (experiment_id, experiment_id), ) finding_rows = _rows( connection, f'''WITH bounded_targets AS ( SELECT id, target_queue_id, manifest_id FROM docker_depth_experiment_targets WHERE experiment_id = ? ), {_RESERVATION_BOUND_BINDINGS_CTE} SELECT rb.target_id, rb.binding_id, rb.scan_id, f.id AS finding_id, f.finding_fingerprint, f.detector_secret_hash, COUNT(a.id) AS attribution_count, SUM(CASE WHEN a.attribution_state = 'exact' AND ml.id IS NOT NULL THEN 1 ELSE 0 END) AS exact_attribution_count, SUM(CASE WHEN a.attribution_state = 'exact' AND ml.id IS NULL THEN 1 ELSE 0 END) AS invalid_exact_attribution_count, SUM(CASE WHEN a.attribution_state = 'unattributed' THEN 1 ELSE 0 END) AS unattributed_attribution_count, SUM(CASE WHEN a.id IS NOT NULL AND a.attribution_state NOT IN ('exact','unattributed') THEN 1 ELSE 0 END) AS invalid_attribution_count FROM reservation_bound_bindings rb JOIN findings f ON f.target_scan_id = rb.scan_id LEFT JOIN docker_finding_layer_attributions a ON a.scan_binding_id = rb.binding_id AND a.finding_id = f.id LEFT JOIN docker_manifest_layers ml ON ml.id = a.manifest_layer_id AND ml.manifest_id = rb.manifest_id AND ml.position_from_base = a.position_from_base AND ml.position_from_top = a.position_from_top AND ml.layer_digest = a.reported_layer_digest GROUP BY rb.target_id, rb.binding_id, rb.scan_id, f.id, f.finding_fingerprint, f.detector_secret_hash ORDER BY rb.target_id, f.id''', (experiment_id,), ) layer_rows = _rows( connection, f'''WITH bounded_targets AS ( SELECT id, target_queue_id, manifest_id FROM docker_depth_experiment_targets WHERE experiment_id = ? ), {_RESERVATION_BOUND_BINDINGS_CTE} SELECT rb.target_id, ml.id AS matched_manifest_layer_id, ml.position_from_base, ml.position_from_top, COUNT(a.id) AS attribution_count FROM reservation_bound_bindings rb JOIN findings f ON f.target_scan_id = rb.scan_id JOIN docker_finding_layer_attributions a ON a.scan_binding_id = rb.binding_id AND a.finding_id = f.id AND a.attribution_state = 'exact' JOIN docker_manifest_layers ml ON ml.id = a.manifest_layer_id AND ml.manifest_id = rb.manifest_id AND ml.position_from_base = a.position_from_base AND ml.position_from_top = a.position_from_top AND ml.layer_digest = a.reported_layer_digest GROUP BY rb.target_id, ml.id, ml.position_from_base, ml.position_from_top ORDER BY rb.target_id, ml.id''', (experiment_id,), ) candidate_rows = _rows( connection, f'''WITH bounded_targets AS ( SELECT id, target_queue_id, manifest_id FROM docker_depth_experiment_targets WHERE experiment_id = ? ), {_RESERVATION_BOUND_BINDINGS_CTE}, bounded_findings AS ( SELECT rb.target_id, rb.scan_id, f.id AS finding_id FROM reservation_bound_bindings rb JOIN findings f ON f.target_scan_id = rb.scan_id ) SELECT bf.target_id, bf.scan_id, bf.finding_id, c.id AS candidate_id, kc.id AS credential_id, c.state AS candidate_state, c.attempts AS candidate_attempts, frozen.id AS frozen_result_id, frozen.status AS frozen_status, frozen.status_group AS frozen_status_group, current_state.credential_id AS current_state_credential_id, current_result.status AS current_status, current_result.status_group AS current_status_group, current_result.id AS matched_current_result_id FROM bounded_findings bf JOIN keycheck_candidates c ON c.finding_id = bf.finding_id AND c.target_scan_id = bf.scan_id JOIN keycheck_credentials kc ON kc.id = c.credential_id LEFT JOIN keycheck_results frozen ON frozen.id = c.keycheck_result_id AND frozen.candidate_id = c.id AND frozen.credential_id = kc.id LEFT JOIN keycheck_current_state current_state ON current_state.credential_id = kc.id LEFT JOIN keycheck_results current_result ON current_result.id = current_state.last_result_id AND current_result.credential_id = current_state.credential_id AND current_result.status = current_state.status AND current_result.status_group = current_state.status_group ORDER BY bf.target_id, c.id''', (experiment_id,), ) error_rows = _rows( connection, f'''WITH bounded_targets AS ( SELECT id, target_queue_id, manifest_id FROM docker_depth_experiment_targets WHERE experiment_id = ? ), {_RESERVATION_BOUND_BINDINGS_CTE} SELECT rb.target_id, rb.binding_id, rb.scan_id, e.id AS error_id, e.category AS error_category FROM reservation_bound_bindings rb JOIN errors e ON e.target_scan_id = rb.scan_id ORDER BY rb.target_id, rb.scan_id, e.id''', (experiment_id,), ) query_contexts = {} for row in query_rows: query_id = _integer(row.get('query_id')) context = query_contexts.setdefault(query_id, { 'query_id': query_id, 'query_ordinal': _integer(row.get('query_ordinal')), 'query_sha256': _sha256(row.get('query_sha256')), 'source': str(row.get('source') or ''), 'required_repository_count': _integer( row.get('required_repository_count') ), 'cohort_repository_count': _integer( row.get('selected_repository_count') ), 'repository_attempts': {}, 'repository_states': {}, 'repository_queue_ids': {}, 'selected_repositories': set(), 'selection_ids': set(), 'target_ranks': {}, }) repository_id = row.get('repository_id') if repository_id is not None: repository_id = _integer(repository_id) context['repository_attempts'].setdefault( repository_id, max(0, _integer(row.get('resolver_attempts'))), ) context['repository_states'].setdefault( repository_id, row.get('repository_state'), ) if row.get('repository_queue_id') is not None: context['repository_queue_ids'].setdefault( repository_id, _integer(row.get('repository_queue_id')), ) if row.get('selection_id') is not None and row.get('target_id') is not None: context['selection_ids'].add(_integer(row.get('selection_id'))) if row.get('repository_queue_id') is not None: context['selected_repositories'].add( _integer(row.get('repository_queue_id')) ) target_id = _integer(row.get('target_id')) rank = _integer(row.get('image_rank')) previous = context['target_ranks'].get(target_id) if previous is None or rank < previous: context['target_ranks'][target_id] = rank physical_target_ranks = {} for row in physical_rows: target_id = _integer(row.get('target_id')) rank = row.get('image_rank') physical_target_ranks.setdefault( target_id, _integer(rank) if rank is not None else None, ) physical = _scope_report( physical_target_ranks, physical_rows, finding_rows, layer_rows, candidate_rows, error_rows, ) repository_queries = defaultdict(set) selected_repository_queries = defaultdict(set) all_selections = set() for query_id, context in query_contexts.items(): for repository_queue_id in set(context['repository_queue_ids'].values()): repository_queries[repository_queue_id].add(query_id) for repository_queue_id in context['selected_repositories']: selected_repository_queries[repository_queue_id].add(query_id) all_selections.update(context['selection_ids']) repository_coverage = _repository_overlap_summary(repository_queries) selected_repository_coverage = _repository_overlap_summary( selected_repository_queries ) skipped_repository_queue_ids = { context['repository_queue_ids'][repository_id] for context in query_contexts.values() for repository_id, state in context['repository_states'].items() if state == 'skipped' and repository_id in context['repository_queue_ids'] } skipped_repository_memberships = sum( state == 'skipped' for context in query_contexts.values() for state in context['repository_states'].values() ) physical['coverage'] = { 'query_count': len(query_contexts), 'queries_with_repositories': sum( bool(item['repository_attempts']) for item in query_contexts.values() ), 'queries_with_selections': sum( bool(item['selection_ids']) for item in query_contexts.values() ), 'repository_count': repository_coverage['physical_count'], 'repository_query_membership_credit_count': ( repository_coverage['query_membership_credit_count'] ), 'repository_overlap_credit_count': repository_coverage['overlap_credit_count'], 'shared_repository_count': repository_coverage['shared_physical_count'], 'selected_repository_count': selected_repository_coverage['physical_count'], 'selected_repository_query_membership_credit_count': ( selected_repository_coverage['query_membership_credit_count'] ), 'selected_repository_overlap_credit_count': ( selected_repository_coverage['overlap_credit_count'] ), 'shared_selected_repository_count': ( selected_repository_coverage['shared_physical_count'] ), 'image_unavailable_repository_count': len(skipped_repository_queue_ids), 'image_unavailable_repository_membership_count': ( skipped_repository_memberships ), 'queries_with_image_unavailable_repositories': sum( any(state == 'skipped' for state in item['repository_states'].values()) for item in query_contexts.values() ), 'selection_count': len(all_selections), 'target_attribution_credit_count': sum( len(item['target_ranks']) for item in query_contexts.values() ), } per_query = {} for context in sorted( query_contexts.values(), key=lambda item: (item['query_ordinal'], item['query_id'])): scope = _scope_report( context['target_ranks'], physical_rows, finding_rows, layer_rows, candidate_rows, error_rows, ) resolver_attempts = context['repository_attempts'].values() scope['query'] = { 'id': context['query_id'], 'ordinal': context['query_ordinal'], 'sha256': context['query_sha256'], 'source': _category(context['source'], DOCKER_DEPTH_SOURCES), } scope['coverage'] = { 'required_repository_count': context['required_repository_count'], 'cohort_repository_count': context['cohort_repository_count'], 'unavailable_repository_count': max( 0, context['required_repository_count'] - context['cohort_repository_count'], ), 'repository_count': len(context['repository_attempts']), 'missing_cohort_repository_count': max( 0, context['cohort_repository_count'] - len(context['repository_attempts']), ), 'selected_repository_count': len(context['selected_repositories']), 'image_unavailable_repository_count': sum( state == 'skipped' for state in context['repository_states'].values() ), 'selection_count': len(context['selection_ids']), 'target_count': len(context['target_ranks']), 'repository_state_counts': _counts( context['repository_states'].values(), REPOSITORY_STATES, ), 'resolver_attempt_count': sum(resolver_attempts), 'resolver_retry_count': sum( max(0, attempts - 1) for attempts in context['repository_attempts'].values() ), } per_query[str(context['query_id'])] = scope target_queries = {target_id: set() for target_id in physical_target_ranks} for query_id, context in query_contexts.items(): for target_id in context['target_ranks']: target_queries.setdefault(target_id, set()).add(query_id) scan_queries = {} for row in physical_rows: if row.get('scan_id') is not None: scan_queries.setdefault( _integer(row.get('scan_id')), set(target_queries.get(_integer(row.get('target_id')), set())), ) finding_queries = defaultdict(set) for row in finding_rows: finding_queries[_finding_identity(row)].update( target_queries.get(_integer(row.get('target_id')), set()) ) credential_queries = defaultdict(set) for row in candidate_rows: if row.get('credential_id') is not None: credential_queries[_integer(row.get('credential_id'))].update( target_queries.get(_integer(row.get('target_id')), set()) ) return { 'experiment': { 'id': experiment_id, 'key': _experiment_key(experiment.get('experiment_key')), 'source': _category(experiment.get('source'), DOCKER_DEPTH_SOURCES), 'state': _category(experiment.get('state'), EXPERIMENT_STATES), 'persisted_counts': { 'queries': _integer(experiment.get('query_count')), 'targets': _integer(experiment.get('target_count')), 'selections': _integer(experiment.get('selection_count')), }, }, 'physical': physical, 'per_query': per_query, 'counting_semantics': { 'physical_totals_are_deduplicated': True, 'per_query_totals_are_attribution_credits': True, 'per_query_totals_are_summable_to_physical': False, }, 'overlap': { 'repositories': repository_coverage, 'selected_repositories': selected_repository_coverage, 'targets': _overlap_summary(target_queries), 'scans': _overlap_summary(scan_queries), 'findings': _overlap_summary(finding_queries), 'credentials': _overlap_summary(credential_queries), }, } def build_docker_depth_report( connection, *, experiment_id=None, experiment_key=None, allow_live=True): """Build one aggregate report from a clean, read-only database snapshot.""" connection = _report_connection(connection) consistency = _begin_report_snapshot(connection) try: report = _build_docker_depth_report_snapshot( connection, experiment_id=experiment_id, experiment_key=experiment_key, ) terminal = report['experiment']['state'] in ('completed', 'released') if not allow_live and not terminal: raise RuntimeError( 'Docker depth report is live; pass allow_live=True explicitly' ) report['snapshot'] = { 'read_only': True, 'consistency': consistency, 'experiment_terminal': terminal, } return report finally: connection.rollback() def main(argv=None): parser = argparse.ArgumentParser( description='Build a secret-safe read-only Docker depth experiment report.', ) parser.add_argument('--config', required=True) selector = parser.add_mutually_exclusive_group(required=True) selector.add_argument('--experiment-id', type=int) selector.add_argument('--experiment-key') parser.add_argument( '--allow-live', action='store_true', help='Allow a coherent snapshot before the experiment is completed.', ) args = parser.parse_args(argv) connection = None try: import yaml from db_backend import ( connect_postgres, database_url_from_env, is_postgres_url, ) from paths import apply_path_config with open(os.path.abspath(args.config), 'r', encoding='utf-8') as handle: config = apply_path_config(yaml.safe_load(handle) or {}, args.config) global_config = config.get('global') or {} database_url = ( global_config.get('dashboard_db_url') or global_config.get('database_url') or database_url_from_env() ) if not is_postgres_url(database_url): raise RuntimeError('Docker depth report requires PostgreSQL') connection = connect_postgres( database_url, connect_timeout_sec=5, statement_timeout_ms=120000, lock_timeout_ms=5000, idle_in_transaction_timeout_ms=120000, tcp_user_timeout_ms=10000, ) connection.execute("SET application_name = 'truf-docker-depth-report'") connection.execute('SET default_transaction_read_only = on') connection.commit() report = build_docker_depth_report( connection, experiment_id=args.experiment_id, experiment_key=args.experiment_key, allow_live=args.allow_live, ) print(json.dumps(report, ensure_ascii=True, sort_keys=True)) return 0 except Exception as exc: raise SystemExit( f'Docker depth report failed closed: {type(exc).__name__}' ) from None finally: if connection is not None: connection.close() if __name__ == '__main__': main()