sashatrask
624 lines
27 KiB
Python
624 lines
27 KiB
Python
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()
|