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 console_runner import scanner class PackagePostmanHarvestLimitTests(unittest.TestCase): @classmethod def setUpClass(cls): scanner.initialize_scanner_runtime(preflight_complete=True, register_cleanup=False) def _cache_patches(self, runtime_dir, cache_dir): return ( mock.patch.object(scanner.scan_config, 'runtime_dir', runtime_dir), mock.patch.object(scanner.scan_config, 'postman_cache_dir', 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), ) def test_thousands_of_matches_stop_at_artifact_cap_and_scan_cache_once(self): with tempfile.TemporaryDirectory() as temp_dir: extract_dir = os.path.join(temp_dir, 'extract') cache_dir = os.path.join(temp_dir, 'cache') os.mkdir(extract_dir) scanner.ensure_private_directory(cache_dir, reject_reparse=True) for index in range(2000): Path(extract_dir, f'{index:04d}.postman_collection.json').write_text( json.dumps({'token': f'value-{index:04d}'}), encoding='utf-8', ) warnings = [] patches = self._cache_patches(temp_dir, cache_dir) with patches[0], patches[1], patches[2], patches[3], patches[4], \ mock.patch.object(scanner, '_prepare_postman_cache_entry', wraps=scanner._prepare_postman_cache_entry) as prepare, \ mock.patch.object(scanner, 'postman_cache_usage', wraps=scanner.postman_cache_usage) as usage: started = time.perf_counter() targets = scanner.find_postman_artifacts( extract_dir, 'npm', {'name': 'fixture', 'version': '1.0.0', 'tarball': 'https://example.invalid/package.tgz'}, cache_dir=cache_dir, max_artifact_size_mb=1, warnings=warnings, max_artifacts=7, max_total_bytes=1024 * 1024, max_elapsed_sec=10, ) elapsed = time.perf_counter() - started self.assertEqual(len(targets), 7) self.assertEqual(prepare.call_count, 7) self.assertEqual(usage.call_count, 1) self.assertEqual(scanner.postman_cache_usage(cache_dir)['items'], 7) self.assertLess(elapsed, 5) self.assertTrue(any('artifact limit of 7' in warning for warning in warnings)) self.assertLessEqual(len(warnings), scanner.POSTMAN_HARVEST_MAX_WARNINGS) self.assertTrue(all(len(warning) <= scanner.POSTMAN_HARVEST_WARNING_MAX_CHARS for warning in warnings)) def test_aggregate_byte_cap_stops_before_reading_or_publishing_excess(self): with tempfile.TemporaryDirectory() as temp_dir: extract_dir = os.path.join(temp_dir, 'extract') cache_dir = os.path.join(temp_dir, 'cache') os.mkdir(extract_dir) scanner.ensure_private_directory(cache_dir, reject_reparse=True) payloads = [] for index in range(6): payload = json.dumps({'value': f'{index:04d}-fixed-width'}).encode('ascii') payloads.append(payload) Path(extract_dir, f'{index}.postman_environment.json').write_bytes(payload) self.assertEqual(len({len(payload) for payload in payloads}), 1) warnings = [] byte_limit = len(payloads[0]) * 2 patches = self._cache_patches(temp_dir, cache_dir) with patches[0], patches[1], patches[2], patches[3], patches[4], \ mock.patch.object(scanner, '_prepare_postman_cache_entry', wraps=scanner._prepare_postman_cache_entry) as prepare, \ mock.patch.object(scanner, 'postman_cache_usage', wraps=scanner.postman_cache_usage) as usage: targets = scanner.find_postman_artifacts( extract_dir, 'pypi', {'name': 'fixture', 'version': '2.0.0', 'artifact': 'https://example.invalid/package.whl'}, cache_dir=cache_dir, max_artifact_size_mb=1, warnings=warnings, max_artifacts=100, max_total_bytes=byte_limit, max_elapsed_sec=10, ) self.assertEqual(len(targets), 2) self.assertEqual(prepare.call_count, 2) self.assertEqual(usage.call_count, 1) self.assertEqual(scanner.postman_cache_usage(cache_dir)['items'], 2) self.assertTrue(any('aggregate byte limit' in warning for warning in warnings)) def test_elapsed_deadline_stops_walking_without_cache_publication(self): with tempfile.TemporaryDirectory() as temp_dir: extract_dir = os.path.join(temp_dir, 'extract') cache_dir = os.path.join(temp_dir, 'cache') os.mkdir(extract_dir) scanner.ensure_private_directory(cache_dir, reject_reparse=True) for index in range(20): Path(extract_dir, f'{index}.postman_collection.json').write_text( json.dumps({'value': index}), encoding='utf-8', ) clock_value = 0.0 def advancing_clock(): nonlocal clock_value clock_value += 0.1 return clock_value warnings = [] patches = self._cache_patches(temp_dir, cache_dir) with patches[0], patches[1], patches[2], patches[3], patches[4], \ mock.patch.object(scanner.time, 'monotonic', side_effect=advancing_clock), \ mock.patch.object(scanner, 'postman_cache_usage', wraps=scanner.postman_cache_usage) as usage: targets = scanner.find_postman_artifacts( extract_dir, 'npm', {'name': 'fixture', 'version': '1.0.0'}, cache_dir=cache_dir, max_artifact_size_mb=1, warnings=warnings, max_artifacts=100, max_total_bytes=1024 * 1024, max_elapsed_sec=0.35, ) self.assertEqual(targets, []) self.assertEqual(usage.call_count, 0) self.assertEqual(scanner.postman_cache_usage(cache_dir)['items'], 0) self.assertTrue(any('elapsed deadline' in warning for warning in warnings)) def test_batch_cache_capacity_failure_is_fail_closed_before_publication(self): with tempfile.TemporaryDirectory() as temp_dir: extract_dir = os.path.join(temp_dir, 'extract') cache_dir = os.path.join(temp_dir, 'cache') os.mkdir(extract_dir) scanner.ensure_private_directory(cache_dir, reject_reparse=True) for index in range(2): Path(extract_dir, f'{index}.postman_collection.json').write_text( json.dumps({'value': index}), encoding='utf-8', ) patches = self._cache_patches(temp_dir, cache_dir) with patches[0], patches[1], patches[2], patches[3], patches[4], \ mock.patch.object(scanner.scan_config, 'postman_cache_max_items', 1), \ mock.patch.object(scanner, 'postman_cache_usage', wraps=scanner.postman_cache_usage) as usage, \ self.assertRaises(scanner.PostmanCacheCapacityError): scanner.find_postman_artifacts( extract_dir, 'npm', {'name': 'fixture', 'version': '1.0.0'}, cache_dir=cache_dir, max_artifact_size_mb=1, max_artifacts=10, max_total_bytes=1024 * 1024, max_elapsed_sec=10, ) self.assertEqual(usage.call_count, 1) self.assertEqual(scanner.postman_cache_usage(cache_dir)['items'], 0) def test_npm_and_pypi_harvest_warnings_preserve_trufflehog_findings(self): finding = {'DetectorName': 'OpenAI', 'Raw': 'fixture-secret', 'Verified': False} def limited_harvest(*args, **kwargs): kwargs['warnings'].append('Optional Postman package harvesting degraded: fixture limit') return [] cases = ( ( 'npm', scanner.scan_npm_package, 'parse_npm_target', {'name': 'fixture', 'version': '1.0.0', 'tarball': 'https://example.invalid/package.tgz'}, 'safe_extract_tar', ), ( 'pypi', scanner.scan_pypi_package, 'parse_pypi_target', {'name': 'fixture', 'version': '1.0.0', 'artifact': 'https://example.invalid/package.whl', 'size': 1}, 'safe_extract_archive', ), ) for source, scan, parser_name, package, extract_name in cases: with self.subTest(source=source), tempfile.TemporaryDirectory() as work_dir, \ mock.patch.object(scanner, parser_name, return_value=package), \ mock.patch.object(scanner, 'create_command_work_dir', return_value=work_dir), \ mock.patch.object(scanner, 'cleanup_command_work_dir'), \ mock.patch.object(scanner, 'download_file', return_value=1), \ mock.patch.object(scanner, extract_name), \ mock.patch.object(scanner, 'harden_private_tree'), \ mock.patch.object(scanner, 'find_postman_artifacts', side_effect=limited_harvest), \ mock.patch.object(scanner, 'run_command_streamed', return_value=scanner.streamed_output_from_text(json.dumps(finding) + '\n', '', 0)), \ mock.patch.object(scanner, 'attach_nearby_context'): result = scan('fixture-target', max_artifact_size_mb=1) self.assertEqual(result['findings'], [finding]) self.assertTrue(any('fixture limit' in warning for warning in result['warnings'])) self.assertIn('postman_package_harvest', result['warning_classes']) self.assertEqual(result['errors'], []) def test_global_config_applies_canonical_harvest_limits(self): configured = console_runner.load_config(str(APP_DIR / 'config.yaml'))['global'] self.assertEqual(configured['postman_package_harvest_max_artifacts'], 100) self.assertEqual(configured['postman_package_harvest_max_bytes'], 128 * 1024 * 1024) self.assertEqual(configured['postman_package_harvest_max_elapsed_sec'], 30) previous = scanner.scan_config.__dict__.copy() try: console_runner.apply_global_config({ 'postman_package_harvest_max_artifacts': 11, 'postman_package_harvest_max_bytes': 12345, 'postman_package_harvest_max_elapsed_sec': 2.5, }) self.assertEqual(scanner.scan_config.postman_package_harvest_max_artifacts, 11) self.assertEqual(scanner.scan_config.postman_package_harvest_max_bytes, 12345) self.assertEqual(scanner.scan_config.postman_package_harvest_max_elapsed_sec, 2.5) finally: scanner.scan_config.__dict__.clear() scanner.scan_config.__dict__.update(previous) if __name__ == '__main__': unittest.main()