sashatrask
246 lines
12 KiB
Python
246 lines
12 KiB
Python
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()
|