sashatrask
667 lines
32 KiB
Python
667 lines
32 KiB
Python
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
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
|
|
from scanner_db import (
|
|
RuntimeSafetySchemaError,
|
|
ScanEventConflictError,
|
|
ScannerDB,
|
|
canonical_git_scan_plan_bytes,
|
|
matching_git_coverage,
|
|
validate_git_resolution,
|
|
)
|
|
|
|
|
|
def api_response(payload=None, status=200, headers=None, raw=None):
|
|
body = raw if raw is not None else json.dumps(payload).encode('utf-8')
|
|
response = mock.Mock()
|
|
response.status_code = status
|
|
response.headers = dict(headers or {})
|
|
response.headers.setdefault('Content-Length', str(len(body)))
|
|
response.encoding = 'utf-8'
|
|
response.text = body.decode('utf-8', errors='replace')
|
|
response.json.return_value = payload
|
|
response.iter_content.return_value = [body]
|
|
return response
|
|
|
|
|
|
def git_plan(mode='baseline', provider='github'):
|
|
head = 'a' * 40
|
|
base = None if mode == 'baseline' else head if mode == 'noop' else 'b' * 40
|
|
host = f'{provider}.com'
|
|
return {
|
|
'version': 1,
|
|
'provider': provider,
|
|
'repo_url': f'https://{host}/Owner/Repo.git',
|
|
'repo_path': 'Owner/Repo',
|
|
'branch': 'Feature/Main',
|
|
'ref': 'refs/heads/Feature/Main',
|
|
'head_sha': head,
|
|
'ref_source': 'explicit',
|
|
'base_sha': base,
|
|
'mode': mode,
|
|
'baseline_depth': 100,
|
|
}
|
|
|
|
|
|
class GitRefResolutionTests(unittest.TestCase):
|
|
def test_github_default_and_explicit_refs_are_resolved_exactly(self):
|
|
token = 'sentinel-github-token'
|
|
head = 'a' * 40
|
|
default_responses = [
|
|
api_response({'default_branch': 'Main'}),
|
|
api_response({'ref': 'refs/heads/Main', 'object': {'type': 'commit', 'sha': head}}),
|
|
]
|
|
with mock.patch.object(scanner, 'api_request', side_effect=default_responses) as request:
|
|
resolved = scanner.resolve_git_scan_target(
|
|
'https://github.com/Owner/Repo.git', 'github', token,
|
|
)
|
|
self.assertEqual(resolved['ref'], 'refs/heads/Main')
|
|
self.assertEqual(resolved['head_sha'], head)
|
|
self.assertEqual(resolved['ref_source'], 'provider_default')
|
|
self.assertEqual(request.call_count, 2)
|
|
self.assertEqual(
|
|
request.call_args_list[1].args[1],
|
|
'https://api.github.com/repos/Owner/Repo/git/ref/heads%2FMain',
|
|
)
|
|
for call in request.call_args_list:
|
|
self.assertEqual(call.kwargs['headers']['Authorization'], f'Bearer {token}')
|
|
self.assertIs(call.kwargs['allow_redirects'], False)
|
|
self.assertIs(call.kwargs['stream'], True)
|
|
self.assertNotIn(token, json.dumps(resolved, sort_keys=True))
|
|
|
|
explicit_target = json.dumps({
|
|
'url': 'https://github.com/Owner/Repo.git',
|
|
'branch': 'Feature/X',
|
|
'ref': 'refs/heads/Feature/X',
|
|
'head_sha': 'f' * 40,
|
|
})
|
|
response = api_response({
|
|
'ref': 'refs/heads/Feature/X',
|
|
'object': {'type': 'commit', 'sha': 'c' * 40},
|
|
})
|
|
with mock.patch.object(scanner, 'api_request', return_value=response) as request:
|
|
resolved = scanner.resolve_git_scan_target(explicit_target, 'github', token)
|
|
self.assertEqual(request.call_count, 1)
|
|
self.assertTrue(request.call_args.args[1].endswith('/git/ref/heads%2FFeature%2FX'))
|
|
self.assertEqual(resolved['head_sha'], 'c' * 40)
|
|
self.assertEqual(resolved['ref_source'], 'explicit')
|
|
|
|
def test_gitlab_default_and_slash_branch_are_encoded_and_validated(self):
|
|
head = 'd' * 40
|
|
responses = [
|
|
api_response({'default_branch': 'release/Next'}),
|
|
api_response({'name': 'release/Next', 'commit': {'id': head}}),
|
|
]
|
|
with mock.patch.object(scanner, 'api_request', side_effect=responses) as request:
|
|
resolved = scanner.resolve_git_scan_target(
|
|
'https://gitlab.com/Group/Sub/Repo.git', 'gitlab', 'gitlab-token',
|
|
)
|
|
self.assertEqual(resolved['repo_path'], 'Group/Sub/Repo')
|
|
self.assertEqual(resolved['ref'], 'refs/heads/release/Next')
|
|
self.assertEqual(request.call_count, 2)
|
|
self.assertEqual(
|
|
request.call_args_list[0].args[1],
|
|
'https://gitlab.com/api/v4/projects/Group%2FSub%2FRepo',
|
|
)
|
|
self.assertTrue(request.call_args_list[1].args[1].endswith('/release%2FNext'))
|
|
self.assertEqual(
|
|
request.call_args_list[0].kwargs['headers']['PRIVATE-TOKEN'], 'gitlab-token',
|
|
)
|
|
|
|
def test_unsafe_targets_and_refs_fail_before_any_authenticated_request(self):
|
|
invalid = [
|
|
('https://user:secret@github.com/Owner/Repo.git', 'github'),
|
|
('https://github.com/Owner/Repo.git?token=secret', 'github'),
|
|
('https://github.com/Owner/Repo.git#fragment', 'github'),
|
|
('https://gitlab.com/Owner/Repo.git', 'github'),
|
|
('https://github.com/Owner%2FRepo.git', 'github'),
|
|
(json.dumps({
|
|
'url': 'https://github.com/Owner/Repo.git',
|
|
'branch': 'main', 'ref': 'refs/heads/other',
|
|
}), 'github'),
|
|
]
|
|
with mock.patch.object(scanner, 'api_request') as request:
|
|
for target, provider in invalid:
|
|
with self.subTest(target=target), self.assertRaises(ValueError):
|
|
scanner.resolve_git_scan_target(target, provider, 'must-not-be-sent')
|
|
request.assert_not_called()
|
|
|
|
def test_malformed_oversized_redirected_and_mismatched_payloads_fail_closed(self):
|
|
cases = [
|
|
api_response(raw=b'{not-json'),
|
|
api_response({'default_branch': 'main'}, headers={'Content-Length': '2000'}),
|
|
api_response({'default_branch': 'main'}, status=302, headers={'Location': 'https://evil.test'}),
|
|
]
|
|
for response in cases:
|
|
with self.subTest(status=response.status_code), \
|
|
mock.patch.object(scanner, 'api_request', return_value=response), \
|
|
self.assertRaises(scanner.ApiRequestError):
|
|
scanner.resolve_github_ref_head(
|
|
'Owner/Repo', max_response_bytes=100,
|
|
)
|
|
response.close.assert_called()
|
|
|
|
mismatched = api_response({
|
|
'ref': 'refs/heads/other',
|
|
'object': {'type': 'tag', 'sha': 'e' * 40},
|
|
})
|
|
with mock.patch.object(scanner, 'api_request', return_value=mismatched), \
|
|
self.assertRaisesRegex(scanner.ApiRequestError, 'mismatched'):
|
|
scanner.resolve_github_ref_head('Owner/Repo', ref_hint='main')
|
|
|
|
def test_provider_error_keeps_rate_limit_category_and_redacts_token(self):
|
|
token = 'sentinel-rate-limit-token'
|
|
response = api_response(
|
|
{'message': f'API rate limit exceeded for {token}'}, status=429,
|
|
headers={'X-RateLimit-Reset': '1800000000'},
|
|
)
|
|
with mock.patch.object(scanner, 'api_request', return_value=response), \
|
|
self.assertRaises(scanner.RateLimitError) as raised:
|
|
scanner.resolve_github_ref_head('Owner/Repo', token, ref_hint='main')
|
|
self.assertEqual(raised.exception.category, 'rate_limit')
|
|
self.assertNotIn(token, str(raised.exception))
|
|
|
|
|
|
class ExactGitExecutionTests(unittest.TestCase):
|
|
@staticmethod
|
|
def run_scan(plan, **kwargs):
|
|
return scanner.scan_git_repo(
|
|
plan['repo_url'], provider=plan['provider'], git_plan=plan,
|
|
token='sentinel-scan-token', max_depth=7, max_commit_age_days=1,
|
|
**kwargs,
|
|
)
|
|
|
|
def test_noop_records_exact_scope_without_launching_command(self):
|
|
plan = git_plan('noop')
|
|
with mock.patch.object(scanner, 'run_command_streamed') as run:
|
|
result = self.run_scan(plan)
|
|
run.assert_not_called()
|
|
self.assertEqual(result['errors'], [])
|
|
self.assertEqual(result['git_scan_execution']['mode'], 'noop')
|
|
self.assertTrue(result['git_scan_execution']['success'])
|
|
self.assertEqual(result['scan_meta']['exact_git_scope']['head_sha'], plan['head_sha'])
|
|
|
|
def test_baseline_is_sha_pinned_and_depth_bounded(self):
|
|
plan = git_plan('baseline')
|
|
output = scanner.streamed_output_from_text('', '{"msg":"finished scanning"}', 0)
|
|
with mock.patch.object(scanner, 'get_trufflehog_cmd', return_value='trufflehog'), \
|
|
mock.patch.object(scanner, 'run_command_streamed', return_value=output) as run:
|
|
result = self.run_scan(plan)
|
|
command, _, environment = run.call_args.args
|
|
self.assertEqual(command[command.index('--branch') + 1], plan['head_sha'])
|
|
self.assertEqual(command[command.index('--max-depth') + 1], '100')
|
|
self.assertNotIn('--since-commit', command)
|
|
self.assertNotIn('sentinel-scan-token', command)
|
|
self.assertEqual(environment['TRUF_GIT_TOKEN'], 'sentinel-scan-token')
|
|
self.assertTrue(result['git_scan_execution']['success'])
|
|
|
|
def test_delta_uses_only_pinned_head_and_covered_base(self):
|
|
plan = git_plan('delta')
|
|
output = scanner.streamed_output_from_text('', '{"msg":"finished scanning"}', 0)
|
|
with mock.patch.object(scanner, 'get_trufflehog_cmd', return_value='trufflehog'), \
|
|
mock.patch.object(scanner, 'run_command_streamed', return_value=output) as run:
|
|
result = self.run_scan(plan)
|
|
command = run.call_args.args[0]
|
|
self.assertEqual(command[command.index('--branch') + 1], plan['head_sha'])
|
|
self.assertEqual(command[command.index('--since-commit') + 1], plan['base_sha'])
|
|
self.assertNotIn('--max-depth', command)
|
|
self.assertEqual(result['git_scan_execution']['mode'], 'delta')
|
|
|
|
def test_unavailable_delta_base_falls_back_to_pinned_bounded_baseline(self):
|
|
plan = git_plan('delta')
|
|
outputs = [
|
|
scanner.streamed_output_from_text('', 'fatal: bad object', 1),
|
|
scanner.streamed_output_from_text('', '{"msg":"finished scanning"}', 0),
|
|
]
|
|
with mock.patch.object(scanner, 'get_trufflehog_cmd', return_value='trufflehog'), \
|
|
mock.patch.object(scanner, 'git_delta_base_unavailable', return_value=True), \
|
|
mock.patch.object(scanner, 'run_command_streamed', side_effect=outputs) as run:
|
|
result = self.run_scan(plan)
|
|
self.assertEqual(run.call_count, 2)
|
|
delta_command = run.call_args_list[0].args[0]
|
|
reset_command = run.call_args_list[1].args[0]
|
|
self.assertIn('--since-commit', delta_command)
|
|
self.assertNotIn('--max-depth', delta_command)
|
|
self.assertNotIn('--since-commit', reset_command)
|
|
self.assertEqual(reset_command[reset_command.index('--max-depth') + 1], '100')
|
|
self.assertEqual(result['git_scan_execution']['mode'], 'baseline_reset')
|
|
self.assertTrue(result['git_scan_execution']['continuity_reset'])
|
|
self.assertTrue(result['git_scan_execution']['success'])
|
|
|
|
|
|
class GitCoverageGateTests(unittest.TestCase):
|
|
@staticmethod
|
|
def reservation(plan):
|
|
encoded = canonical_git_scan_plan_bytes(plan)
|
|
return {
|
|
'git_scan_plan_json': encoded.decode('ascii'),
|
|
'git_scan_plan_sha256': hashlib.sha256(encoded).hexdigest(),
|
|
}
|
|
|
|
@staticmethod
|
|
def metadata(plan, execution_mode=None, **overrides):
|
|
encoded = canonical_git_scan_plan_bytes(plan)
|
|
execution = {
|
|
'mode': execution_mode or plan['mode'],
|
|
'pinned': True,
|
|
'success': True,
|
|
'continuity_reset': False,
|
|
'plan_sha256': hashlib.sha256(encoded).hexdigest(),
|
|
}
|
|
execution.update(overrides)
|
|
return {'git_scan_plan': plan, 'git_scan_execution': execution}
|
|
|
|
def test_matching_successful_modes_advance_only_to_bound_head(self):
|
|
for mode, execution_mode, reset in (
|
|
('baseline', 'baseline', False),
|
|
('delta', 'delta', False),
|
|
('delta', 'baseline_reset', True),
|
|
('noop', 'noop', False),
|
|
):
|
|
with self.subTest(mode=mode, execution_mode=execution_mode):
|
|
plan = git_plan(mode)
|
|
metadata = self.metadata(
|
|
plan, execution_mode, continuity_reset=reset,
|
|
)
|
|
self.assertEqual(
|
|
matching_git_coverage(self.reservation(plan), metadata, 'done', 0),
|
|
(True, plan['ref'], plan['head_sha']),
|
|
)
|
|
|
|
def test_failed_deferred_or_unpinned_execution_never_advances(self):
|
|
plan = git_plan('delta')
|
|
reservation = self.reservation(plan)
|
|
metadata = self.metadata(plan)
|
|
self.assertEqual(
|
|
matching_git_coverage(reservation, metadata, 'deferred', 1),
|
|
(False, None, None),
|
|
)
|
|
metadata['git_scan_execution']['pinned'] = False
|
|
self.assertEqual(
|
|
matching_git_coverage(reservation, metadata, 'done', 0),
|
|
(False, None, None),
|
|
)
|
|
|
|
def test_plan_mismatch_and_corrupt_storage_fail_closed(self):
|
|
plan = git_plan('baseline')
|
|
reservation = self.reservation(plan)
|
|
changed = dict(plan, head_sha='f' * 40)
|
|
with self.assertRaises(ScanEventConflictError):
|
|
matching_git_coverage(
|
|
reservation, self.metadata(changed), 'done', 0,
|
|
)
|
|
with self.assertRaises(RuntimeSafetySchemaError):
|
|
matching_git_coverage({
|
|
'git_scan_plan_json': '\N{SNOWMAN}',
|
|
'git_scan_plan_sha256': '0' * 64,
|
|
}, {}, 'done', 0)
|
|
|
|
def test_resolution_repository_url_must_match_canonical_path(self):
|
|
resolved = {key: value for key, value in git_plan().items() if key in {
|
|
'provider', 'repo_url', 'repo_path', 'branch', 'ref', 'head_sha', 'ref_source',
|
|
}}
|
|
validate_git_resolution(resolved)
|
|
with self.assertRaisesRegex(ValueError, 'canonical repository'):
|
|
validate_git_resolution(dict(resolved, repo_url='https://github.com/Other/Repo.git'))
|
|
|
|
|
|
class ExactGitWiringTests(unittest.TestCase):
|
|
def test_core_source_config_enables_bounded_exact_planning(self):
|
|
config = console_runner.load_config(str(APP_DIR / 'config.yaml'))
|
|
for source in ('github', 'gitlab'):
|
|
args = console_runner.build_args_from_source_config(
|
|
source, config['sources'][source], config['global'], 'fixture',
|
|
)
|
|
self.assertTrue(args.exact_git_planning_enabled)
|
|
self.assertEqual(args.git_baseline_depth, 100)
|
|
self.assertEqual(args.git_ref_resolution_attempts, 2)
|
|
self.assertEqual(args.git_ref_resolution_timeout_sec, 10)
|
|
self.assertEqual(args.git_ref_resolution_max_bytes, 1 << 20)
|
|
self.assertEqual(args.admission_resolution_attempts, 300)
|
|
self.assertEqual(args.admission_resolution_seconds, 300)
|
|
self.assertEqual(args.admission_resolution_retry_delay_sec, 1)
|
|
|
|
def test_post_claim_helper_resolves_then_binds_with_dedicated_connection(self):
|
|
args = SimpleNamespace(
|
|
platform='github', git_ref_resolution_attempts=2,
|
|
git_ref_resolution_timeout_sec=9, git_ref_resolution_max_bytes=4096,
|
|
git_baseline_depth=77,
|
|
)
|
|
claim = {
|
|
'target': 'https://github.com/Owner/Repo.git',
|
|
'reservation_id': 12, 'claim_lease_token': 'lease-token',
|
|
}
|
|
resolved = {key: value for key, value in git_plan().items() if key in {
|
|
'provider', 'repo_url', 'repo_path', 'branch', 'ref', 'head_sha', 'ref_source',
|
|
}}
|
|
bound = git_plan()
|
|
planning_db = mock.Mock(enabled=True)
|
|
planning_db.bind_git_scan_plan.return_value = bound
|
|
with mock.patch.object(console_runner, 'resolve_git_scan_target', return_value=resolved) as resolve, \
|
|
mock.patch.object(console_runner, 'ScannerDB', return_value=planning_db) as db_class:
|
|
result = console_runner.resolve_and_bind_git_claim(
|
|
args, 'postgresql://fixture', 'github', claim,
|
|
{'token': 'sentinel-token'},
|
|
)
|
|
self.assertIs(result, bound)
|
|
resolve.assert_called_once_with(
|
|
claim['target'], 'github', 'sentinel-token', request_attempts=2,
|
|
timeout_sec=9.0, max_response_bytes=4096,
|
|
)
|
|
db_class.assert_called_once_with(db_url='postgresql://fixture', initialize=False)
|
|
planning_db.bind_git_scan_plan.assert_called_once_with(
|
|
12, 'lease-token', resolved, 77,
|
|
)
|
|
planning_db.close.assert_called_once_with()
|
|
|
|
def test_resolver_failures_have_durable_queue_policy_without_token_leakage(self):
|
|
args = SimpleNamespace(platform='github')
|
|
claim = {
|
|
'target': 'https://github.com/Owner/Repo.git',
|
|
'scan_event_id': 'event-id',
|
|
}
|
|
token = 'sentinel-resolution-token'
|
|
cases = [
|
|
(
|
|
scanner.RateLimitError(
|
|
'github', f'HTTP 404 for {token}', category='not_found',
|
|
retryable=False, auth_related=False,
|
|
),
|
|
False, False, 'not_found', 'not_found',
|
|
),
|
|
(
|
|
scanner.RateLimitError(
|
|
'github', f'rate limited {token}', category='rate_limit',
|
|
retryable=True, auth_related=True,
|
|
),
|
|
True, True, 'source_auth', 'rate_limit',
|
|
),
|
|
(
|
|
scanner.ApiRequestError(f'transport failed with {token}'),
|
|
True, True, 'remote_transient', 'remote_transient',
|
|
),
|
|
]
|
|
for error, source_failure, retryable, error_class, category in cases:
|
|
with self.subTest(category=category):
|
|
failure = console_runner._GitResolutionFailure(error)
|
|
result = console_runner.git_resolution_failure_result(
|
|
args, claim, {'token': token}, failure,
|
|
)
|
|
self.assertEqual(result['source_failure'], source_failure)
|
|
self.assertEqual(result['retryable'], retryable)
|
|
self.assertEqual(result['error_class'], error_class)
|
|
self.assertEqual(
|
|
result['scan_meta']['exact_git_resolution']['category'], category,
|
|
)
|
|
self.assertNotIn(token, json.dumps(result, sort_keys=True))
|
|
|
|
invalid = console_runner.git_resolution_failure_result(
|
|
args, claim, {'token': token},
|
|
console_runner._GitResolutionFailure(
|
|
ValueError('unsafe target'), invalid_target=True,
|
|
),
|
|
)
|
|
self.assertFalse(invalid['source_failure'])
|
|
self.assertFalse(invalid['retryable'])
|
|
self.assertEqual(invalid['error_class'], 'invalid_target')
|
|
|
|
def test_only_resolver_failures_are_wrapped_before_plan_binding(self):
|
|
args = SimpleNamespace(
|
|
platform='github', git_ref_resolution_attempts=2,
|
|
git_ref_resolution_timeout_sec=9, git_ref_resolution_max_bytes=4096,
|
|
)
|
|
claim = {
|
|
'target': 'https://github.com/Owner/Repo.git',
|
|
'reservation_id': 12, 'claim_lease_token': 'lease-token',
|
|
}
|
|
rate_limit = scanner.RateLimitError(
|
|
'github', 'limited', category='rate_limit',
|
|
retryable=True, auth_related=True,
|
|
)
|
|
with mock.patch.object(console_runner, 'resolve_git_scan_target', side_effect=rate_limit), \
|
|
mock.patch.object(console_runner, 'ScannerDB') as db_class, \
|
|
self.assertRaises(console_runner._GitResolutionFailure) as raised:
|
|
console_runner.resolve_and_bind_git_claim(
|
|
args, 'postgresql://fixture', 'github', claim, {'token': 'token'},
|
|
)
|
|
self.assertIs(raised.exception.error, rate_limit)
|
|
db_class.assert_not_called()
|
|
|
|
invalid_claim = dict(claim, target='https://user:secret@github.com/Owner/Repo.git')
|
|
with mock.patch.object(console_runner, 'resolve_git_scan_target') as resolve, \
|
|
self.assertRaises(console_runner._GitResolutionFailure) as raised:
|
|
console_runner.resolve_and_bind_git_claim(
|
|
args, 'postgresql://fixture', 'github', invalid_claim, {'token': 'token'},
|
|
)
|
|
self.assertTrue(raised.exception.invalid_target)
|
|
resolve.assert_not_called()
|
|
|
|
def test_bind_fence_failure_is_not_downgraded_to_a_target_result(self):
|
|
args = SimpleNamespace(
|
|
platform='github', git_ref_resolution_attempts=2,
|
|
git_ref_resolution_timeout_sec=9, git_ref_resolution_max_bytes=4096,
|
|
git_baseline_depth=77,
|
|
)
|
|
claim = {
|
|
'target': 'https://github.com/Owner/Repo.git',
|
|
'reservation_id': 12, 'claim_lease_token': 'lease-token',
|
|
}
|
|
resolved = {key: value for key, value in git_plan().items() if key in {
|
|
'provider', 'repo_url', 'repo_path', 'branch', 'ref', 'head_sha', 'ref_source',
|
|
}}
|
|
planning_db = mock.Mock(enabled=True)
|
|
planning_db.bind_git_scan_plan.side_effect = ScanEventConflictError('stale fence')
|
|
with mock.patch.object(console_runner, 'resolve_git_scan_target', return_value=resolved), \
|
|
mock.patch.object(console_runner, 'ScannerDB', return_value=planning_db), \
|
|
self.assertRaisesRegex(ScanEventConflictError, 'stale fence'):
|
|
console_runner.resolve_and_bind_git_claim(
|
|
args, 'postgresql://fixture', 'github', claim, {'token': 'token'},
|
|
)
|
|
planning_db.close.assert_called_once_with()
|
|
|
|
def test_expected_resolver_failures_are_staged_instead_of_infrastructure_holds(self):
|
|
class Connection:
|
|
is_postgres = True
|
|
|
|
class DB:
|
|
url = 'postgresql://fixture'
|
|
last_error = ''
|
|
conn = Connection()
|
|
|
|
@staticmethod
|
|
def require_runtime_safety_schema():
|
|
return True
|
|
|
|
@staticmethod
|
|
def require_final_cutover():
|
|
return True
|
|
|
|
@staticmethod
|
|
def has_claimable_targets_v2(*_args, **_kwargs):
|
|
return True
|
|
|
|
@staticmethod
|
|
def finish_source_cycle(*_args, **_kwargs):
|
|
return True
|
|
|
|
class Lease:
|
|
def release(self):
|
|
return None
|
|
|
|
args = SimpleNamespace(
|
|
platform='github', workers=1, max_targets=1, timeout=10,
|
|
target_retry_max_attempts=3, target_retry_base_delay_sec=60,
|
|
target_retry_max_delay_sec=3600, refresh_registry=False,
|
|
target_claim_order='oldest', result_bundle_min_free_bytes=0,
|
|
result_bundle_max_event_bytes=1024 * 1024,
|
|
projection_backlog_max_items=10,
|
|
projection_backlog_max_bytes=8 * 1024 * 1024,
|
|
projection_backlog_headroom_bytes=2 * 1024 * 1024,
|
|
result_spool_wait_sec=0.01, exact_git_planning_enabled=True,
|
|
)
|
|
claim = {
|
|
'target': 'https://github.com/Owner/Repo.git',
|
|
'reservation_id': 12, 'reservation_token': 'reservation-token',
|
|
'bundle_id': 'bundle-id', 'scan_event_id': 'event-id',
|
|
'claim_lease_token': 'lease-token', 'ready_relative_path': 'ready/bundle',
|
|
'attempts': 1,
|
|
}
|
|
identity = SimpleNamespace(as_dict=lambda: {
|
|
'pid': 1, 'creation_time': 'fixture', 'executable': 'python',
|
|
})
|
|
|
|
cases = [
|
|
(
|
|
scanner.RateLimitError(
|
|
'github', 'missing', category='not_found',
|
|
retryable=False, auth_related=False,
|
|
),
|
|
'failed', 0,
|
|
),
|
|
(
|
|
scanner.RateLimitError(
|
|
'github', 'limited', category='rate_limit',
|
|
retryable=True, auth_related=True,
|
|
),
|
|
'deferred', 1,
|
|
),
|
|
]
|
|
for error, expected_queue_status, expected_source_failures in cases:
|
|
captured = {}
|
|
|
|
def stage(result, _claim, _root, _options, disposition, **_kwargs):
|
|
captured['result'] = result
|
|
captured['disposition'] = disposition
|
|
return scanner.StagedResult(
|
|
target=result['target'], scan_event_id='event-id', bundle_id='bundle-id',
|
|
reservation_id=12, scan_event_hash='hash', actual_bytes=1,
|
|
relative_path='ready/bundle', frame_count=1, finding_count=0,
|
|
error_count=1, candidate_count=0,
|
|
queue_status=disposition['queue_status'],
|
|
source_failure=bool(result.get('source_failure')),
|
|
source_failure_category=str(result.get('source_failure_category') or ''),
|
|
source_failure_auth_related=bool(result.get('source_failure_auth_related')),
|
|
first_error=str(result['errors'][0]),
|
|
)
|
|
|
|
notification = mock.Mock(enabled=True)
|
|
notification.mark_result_bundle_ready.return_value = True
|
|
with self.subTest(category=error.category), tempfile.TemporaryDirectory() as temp_dir, \
|
|
mock.patch.dict(os.environ, {'TRUF_SUPERVISOR_INSTANCE_ID': 'fixture'}), \
|
|
mock.patch.object(scanner.scan_config, 'max_active_scans', 1), \
|
|
mock.patch.object(console_runner, 'require_private_directory', return_value=temp_dir), \
|
|
mock.patch.object(console_runner, 'current_process_identity', return_value=identity), \
|
|
mock.patch.object(console_runner, 'queue_files_for_args', return_value=(None, None)), \
|
|
mock.patch.object(console_runner, 'prepare_scan_options', return_value={'token': 'token'}), \
|
|
mock.patch.object(console_runner, 'acquire_scan_slot', return_value=Lease()), \
|
|
mock.patch.object(console_runner, 'scan_slot_scope', return_value=mock.MagicMock()), \
|
|
mock.patch.object(console_runner, 'ensure_bundle_reservation_paths'), \
|
|
mock.patch.object(console_runner, 'reserve_v2_admission_with_recovery', return_value=
|
|
console_runner.V2AdmissionOutcome(claim, False)), \
|
|
mock.patch.object(console_runner, 'resolve_and_bind_git_claim', side_effect=
|
|
console_runner._GitResolutionFailure(error)), \
|
|
mock.patch.object(console_runner, 'scan_target_result') as scan, \
|
|
mock.patch.object(console_runner, 'stage_result_bundle', side_effect=stage), \
|
|
mock.patch.object(console_runner, 'ScannerDB', return_value=notification), \
|
|
mock.patch.object(console_runner, 'supervised_spool_stop_requested', return_value=False):
|
|
metrics = console_runner.run_cycle_v2(args, DB(), 1, 2, 'github')
|
|
|
|
self.assertEqual(captured['disposition']['queue_status'], expected_queue_status)
|
|
self.assertEqual(metrics['staged_count'], 1)
|
|
self.assertEqual(metrics['source_failure_count'], expected_source_failures)
|
|
scan.assert_not_called()
|
|
|
|
def test_fresh_schema_contains_plan_and_coverage_columns(self):
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
db = ScannerDB(db_path=os.path.join(temp_dir, 'scanner.db'), db_url='')
|
|
try:
|
|
reservations = {
|
|
row['name'] for row in db.conn.execute(
|
|
'PRAGMA table_info(result_reservations)'
|
|
).fetchall()
|
|
}
|
|
queue = {
|
|
row['name'] for row in db.conn.execute(
|
|
'PRAGMA table_info(target_queue)'
|
|
).fetchall()
|
|
}
|
|
finally:
|
|
db.close()
|
|
self.assertTrue({'git_scan_plan_json', 'git_scan_plan_sha256'} <= reservations)
|
|
self.assertTrue({'covered_ref', 'covered_head'} <= queue)
|
|
|
|
|
|
class InstalledTruffleHogContractTests(unittest.TestCase):
|
|
def test_installed_binary_accepts_commit_sha_as_branch(self):
|
|
executable = Path(r'C:\Tools\trufflehog.exe')
|
|
if not executable.is_file():
|
|
self.skipTest('configured TruffleHog binary is not installed')
|
|
with tempfile.TemporaryDirectory(dir=ROOT / 'tmp') as temp_dir:
|
|
repo = Path(temp_dir) / 'repo'
|
|
repo.mkdir()
|
|
home = Path(temp_dir) / 'home'
|
|
home.mkdir()
|
|
env = {key: os.environ[key] for key in (
|
|
'PATH', 'SystemRoot', 'WINDIR', 'COMSPEC', 'PATHEXT',
|
|
) if key in os.environ}
|
|
env.update(
|
|
TMP=temp_dir, TEMP=temp_dir, TMPDIR=temp_dir, HOME=str(home), USERPROFILE=str(home),
|
|
GIT_CONFIG_NOSYSTEM='1', GIT_CONFIG_GLOBAL=os.devnull, GIT_TERMINAL_PROMPT='0',
|
|
GIT_ALLOW_PROTOCOL='file', HTTP_PROXY='http://127.0.0.1:9',
|
|
HTTPS_PROXY='http://127.0.0.1:9', ALL_PROXY='http://127.0.0.1:9',
|
|
)
|
|
git = ['git', '-c', f'core.hooksPath={home}', '-c', 'user.name=Fixture',
|
|
'-c', 'user.email=fixture@example.test', '-C', str(repo)]
|
|
subprocess.run(git + ['init', '-q', f'--template={home}'], env=env, check=True)
|
|
revisions = []
|
|
for number in range(3):
|
|
(repo / 'README.md').write_text(f'exact revision fixture {number}\n', encoding='ascii')
|
|
subprocess.run(git + ['add', 'README.md'], env=env, check=True)
|
|
subprocess.run(git + ['commit', '-q', '-m', 'fixture'], env=env, check=True)
|
|
revisions.append(subprocess.check_output(git + ['rev-parse', 'HEAD'], env=env, text=True).strip())
|
|
uri = 'file://' + repo.as_posix() if os.name == 'nt' else repo.as_uri()
|
|
for label, head, options, success in (
|
|
('baseline', revisions[1], ['--max-depth', '2'], True),
|
|
('delta', revisions[1], ['--since-commit', revisions[0]], True),
|
|
('missing_head', 'f' * 40, ['--max-depth', '2'], False),
|
|
('missing_base', revisions[1], ['--since-commit', 'f' * 40], False),
|
|
):
|
|
with self.subTest(case=label):
|
|
completed = subprocess.run(
|
|
[str(executable), 'git', uri, '--json', '--no-update', '--no-verification',
|
|
'--local-dev', '--concurrency', '1', '--branch', head, *options],
|
|
cwd=temp_dir, env=env, capture_output=True, text=True, timeout=120,
|
|
)
|
|
result = scanner.apply_trufflehog_diagnostics(
|
|
{'errors': []}, completed.stderr, completed.returncode, 'git', require_completion=True,
|
|
)
|
|
if not success:
|
|
self.assertGreater(len(result['errors']), 0)
|
|
self.assertIn('unable to resolve commit: object not found', completed.stderr)
|
|
continue
|
|
self.assertEqual(completed.returncode, 0)
|
|
self.assertEqual(len(result['errors']), 0)
|
|
self.assertTrue(result['scan_meta']['trufflehog_finished'])
|
|
messages = [json.loads(line) for line in completed.stderr.splitlines() if line.strip().startswith('{')]
|
|
finished = [message for message in messages if message.get('msg') == 'finished scanning']
|
|
self.assertTrue(any(message.get('chunks', 0) > 0 and message.get('bytes', 0) > 0 for message in finished))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|