import builtins import hashlib import json import os import sys import tempfile import unittest from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parents[1] APP_DIR = ROOT / 'app' sys.path.insert(0, str(APP_DIR)) import scanner import scanner_db from keycheckers import keycheck_common class FindingLineSafetyTests(unittest.TestCase): @classmethod def setUpClass(cls): scanner.initialize_scanner_runtime(preflight_complete=True, register_cleanup=False) @staticmethod def reader_env(state_dir): return mock.patch.dict(os.environ, { 'KEYCHECK_OUTPUT_DIR': state_dir, 'KEYCHECK_STATE_DIR': state_dir, 'KEYCHECK_SERVICE': 'fixture', 'KEYCHECK_INPUT_TAIL_BYTES': '0', 'KEYCHECK_INPUT_MAX_LINE_BYTES': '1024', }) def test_oversized_finding_projects_secret_free_marker_and_later_row_is_readable(self): sentinel = 'DO-NOT-PROJECT-THIS-SECRET' oversized_raw = sentinel * 100 oversized_uid = 'finding-oversized-0001' later_uid = 'finding-later-0002' result = { 'scan_event_id': 'scan-oversized-0001', 'target': 'fixture-target', 'scan_type': 'filesystem', 'timestamp': '2026-07-19T00:00:00+00:00', 'findings': [{ 'finding_uid': oversized_uid, 'DetectorName': 'OpenAI', 'Raw': oversized_raw, 'RawV2': sentinel, 'StructuredData': {'token': sentinel}, 'ScannerContext': {'nearby': sentinel, 'file': 'artifact.txt'}, 'PostmanContext': {'endpoint': sentinel}, 'SourceMetadata': {'Data': {'Filesystem': { 'file': 'artifact.txt', 'line': 7, 'commit': 'abc123', }}}, }, { 'finding_uid': later_uid, 'DetectorName': 'Anthropic', 'Raw': 'small-later-secret', }], 'errors': [], } with tempfile.TemporaryDirectory() as temp_dir: scanner.ensure_private_directory(temp_dir, reject_reparse=True) results_dir = os.path.join(temp_dir, 'results') state_dir = os.path.join(temp_dir, 'state') scanner.ensure_private_directory(results_dir, reject_reparse=True) scanner.ensure_private_directory(state_dir, reject_reparse=True) with mock.patch.object(scanner.scan_config, 'results_dir', results_dir), \ mock.patch.object(scanner.scan_config, 'jsonl_rotation_enabled', False), \ mock.patch.object(scanner.scan_config, 'keycheck_input_max_line_bytes', 1024), \ mock.patch.object(scanner, 'write_foundry_keycheck_candidates_from_findings', return_value=0): self.assertTrue(scanner.save_scan_result(result)) findings_path = os.path.join(results_dir, 'found_secrets.jsonl') raw_lines = Path(findings_path).read_bytes().splitlines(keepends=True) self.assertEqual(len(raw_lines), 2) self.assertTrue(all(len(line) <= 1024 for line in raw_lines)) marker = json.loads(raw_lines[0]) self.assertEqual(marker['finding_uid'], oversized_uid) self.assertEqual(marker['DetectorName'], 'OpenAI') self.assertTrue(marker['finding_omitted']) self.assertTrue(marker['keycheck_uncheckable']) self.assertEqual(marker['secret_sha256'], hashlib.sha256(sentinel.encode()).hexdigest()) self.assertEqual(marker['SourceIdentity']['file'], 'artifact.txt') self.assertNotIn(sentinel, raw_lines[0].decode('utf-8')) for forbidden in ('Raw', 'RawV2', 'StructuredData', 'ScannerContext', 'PostmanContext'): self.assertNotIn(forbidden, marker) scan_row = json.loads(Path(os.path.join(results_dir, 'scan_results.jsonl')).read_text(encoding='utf-8')) self.assertTrue(any('oversized' in warning.lower() for warning in scan_row['warnings'])) self.assertNotIn(sentinel, json.dumps(scan_row)) self.assertIn(sentinel, result['findings'][0]['Raw']) with self.reader_env(state_dir): rows = [item['data'] for item in keycheck_common.iter_jsonl_input(findings_path)] self.assertEqual([row['finding_uid'] for row in rows], [oversized_uid, later_uid]) def test_plain_legacy_oversized_row_is_skipped_and_checkpointed(self): with tempfile.TemporaryDirectory() as temp_dir: state_dir = os.path.join(temp_dir, 'state') scanner.ensure_private_directory(state_dir, reject_reparse=True) path = os.path.join(temp_dir, 'found_secrets.jsonl') oversized = json.dumps({'finding_uid': 'legacy', 'Raw': 'x' * 3000}).encode() + b'\n' later = b'{"finding_uid":"later"}\n' Path(path).write_bytes(oversized + later) scanner.harden_private_file(path) with self.reader_env(state_dir): rows = [item['data'] for item in keycheck_common.iter_jsonl_input(path)] self.assertEqual(rows, [{'finding_uid': 'later'}]) state = json.loads(Path(os.path.join(state_dir, 'input_state.json')).read_text(encoding='utf-8')) self.assertEqual(state['offset'], os.path.getsize(path)) self.assertEqual(state['skipped_oversized'], 1) def test_segmented_legacy_oversized_row_does_not_block_segment_retirement(self): with tempfile.TemporaryDirectory() as temp_dir: state_dir = os.path.join(temp_dir, 'state') scanner.ensure_private_directory(state_dir, reject_reparse=True) current = os.path.join(temp_dir, 'found_secrets.jsonl') segment = os.path.join(temp_dir, 'found_secrets.000001.jsonl') oversized = json.dumps({'finding_uid': 'legacy', 'Raw': 'x' * 3000}).encode() + b'\n' Path(segment).write_bytes(oversized + b'{"finding_uid":"segment-later"}\n') Path(current).write_bytes(b'{"finding_uid":"current-later"}\n') scanner.harden_private_file(segment) scanner.harden_private_file(current) with self.reader_env(state_dir): rows = [item['data'] for item in keycheck_common.iter_jsonl_input(current)] self.assertEqual( [row['finding_uid'] for row in rows], ['segment-later', 'current-later'], ) state = json.loads(Path(os.path.join(state_dir, 'input_state.json')).read_text(encoding='utf-8')) segment_state = state['files'][os.path.abspath(segment)] self.assertTrue(segment_state['done']) self.assertEqual(segment_state['offset'], os.path.getsize(segment)) self.assertEqual(segment_state['skipped_oversized'], 1) def test_torn_oversized_row_remains_fail_closed(self): with tempfile.TemporaryDirectory() as temp_dir: state_dir = os.path.join(temp_dir, 'state') scanner.ensure_private_directory(state_dir, reject_reparse=True) path = os.path.join(temp_dir, 'found_secrets.jsonl') Path(path).write_bytes(b'{"finding_uid":"torn","Raw":"' + (b'x' * 3000)) scanner.harden_private_file(path) with self.reader_env(state_dir): with self.assertRaisesRegex(RuntimeError, 'torn oversized'): list(keycheck_common.iter_jsonl_input(path)) self.assertFalse(os.path.exists(os.path.join(state_dir, 'input_state.json'))) def test_only_exact_resolved_corrupt_record_is_skipped(self): with tempfile.TemporaryDirectory() as temp_dir: scanner.ensure_private_directory(temp_dir, reject_reparse=True) state_dir = os.path.join(temp_dir, 'state') scanner.ensure_private_directory(state_dir, reject_reparse=True) current = os.path.join(temp_dir, 'found_secrets.jsonl') segment = os.path.join(temp_dir, 'found_secrets.000001.jsonl') first = b'{"finding_uid":"first"}\n' corrupt = b'1, "legacy":"reviewed"}\n' last = b'{"finding_uid":"last"}\n' Path(segment).write_bytes(first + corrupt + last) Path(current).write_bytes(b'{"finding_uid":"current"}\n') scanner.harden_private_file(segment) scanner.harden_private_file(current) offset = len(first) digest = hashlib.sha256(corrupt).hexdigest() with self.assertRaisesRegex(scanner.JsonlProjectionReconciliationRequired, 'explicit review'): scanner.reconcile_projection_ledger_batch(current, 'finding_uid') scanner.approve_projection_reconciliation_issue( current, 'finding_uid', os.path.basename(segment), offset, digest, ) self.assertTrue(scanner.reconcile_projection_ledger_batch(current, 'finding_uid')['complete']) with self.reader_env(state_dir): rows = [item['data']['finding_uid'] for item in keycheck_common.iter_jsonl_input(current)] self.assertEqual(rows, ['first', 'last', 'current']) state = json.loads(Path(state_dir, 'input_state.json').read_text(encoding='utf-8')) self.assertEqual(state['skipped_reviewed_corrupt'], 1) def test_unreviewed_corrupt_record_remains_fail_closed(self): with tempfile.TemporaryDirectory() as temp_dir: state_dir = os.path.join(temp_dir, 'state') scanner.ensure_private_directory(state_dir, reject_reparse=True) path = os.path.join(temp_dir, 'found_secrets.jsonl') Path(path).write_bytes(b'{"finding_uid":"first"}\ninvalid-json\n') scanner.harden_private_file(path) with self.reader_env(state_dir): with self.assertRaisesRegex(RuntimeError, 'invalid committed keycheck input'): list(keycheck_common.iter_jsonl_input(path)) def test_reviewed_corrupt_record_mutation_remains_fail_closed(self): with tempfile.TemporaryDirectory() as temp_dir: scanner.ensure_private_directory(temp_dir, reject_reparse=True) state_dir = os.path.join(temp_dir, 'state') scanner.ensure_private_directory(state_dir, reject_reparse=True) path = os.path.join(temp_dir, 'found_secrets.jsonl') first = b'{"finding_uid":"first"}\n' corrupt = b'invalid-one\n' replacement = b'invalid-two\n' self.assertEqual(len(corrupt), len(replacement)) Path(path).write_bytes(first + corrupt) scanner.harden_private_file(path) offset = len(first) digest = hashlib.sha256(corrupt).hexdigest() with self.assertRaises(scanner.JsonlProjectionReconciliationRequired): scanner.reconcile_projection_ledger_batch(path, 'finding_uid') scanner.approve_projection_reconciliation_issue( path, 'finding_uid', os.path.basename(path), offset, digest, ) self.assertTrue(scanner.reconcile_projection_ledger_batch(path, 'finding_uid')['complete']) identity = os.stat(path, follow_symlinks=False) with open(path, 'r+b') as handle: handle.seek(offset) handle.write(replacement) handle.flush() os.fsync(handle.fileno()) os.utime(path, ns=(identity.st_atime_ns, identity.st_mtime_ns)) with self.reader_env(state_dir): with self.assertRaisesRegex(RuntimeError, 'reviewed keycheck corruption record changed'): list(keycheck_common.iter_jsonl_input(path)) class OptionalContextSafetyTests(unittest.TestCase): @classmethod def setUpClass(cls): scanner.initialize_scanner_runtime(preflight_complete=True, register_cleanup=False) def test_postman_parse_exception_retains_successful_trufflehog_finding(self): finding = {'DetectorName': 'OpenAI', 'Raw': 'fixture-postman-secret'} stdout = json.dumps(finding) + '\n' with tempfile.TemporaryDirectory() as temp_dir: cache_path = os.path.join(temp_dir, 'cache.json') work_dir = os.path.join(temp_dir, 'work') Path(cache_path).write_text('{"token":"fixture-postman-secret"}', encoding='utf-8') scanner.ensure_private_directory(work_dir, reject_reparse=True) target_data = { 'cache_path': cache_path, 'sha256': 'a' * 64, 'size': os.path.getsize(cache_path), 'kind': 'collection', } with mock.patch.object(scanner, 'parse_postman_target', return_value=target_data), \ mock.patch.object(scanner, 'validate_postman_cache_artifact', return_value=(cache_path, os.path.getsize(cache_path))), \ mock.patch.object(scanner, 'create_command_work_dir', return_value=work_dir), \ mock.patch.object(scanner, 'cleanup_command_work_dir'), \ mock.patch.object(scanner, 'get_trufflehog_cmd', return_value='trufflehog'), \ mock.patch.object(scanner, 'run_command_streamed', return_value=scanner.streamed_output_from_text(stdout, '', 0)), \ mock.patch.object(scanner, 'load_postman_context', side_effect=scanner.PostmanCacheValidationError('fixture')): result = scanner.scan_postman_target('fixture-target') self.assertEqual(result['findings'], [finding]) self.assertFalse(result.get('structured_keycheck_pending', False)) self.assertTrue(result['context_enrichment_degraded']) self.assertTrue(any('Postman JSON' in warning for warning in result['warnings'])) def test_bruno_text_artifact_skips_postman_json_context(self): with tempfile.TemporaryDirectory() as temp_dir: cache_path = os.path.join(temp_dir, 'request.bru') work_dir = os.path.join(temp_dir, 'work') Path(cache_path).write_text('meta {\n name: fixture\n}\n', encoding='utf-8') scanner.ensure_private_directory(work_dir, reject_reparse=True) target_data = { 'cache_path': cache_path, 'path': 'collection/request.bru', 'sha256': 'a' * 64, 'size': os.path.getsize(cache_path), 'kind': 'bruno', } with mock.patch.object(scanner, 'parse_postman_target', return_value=target_data), \ mock.patch.object(scanner, 'validate_postman_cache_artifact', return_value=(cache_path, os.path.getsize(cache_path))), \ mock.patch.object(scanner, 'create_command_work_dir', return_value=work_dir), \ mock.patch.object(scanner, 'cleanup_command_work_dir'), \ mock.patch.object(scanner, 'get_trufflehog_cmd', return_value='trufflehog'), \ mock.patch.object(scanner, 'run_command_streamed', return_value=scanner.streamed_output_from_text('', '', 0)), \ mock.patch.object(scanner, 'load_postman_context', side_effect=AssertionError('JSON parser must not run')): result = scanner.scan_postman_target('fixture-target') self.assertEqual(result['findings'], []) self.assertEqual(result['errors'], []) self.assertFalse(result.get('context_enrichment_degraded', False)) self.assertFalse(result.get('degraded', False)) def test_nearby_context_reads_shared_file_once_for_many_findings(self): with tempfile.TemporaryDirectory() as temp_dir: source_path = os.path.join(temp_dir, 'source.txt') Path(source_path).write_text(''.join(f'line-{index}\n' for index in range(100)), encoding='utf-8') findings = [{ 'DetectorName': 'OpenAI', 'Raw': f'secret-{index}', 'SourceMetadata': {'Data': {'Filesystem': {'file': source_path, 'line': index + 1}}}, } for index in range(20)] result = {'findings': findings, 'errors': []} real_open = builtins.open source_reads = [] def counting_open(path, mode='r', *args, **kwargs): if os.path.normcase(os.path.abspath(os.fspath(path))) == os.path.normcase(os.path.abspath(source_path)) and mode == 'rb': source_reads.append(path) return real_open(path, mode, *args, **kwargs) with mock.patch.object(scanner.scan_config, 'context_enrichment_max_source_bytes', 4096), \ mock.patch.object(scanner.scan_config, 'context_enrichment_max_findings', 100), \ mock.patch.object(scanner.scan_config, 'context_enrichment_max_elapsed_sec', 30), \ mock.patch('builtins.open', side_effect=counting_open): scanner.attach_nearby_context(result) self.assertEqual(len(source_reads), 1) self.assertTrue(all(finding.get('ScannerContext') for finding in findings)) def test_postman_comparison_budget_stops_work_without_dropping_findings(self): with tempfile.TemporaryDirectory() as temp_dir: cache_path = os.path.join(temp_dir, 'cache.json') Path(cache_path).write_text('{}', encoding='utf-8') findings = [ {'DetectorName': 'OpenAI', 'Raw': f'unmatched-secret-{index}'} for index in range(5) ] original = json.loads(json.dumps(findings)) contexts = [{ 'path': f'$.values[{index}]', 'key': 'token', 'value': f'different-value-{index}', 'endpoint': '', 'host': '', 'auth_type': '', 'location': 'value', } for index in range(10)] with mock.patch.object(scanner.scan_config, 'context_enrichment_max_source_bytes', 1024), \ mock.patch.object(scanner.scan_config, 'context_enrichment_max_findings', 100), \ mock.patch.object(scanner.scan_config, 'context_enrichment_max_postman_comparisons', 3), \ mock.patch.object(scanner.scan_config, 'context_enrichment_max_elapsed_sec', 30): budget = scanner.context_enrichment_budget() with mock.patch.object(scanner, 'load_postman_context', return_value=contexts): result = scanner.attach_postman_context({'findings': findings, 'errors': []}, cache_path, budget) self.assertEqual(budget['postman_comparisons'], 3) self.assertEqual(result['findings'], original) self.assertTrue(result['structured_keycheck_pending']) budget_warnings = [warning for warning in result['warnings'] if 'comparison budget' in warning] self.assertEqual(len(budget_warnings), 1) class PostmanEndpointSanitizationTests(unittest.TestCase): USER_SENTINEL = 'POSTMAN-URL-USER-SENTINEL' PASSWORD_SENTINEL = 'POSTMAN-URL-PASSWORD-SENTINEL' API_KEY_SENTINEL = 'POSTMAN-QUERY-API-KEY-SENTINEL' TOKEN_SENTINEL = 'POSTMAN-QUERY-TOKEN-SENTINEL' QUERY_PASSWORD_SENTINEL = 'POSTMAN-QUERY-PASSWORD-SENTINEL' FRAGMENT_SENTINEL = 'POSTMAN-FRAGMENT-SENTINEL' SAFE_ENDPOINT = 'https://normal.openai.azure.com:443/openai/deployments/demo' @classmethod def setUpClass(cls): scanner.initialize_scanner_runtime(preflight_complete=True, register_cleanup=False) @classmethod def credentialed_endpoint(cls): return ( f'https://{cls.USER_SENTINEL}:{cls.PASSWORD_SENTINEL}' '@normal.openai.azure.com:443/openai/deployments/demo' f'?api_key={cls.API_KEY_SENTINEL}&token={cls.TOKEN_SENTINEL}' f'&password={cls.QUERY_PASSWORD_SENTINEL}#{cls.FRAGMENT_SENTINEL}' ) @classmethod def sentinels(cls): return ( cls.USER_SENTINEL, cls.PASSWORD_SENTINEL, cls.API_KEY_SENTINEL, cls.TOKEN_SENTINEL, cls.QUERY_PASSWORD_SENTINEL, cls.FRAGMENT_SENTINEL, ) def test_scanner_context_removes_url_credentials_but_keeps_candidate_host(self): detected_secret = 'detected-postman-secret' contexts = [{ 'path': '$.item[0].request.auth', 'key': 'api_key', 'value': detected_secret, 'endpoint': self.credentialed_endpoint(), 'host': 'normal.openai.azure.com', 'auth_type': 'apikey', 'location': 'header', }] with tempfile.TemporaryDirectory() as temp_dir: cache_path = os.path.join(temp_dir, 'collection.json') Path(cache_path).write_text('{}', encoding='utf-8') with mock.patch.object(scanner, 'load_postman_context', return_value=contexts): result = scanner.attach_postman_context({ 'findings': [{'DetectorName': 'OpenAI', 'Raw': detected_secret}], 'errors': [], }, cache_path) context = result['findings'][0]['PostmanContext'] self.assertEqual(context['endpoint'], self.SAFE_ENDPOINT) self.assertEqual(context['host'], 'normal.openai.azure.com') persisted_finding = json.dumps(result['findings'][0]) self.assertTrue(all(sentinel not in persisted_finding for sentinel in self.sentinels())) endpoints, _ = scanner.context_values_for_pairing(contexts) self.assertIn('normal.openai.azure.com', endpoints) def test_endpoint_sanitizer_validates_ports_and_caps_paths(self): self.assertEqual( scanner_db.sanitize_endpoint( 'https://user:password@normal.openai.azure.com:70000/path?token=secret' ), '', ) self.assertEqual( scanner_db.sanitize_endpoint('normal.openai.azure.com:not-a-port/path?token=secret'), '', ) bounded = scanner_db.sanitize_endpoint( 'https://normal.openai.azure.com/' + ('a' * 5000) + '?token=PATH-QUERY-SENTINEL' ) self.assertLessEqual(len(bounded), scanner_db.ENDPOINT_METADATA_MAX_CHARS) self.assertTrue(bounded.startswith('https://normal.openai.azure.com/')) self.assertNotIn('PATH-QUERY-SENTINEL', bounded) self.assertEqual(scanner_db.sanitize_endpoint('not endpoint metadata'), '') def test_enrichment_and_database_re_sanitize_imported_postman_context(self): raw_endpoint = self.credentialed_endpoint() finding = { 'DetectorName': 'OpenAI', 'Raw': 'detected-postman-secret', 'PostmanContext': { 'provider': 'openai', 'credential_kind': 'api_key', 'credential_confidence': 'detector_match', 'endpoint': raw_endpoint, 'host': raw_endpoint, 'json_path': raw_endpoint, 'legacy_url': raw_endpoint, }, } enriched = scanner_db.enrich_finding(finding) self.assertEqual(enriched['endpoint'], self.SAFE_ENDPOINT) self.assertEqual(enriched['resource'], self.SAFE_ENDPOINT) self.assertTrue(all(sentinel not in json.dumps(enriched) for sentinel in self.sentinels())) with tempfile.TemporaryDirectory() as temp_dir, mock.patch.dict(os.environ, { 'SCANNER_DB_URL': '', 'DATABASE_URL': '', 'TRUF_MANAGED_POSTGRES_DSN': '', }): db = scanner_db.ScannerDB(db_path=os.path.join(temp_dir, 'scanner.db'), db_url='') try: run_id = db.start_run('test', ['test']) cycle_id = db.start_source_cycle( run_id, 'fixture', 'postman', 'search', 'q', 1, 1, None, {}, {}, ) db.record_target_result(run_id, cycle_id, 'fixture', 'q', 'fixture-target', { 'findings': [finding], 'errors': [], 'scan_type': 'postman', }) row = dict(db.conn.execute( '''SELECT endpoint, resource, enrichment_json, raw_finding_json FROM findings''' ).fetchone()) raw_result = db.conn.execute( 'SELECT raw_result_json FROM target_scans' ).fetchone()['raw_result_json'] finally: db.close() self.assertEqual(row['endpoint'], self.SAFE_ENDPOINT) self.assertEqual(row['resource'], self.SAFE_ENDPOINT) persisted = json.dumps(row) + raw_result self.assertTrue(all(sentinel not in persisted for sentinel in self.sentinels())) self.assertEqual(json.loads(row['enrichment_json'])['endpoint'], self.SAFE_ENDPOINT) if __name__ == '__main__': unittest.main()