import json import os from pathlib import Path import sqlite3 import sys import tempfile import unittest from unittest import mock ROOT = Path(__file__).resolve().parents[1] APP_DIR = ROOT / 'app' sys.path.insert(0, str(APP_DIR)) import docker_depth_report from docker_depth_report import build_docker_depth_report from scanner_db import ScannerDB, utc_now_iso SCHEMA = ''' CREATE TABLE docker_depth_experiments ( id INTEGER PRIMARY KEY, experiment_key TEXT NOT NULL, source TEXT NOT NULL, state TEXT NOT NULL, query_count INTEGER NOT NULL, target_count INTEGER NOT NULL, selection_count INTEGER NOT NULL ); CREATE TABLE docker_depth_experiment_queries ( id INTEGER PRIMARY KEY, experiment_id INTEGER NOT NULL, source TEXT NOT NULL, query_ordinal INTEGER NOT NULL, query TEXT NOT NULL, query_sha256 TEXT NOT NULL, required_repository_count INTEGER NOT NULL, selected_repository_count INTEGER NOT NULL ); CREATE TABLE docker_depth_experiment_repositories ( id INTEGER PRIMARY KEY, experiment_id INTEGER NOT NULL, query_ordinal INTEGER NOT NULL, source TEXT NOT NULL, query TEXT NOT NULL, repository_queue_id INTEGER NOT NULL, replacement_repository_queue_id INTEGER, repository_rank INTEGER NOT NULL, work_state TEXT NOT NULL, resolver_attempts INTEGER NOT NULL ); CREATE TABLE docker_depth_experiment_targets ( id INTEGER PRIMARY KEY, experiment_id INTEGER NOT NULL, target_queue_id INTEGER NOT NULL, manifest_id INTEGER NOT NULL, state TEXT NOT NULL ); CREATE TABLE docker_depth_experiment_selections ( id INTEGER PRIMARY KEY AUTOINCREMENT, experiment_id INTEGER NOT NULL, query_ordinal INTEGER NOT NULL, experiment_repository_id INTEGER NOT NULL, experiment_target_id INTEGER NOT NULL, image_rank INTEGER NOT NULL ); CREATE TABLE docker_depth_experiment_scan_bindings ( id INTEGER PRIMARY KEY, experiment_target_id INTEGER NOT NULL, reservation_id INTEGER NOT NULL, target_scan_id INTEGER, attempt INTEGER NOT NULL, state TEXT NOT NULL ); CREATE TABLE target_queue ( id INTEGER PRIMARY KEY, source TEXT NOT NULL, platform TEXT NOT NULL, query TEXT, target TEXT NOT NULL, normalized_target TEXT NOT NULL ); CREATE TABLE result_reservations ( id INTEGER PRIMARY KEY, queue_id INTEGER NOT NULL, source TEXT NOT NULL, platform TEXT NOT NULL, query TEXT, target TEXT NOT NULL, normalized_target TEXT NOT NULL, claim_lease_token TEXT NOT NULL, scan_event_id TEXT NOT NULL, state TEXT NOT NULL ); CREATE TABLE target_scans ( id INTEGER PRIMARY KEY, scan_event_id TEXT, queue_id INTEGER, result_reservation_id INTEGER, claim_lease_token TEXT, source TEXT, normalized_target TEXT, scan_type TEXT, status TEXT, duration_sec REAL, error_count INTEGER, query TEXT, target TEXT, raw_result_json TEXT ); CREATE TABLE docker_image_manifests ( id INTEGER PRIMARY KEY, target_queue_id INTEGER NOT NULL, repository TEXT, layer_count INTEGER NOT NULL ); CREATE TABLE docker_manifest_layers ( id INTEGER PRIMARY KEY, manifest_id INTEGER NOT NULL, position_from_base INTEGER NOT NULL, position_from_top INTEGER NOT NULL, layer_digest TEXT NOT NULL ); CREATE TABLE findings ( id INTEGER PRIMARY KEY, target_scan_id INTEGER NOT NULL, finding_fingerprint TEXT, detector_secret_hash TEXT, raw_secret TEXT, redacted_secret TEXT, raw_finding_json TEXT ); CREATE TABLE docker_finding_layer_attributions ( id INTEGER PRIMARY KEY, scan_binding_id INTEGER NOT NULL, finding_id INTEGER NOT NULL, manifest_layer_id INTEGER, attribution_state TEXT NOT NULL, reported_layer_digest TEXT, position_from_base INTEGER, position_from_top INTEGER ); CREATE TABLE keycheck_credentials ( id INTEGER PRIMARY KEY, secret_text TEXT, secret_json TEXT, key_masked TEXT ); CREATE TABLE keycheck_candidates ( id INTEGER PRIMARY KEY, credential_id INTEGER NOT NULL, finding_id INTEGER NOT NULL, target_scan_id INTEGER NOT NULL, state TEXT NOT NULL, attempts INTEGER NOT NULL, keycheck_result_id INTEGER, metadata_json TEXT ); CREATE TABLE keycheck_results ( id INTEGER PRIMARY KEY, candidate_id INTEGER, credential_id INTEGER, status TEXT NOT NULL, status_group TEXT NOT NULL, message TEXT, source_line TEXT, metadata_json TEXT ); CREATE TABLE keycheck_current_state ( credential_id INTEGER PRIMARY KEY, status TEXT NOT NULL, status_group TEXT NOT NULL, last_result_id INTEGER NOT NULL, metadata_json TEXT ); CREATE TABLE errors ( id INTEGER PRIMARY KEY, target_scan_id INTEGER NOT NULL, category TEXT, summary TEXT, raw_error TEXT ); ''' class DockerDepthReportTests(unittest.TestCase): def setUp(self): self.conn = sqlite3.connect(':memory:') self.conn.row_factory = sqlite3.Row self.conn.executescript(SCHEMA) self.conn.execute( '''INSERT INTO docker_depth_experiments( id, experiment_key, source, state, query_count, target_count, selection_count ) VALUES (1, 'depth-fixture', 'dockerhub', 'completed', 0, 0, 0)''' ) def tearDown(self): self.conn.close() def test_rows_fail_closed_before_materialization_bound_is_exceeded(self): with self.assertRaisesRegex(RuntimeError, 'row bound is exceeded'): docker_depth_report._rows( self.conn, 'SELECT 1 AS value UNION ALL SELECT 2 UNION ALL SELECT 3', (), max_rows=2, ) def add_query( self, query_id, ordinal, *, query=None, attempts=1, required=1, selected=None, repository_queue_id=None, add_repository=True, repository_state='resolved'): query = query or f'query-{query_id}' selected = required if selected is None else selected repository_queue_id = repository_queue_id or 2000 + query_id self.conn.execute( '''INSERT INTO docker_depth_experiment_queries( id, experiment_id, source, query_ordinal, query, query_sha256, required_repository_count, selected_repository_count ) VALUES (?, 1, 'dockerhub', ?, ?, ?, ?, ?)''', (query_id, ordinal, query, f'{query_id:064x}', required, selected), ) if add_repository: self.conn.execute( '''INSERT INTO docker_depth_experiment_repositories( id, experiment_id, query_ordinal, source, query, repository_queue_id, repository_rank, work_state, resolver_attempts ) VALUES (?, 1, ?, 'dockerhub', ?, ?, 1, ?, ?)''', ( query_id, ordinal, query, repository_queue_id, repository_state, attempts, ), ) self.conn.execute( 'UPDATE docker_depth_experiments SET query_count = query_count + 1 WHERE id = 1' ) def add_target(self, target_id, query_ranks, *, layer_count=1, repository='repo'): target_queue_id = 1000 + target_id target = f'{repository}@sha256:{target_id:064x}' self.conn.execute( '''INSERT INTO target_queue( id, source, platform, query, target, normalized_target ) VALUES (?, 'dockerhub', 'docker', 'fixture', ?, ?)''', (target_queue_id, target, target), ) self.conn.execute( '''INSERT INTO docker_image_manifests( id, target_queue_id, repository, layer_count ) VALUES (?, ?, ?, ?)''', (target_id, target_queue_id, repository, layer_count), ) self.conn.execute( '''INSERT INTO docker_depth_experiment_targets( id, experiment_id, target_queue_id, manifest_id, state ) VALUES (?, 1, ?, ?, 'done')''', (target_id, target_queue_id, target_id), ) for query_id, rank in query_ranks.items(): query = self.conn.execute( '''SELECT query_ordinal FROM docker_depth_experiment_queries WHERE id = ?''', (query_id,), ).fetchone() self.conn.execute( '''INSERT INTO docker_depth_experiment_selections( experiment_id, query_ordinal, experiment_repository_id, experiment_target_id, image_rank ) VALUES (1, ?, ?, ?, ?)''', (query['query_ordinal'], query_id, target_id, rank), ) self.conn.execute( '''UPDATE docker_depth_experiments SET target_count = target_count + 1, selection_count = selection_count + ? WHERE id = 1''', (len(query_ranks),), ) def add_scan( self, target_id, scan_id, *, binding_id=None, attempt=1, duration=1.0, error_count=0, status='clean', target=None, binding_state='completed', reservation_state='acknowledged'): binding_id = scan_id if binding_id is None else binding_id queue_id = 1000 + target_id reservation_id = 5000 + binding_id queue = self.conn.execute( 'SELECT * FROM target_queue WHERE id = ?', (queue_id,), ).fetchone() target = str(target or queue['target']) normalized_target = target.lower() self.conn.execute( '''UPDATE target_queue SET target = ?, normalized_target = ? WHERE id = ?''', (target, normalized_target, queue_id), ) self.conn.execute( '''INSERT INTO result_reservations( id, queue_id, source, platform, query, target, normalized_target, claim_lease_token, scan_event_id, state ) VALUES (?, ?, 'dockerhub', 'docker', 'fixture', ?, ?, ?, ?, ?)''', ( reservation_id, queue_id, target, normalized_target, f'lease-{binding_id}', f'event-{binding_id}', reservation_state, ), ) self.conn.execute( '''INSERT INTO target_scans( id, scan_event_id, queue_id, result_reservation_id, claim_lease_token, source, query, target, normalized_target, scan_type, status, duration_sec, error_count, raw_result_json ) VALUES (?, ?, ?, ?, ?, 'dockerhub', 'fixture', ?, ?, 'docker', ?, ?, ?, 'private result')''', ( scan_id, f'event-{binding_id}', queue_id, reservation_id, f'lease-{binding_id}', target, normalized_target, status, duration, error_count, ), ) self.conn.execute( '''INSERT INTO docker_depth_experiment_scan_bindings( id, experiment_target_id, reservation_id, target_scan_id, attempt, state ) VALUES (?, ?, ?, ?, ?, ?)''', ( binding_id, target_id, reservation_id, scan_id, attempt, binding_state, ), ) return binding_id def add_scanless_binding( self, target_id, binding_id, *, attempt, binding_state, reservation_state): queue_id = 1000 + target_id reservation_id = 5000 + binding_id queue = self.conn.execute( 'SELECT * FROM target_queue WHERE id = ?', (queue_id,), ).fetchone() self.conn.execute( '''INSERT INTO result_reservations( id, queue_id, source, platform, query, target, normalized_target, claim_lease_token, scan_event_id, state ) VALUES (?, ?, 'dockerhub', 'docker', 'fixture', ?, ?, ?, ?, ?)''', ( reservation_id, queue_id, queue['target'], queue['normalized_target'], f'lease-{binding_id}', f'event-{binding_id}', reservation_state, ), ) self.conn.execute( '''INSERT INTO docker_depth_experiment_scan_bindings( id, experiment_target_id, reservation_id, target_scan_id, attempt, state ) VALUES (?, ?, ?, NULL, ?, ?)''', (binding_id, target_id, reservation_id, attempt, binding_state), ) def add_finding( self, finding_id, scan_id, fingerprint, detector_hash, *, secret='private-secret'): self.conn.execute( '''INSERT INTO findings( id, target_scan_id, finding_fingerprint, detector_secret_hash, raw_secret, redacted_secret, raw_finding_json ) VALUES (?, ?, ?, ?, ?, ?, ?)''', ( finding_id, scan_id, fingerprint, detector_hash, secret, f'redacted-{secret}', json.dumps({'excerpt': secret}), ), ) def add_layer(self, layer_id, target_id, base, top, digest='sha256:layer'): self.conn.execute( '''INSERT INTO docker_manifest_layers( id, manifest_id, position_from_base, position_from_top, layer_digest ) VALUES (?, ?, ?, ?, ?)''', (layer_id, target_id, base, top, digest), ) def add_attribution( self, attribution_id, binding_id, finding_id, *, state, layer_id=None, base=None, top=None, digest=None): self.conn.execute( '''INSERT INTO docker_finding_layer_attributions( id, scan_binding_id, finding_id, manifest_layer_id, attribution_state, reported_layer_digest, position_from_base, position_from_top ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)''', ( attribution_id, binding_id, finding_id, layer_id, state, digest, base, top, ), ) def add_credential(self, credential_id, secret='private-credential'): self.conn.execute( '''INSERT INTO keycheck_credentials( id, secret_text, secret_json, key_masked ) VALUES (?, ?, ?, ?)''', (credential_id, secret, json.dumps({'secret': secret}), f'***{secret}'), ) def add_result( self, result_id, credential_id, status, status_group, *, candidate_id=None, sensitive='private-result-evidence'): self.conn.execute( '''INSERT INTO keycheck_results( id, candidate_id, credential_id, status, status_group, message, source_line, metadata_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)''', ( result_id, candidate_id, credential_id, status, status_group, sensitive, sensitive, json.dumps({'evidence': sensitive}), ), ) def add_candidate( self, candidate_id, credential_id, finding_id, scan_id, *, state='completed', attempts=1, result_id=None): self.conn.execute( '''INSERT INTO keycheck_candidates( id, credential_id, finding_id, target_scan_id, state, attempts, keycheck_result_id, metadata_json ) VALUES (?, ?, ?, ?, ?, ?, ?, 'private candidate evidence')''', ( candidate_id, credential_id, finding_id, scan_id, state, attempts, result_id, ), ) def report(self, **selector): self.conn.commit() if not selector: selector = {'experiment_id': 1} return build_docker_depth_report(self.conn, **selector) def test_shared_attribution_is_separate_from_physical_totals(self): self.add_query(1, 0) self.add_query(2, 1) self.add_target(10, {1: 3, 2: 4}) self.add_scan(10, 100) report = self.report(experiment_key='depth-fixture') self.assertEqual(report['physical']['totals']['target_count'], 1) self.assertEqual( report['physical']['rank_buckets']['ranks_1_3']['target_count'], 1, ) self.assertEqual( report['physical']['rank_buckets']['ranks_4_10']['target_count'], 0, ) self.assertEqual(report['per_query']['1']['totals']['target_count'], 1) self.assertEqual(report['per_query']['2']['totals']['target_count'], 1) self.assertEqual( report['per_query']['1']['rank_buckets']['ranks_1_3']['target_count'], 1, ) self.assertEqual( report['per_query']['2']['rank_buckets']['ranks_4_10']['target_count'], 1, ) self.assertEqual(report['overlap']['targets']['shared_physical_count'], 1) self.assertEqual(report['overlap']['targets']['attribution_credit_count'], 2) self.assertEqual(report['overlap']['scans']['overlap_credit_count'], 1) def test_zero_member_query_reports_scarcity_without_missing_planned_work(self): self.add_query( 1, 0, required=10, selected=0, add_repository=False, ) coverage = self.report()['per_query']['1']['coverage'] self.assertEqual(coverage['required_repository_count'], 10) self.assertEqual(coverage['cohort_repository_count'], 0) self.assertEqual(coverage['unavailable_repository_count'], 10) self.assertEqual(coverage['repository_count'], 0) self.assertEqual(coverage['missing_cohort_repository_count'], 0) def test_image_unavailable_repository_is_reported_without_a_target(self): self.add_query( 1, 0, required=10, selected=1, repository_state='skipped', ) report = self.report() physical = report['physical']['coverage'] query = report['per_query']['1']['coverage'] self.assertEqual(physical['image_unavailable_repository_count'], 1) self.assertEqual( physical['image_unavailable_repository_membership_count'], 1, ) self.assertEqual( physical['queries_with_image_unavailable_repositories'], 1, ) self.assertEqual(query['image_unavailable_repository_count'], 1) self.assertEqual(query['target_count'], 0) def test_shared_repository_uses_physical_queue_identity_and_membership_credits(self): self.add_query(1, 0, repository_queue_id=9000) self.add_query(2, 1, repository_queue_id=9000) self.add_target(10, {1: 2, 2: 3}) report = self.report() coverage = report['physical']['coverage'] self.assertEqual(coverage['repository_count'], 1) self.assertEqual(coverage['repository_query_membership_credit_count'], 2) self.assertEqual(coverage['repository_overlap_credit_count'], 1) self.assertEqual(coverage['shared_repository_count'], 1) self.assertEqual(coverage['selected_repository_count'], 1) self.assertEqual( coverage['selected_repository_query_membership_credit_count'], 2, ) self.assertEqual(coverage['selected_repository_overlap_credit_count'], 1) self.assertEqual(report['per_query']['1']['coverage']['repository_count'], 1) self.assertEqual(report['per_query']['2']['coverage']['repository_count'], 1) self.assertEqual( report['overlap']['repositories'], { 'physical_count': 1, 'query_membership_credit_count': 2, 'overlap_credit_count': 1, 'shared_physical_count': 1, }, ) def test_replacement_repository_is_used_for_physical_overlap(self): self.add_query(1, 0, repository_queue_id=9001) self.add_query(2, 1, repository_queue_id=9002) self.conn.execute( '''UPDATE docker_depth_experiment_repositories SET replacement_repository_queue_id = 9999 WHERE id IN (1, 2)''' ) self.add_target(10, {1: 1, 2: 1}) coverage = self.report()['physical']['coverage'] self.assertEqual(coverage['repository_count'], 1) self.assertEqual(coverage['shared_repository_count'], 1) self.assertEqual(coverage['selected_repository_count'], 1) def test_queries_match_the_current_scanner_schema(self): with tempfile.TemporaryDirectory() as directory, mock.patch.dict( os.environ, {'SCANNER_DB_URL': '', 'DATABASE_URL': ''}): db = ScannerDB(db_path=os.path.join(directory, 'scanner.db'), db_url='') try: now = utc_now_iso() digest = 'a' * 64 experiment_id = db.conn.execute( '''INSERT INTO docker_depth_experiments( experiment_key, source, state, config_sha256, ordered_queries_sha256, selector_version, selector_sha256, provenance_policy_sha256, query_count, repositories_per_query, images_per_repository, target_limit, created_at, updated_at ) VALUES ( 'schema-smoke', 'dockerhub', 'held', ?, ?, 'selector-v1', ?, ?, 1, 1, 1, 1, ?, ? )''', (digest, digest, digest, digest, now, now), ).lastrowid db.conn.commit() report = build_docker_depth_report( db.conn, experiment_id=experiment_id, ) self.assertEqual(report['experiment']['id'], experiment_id) self.assertEqual(report['physical']['totals']['target_count'], 0) self.assertEqual(report['per_query'], {}) finally: db.close() def test_rank_three_and_four_are_in_different_buckets(self): self.add_query(1, 0) self.add_target(10, {1: 1}) self.add_target(20, {1: 3}) self.add_target(30, {1: 4}) self.add_target(40, {1: 10}) report = self.report() buckets = report['physical']['rank_buckets'] self.assertEqual(buckets['ranks_1_3']['target_count'], 2) self.assertEqual(buckets['ranks_4_10']['target_count'], 2) def test_minimum_rank_yield_uses_rank_three_at_the_bucket_boundary(self): self.add_query(1, 0) self.add_target(10, {1: 4}) self.add_target(20, {1: 3}) self.add_scan(10, 100) self.add_scan(20, 200) fingerprint = 'f' * 64 detector_hash = 'd' * 64 self.add_finding(1000, 100, fingerprint, detector_hash) self.add_finding(2000, 200, fingerprint, detector_hash) self.add_credential(50) self.add_candidate(1, 50, 1000, 100) self.add_candidate(2, 50, 2000, 200) marginal = self.report()['physical']['marginal_minimum_rank_yield'] for metric in ('findings', 'detector_secret_identities', 'credentials'): self.assertEqual(marginal[metric]['deduplicated_total'], 1) self.assertEqual(marginal[metric]['ranks_1_3'], 1) self.assertEqual(marginal[metric]['ranks_4_10'], 0) def test_duplicate_identities_have_one_minimum_rank_yield(self): self.add_query(1, 0) self.add_target(10, {1: 2}) self.add_target(20, {1: 7}) self.add_scan(10, 100) self.add_scan(20, 200) fingerprint = 'f' * 64 detector_hash = 'd' * 64 self.add_finding(1000, 100, fingerprint, detector_hash) self.add_finding(2000, 200, fingerprint, detector_hash) self.add_credential(50) self.add_candidate(1, 50, 1000, 100, state='pending') self.add_candidate(2, 50, 2000, 200, state='pending') report = self.report() totals = report['physical']['totals'] marginal = report['physical']['marginal_minimum_rank_yield'] self.assertEqual(totals['findings']['occurrence_count'], 2) self.assertEqual(totals['findings']['deduplicated_count'], 1) self.assertEqual(totals['credentials']['deduplicated_count'], 1) self.assertEqual(marginal['findings']['ranks_1_3'], 1) self.assertEqual(marginal['findings']['ranks_4_10'], 0) self.assertEqual(marginal['detector_secret_identities']['ranks_1_3'], 1) self.assertEqual(marginal['credentials']['ranks_1_3'], 1) self.assertEqual(marginal['credentials']['ranks_4_10'], 0) def test_pending_frozen_and_current_verification_are_separate(self): self.add_query(1, 0) self.add_target(10, {1: 1}) self.add_scan(10, 100) self.add_finding(1000, 100, 'a' * 64, 'b' * 64) self.add_finding(1001, 100, 'c' * 64, 'd' * 64) self.add_credential(50) self.add_credential(60) self.add_result(500, 50, 'VALID', 'alive', candidate_id=5) self.add_result(501, 50, 'DEAD', 'dead') self.add_candidate(5, 50, 1000, 100, result_id=500) self.add_candidate(6, 60, 1001, 100, state='pending', attempts=0) self.conn.execute( '''INSERT INTO keycheck_current_state( credential_id, status, status_group, last_result_id, metadata_json ) VALUES (50, 'DEAD', 'dead', 501, 'private current evidence')''' ) keychecks = self.report()['physical']['totals']['keychecks'] self.assertEqual(keychecks['pending_candidate_count'], 1) self.assertEqual(keychecks['missing_frozen_result_candidate_count'], 1) self.assertEqual(keychecks['frozen']['credential_count'], 1) self.assertEqual(keychecks['frozen']['missing_credential_count'], 1) self.assertEqual(keychecks['frozen']['status_counts'], {'VALID': 1}) self.assertEqual(keychecks['current']['credential_count'], 1) self.assertEqual(keychecks['current']['missing_credential_count'], 1) self.assertEqual(keychecks['current']['missing_backing_result_count'], 0) self.assertEqual(keychecks['current']['status_counts'], {'DEAD': 1}) def test_keycheck_outcomes_require_candidate_and_credential_identity(self): self.add_query(1, 0) self.add_target(10, {1: 1}) self.add_scan(10, 100) self.add_finding(1000, 100, 'a' * 64, 'b' * 64) self.add_credential(50) self.add_credential(60) self.add_result(500, 60, 'VALID', 'alive', candidate_id=5) self.add_result(501, 60, 'DEAD', 'dead', candidate_id=6) self.add_candidate(5, 50, 1000, 100, result_id=500) self.conn.execute( '''INSERT INTO keycheck_current_state( credential_id, status, status_group, last_result_id, metadata_json ) VALUES (50, 'DEAD', 'dead', 501, 'private current evidence')''' ) keychecks = self.report()['physical']['totals']['keychecks'] self.assertEqual(keychecks['missing_frozen_result_candidate_count'], 1) self.assertEqual(keychecks['frozen']['result_count'], 0) self.assertEqual(keychecks['frozen']['credential_count'], 0) self.assertEqual(keychecks['current']['credential_count'], 1) self.assertEqual(keychecks['current']['missing_backing_result_count'], 1) self.assertEqual( keychecks['current']['status_counts'], {'invalid_or_unknown': 1}, ) def test_duplicate_layer_digest_positions_remain_exact_rows(self): self.add_query(1, 0) self.add_target(10, {1: 1}, layer_count=3) binding_id = self.add_scan(10, 100) self.add_finding(1000, 100, 'a' * 64, 'b' * 64) digest = 'sha256:duplicate' self.add_layer(1, 10, 1, 3, digest) self.add_layer(2, 10, 3, 1, digest) self.add_attribution( 1, binding_id, 1000, state='exact', layer_id=1, base=1, top=3, digest=digest, ) self.add_attribution( 2, binding_id, 1000, state='exact', layer_id=2, base=3, top=1, digest=digest, ) totals = self.report()['physical']['totals'] layers = totals['layers'] self.assertEqual(totals['declared_manifest_layer_count'], 3) self.assertEqual(layers['exact_attribution_count'], 2) self.assertEqual(layers['exact_manifest_layer_count'], 2) self.assertEqual(layers['positions_from_base'], {'1': 1, '3': 1}) self.assertEqual(layers['positions_from_top'], {'1': 1, '3': 1}) self.assertEqual(layers['unattributed_finding_occurrence_count'], 0) def test_stale_or_cross_queue_bindings_cannot_admit_evidence(self): self.add_query(1, 0) for target_id, rank in ((10, 1), (20, 2), (30, 3), (40, 4)): self.add_target(target_id, {1: rank}) self.add_scan(target_id, target_id * 10) self.add_finding( target_id * 100, target_id * 10, f'{target_id:064x}', f'{target_id + 1:064x}', ) self.conn.execute( 'UPDATE result_reservations SET queue_id = 999999 WHERE id = 5200' ) self.conn.execute('UPDATE target_scans SET queue_id = 999998 WHERE id = 300') self.conn.execute( 'UPDATE target_scans SET result_reservation_id = 5100 WHERE id = 400' ) self.add_credential(50) self.add_candidate(1, 50, 2000, 200) self.conn.execute( '''INSERT INTO errors(id, target_scan_id, category, summary, raw_error) VALUES (1, 100, 'timeout', 'private', 'private'), (2, 200, 'network', 'private', 'private')''' ) totals = self.report()['physical']['totals'] self.assertEqual(totals['target_count'], 4) self.assertEqual(totals['targets_with_bindings'], 3) self.assertEqual(totals['targets_with_scans'], 1) self.assertEqual(totals['scan_bindings']['attempt_count'], 3) self.assertEqual(totals['scans']['count'], 1) self.assertEqual(totals['findings']['occurrence_count'], 1) self.assertEqual(totals['keychecks']['candidate_count'], 0) self.assertEqual(totals['scans']['errors']['row_count'], 1) self.assertEqual( totals['scans']['errors']['category_counts'], {'timeout': 1}, ) def test_retries_duration_errors_and_coverage_are_aggregated_once(self): self.add_query(1, 0, attempts=3) self.add_target(10, {1: 1}) self.add_scan(10, 100, attempt=1, duration=2.0, error_count=1) self.add_scan(10, 200, attempt=2, duration=3.0) self.conn.execute( '''INSERT INTO errors(id, target_scan_id, summary, raw_error) VALUES (1, 100, 'private summary', 'private raw error')''' ) report = self.report() totals = report['physical']['totals'] coverage = report['per_query']['1']['coverage'] self.assertEqual(totals['scan_bindings']['attempt_count'], 2) self.assertEqual(totals['scan_bindings']['retry_count'], 1) self.assertEqual(totals['scan_bindings']['maximum_attempt'], 2) self.assertEqual(totals['scans']['count'], 2) self.assertEqual(totals['scans']['duration']['total_seconds'], 5.0) self.assertEqual(totals['scans']['errors']['row_count'], 1) self.assertEqual(totals['scans']['errors']['reported_count'], 1) self.assertEqual(totals['scans']['errors']['scan_count'], 1) self.assertEqual(coverage['repository_count'], 1) self.assertEqual(coverage['selected_repository_count'], 1) self.assertEqual(coverage['resolver_attempt_count'], 3) self.assertEqual(coverage['resolver_retry_count'], 2) def test_scanless_refund_and_retry_attempts_remain_visible(self): self.add_query(1, 0) self.add_target(10, {1: 1}) self.add_scanless_binding( 10, 100, attempt=1, binding_state='released', reservation_state='refunded', ) self.add_scanless_binding( 10, 200, attempt=2, binding_state='scanning', reservation_state='ready', ) self.conn.execute( "UPDATE docker_depth_experiment_targets SET state = 'scanning' WHERE id = 10" ) totals = self.report()['physical']['totals'] self.assertEqual(totals['targets_with_bindings'], 1) self.assertEqual(totals['targets_with_scans'], 0) self.assertEqual(totals['scan_bindings']['attempt_count'], 2) self.assertEqual(totals['scan_bindings']['retry_count'], 1) self.assertEqual(totals['scan_bindings']['refunded_attempt_count'], 1) self.assertEqual(totals['scan_bindings']['maximum_attempt'], 2) self.assertEqual( totals['scan_bindings']['state_counts'], {'released': 1, 'scanning': 1}, ) self.assertEqual( totals['scan_bindings']['reservation_state_counts'], {'ready': 1, 'refunded': 1}, ) def test_canonical_unknown_and_foundry_statuses_are_not_collapsed(self): self.add_query(1, 0) self.add_target(10, {1: 1}) self.add_scan(10, 100) self.add_finding(1000, 100, 'a' * 64, 'b' * 64) self.add_finding(1001, 100, 'c' * 64, 'd' * 64) self.add_credential(50) self.add_credential(60) self.add_result(500, 50, 'UNKNOWN', 'unknown', candidate_id=5) self.add_result( 600, 60, 'FOUNDRY_BAD_ENDPOINT', 'unknown', candidate_id=6, ) self.add_candidate(5, 50, 1000, 100, result_id=500) self.add_candidate(6, 60, 1001, 100, result_id=600) self.conn.execute( '''INSERT INTO keycheck_current_state( credential_id, status, status_group, last_result_id, metadata_json ) VALUES (50, 'UNKNOWN', 'unknown', 500, '{}'), (60, 'FOUNDRY_BAD_ENDPOINT', 'unknown', 600, '{}')''' ) self.conn.execute( "INSERT INTO errors(id, target_scan_id, category) VALUES (1, 100, 'unknown')" ) totals = self.report()['physical']['totals'] self.assertEqual( totals['keychecks']['frozen']['status_counts'], {'FOUNDRY_BAD_ENDPOINT': 1, 'UNKNOWN': 1}, ) self.assertEqual( totals['keychecks']['current']['status_group_counts'], {'unknown': 2}, ) self.assertEqual( totals['scans']['errors']['category_counts'], {'unknown': 1}, ) def test_unattributed_findings_and_secret_sentinels_never_leak(self): sentinel = 'SECRET_SENTINEL_DO_NOT_LEAK' self.add_query(1, 0, query=f'query-{sentinel}') self.add_target(10, {1: 1}, repository=f'repository-{sentinel}') binding_id = self.add_scan(10, 100, target=f'target-{sentinel}') self.add_finding(1000, 100, 'a' * 64, 'b' * 64, secret=sentinel) self.add_attribution( 1, binding_id, 1000, state='unattributed', ) self.add_credential(50, secret=sentinel) self.add_result( 500, 50, 'VALID', 'alive', candidate_id=5, sensitive=sentinel, ) self.add_candidate(5, 50, 1000, 100, result_id=500) self.conn.execute( 'INSERT INTO errors(id, target_scan_id, summary, raw_error) VALUES (1, 100, ?, ?)', (sentinel, sentinel), ) report = self.report() serialized = json.dumps(report, sort_keys=True) self.assertNotIn(sentinel, serialized) layers = report['physical']['totals']['layers'] self.assertEqual(layers['unattributed_attribution_count'], 1) self.assertEqual(layers['unattributed_finding_occurrence_count'], 1) self.assertEqual(layers['unattributed_deduplicated_finding_count'], 1) def test_untrusted_enum_and_error_sentinel_is_collapsed(self): sentinel = 'SECRET_SENTINEL_DO_NOT_LEAK' self.add_query(1, 0) self.add_target(10, {1: 1}) binding_id = self.add_scan(10, 100) self.add_finding(1000, 100, 'a' * 64, 'b' * 64) self.add_attribution(1, binding_id, 1000, state=sentinel) self.add_credential(50) self.add_result(500, 50, sentinel, sentinel, candidate_id=5) self.add_candidate(5, 50, 1000, 100, state=sentinel, result_id=500) self.conn.execute( '''INSERT INTO keycheck_current_state( credential_id, status, status_group, last_result_id, metadata_json ) VALUES (50, ?, ?, 500, 'private current evidence')''', (sentinel, sentinel), ) self.conn.execute( '''INSERT INTO errors(id, target_scan_id, category, summary, raw_error) VALUES (1, 100, ?, 'private', 'private')''', (sentinel,), ) self.conn.execute( '''UPDATE docker_depth_experiments SET experiment_key = ?, source = ?, state = ? WHERE id = 1''', (sentinel, sentinel, sentinel), ) self.conn.execute( '''UPDATE docker_depth_experiment_queries SET source = ?, query_sha256 = ? WHERE id = 1''', (sentinel, sentinel), ) self.conn.execute( '''UPDATE docker_depth_experiment_repositories SET source = ?, work_state = ? WHERE id = 1''', (sentinel, sentinel), ) self.conn.execute( 'UPDATE docker_depth_experiment_targets SET state = ? WHERE id = 10', (sentinel,), ) self.conn.execute( '''UPDATE docker_depth_experiment_scan_bindings SET state = ? WHERE id = ?''', (sentinel, binding_id), ) self.conn.execute( 'UPDATE target_scans SET status = ? WHERE id = 100', (sentinel,), ) report = self.report() totals = report['physical']['totals'] self.assertNotIn(sentinel, json.dumps(report, sort_keys=True)) self.assertEqual(report['experiment']['key'], 'invalid_or_unknown') self.assertEqual(report['experiment']['source'], 'invalid_or_unknown') self.assertEqual(report['experiment']['state'], 'invalid_or_unknown') self.assertEqual(report['per_query']['1']['query']['source'], 'invalid_or_unknown') self.assertEqual(report['per_query']['1']['query']['sha256'], 'invalid_or_unknown') self.assertEqual(totals['target_state_counts'], {'invalid_or_unknown': 1}) self.assertEqual( totals['scan_bindings']['state_counts'], {'invalid_or_unknown': 1}, ) self.assertEqual(totals['scans']['status_counts'], {'invalid_or_unknown': 1}) self.assertEqual( totals['scans']['errors']['category_counts'], {'invalid_or_unknown': 1}, ) self.assertEqual( totals['keychecks']['candidate_state_counts'], {'invalid_or_unknown': 1}, ) self.assertEqual( totals['keychecks']['frozen']['status_counts'], {'invalid_or_unknown': 1}, ) self.assertEqual( totals['keychecks']['current']['status_group_counts'], {'invalid_or_unknown': 1}, ) self.assertEqual(totals['layers']['invalid_or_unknown_attribution_count'], 1) def test_sqlite_report_uses_one_snapshot_during_concurrent_commit(self): with tempfile.TemporaryDirectory() as directory: path = os.path.join(directory, 'report.db') reader = sqlite3.connect(path) reader.row_factory = sqlite3.Row reader.execute('PRAGMA journal_mode=WAL') reader.executescript(SCHEMA) reader.execute( '''INSERT INTO docker_depth_experiments( id, experiment_key, source, state, query_count, target_count, selection_count ) VALUES (1, 'snapshot', 'dockerhub', 'completed', 0, 0, 0)''' ) reader.commit() writer = sqlite3.connect(path) writer.row_factory = sqlite3.Row class ConcurrentConnection: def __init__(self): self.changed = False @property def in_transaction(self): return reader.in_transaction def execute(self, sql, params=None): cursor = reader.execute(sql, tuple(params or ())) if ( not self.changed and 'FROM docker_depth_experiment_queries' in sql ): writer.execute( 'UPDATE docker_depth_experiments SET target_count = 99 WHERE id = 1' ) writer.commit() self.changed = True return cursor def rollback(self): return reader.rollback() try: report = build_docker_depth_report( ConcurrentConnection(), experiment_id=1, ) self.assertEqual( report['experiment']['persisted_counts']['targets'], 0, ) self.assertEqual( writer.execute( 'SELECT target_count FROM docker_depth_experiments WHERE id = 1' ).fetchone()['target_count'], 99, ) self.assertFalse(reader.in_transaction) self.assertEqual( report['snapshot']['consistency'], 'sqlite_transaction', ) finally: writer.close() reader.close() def test_postgres_report_transaction_is_read_only_and_rolled_back(self): class Connection: is_postgres = True def __init__(self): self.statements = [] self.rollbacks = 0 def execute(self, sql, params=None): self.statements.append(str(sql)) return object() def rollback(self): self.rollbacks += 1 connection = Connection() fixture = { 'experiment': {'state': 'completed'}, 'physical': {}, 'per_query': {}, 'overlap': {}, } with mock.patch.object( docker_depth_report, '_build_docker_depth_report_snapshot', return_value=fixture, ): report = build_docker_depth_report(connection, experiment_id=1) self.assertEqual( connection.statements, ['BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY'], ) self.assertEqual(connection.rollbacks, 1) self.assertEqual(report['snapshot']['consistency'], 'repeatable_read') def test_raw_postgres_connection_is_rejected(self): class Connection: def execute(self, _sql, _params=None): raise AssertionError('raw connection must be rejected before SQL') Connection.__module__ = 'psycopg' with self.assertRaisesRegex(ValueError, 'DatabaseConnection'): build_docker_depth_report(Connection(), experiment_id=1) def test_finding_position_fanout_is_aggregated_before_materialization(self): source = (APP_DIR / 'docker_depth_report.py').read_text(encoding='utf-8') self.assertIn('cursor.fetchmany(512)', source) self.assertIn('COUNT(a.id) AS attribution_count', source) self.assertIn( 'GROUP BY rb.target_id, ml.id, ml.position_from_base,', source, ) self.assertNotIn('ORDER BY rb.target_id, f.id, a.id', source) if __name__ == '__main__': unittest.main()