import base64 from contextlib import contextmanager import json import os from pathlib import Path import sys import tempfile import time import unittest from unittest import mock ROOT = Path(__file__).resolve().parents[1] APP_DIR = ROOT / 'app' sys.path.insert(0, str(APP_DIR)) import scanner import console_runner class Response: def __init__(self, payload=None, content=b'', status_code=200): self._payload = payload self.content = content self.status_code = status_code self.headers = {} self.text = '' def json(self): return self._payload def raise_for_status(self): return None def close(self): return None class ScannerDiagnosticBoundTests(unittest.TestCase): @staticmethod def trufflehog_message(message, level='info-0'): return json.dumps({'level': level, 'msg': message}) def test_gitlab_missing_repository_is_a_terminal_target_skip(self): response = mock.Mock(status_code=404, text='404 Repository Not Found', headers={}) response.iter_content.return_value = [b'{"message":"404 Repository Not Found"}'] response.raise_for_status.side_effect = scanner.requests.exceptions.HTTPError(response=response) with mock.patch.object(scanner, 'api_request', return_value=response), \ mock.patch.object(scanner, 'run_command_streamed') as command: result = scanner.scan_git_repo( 'https://gitlab.com/example/deleted.git', provider='gitlab', max_commit_age_days=90, skip_if_commit_lookup_fails=False, ) self.assertIn('no repository', result['skipped']) self.assertFalse(result.get('source_failure', False)) self.assertFalse(result['errors']) command.assert_not_called() def test_go_memory_limit_is_terminal_for_one_target(self): stderr = 'fatal error: out of memory\n' + ('runtime stack frame\n' * 100) result = scanner.apply_trufflehog_diagnostics({}, stderr, 2, 'docker') self.assertEqual(result['error_class'], 'memory_limit') self.assertFalse(result['retryable']) self.assertFalse(result.get('source_failure', False)) self.assertEqual(len(result['errors']), 1) def test_docker_manifest_unauthorized_is_target_scoped(self): stderr = json.dumps({ 'level': 'error', 'msg': 'error processing image', 'error': 'GET https://index.docker.io/v2/example/image/manifests/sha256:abc: UNAUTHORIZED: authentication required', }) result = scanner.apply_trufflehog_diagnostics({}, stderr, 1, 'docker') self.assertEqual(result['error_class'], 'docker_registry_access') self.assertFalse(result['retryable']) self.assertFalse(result.get('source_failure', False)) def test_non_docker_unauthorized_remains_source_auth_failure(self): result = scanner.apply_trufflehog_diagnostics( {}, 'authentication failed: bad credentials', 1, 'git', ) self.assertEqual(result['error_class'], 'source_auth') self.assertTrue(result['source_failure']) self.assertTrue(result['source_failure_auth_related']) def test_command_output_capacity_error_is_terminal_for_one_target(self): severity, error_class, retryable = scanner._trufflehog_diagnostic_policy( 'Error: TruffleHog stdout exceeded 33554432 bytes', 'git', -1, ) self.assertEqual((severity, error_class, retryable), ('error', 'output_limit', False)) status, available_after, attempts, max_attempts = console_runner.queue_error_disposition( None, 'github', 'github', 'target', {'errors': ['oversized'], 'error_class': 'output_limit', 'retryable': False}, mock.Mock(target_retry_max_attempts=3), {'attempts': 1}, ) self.assertEqual((status, available_after, attempts, max_attempts), ('failed', None, 1, 3)) def test_complete_docker_process_records_completion(self): result = scanner.apply_trufflehog_diagnostics( {'findings': [], 'errors': []}, self.trufflehog_message('finished scanning'), 0, 'docker', ) self.assertFalse(result['errors']) self.assertTrue(result['scan_meta']['trufflehog_finished']) def test_docker_zero_exit_without_completion_is_retryable(self): result = scanner.apply_trufflehog_diagnostics( {'findings': [], 'errors': []}, self.trufflehog_message('running source'), 0, 'docker', ) self.assertEqual(result['error_class'], 'command_incomplete') self.assertTrue(result['retryable']) self.assertFalse(result['scan_meta']['trufflehog_finished']) def test_docker_one_exit_without_completion_preserves_findings_and_retries(self): finding = {'DetectorName': 'Example'} result = scanner.apply_trufflehog_diagnostics( {'findings': [finding], 'errors': []}, self.trufflehog_message('running source'), 1, 'docker', ) self.assertEqual(result['findings'], [finding]) self.assertEqual(result['error_class'], 'command_incomplete') self.assertTrue(result['retryable']) args = mock.Mock( target_retry_max_attempts=3, target_retry_base_delay_sec=60, target_retry_max_delay_sec=3600, ) first = console_runner.queue_error_disposition( None, 'dockerhub', 'docker', 'target', result, args, {'attempts': 1}, ) exhausted = console_runner.queue_error_disposition( None, 'dockerhub', 'docker', 'target', result, args, {'attempts': 3}, ) self.assertEqual((first[0], first[2], first[3]), ('deferred', 1, 3)) self.assertIsNotNone(first[1]) self.assertEqual(exhausted, ('failed', None, 3, 3)) def test_docker_timeout_preserves_findings_and_stops_at_attempt_limit(self): finding = {'DetectorName': 'PartialFinding'} result = scanner.apply_trufflehog_diagnostics( {'findings': [finding], 'errors': []}, 'Command timed out after 600 seconds', -1, 'docker', ) args = mock.Mock( target_retry_max_attempts=3, target_timeout_retry_delay_sec=21600, ) first = console_runner.queue_error_disposition( None, 'dockerhub', 'docker', 'target', result, args, {'attempts': 1}, ) exhausted = console_runner.queue_error_disposition( None, 'dockerhub', 'docker', 'target', result, args, {'attempts': 3}, ) self.assertEqual(result['findings'], [finding]) self.assertEqual(result['error_class'], 'timeout') self.assertEqual((first[0], first[2], first[3]), ('deferred', 1, 3)) self.assertIsNotNone(first[1]) self.assertEqual(exhausted, ('failed', None, 3, 3)) self.assertFalse(console_runner.queue_result_resets_attempts(result)) self.assertTrue(console_runner.queue_result_resets_attempts({ 'source_failure': True, 'retryable': True, 'error_class': 'remote_transient', })) def test_unexplained_exit_after_nonfatal_warning_is_retryable(self): result = scanner.apply_trufflehog_diagnostics( {'findings': [], 'errors': []}, self.trufflehog_message('a detector ignored the context timeout', level='error'), 1, 'git', ) self.assertEqual(result['error_class'], 'command_exit') self.assertTrue(result['retryable']) self.assertEqual(result['warning_classes'], ['detector_timeout']) def test_docker_nonzero_exit_after_completion_is_retryable_wrapper_exit(self): result = scanner.apply_trufflehog_diagnostics( {'findings': [], 'errors': []}, self.trufflehog_message('finished scanning'), 1, 'docker', ) self.assertEqual(result['error_class'], 'wrapper_exit') self.assertTrue(result['retryable']) self.assertTrue(result['scan_meta']['trufflehog_finished']) def test_coverage_warnings_preserve_fatal_error_retry_policy(self): finding = {'DetectorName': 'PartialFinding'} for message, detail, source, warning_class in ( ('error processing layer', 'gzip: invalid header', 'docker', 'docker_layer_gzip'), ('non-critical error processing chunk', 'invalid archive', 'git', 'chunk_processing'), ): for error, error_class, retryable in ( ('connection reset', 'network', True), ('unknown flag', 'source_configuration', False), ): with self.subTest(warning_class=warning_class, error_class=error_class): warning = json.dumps({'level': 'error', 'msg': message, 'error': detail}) fatal = json.dumps({'level': 'error', 'msg': 'source failed', 'error': error}) stderr = '\n'.join((warning, fatal, self.trufflehog_message('finished scanning'))) result = scanner.apply_trufflehog_diagnostics( {'findings': [finding], 'errors': []}, stderr, 0, source, ) self.assertEqual(result['error_class'], error_class) self.assertEqual(result['retryable'], retryable) self.assertEqual(result['warning_classes'], [warning_class]) self.assertTrue(result['degraded']) self.assertEqual(result['findings'], [finding]) self.assertFalse(result['scan_meta']['trufflehog_warnings_retryable']) def test_docker_detector_context_timeout_is_nonfatal_degraded_coverage(self): stderr = '\n'.join(( self.trufflehog_message('a detector ignored the context timeout', level='error'), self.trufflehog_message('finished scanning'), )) result = scanner.apply_trufflehog_diagnostics( {'findings': [], 'errors': []}, stderr, 0, 'docker', ) self.assertFalse(result['errors']) self.assertTrue(result['degraded']) self.assertEqual(result['warning_classes'], ['detector_timeout']) self.assertTrue(result['scan_meta']['trufflehog_finished']) self.assertEqual(scanner.target_status(result), 'degraded') def test_other_docker_timeout_remains_retryable_error(self): stderr = '\n'.join(( json.dumps({'level': 'error', 'msg': 'error processing image', 'error': 'request timed out'}), self.trufflehog_message('finished scanning'), )) result = scanner.apply_trufflehog_diagnostics( {'findings': [], 'errors': []}, stderr, 0, 'docker', ) self.assertEqual(result['error_class'], 'timeout') self.assertTrue(result['retryable']) def test_non_docker_unexplained_exit_uses_bounded_retry_policy(self): result = scanner.apply_trufflehog_diagnostics( {'findings': [], 'errors': []}, self.trufflehog_message('running source'), 1, 'git', ) self.assertEqual(result['error_class'], 'command_exit') self.assertTrue(result['retryable']) self.assertEqual(result['scan_meta']['trufflehog_returncode'], 1) self.assertEqual(result['scan_meta']['diagnostic_lines_processed'], 1) args = mock.Mock( target_retry_max_attempts=3, target_retry_base_delay_sec=60, target_retry_max_delay_sec=3600, ) first = console_runner.queue_error_disposition( None, 'github', 'github', 'target', result, args, {'attempts': 1}, ) exhausted = console_runner.queue_error_disposition( None, 'github', 'github', 'target', result, args, {'attempts': 3}, ) self.assertEqual((first[0], first[2], first[3]), ('deferred', 1, 3)) self.assertIsNotNone(first[1]) self.assertEqual(exhausted, ('failed', None, 3, 3)) @classmethod def setUpClass(cls): scanner.initialize_scanner_runtime(preflight_complete=True, register_cleanup=False) def test_sixteen_mib_repeated_errors_have_bounded_policy_work_and_projection(self): class NoSplitLines(str): def splitlines(self, *args, **kwargs): raise AssertionError('splitlines materialized diagnostic output') repetitions = (16 * 1024 * 1024 // len('error\n')) + 1 stderr = NoSplitLines('error\n' * repetitions) with mock.patch.object( scanner, '_trufflehog_diagnostic_policy', wraps=scanner._trufflehog_diagnostic_policy, ) as policy: result = scanner.apply_trufflehog_diagnostics({}, stderr, 1, 'git') self.assertLessEqual(policy.call_count, scanner.scan_config.trufflehog_diagnostic_max_errors + 1) self.assertLessEqual(len(result['errors']), scanner.scan_config.trufflehog_diagnostic_max_errors) self.assertEqual(sum('output_limit' in line for line in result['errors']), 1) self.assertTrue(all(len(line) <= scanner.scan_config.trufflehog_diagnostic_max_line_chars for line in result['errors'])) self.assertEqual(result['error_class'], 'source_resource') self.assertEqual(result['source_failure_category'], 'source_resource') self.assertTrue(result['scan_meta']['diagnostic_output_limited']) def test_oversized_single_line_is_not_classified_or_retained(self): stderr = 'x' * (16 * 1024 * 1024) with mock.patch.object(scanner, '_trufflehog_diagnostic_policy') as policy: result = scanner.apply_trufflehog_diagnostics({}, stderr, 1, 'git') policy.assert_not_called() self.assertEqual(len(result['errors']), 1) self.assertIn('output_limit', result['errors'][0]) self.assertNotIn('x' * 100, result['errors'][0]) def test_canonical_limits_are_loaded_and_applied(self): config = console_runner.load_config(str(APP_DIR / 'config.yaml')) configured = config['global'] self.assertEqual(configured['trufflehog_diagnostic_max_lines'], 2000) self.assertEqual(configured['postman_discovery_max_artifacts_per_page'], 100) self.assertEqual(configured['postman_cache_lock_timeout_sec'], 300) self.assertEqual(configured['target_claim_batch_size'], 1) self.assertEqual(config['sources']['github']['workers'], 1) self.assertFalse(config['sources']['github']['enabled']) self.assertEqual(config['sources']['gitlab']['workers'], 1) self.assertEqual(config['sources']['huggingface']['workers'], 1) self.assertEqual(config['sources']['huggingface']['discovery_request_attempts'], 3) self.assertEqual(config['sources']['huggingface']['discovery_retry_delay'], 5) self.assertEqual(config['sources']['dockerhub']['workers'], 2) self.assertEqual(config['sources']['dockerhub']['pages'], 30) self.assertEqual(config['sources']['dockerhub']['per_page'], 100) self.assertEqual( config['sources']['dockerhub']['docker_repository_refresh_max_per_cycle'], 0, ) self.assertEqual(config['sources']['dockerhub']['tag_fetch_workers'], 4) self.assertEqual(config['sources']['dockerhub']['timeout'], 600) self.assertEqual(config['sources']['dockerhub']['trufflehog_concurrency'], 4) self.assertTrue(config['sources']['dockerhub']['refresh_registry']) self.assertEqual(config['sources']['dockerhub']['trufflehog_job_memory_limit_bytes'], 6 * 1024 * 1024 * 1024) self.assertFalse(config['sources']['package_git']['enabled']) self.assertEqual(config['sources']['package_git']['max_targets'], 30) self.assertEqual(config['sources']['package_git']['target_claim_order'], 'balanced') self.assertFalse(config['sources']['github_actions']['enabled']) self.assertEqual(config['sources']['github_actions']['workers'], 1) self.assertTrue(config['sources']['github_actions']['refresh_registry']) self.assertEqual(config['sources']['github_actions']['target_claim_order'], 'balanced') self.assertFalse(config['supervisor']['sources']['github_actions']['enabled']) self.assertNotIn('github_actions', config['supervisor']['enabled_sources']) self.assertNotIn('github', config['supervisor']['enabled_sources']) enabled_source_count = sum( source != 'pypi' and bool(source_config.get('enabled', True)) for source, source_config in console_runner.load_config(str(APP_DIR / 'config.yaml'))['sources'].items() ) self.assertGreaterEqual( configured['result_bundle_max_total_bytes'], 3 * configured['result_bundle_max_event_bytes'], ) previous = scanner.scan_config.__dict__.copy() try: console_runner.apply_global_config({ 'trufflehog_diagnostic_max_errors': 17, 'postman_discovery_max_elapsed_sec': 2.5, }) self.assertEqual(scanner.scan_config.trufflehog_diagnostic_max_errors, 17) self.assertEqual(scanner.scan_config.postman_discovery_max_elapsed_sec, 2.5) finally: scanner.scan_config.__dict__.clear() scanner.scan_config.__dict__.update(previous) class NonPackagePostmanDiscoveryBoundTests(unittest.TestCase): @classmethod def setUpClass(cls): scanner.initialize_scanner_runtime(preflight_complete=True, register_cleanup=False) def setUp(self): self.temp_dir = tempfile.TemporaryDirectory() self.addCleanup(self.temp_dir.cleanup) self.cache_dir = os.path.join(self.temp_dir.name, 'cache') scanner.ensure_private_directory(self.cache_dir, reject_reparse=True) self.patchers = ( mock.patch.object(scanner.scan_config, 'runtime_dir', self.temp_dir.name), mock.patch.object(scanner.scan_config, 'postman_cache_dir', self.cache_dir), mock.patch.object(scanner.scan_config, 'postman_cache_max_items', 10000), mock.patch.object(scanner.scan_config, 'postman_cache_max_bytes', 64 * 1024 * 1024), mock.patch.object(scanner.scan_config, 'postman_cache_min_free_bytes', 0), ) for patcher in self.patchers: patcher.start() self.addCleanup(patcher.stop) @staticmethod def _json_content(index): return json.dumps({'value': index}, separators=(',', ':')).encode('ascii') def test_code_search_batches_one_cache_scan_and_stops_at_shared_count(self): items = [ { 'repository': {'full_name': 'owner/repo'}, 'path': f'{index}.postman_collection.json', 'sha': f'sha-{index}', 'url': f'https://api.github.test/content/{index}', 'html_url': f'https://github.test/owner/repo/{index}', } for index in range(20) ] def request(_pool, _method, url, **kwargs): if url.endswith('/search/code'): self.assertIsNone(kwargs.get('use_proxy')) return Response({'items': items}) self.assertIs(kwargs['use_proxy'], False) index = int(url.rsplit('/', 1)[-1]) content = self._json_content(index) return Response({ 'size': len(content), 'encoding': 'base64', 'content': base64.b64encode(content).decode('ascii'), }) with mock.patch.object(scanner.GitHubTokenPool, 'request', autospec=True, side_effect=request) as requests, \ mock.patch.object(scanner, 'postman_cache_usage', wraps=scanner.postman_cache_usage) as usage: targets = scanner.fetch_github_postman_targets( 'fixture', pages=1, per_page=100, token='token', search_kinds='collection', cache_dir=self.cache_dir, max_file_age_days=0, max_artifact_size_mb=1, discovery_max_artifacts=5, discovery_max_artifacts_per_page=5, discovery_max_bytes=1024 * 1024, discovery_max_elapsed_sec=10, ) self.assertEqual(len(targets), 5) self.assertEqual(requests.call_count, 6) self.assertEqual(usage.call_count, 1) def test_gist_page_batches_one_cache_scan_and_bounds_raw_fetches(self): files = { f'.env-{index}': { 'size': 20, 'raw_url': f'https://gist.test/raw/{index}', 'type': 'text/plain', } for index in range(20) } gists = [{'id': 'gist-1', 'files': files, 'html_url': 'https://gist.test/gist-1'}] def request(_method, url, **kwargs): if url.endswith('/gists/public'): self.assertIsNone(kwargs.get('use_proxy')) return Response(gists) self.assertIs(kwargs['use_proxy'], False) return Response(content=self._json_content(int(url.rsplit('/', 1)[-1]))) with mock.patch.object(scanner, 'api_request', side_effect=request) as requests, \ mock.patch.object(scanner, 'postman_cache_usage', wraps=scanner.postman_cache_usage) as usage: targets = scanner.fetch_github_gist_targets( pages=1, cache_dir=self.cache_dir, max_file_size_mb=1, discovery_max_artifacts=10, discovery_max_artifacts_per_page=4, discovery_max_bytes=1024 * 1024, discovery_max_elapsed_sec=10, ) self.assertEqual(len(targets), 4) self.assertEqual(requests.call_count, 5) self.assertEqual(usage.call_count, 1) def test_gharchive_hour_batches_one_cache_scan_and_bounds_raw_fetches(self): event = { 'type': 'PushEvent', 'repo': {'name': 'owner/repo'}, 'payload': { 'commits': [{ 'sha': 'abc123', 'added': [f'.env-{index}' for index in range(20)], }], }, } @contextmanager def archive_reader(*args, **kwargs): yield 'fixture-archive' def request(_method, url, **kwargs): index = int(url.rsplit('-', 1)[-1]) self.assertIs(kwargs['use_proxy'], False) return Response(content=self._json_content(index)) with mock.patch.object(scanner, 'cached_gharchive_hour_reader', side_effect=archive_reader), \ mock.patch.object(scanner, 'iter_gharchive_lines', return_value=[json.dumps(event).encode('utf-8')]), \ mock.patch.object(scanner, 'api_request', side_effect=request) as requests, \ mock.patch.object(scanner, 'postman_cache_usage', wraps=scanner.postman_cache_usage) as usage: targets = scanner.fetch_github_archive_file_targets( hours_back=1, max_files=20, cache_dir=self.cache_dir, max_file_size_mb=1, discovery_max_artifacts=10, discovery_max_artifacts_per_page=3, discovery_max_bytes=1024 * 1024, discovery_max_elapsed_sec=10, ) self.assertEqual(len(targets), 3) self.assertEqual(requests.call_count, 3) self.assertEqual(usage.call_count, 1) def test_cache_scan_receives_deadline_and_deadline_is_fail_soft(self): gists = [{ 'id': 'gist-1', 'files': {'.env': {'size': 10, 'raw_url': 'https://gist.test/raw/1', 'type': 'text/plain'}}, }] def request(_method, url, **kwargs): return Response(gists) if url.endswith('/gists/public') else Response(content=b'{"ok":1}') deadlines = [] def expired_usage(root, deadline=None): deadlines.append(deadline) raise scanner._PostmanHarvestDeadlineReached() with mock.patch.object(scanner, 'api_request', side_effect=request), \ mock.patch.object(scanner, 'postman_cache_usage', side_effect=expired_usage) as usage, \ self.assertLogs(scanner.logger, level='WARNING') as captured: targets = scanner.fetch_github_gist_targets( pages=1, cache_dir=self.cache_dir, max_file_size_mb=1, discovery_max_elapsed_sec=10, ) self.assertEqual(targets, []) self.assertEqual(usage.call_count, 1) self.assertTrue(deadlines[0] is not None) self.assertTrue(any('deadline' in line.lower() for line in captured.output)) def test_hundred_thousand_entry_cache_scan_stops_after_bounded_operations(self): class Entry: name = 'artifact.json' path = os.path.join(self.cache_dir, 'artifact.json') @staticmethod def is_symlink(): return False @staticmethod def is_dir(follow_symlinks=False): return False @staticmethod def is_file(follow_symlinks=False): return True @staticmethod def stat(follow_symlinks=False): return type('Stat', (), {'st_size': 1})() class Entries: def __init__(self): self.calls = 0 def __enter__(self): return self def __exit__(self, *args): return False def __iter__(self): return self def __next__(self): if self.calls >= 100000: raise StopIteration self.calls += 1 return Entry() entries = Entries() clock = 0.0 def advancing_clock(): nonlocal clock clock += 0.001 return clock started = time.perf_counter() with mock.patch.object(scanner.os, 'scandir', return_value=entries), \ mock.patch.object(scanner, 'is_reparse_point', return_value=False), \ mock.patch.object(scanner.time, 'monotonic', side_effect=advancing_clock), \ self.assertRaises(scanner._PostmanHarvestDeadlineReached): scanner.postman_cache_usage(self.cache_dir, deadline=0.01) self.assertLess(entries.calls, 20) self.assertLess(time.perf_counter() - started, 0.5) def test_cache_capacity_corruption_remains_fail_closed(self): gists = [{ 'id': 'gist-1', 'files': {'.env': {'size': 10, 'raw_url': 'https://gist.test/raw/1', 'type': 'text/plain'}}, }] def request(_method, url, **kwargs): return Response(gists) if url.endswith('/gists/public') else Response(content=b'{"ok":1}') with mock.patch.object(scanner, 'api_request', side_effect=request), \ mock.patch.object( scanner, 'postman_cache_usage', side_effect=scanner.PostmanCacheCapacityError('corrupt linked cache entry'), ), self.assertRaisesRegex(scanner.PostmanCacheCapacityError, 'corrupt'): scanner.fetch_github_gist_targets( pages=1, cache_dir=self.cache_dir, max_file_size_mb=1, discovery_max_elapsed_sec=10, ) if __name__ == '__main__': unittest.main()