sashatrask
676 lines
32 KiB
Python
676 lines
32 KiB
Python
import contextlib
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / 'app'))
|
|
import scanner
|
|
from scanner_db import canonical_git_scan_plan_bytes, matching_git_coverage
|
|
from test_docker_staging_bounds import command_harness
|
|
|
|
|
|
FINISHED = '{"level":"info-0","msg":"finished scanning"}'
|
|
FINDING = {'DetectorName': 'SyntheticInitial'}
|
|
|
|
|
|
def checkout_error(cause="error: invalid path 'invalid:name.txt'"):
|
|
return json.dumps({
|
|
'level': 'error', 'msg': 'error running scan',
|
|
'error': 'failed to scan Git: error preparing repo: failed to clone: '
|
|
'error executing git clone: exit status 128, ' + cause
|
|
+ '\nwarning: Clone succeeded, but checkout failed.\n',
|
|
})
|
|
|
|
|
|
def checkout_clone_failure(cause="error: invalid path 'invalid:name.txt'", **changes):
|
|
payload = {
|
|
'level': 'info-0', 'ts': '2026-01-01T00:00:00Z',
|
|
'logger': 'trufflehog', 'msg': 'git clone failed',
|
|
'subcommand': 'git clone', 'repo': plan()['repo_url'],
|
|
'path': r'C:\fixture\private\repo', 'args': [],
|
|
'error': 'failed to clone: error executing git clone: exit status 128, '
|
|
+ cause + '\nwarning: Clone succeeded, but checkout failed.\n',
|
|
}
|
|
payload.update(changes)
|
|
return json.dumps(payload)
|
|
|
|
|
|
def checkout_error_pair(cause="error: invalid path 'invalid:name.txt'", **changes):
|
|
return checkout_clone_failure(cause, **changes) + '\n' + checkout_error(cause)
|
|
|
|
|
|
def plan(mode='baseline', depth=2):
|
|
return {
|
|
'version': 1, 'provider': 'gitlab', 'repo_url': 'https://gitlab.com/Fixture/Only.git',
|
|
'repo_path': 'Fixture/Only', 'branch': 'main', 'ref': 'refs/heads/main',
|
|
'head_sha': 'a' * 40, 'base_sha': 'b' * 40 if mode == 'delta' else None,
|
|
'mode': mode, 'baseline_depth': depth, 'ref_source': 'explicit',
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def recovery(tmp_path, monkeypatch):
|
|
state = SimpleNamespace(clock=100.0, calls=[], cleanup=[], outcomes=[], root=tmp_path / 'private # parent',
|
|
after_command=lambda index: None, after_cleanup=lambda: None)
|
|
monkeypatch.setattr(scanner.time, 'monotonic', lambda: state.clock)
|
|
monkeypatch.setattr(scanner, 'get_trufflehog_cmd', lambda: 'synthetic-th')
|
|
monkeypatch.setattr(scanner, 'get_git_cmd', lambda: r'C:\fixture\git.exe')
|
|
monkeypatch.setattr(scanner, 'create_command_work_dir', lambda: str(state.root))
|
|
|
|
def cleanup(path):
|
|
state.cleanup.append(path)
|
|
state.after_cleanup()
|
|
|
|
monkeypatch.setattr(scanner, 'cleanup_command_work_dir', cleanup)
|
|
monkeypatch.setattr(scanner, 'apply_finding_filters', lambda result, target: result)
|
|
|
|
@contextlib.contextmanager
|
|
def command(argv, timeout, environment, **kwargs):
|
|
index = len(state.calls)
|
|
state.calls.append((argv, timeout, dict(environment), kwargs))
|
|
outcome = state.outcomes[index]
|
|
if isinstance(outcome, BaseException):
|
|
raise outcome
|
|
stdout, stderr, code = outcome
|
|
with scanner.streamed_output_from_text(stdout, stderr, code) as output:
|
|
yield output
|
|
state.after_command(index)
|
|
|
|
monkeypatch.setattr(scanner, 'run_command_streamed', command)
|
|
|
|
def run(mode='baseline', **kwargs):
|
|
bound = plan(mode)
|
|
digest = hashlib.sha256(canonical_git_scan_plan_bytes(bound)).hexdigest()
|
|
return scanner.scan_exact_git_plan(
|
|
bound['repo_url'], bound, digest, 30, 'CustomRegex', 'IgnoredFixture', True,
|
|
'fixture-policy.yaml', 'synthetic-token', True, **kwargs,
|
|
)
|
|
|
|
state.run = run
|
|
state.outcomes = [(json.dumps(FINDING), checkout_error(), 1), ('', '', 0),
|
|
('{"DetectorName":"SyntheticRecovered"}', FINISHED, 0)]
|
|
return state
|
|
|
|
|
|
@pytest.mark.parametrize('mode', ['baseline', 'delta'])
|
|
def test_recovery_preserves_pinning_options_identity_findings_and_deadline(recovery, mode):
|
|
result = recovery.run(mode)
|
|
assert not result['errors']
|
|
assert len(result['findings']) == 2
|
|
assert result['git_scan_execution']['success']
|
|
assert result['git_scan_execution']['coverage_complete']
|
|
assert result['scan_meta']['git_checkout_recovery'] == {
|
|
'attempted': True, 'clone_succeeded': True, 'coverage_complete': True,
|
|
}
|
|
assert len(recovery.calls) == 3
|
|
first, clone, local = recovery.calls
|
|
assert clone[0] == [r'C:\fixture\git.exe', 'clone', '--no-checkout', '--no-recurse-submodules', '--',
|
|
plan()['repo_url'], str(recovery.root / 'repo')]
|
|
assert clone[3]['native_git_clone'] is True
|
|
assert clone[3]['staging_roots'] == local[3]['staging_roots'] == (str(recovery.root),)
|
|
assert '%23' in local[0][2] and '%20' in local[0][2]
|
|
assert local[0][2].startswith('file://') and not local[0][2].startswith('file:///')
|
|
assert first[0][3:] == local[0][3:]
|
|
assert local[0][local[0].index('--branch') + 1] == plan()['head_sha']
|
|
if mode == 'delta':
|
|
assert local[0][local[0].index('--since-commit') + 1] == plan('delta')['base_sha']
|
|
else:
|
|
assert local[0][local[0].index('--max-depth') + 1] == '2'
|
|
for argv, timeout, environment, kwargs in recovery.calls:
|
|
assert 'synthetic-token' not in ' '.join(argv)
|
|
assert environment['TRUF_GIT_TOKEN'] == 'synthetic-token'
|
|
assert environment['TRUF_GIT_USERNAME'] == 'oauth2'
|
|
assert kwargs['deadline'] == 130.0 and timeout <= 30
|
|
assert recovery.cleanup == [str(recovery.root)]
|
|
|
|
|
|
def test_exact_git_commands_share_appended_windows_longpaths(recovery, monkeypatch):
|
|
appended = []
|
|
|
|
def append_longpaths(environment, operation):
|
|
appended.append((environment, operation))
|
|
environment.update({
|
|
'GIT_CONFIG_COUNT': '2',
|
|
'GIT_CONFIG_KEY_1': 'core.longpaths',
|
|
'GIT_CONFIG_VALUE_1': 'true',
|
|
})
|
|
|
|
monkeypatch.setattr(scanner, '_append_windows_git_longpaths', append_longpaths)
|
|
result = recovery.run()
|
|
|
|
assert not result['errors']
|
|
assert len(appended) == 1 and appended[0][1] == 'Exact Git'
|
|
assert all(call[2]['GIT_CONFIG_COUNT'] == '2' for call in recovery.calls)
|
|
assert all(call[2]['GIT_CONFIG_KEY_1'] == 'core.longpaths' for call in recovery.calls)
|
|
assert all(call[2]['GIT_CONFIG_VALUE_1'] == 'true' for call in recovery.calls)
|
|
|
|
|
|
def test_healthy_scan_does_not_prepare_or_clone(recovery):
|
|
recovery.outcomes = [(json.dumps(FINDING), FINISHED, 0)]
|
|
result = recovery.run()
|
|
assert result['git_scan_execution']['coverage_complete']
|
|
assert len(recovery.calls) == 1 and not recovery.cleanup
|
|
assert 'git_checkout_recovery' not in result['scan_meta']
|
|
|
|
|
|
@pytest.mark.parametrize('cause', [
|
|
"error: invalid path 'invalid:name.txt'",
|
|
"error: invalid path 'fatal:notes.txt'",
|
|
"error: invalid path 'error:notes.txt'",
|
|
"error: invalid path 'fatal: notes.txt'",
|
|
'error: invalid path "error: notes.txt"',
|
|
"error: invalid path 'trailing. /file.txt'",
|
|
'error: unable to create file synthetic/file: Filename too long',
|
|
"fatal: cannot create directory at 'synthetic/path': Filename too long",
|
|
])
|
|
def test_proven_checkout_path_families_allow_recovery(recovery, cause):
|
|
recovery.outcomes[0] = ('', checkout_error(cause), 1)
|
|
assert not recovery.run()['errors']
|
|
assert len(recovery.calls) == 3
|
|
|
|
|
|
@pytest.mark.parametrize('cause', [
|
|
"error: invalid path 'invalid:name.txt'",
|
|
'error: unable to create file synthetic/file: Filename too long',
|
|
])
|
|
def test_real_clone_failure_companion_allows_checkout_recovery(recovery, cause):
|
|
recovery.outcomes[0] = ('', checkout_error_pair(cause), 1)
|
|
assert not recovery.run()['errors']
|
|
assert len(recovery.calls) == 3
|
|
|
|
|
|
@pytest.mark.parametrize('changes', [
|
|
{'level': 'error'},
|
|
{'logger': 'other'},
|
|
{'msg': 'other failure'},
|
|
{'subcommand': 'git fetch'},
|
|
{'args': ['unexpected']},
|
|
{'error': 'unrelated failure'},
|
|
{'extra': 'unexpected'},
|
|
])
|
|
def test_unproven_clone_failure_companion_never_launches_recovery(recovery, changes):
|
|
recovery.outcomes[0] = ('', checkout_error_pair(**changes), 1)
|
|
result = recovery.run()
|
|
assert result['errors']
|
|
assert len(recovery.calls) == 1 and not recovery.cleanup
|
|
|
|
|
|
@pytest.mark.parametrize('stderr,code', [
|
|
('fatal: exit status 128', 1),
|
|
(checkout_error().replace('Clone succeeded, but checkout failed', 'checkout failed'), 1),
|
|
(checkout_error().replace('error preparing repo', 'other operation'), 1),
|
|
(checkout_error('fatal: repository not found'), 1),
|
|
(checkout_error('error: invalid path fixture\nfatal: unclassified additional failure'), 1),
|
|
(checkout_error("error: invalid path 'fatal:notes.txt'\nfatal: unclassified additional failure"), 1),
|
|
(checkout_error("error: invalid path 'error:notes.txt' error: additional failure"), 1),
|
|
(checkout_error("error: invalid path 'error:notes.txt'\nremote: fatal: additional failure"), 1),
|
|
(checkout_error("error: invalid path 'unterminated fatal: notes.txt"), 1),
|
|
(checkout_error('error: invalid path fixture\nerror: unknown flag'), 1),
|
|
(checkout_error('error: invalid path fixture\nfatal: connection reset'), 1),
|
|
(checkout_error('error: invalid path fixture\nfatal: no space left on device'), 1),
|
|
(checkout_error('error: invalid path fixture\nfatal: unauthorized'), 1),
|
|
(checkout_error('error: invalid path fixture\nfatal: context deadline exceeded'), 1),
|
|
(checkout_error() + '\n' + '{"level":"error","msg":"unclassified failure"}', 1),
|
|
(checkout_error() + '\n' + FINISHED, 1),
|
|
(checkout_error(), 0),
|
|
(checkout_error(), -1),
|
|
(checkout_error().replace('"level": "error"', '"level": "info"'), 1),
|
|
])
|
|
def test_non_checkout_or_mixed_failures_never_launch_recovery(recovery, stderr, code):
|
|
recovery.outcomes = [(json.dumps(FINDING), stderr, code)]
|
|
result = recovery.run()
|
|
assert result['errors']
|
|
assert len(result['findings']) == 1
|
|
assert len(recovery.calls) == 1 and not recovery.cleanup
|
|
assert not result['git_scan_execution']['coverage_complete']
|
|
|
|
|
|
@pytest.mark.parametrize('field', ['source_failure', 'degraded', 'warnings'])
|
|
def test_gate_rejects_other_result_failure_evidence(field):
|
|
result = scanner.apply_trufflehog_diagnostics({'errors': []}, checkout_error(), 1, 'git', True)
|
|
result[field] = True
|
|
assert not scanner.git_checkout_recovery_allowed(result)
|
|
|
|
|
|
@pytest.mark.parametrize('stderr,code', [
|
|
('fatal: authentication failed', 128), ('fatal: no space left on device', 128),
|
|
('fatal: unclassified clone failure', 128),
|
|
])
|
|
def test_failed_native_clone_preserves_initial_error_and_findings(recovery, stderr, code):
|
|
recovery.outcomes[1] = ('', stderr, code)
|
|
result = recovery.run()
|
|
assert checkout_error() in result['errors'] and len(result['errors']) >= 2
|
|
assert len(result['findings']) == 1
|
|
assert len(recovery.calls) == 2 and recovery.cleanup
|
|
assert not result['scan_meta']['git_checkout_recovery']['coverage_complete']
|
|
|
|
|
|
@pytest.mark.parametrize('stderr,code', [
|
|
('', 0), ('{"level":"error","msg":"source failed","error":"connection reset"}', 1),
|
|
('{"level":"error","msg":"non-critical error processing chunk","error":"invalid archive"}\n' + FINISHED, 0),
|
|
])
|
|
def test_incomplete_local_scan_cannot_clear_initial_errors(recovery, stderr, code):
|
|
recovery.outcomes[2] = ('{"DetectorName":"SyntheticPartial"}', stderr, code)
|
|
result = recovery.run()
|
|
assert checkout_error() in result['errors']
|
|
assert len(result['findings']) == 2
|
|
assert not result['git_scan_execution']['coverage_complete']
|
|
|
|
|
|
def test_existing_base_reset_reuses_prepared_clone(recovery):
|
|
recovery.outcomes[2] = ('', 'fatal: bad object', 1)
|
|
recovery.outcomes.append(('', FINISHED, 0))
|
|
result = recovery.run('delta')
|
|
assert not result['errors']
|
|
assert len(recovery.calls) == 4
|
|
assert sum(bool(call[3].get('native_git_clone')) for call in recovery.calls) == 1
|
|
assert recovery.calls[2][0][2] == recovery.calls[3][0][2]
|
|
assert '--since-commit' not in recovery.calls[3][0]
|
|
assert '--max-depth' in recovery.calls[3][0]
|
|
assert result['git_scan_execution']['mode'] == 'baseline_reset'
|
|
assert all(call[3]['deadline'] == 130 for call in recovery.calls)
|
|
|
|
|
|
def test_clone_transport_object_error_is_not_a_delta_boundary_reset(recovery):
|
|
recovery.outcomes[1] = ('', 'fatal: object not found during clone transport', 128)
|
|
result = recovery.run('delta')
|
|
assert len(recovery.calls) == 2
|
|
assert result['git_scan_execution']['mode'] == 'delta'
|
|
assert not result['git_scan_execution']['coverage_complete']
|
|
assert checkout_error() in result['errors']
|
|
|
|
|
|
def test_remaining_budget_decreases_across_commands(recovery):
|
|
recovery.after_command = lambda index: setattr(recovery, 'clock', recovery.clock + 4)
|
|
result = recovery.run()
|
|
assert not result['errors']
|
|
assert [call[1] for call in recovery.calls] == [30, 26, 22]
|
|
assert all(call[3]['deadline'] == 130 for call in recovery.calls)
|
|
|
|
|
|
@pytest.mark.parametrize('flag', ['diagnostic_output_limited', 'command_timed_out', 'trufflehog_finished'])
|
|
def test_gate_rejects_incomplete_or_inconsistent_diagnostic_evidence(flag):
|
|
result = scanner.apply_trufflehog_diagnostics({'errors': []}, checkout_error(), 1, 'git', True)
|
|
result['scan_meta'][flag] = True
|
|
assert not scanner.git_checkout_recovery_allowed(result)
|
|
|
|
|
|
def test_oversized_checkout_evidence_is_not_reprocessed():
|
|
result = scanner.apply_trufflehog_diagnostics({'errors': []}, checkout_error(), 1, 'git', True)
|
|
result['errors'][0] += 'x' * 8192
|
|
assert not scanner.git_checkout_recovery_allowed(result)
|
|
|
|
|
|
def test_clone_stdout_bounds_preserve_initial_failure(recovery):
|
|
recovery.outcomes[1] = ('x' * 8193, '', 0)
|
|
result = recovery.run()
|
|
assert len(recovery.calls) == 2
|
|
assert result['error_class'] == 'output_limit'
|
|
assert result['retryable'] is False
|
|
assert checkout_error() in result['errors']
|
|
assert len(result['findings']) == 1
|
|
|
|
|
|
def test_finding_bound_is_shared_across_initial_and_recovered_scans(recovery, monkeypatch):
|
|
monkeypatch.setenv('TRUFFLEHOG_MAX_FINDINGS_PER_TARGET', '1')
|
|
result = recovery.run()
|
|
assert result['findings'] == [FINDING]
|
|
assert result['error_class'] == 'output_limit'
|
|
assert result['retryable'] is False
|
|
assert checkout_error() in result['errors']
|
|
assert not result['git_scan_execution']['coverage_complete']
|
|
|
|
|
|
@pytest.mark.parametrize('expired_after', [0, 1, 2])
|
|
def test_shared_deadline_never_restarts_for_recovery(recovery, expired_after):
|
|
recovery.after_command = lambda index: setattr(recovery, 'clock', 131.0) if index == expired_after else None
|
|
result = recovery.run()
|
|
assert result['errors'] and result['scan_meta']['git_deadline_exceeded']
|
|
assert len(recovery.calls) == expired_after + 1
|
|
assert len(result['findings']) >= 1
|
|
assert not result['git_scan_execution']['coverage_complete']
|
|
|
|
|
|
@pytest.mark.parametrize('stage', ['cleanup', 'filter'])
|
|
def test_deadline_includes_cleanup_and_filters(recovery, monkeypatch, stage):
|
|
if stage == 'cleanup':
|
|
recovery.after_cleanup = lambda: setattr(recovery, 'clock', 131.0)
|
|
else:
|
|
def filter_result(result, target):
|
|
recovery.clock = 131.0
|
|
return result
|
|
monkeypatch.setattr(scanner, 'apply_finding_filters', filter_result)
|
|
result = recovery.run()
|
|
assert checkout_error() in result['errors']
|
|
assert result['scan_meta']['git_deadline_exceeded']
|
|
assert not result['git_scan_execution']['coverage_complete']
|
|
assert len(result['findings']) == 2
|
|
|
|
|
|
def test_optional_post_scan_warning_does_not_undo_recovered_coverage(recovery, monkeypatch):
|
|
def filter_result(result, target):
|
|
result.update(degraded=True, warnings=['optional candidate staging unavailable'])
|
|
return result
|
|
monkeypatch.setattr(scanner, 'apply_finding_filters', filter_result)
|
|
result = recovery.run()
|
|
assert not result['errors']
|
|
assert result['git_scan_execution']['coverage_complete']
|
|
encoded = canonical_git_scan_plan_bytes(result['git_scan_plan'])
|
|
reservation = {'git_scan_plan_json': encoded.decode(), 'git_scan_plan_sha256': hashlib.sha256(encoded).hexdigest()}
|
|
assert matching_git_coverage(reservation, result, 'done', 0)[0]
|
|
|
|
|
|
def test_unconfirmed_child_retains_parent_for_authenticated_cleanup(recovery):
|
|
recovery.outcomes[1] = scanner.ScanSlotFatalError('synthetic unconfirmed child')
|
|
with pytest.raises(scanner.ScanSlotFatalError):
|
|
recovery.run()
|
|
assert not recovery.cleanup
|
|
|
|
|
|
def clone_command(state):
|
|
return [str(Path(shutil.which('git')).resolve()), 'clone', '--no-checkout', '--no-recurse-submodules', '--',
|
|
plan()['repo_url'], str(state.blobs / 'repo')]
|
|
|
|
|
|
def test_native_runner_dispatches_real_git_helper_without_weakening_th_default(command_harness, monkeypatch):
|
|
state = command_harness
|
|
state.completed = True
|
|
argv = clone_command(state)
|
|
authority = mock.Mock(return_value={'code_manifest': {'executables': {'git': {'path': argv[0]}}}})
|
|
monkeypatch.setattr(scanner, 'require_active_supervisor_child', authority)
|
|
th_guard = mock.Mock(side_effect=RuntimeError('default TH guard rejects Git'))
|
|
monkeypatch.setattr(scanner, 'require_trufflehog_launch_authority', th_guard)
|
|
with scanner.run_command_streamed(argv, 30, staging_roots=(str(state.blobs),), native_git_clone=True) as output:
|
|
assert output.returncode == 0
|
|
authority.assert_called_once_with(child_kind='scanner', require_dsn=True)
|
|
th_guard.assert_not_called()
|
|
with pytest.raises(RuntimeError, match='default TH guard'):
|
|
with scanner.run_command_streamed(argv, 30, staging_roots=(str(state.blobs),)):
|
|
pass
|
|
|
|
|
|
def test_native_runner_keeps_borrowed_slot_watchdog_and_environment(command_harness, monkeypatch):
|
|
state = command_harness
|
|
state.completed = True
|
|
monkeypatch.setattr(scanner, 'scoped_scan_slot_lease', lambda: (True, state.slot))
|
|
monkeypatch.setattr(scanner, 'require_git_clone_launch_authority', lambda cmd: None)
|
|
monkeypatch.setattr(scanner, 'harden_private_file', lambda path: None)
|
|
acquire = mock.Mock(side_effect=AssertionError('borrowed lease must not reacquire'))
|
|
monkeypatch.setattr(scanner, 'acquire_scan_slot', acquire)
|
|
watchdog = mock.Mock(return_value='')
|
|
monkeypatch.setattr(scanner, '_check_command_staging', watchdog)
|
|
with scanner.run_command_streamed(clone_command(state), 30, {
|
|
'TRUF_GIT_TOKEN': 'synthetic-token', 'TRUF_GIT_USERNAME': 'oauth2',
|
|
}, deadline=125.0, staging_roots=(str(state.blobs),), native_git_clone=True):
|
|
pass
|
|
assert state.options['env']['GIT_TERMINAL_PROMPT'] == '0'
|
|
assert state.options['env']['TEMP'] == str(state.command_dir)
|
|
assert state.options['env']['GIT_ASKPASS'].endswith('git-askpass.cmd')
|
|
assert state.options['env']['NO_PROXY'] == '*'
|
|
assert state.options['job_memory_limit_bytes'] == 4 * 1024 ** 3
|
|
assert watchdog.call_count >= 2
|
|
assert all(call.args[1] == 125.0 for call in watchdog.call_args_list)
|
|
state.slot.release.assert_not_called()
|
|
state.slot.set_child_pid.assert_called_once_with(41)
|
|
|
|
|
|
@pytest.mark.parametrize('provided', [False, True])
|
|
def test_command_clears_case_insensitive_proxies_after_environment_copy(command_harness, monkeypatch, provided):
|
|
state = command_harness
|
|
state.completed = True
|
|
poison = {
|
|
'HTTP_PROXY': 'http://fixture.invalid:8080',
|
|
'https_proxy': 'http://fixture.invalid:8080',
|
|
'AlL_pRoXy': 'http://fixture.invalid:8080',
|
|
'no_proxy': 'fixture.invalid',
|
|
'NO_PROXY': 'other.invalid',
|
|
'TRUF_GIT_TOKEN': 'synthetic-token', 'TRUF_GIT_USERNAME': 'oauth2',
|
|
'PGPASSWORD': 'synthetic-supervisor-secret',
|
|
}
|
|
monkeypatch.setattr(scanner, 'harden_private_file', lambda path: None)
|
|
for key, value in poison.items():
|
|
monkeypatch.setenv(key, value)
|
|
environment = dict(os.environ)
|
|
supplied = dict(poison) if provided else None
|
|
with scanner.run_command_streamed(['fixture'], 30, env=supplied):
|
|
pass
|
|
child = state.options['env']
|
|
assert {key.lower(): value for key, value in child.items() if key.lower().endswith('_proxy')} == {'no_proxy': '*'}
|
|
assert child['TRUF_GIT_TOKEN'] == 'synthetic-token'
|
|
assert child['TRUF_GIT_USERNAME'] == 'oauth2'
|
|
assert 'PGPASSWORD' not in child
|
|
assert child['GIT_ASKPASS'].endswith('git-askpass.cmd')
|
|
assert child['GIT_TERMINAL_PROMPT'] == '0'
|
|
assert dict(os.environ) == environment
|
|
assert supplied == (poison if provided else None)
|
|
state.slot.set_child_pid.assert_called_once_with(41)
|
|
state.slot.release.assert_called_once()
|
|
|
|
|
|
def test_native_authority_fingerprint_time_is_inside_deadline(command_harness, monkeypatch):
|
|
state = command_harness
|
|
def authority(cmd):
|
|
state.clock = 140.0
|
|
monkeypatch.setattr(scanner, 'require_git_clone_launch_authority', authority)
|
|
launch = mock.Mock(side_effect=AssertionError('expired authorization must not launch'))
|
|
monkeypatch.setattr(scanner, 'OwnedProcess', launch)
|
|
with scanner.run_command_streamed(clone_command(state), 30, deadline=130,
|
|
staging_roots=(str(state.blobs),), native_git_clone=True) as output:
|
|
assert output.returncode == -1
|
|
launch.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize('selector', [None, 1, 'true'])
|
|
def test_native_selector_requires_explicit_boolean(selector):
|
|
with pytest.raises(ValueError):
|
|
with scanner.run_command_streamed([], 1, native_git_clone=selector):
|
|
pass
|
|
|
|
|
|
def test_native_runner_rejects_unmonitored_or_outside_destination(command_harness, monkeypatch):
|
|
state = command_harness
|
|
monkeypatch.setattr(scanner, 'require_git_clone_launch_authority', lambda cmd: None)
|
|
with pytest.raises(ValueError):
|
|
with scanner.run_command_streamed(clone_command(state), 30, native_git_clone=True):
|
|
pass
|
|
argv = clone_command(state)
|
|
argv[-1] = str(state.blobs.parent / 'outside' / 'repo')
|
|
with pytest.raises(RuntimeError, match='outside its private staging parent'):
|
|
with scanner.run_command_streamed(argv, 30, staging_roots=(str(state.blobs),), native_git_clone=True):
|
|
pass
|
|
|
|
|
|
@pytest.mark.parametrize('unkillable', [False, True])
|
|
def test_native_watchdog_preserves_job_termination_and_fail_closed_capacity(command_harness, monkeypatch, unkillable):
|
|
state = command_harness
|
|
state.unkillable = unkillable
|
|
monkeypatch.setattr(scanner, 'require_git_clone_launch_authority', lambda cmd: None)
|
|
monkeypatch.setattr(scanner, '_check_command_staging', mock.Mock(side_effect=['', 'TruffleHog staging limit exceeded']))
|
|
|
|
def run():
|
|
with scanner.run_command_streamed(clone_command(state), 30,
|
|
staging_roots=(str(state.blobs),), native_git_clone=True) as output:
|
|
assert output.returncode == -1
|
|
|
|
if unkillable:
|
|
with pytest.raises(scanner.ScanSlotFatalError):
|
|
run()
|
|
state.slot.release.assert_not_called()
|
|
state.slot.mark_non_releasable.assert_called()
|
|
scanner.cleanup_command_work_dir.assert_not_called()
|
|
else:
|
|
run()
|
|
state.slot.release.assert_called_once()
|
|
assert state.kill_calls and state.wait_calls
|
|
|
|
|
|
def test_git_executable_fingerprint_budget_does_not_allow_clone(recovery, monkeypatch):
|
|
def get_git():
|
|
recovery.clock = 131.0
|
|
return r'C:\fixture\git.exe'
|
|
monkeypatch.setattr(scanner, 'get_git_cmd', get_git)
|
|
result = recovery.run()
|
|
assert len(recovery.calls) == 1
|
|
assert recovery.cleanup
|
|
assert checkout_error() in result['errors']
|
|
assert result['scan_meta']['git_deadline_exceeded']
|
|
|
|
|
|
@pytest.mark.parametrize('failure', [OSError('synthetic cleanup failure'), RuntimeError('synthetic cleanup failure')])
|
|
def test_cleanup_failure_never_clears_initial_error_or_findings(recovery, monkeypatch, failure):
|
|
monkeypatch.setattr(scanner, 'cleanup_command_work_dir', mock.Mock(side_effect=failure))
|
|
result = recovery.run()
|
|
assert checkout_error() in result['errors']
|
|
assert len(result['findings']) == 2
|
|
assert result['error_class'] == 'source_resource'
|
|
assert not result['git_scan_execution']['coverage_complete']
|
|
|
|
|
|
@pytest.fixture(params=['healthy', 'forbidden_character', 'fatal_filename', 'error_filename', 'trailing_dot_space', 'long_path'])
|
|
def installed_fixture(request, monkeypatch):
|
|
executable = Path(r'C:\Tools\trufflehog.exe')
|
|
git_executable = shutil.which('git')
|
|
if os.name != 'nt' or not executable.is_file() or not git_executable:
|
|
pytest.skip('Windows installed Git/TH contract test')
|
|
with tempfile.TemporaryDirectory(prefix='git-checkout-test-', dir=ROOT / 'tmp') as temp_dir:
|
|
root = Path(temp_dir)
|
|
home = root / 'home'
|
|
home.mkdir()
|
|
command_root = root / 'commands'
|
|
scanner.ensure_private_directory(str(command_root))
|
|
source = root / 'fixture.git'
|
|
policy = root / 'marker.yaml'
|
|
policy.write_text(
|
|
'detectors:\n - name: OfflineGitScopeFixture\n keywords: [GITSCOPEFIXTURE]\n'
|
|
" regex:\n marker: 'GITSCOPEFIXTURE_[A-Z]{8}_[0-9]{4}'\n", encoding='ascii',
|
|
)
|
|
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', GIT_LFS_SKIP_SMUDGE='1', HTTP_PROXY='http://127.0.0.1:9',
|
|
HTTPS_PROXY='http://127.0.0.1:9', ALL_PROXY='http://127.0.0.1:9', NO_PROXY='',
|
|
)
|
|
|
|
def git(*args, data=None):
|
|
completed = subprocess.run(
|
|
[git_executable, '-c', f'core.hooksPath={home}', *args], input=data, capture_output=True,
|
|
env=env, cwd=root, timeout=30,
|
|
)
|
|
assert completed.returncode == 0, 'synthetic fixture Git setup failed'
|
|
return completed.stdout.decode().strip()
|
|
|
|
git('init', '--bare', '--quiet', f'--template={home}', str(source))
|
|
problematic = {
|
|
'healthy': 'head_marker.txt', 'forbidden_character': 'invalid:name.txt',
|
|
'fatal_filename': 'fatal:notes.txt', 'error_filename': 'error:notes.txt',
|
|
'trailing_dot_space': 'trailing. /head_marker.txt',
|
|
'long_path': '/'.join(['segment_' + 'x' * 22] * 11 + ['head_marker.txt']),
|
|
}[request.param]
|
|
markers = {
|
|
'earlier': 'GITSCOPEFIXTURE_EARLIERX_1001', 'base': 'GITSCOPEFIXTURE_BASEONLY_1002',
|
|
'head': 'GITSCOPEFIXTURE_HEADONLY_1003', 'later': 'GITSCOPEFIXTURE_LATERXXX_1004',
|
|
'side': 'GITSCOPEFIXTURE_SIDEONLY_1005',
|
|
}
|
|
stream = bytearray()
|
|
for index, (role, path, parent, branch) in enumerate((
|
|
('earlier', 'earlier.txt', None, 'main'), ('base', 'base.txt', 1, 'main'),
|
|
('head', problematic, 2, 'main'), ('later', 'later.txt', 3, 'main'),
|
|
('side', 'side.txt', 2, 'side'),
|
|
), 1):
|
|
content = (markers[role] + '\n').encode('ascii')
|
|
stream.extend((f'commit refs/heads/{branch}\nmark :{index}\n'
|
|
f'committer Fixture <fixture@example.test> {1700000000 + index} +0000\n'
|
|
'data 7\nfixture\n').encode('ascii'))
|
|
if parent:
|
|
stream.extend(f'from :{parent}\n'.encode('ascii'))
|
|
stream.extend(f'M 100644 inline {json.dumps(path)}\ndata {len(content)}\n'.encode('ascii'))
|
|
stream.extend(content + b'\n')
|
|
git('-C', str(source), 'fast-import', '--quiet', data=bytes(stream) + b'done\n')
|
|
git('-C', str(source), 'symbolic-ref', 'HEAD', 'refs/heads/main')
|
|
refs = {role: git('-C', str(source), 'rev-parse', revision) for role, revision in (
|
|
('earlier', 'main~3'), ('base', 'main~2'), ('head', 'main~1'), ('later', 'main'), ('side', 'side'),
|
|
)}
|
|
env.update(
|
|
GIT_CONFIG_COUNT='3', GIT_CONFIG_KEY_0=f'url.{source.as_uri()}.insteadOf',
|
|
GIT_CONFIG_VALUE_0=plan()['repo_url'], GIT_CONFIG_KEY_1='core.longpaths', GIT_CONFIG_VALUE_1='false',
|
|
GIT_CONFIG_KEY_2='core.protectNTFS', GIT_CONFIG_VALUE_2='true',
|
|
)
|
|
# Bootstrap authority/lease are isolated; the real path/argv guards and OwnedProcess run below.
|
|
metadata = {'code_manifest': {'executables': {
|
|
'git': {'path': git_executable}, 'trufflehog': {'path': str(executable)},
|
|
}, 'assets': {str(policy): {'path': str(policy)}}}}
|
|
monkeypatch.setattr(scanner, 'require_active_supervisor_child', lambda **kwargs: metadata)
|
|
monkeypatch.setattr(scanner.os, 'environ', env)
|
|
monkeypatch.setattr(scanner, '_runtime_initialized', True)
|
|
monkeypatch.setattr(scanner.scan_config, 'work_dir', str(command_root))
|
|
monkeypatch.setattr(scanner.scan_config, 'trufflehog_path', str(executable))
|
|
monkeypatch.setattr(scanner.scan_config, 'min_free_gb', 0)
|
|
monkeypatch.setattr(scanner.scan_config, 'trufflehog_job_memory_limit_bytes', 2 * 1024 ** 3)
|
|
slot = mock.Mock(releasable=True)
|
|
monkeypatch.setattr(scanner, 'scoped_scan_slot_lease', lambda: (True, slot))
|
|
calls = []
|
|
original = scanner.run_command_streamed
|
|
|
|
@contextlib.contextmanager
|
|
def command(*args, **kwargs):
|
|
calls.append((args, kwargs))
|
|
with original(*args, **kwargs) as output:
|
|
yield output
|
|
|
|
monkeypatch.setattr(scanner, 'run_command_streamed', command)
|
|
yield SimpleNamespace(family=request.param, markers=markers, refs=refs, policy=policy,
|
|
calls=calls, slot=slot, command_root=command_root)
|
|
|
|
|
|
@pytest.mark.parametrize('scope,depth,expected', [
|
|
('baseline', 1, {'head'}), ('baseline', 2, {'base', 'head'}),
|
|
('delta', 2, {'head'}), ('invalid_head', 2, set()),
|
|
])
|
|
def test_installed_checkout_recovery_scans_exact_object_scope(installed_fixture, scope, depth, expected):
|
|
fixture = installed_fixture
|
|
bound = plan('delta' if scope == 'delta' else 'baseline', depth)
|
|
bound['head_sha'] = 'f' * 40 if scope == 'invalid_head' else fixture.refs['head']
|
|
if scope == 'delta':
|
|
bound['base_sha'] = fixture.refs['base']
|
|
digest = hashlib.sha256(canonical_git_scan_plan_bytes(bound)).hexdigest()
|
|
result = scanner.scan_exact_git_plan(
|
|
bound['repo_url'], bound, digest, 60, None, None, True, str(fixture.policy), '', True,
|
|
)
|
|
seen = set()
|
|
correct_metadata = True
|
|
for finding in result['findings']:
|
|
raw = str(finding.get('Raw', '')) + str(finding.get('RawV2', ''))
|
|
roles = {role for role, marker in fixture.markers.items() if marker in raw}
|
|
seen.update(roles)
|
|
if roles:
|
|
metadata = finding['SourceMetadata']['Data']['Git']
|
|
correct_metadata &= metadata['repository'] == bound['repo_url']
|
|
correct_metadata &= all(metadata['commit'] == fixture.refs[role] for role in roles)
|
|
assert seen == expected
|
|
assert correct_metadata
|
|
if scope == 'invalid_head':
|
|
assert len(result['errors']) > 0
|
|
assert not result['git_scan_execution']['coverage_complete']
|
|
else:
|
|
assert len(result['errors']) == 0, 'synthetic recovery retained errors'
|
|
assert result['scan_meta']['trufflehog_finished']
|
|
assert result['git_scan_execution']['coverage_complete']
|
|
assert result['git_scan_execution']['plan_sha256'] == digest
|
|
clones = sum(bool(kwargs.get('native_git_clone')) for _, kwargs in fixture.calls)
|
|
assert clones == (0 if fixture.family in {'healthy', 'long_path'} else 1)
|
|
assert len({kwargs['deadline'] for _, kwargs in fixture.calls}) == 1
|
|
fixture.slot.release.assert_not_called()
|
|
assert fixture.slot.set_child_pid.call_count == len(fixture.calls)
|
|
assert not list(fixture.command_root.iterdir()), 'owned command/recovery directories were not cleaned'
|