sashatrask
211 lines
8.2 KiB
Python
211 lines
8.2 KiB
Python
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'app'))
|
|
import scanner
|
|
from test_docker_staging_bounds import command_harness
|
|
|
|
|
|
DOCKER_TARGET = 'docker.io/library/alpine@sha256:' + ('a' * 64)
|
|
|
|
|
|
def _json_response(status, payload):
|
|
encoded = json.dumps(payload).encode('utf-8')
|
|
response = mock.Mock(
|
|
status_code=status,
|
|
headers={'Content-Length': str(len(encoded))},
|
|
)
|
|
response.iter_content.return_value = [encoded]
|
|
return response
|
|
|
|
|
|
@pytest.mark.parametrize('planning_kind', [
|
|
'docker_direct_v1', 'huggingface_space_v1',
|
|
])
|
|
def test_direct_child_environment_preserves_operator_provider_settings(
|
|
command_harness, monkeypatch, planning_kind,
|
|
):
|
|
state = command_harness
|
|
state.completed = True
|
|
operator_environment = {
|
|
'PATH': os.environ.get('PATH', ''),
|
|
'UNRELATED_SETTING': 'preserved',
|
|
'DOCKER_CONFIG': 'operator-docker-config',
|
|
'DOCKER_AUTH_CONFIG': 'operator-docker-auth',
|
|
'REGISTRY_AUTH_FILE': 'operator-registry-auth',
|
|
'HF_HOME': 'operator-hf-home',
|
|
'HF_TOKEN_PATH': 'operator-hf-token-path',
|
|
'HF_TOKEN': 'operator-hf-token',
|
|
'HUGGINGFACE_TOKEN': 'operator-huggingface-token',
|
|
'GIT_CONFIG_PARAMETERS': 'credential.helper=operator-helper',
|
|
'GH_TOKEN': 'operator-github-token',
|
|
'GITLAB_TOKEN': 'operator-gitlab-token',
|
|
'HOME': 'operator-home',
|
|
'USERPROFILE': 'operator-profile',
|
|
'XDG_CONFIG_HOME': 'operator-xdg-config',
|
|
'HTTP_PROXY': 'http://operator-proxy.invalid',
|
|
}
|
|
monkeypatch.setattr(scanner.os, 'environ', operator_environment)
|
|
manifest_token = scanner._client_scan_manifest.set({
|
|
'executables': {'git': {'path': sys.executable}},
|
|
})
|
|
try:
|
|
with scanner.client_remote_execution_binding(planning_kind):
|
|
with scanner.run_command_streamed(['fixture'], 30):
|
|
pass
|
|
child = state.options['env']
|
|
finally:
|
|
scanner._client_scan_manifest.reset(manifest_token)
|
|
|
|
for name in (
|
|
'UNRELATED_SETTING', 'DOCKER_CONFIG', 'DOCKER_AUTH_CONFIG',
|
|
'REGISTRY_AUTH_FILE', 'HF_HOME', 'HF_TOKEN_PATH', 'HF_TOKEN',
|
|
'HUGGINGFACE_TOKEN', 'GIT_CONFIG_PARAMETERS', 'GH_TOKEN',
|
|
'GITLAB_TOKEN', 'HOME', 'USERPROFILE', 'XDG_CONFIG_HOME',
|
|
):
|
|
assert child[name] == operator_environment[name]
|
|
assert 'HTTP_PROXY' not in child
|
|
assert child['NO_PROXY'] == '*'
|
|
assert all(child[name] == str(state.command_dir) for name in ('TEMP', 'TMP', 'TMPDIR'))
|
|
assert child['GIT_TERMINAL_PROMPT'] == '0'
|
|
assert child['GIT_ASKPASS'] == 'true'
|
|
assert scanner._client_remote_execution_kind.get() is None
|
|
|
|
|
|
def test_direct_scanner_entries_reject_credentials_and_skip_docker_pool(monkeypatch):
|
|
manifest_token = scanner._client_scan_manifest.set({
|
|
'executables': {'git': {'path': sys.executable}},
|
|
})
|
|
try:
|
|
with scanner.client_remote_execution_binding('docker_direct_v1'), \
|
|
mock.patch.object(scanner.docker_token_manager, 'get_next_config') as next_config, \
|
|
mock.patch.object(
|
|
scanner, 'scan_docker_image',
|
|
return_value={'findings': [], 'errors': []},
|
|
) as docker_scan:
|
|
result = scanner.scan_target_result(
|
|
DOCKER_TARGET, 'docker', 'fixture-event', {},
|
|
)
|
|
assert not result['errors']
|
|
next_config.assert_not_called()
|
|
assert docker_scan.call_args.kwargs['config_dir'] is None
|
|
|
|
with scanner.client_remote_execution_binding('huggingface_space_v1'):
|
|
with pytest.raises(RuntimeError, match='cannot use a token'):
|
|
scanner.scan_huggingface_space('Owner/Space', token='private-token')
|
|
finally:
|
|
scanner._client_scan_manifest.reset(manifest_token)
|
|
|
|
|
|
def test_anonymous_docker_bearer_never_reads_account_pool(monkeypatch):
|
|
response = mock.Mock(status_code=200, headers={})
|
|
response.iter_content.return_value = [b'{"token":"anonymous-bearer"}']
|
|
monkeypatch.setattr(scanner, 'api_request', mock.Mock(return_value=response))
|
|
monkeypatch.setattr(
|
|
scanner.docker_token_manager, 'has_accounts',
|
|
mock.Mock(side_effect=AssertionError('anonymous path must not read account pool')),
|
|
)
|
|
auth = scanner.docker_registry_bearer_token(
|
|
'Bearer realm="https://auth.docker.io/token",service="registry.docker.io"',
|
|
'library/alpine', anonymous_only=True,
|
|
)
|
|
assert auth.token == 'anonymous-bearer'
|
|
assert auth.account_name == ''
|
|
|
|
|
|
def test_direct_anonymous_docker_auth_failure_is_permanent(monkeypatch):
|
|
monkeypatch.setattr(
|
|
scanner, 'resolve_docker_content_manifest',
|
|
mock.Mock(side_effect=scanner.DockerRemoteAccessError(
|
|
'private provider detail', status='auth_failed', remote_attempted=True,
|
|
)),
|
|
)
|
|
result = scanner._recover_docker_image_contents(
|
|
DOCKER_TARGET, scanner.time.monotonic() + 30, None, None, 0,
|
|
None, None, False, None, anonymous_public_client=True,
|
|
)
|
|
assert result['error_class'] == 'docker_registry_access'
|
|
assert result['retryable'] is False
|
|
assert not result.get('source_failure')
|
|
assert 'private provider detail' not in json.dumps(result)
|
|
|
|
|
|
def test_huggingface_discovery_errors_never_include_response_body(monkeypatch):
|
|
secret = 'provider-response-secret-must-not-appear'
|
|
for return_metadata in (False, True):
|
|
response = _json_response(403, {'error': secret})
|
|
monkeypatch.setattr(scanner, 'api_request', mock.Mock(return_value=response))
|
|
with pytest.raises(scanner.RateLimitError) as captured:
|
|
scanner.fetch_huggingface_spaces(
|
|
pages=1, token='discovery-token',
|
|
return_metadata=return_metadata,
|
|
)
|
|
assert secret not in str(captured.value)
|
|
response.close.assert_called_once()
|
|
response.iter_content.assert_not_called()
|
|
|
|
|
|
def test_huggingface_missing_repository_is_permanent_and_not_retryable():
|
|
diagnostic = json.dumps({
|
|
'level': 'error', 'msg': 'failed to enumerate source',
|
|
'error': 'no repo found for repo',
|
|
})
|
|
result = scanner.apply_trufflehog_diagnostics(
|
|
{'findings': [], 'errors': []}, diagnostic, 1, 'huggingface',
|
|
)
|
|
assert result['skipped'] == 'HuggingFace Space repository is unavailable'
|
|
assert result['error_class'] == 'huggingface_no_repo'
|
|
assert result['retryable'] is False
|
|
assert result['errors'] == []
|
|
|
|
|
|
@pytest.mark.parametrize('planning_kind,source,detail,error_class,skipped', [
|
|
(
|
|
'huggingface_space_v1', 'huggingface', 'permission denied',
|
|
'huggingface_inaccessible', 'HuggingFace Space repository is unavailable',
|
|
),
|
|
(
|
|
'huggingface_space_v1', 'huggingface', 'HTTP 401 unauthorized',
|
|
'huggingface_inaccessible', 'HuggingFace Space repository is unavailable',
|
|
),
|
|
(
|
|
'huggingface_space_v1', 'huggingface', 'invalid API key',
|
|
'huggingface_inaccessible', 'HuggingFace Space repository is unavailable',
|
|
),
|
|
(
|
|
'docker_direct_v1', 'docker', 'pull access denied',
|
|
'docker_registry_access', 'Docker image is unavailable to the worker',
|
|
),
|
|
(
|
|
'docker_direct_v1', 'docker', 'manifest unknown: HTTP 404',
|
|
'docker_registry_access', 'Docker image is unavailable to the worker',
|
|
),
|
|
])
|
|
def test_direct_provider_access_failures_are_permanent(
|
|
planning_kind, source, detail, error_class, skipped,
|
|
):
|
|
manifest_token = scanner._client_scan_manifest.set({
|
|
'executables': {'git': {'path': sys.executable}},
|
|
})
|
|
try:
|
|
diagnostic = json.dumps({
|
|
'level': 'error', 'msg': 'provider access failed', 'error': detail,
|
|
})
|
|
with scanner.client_remote_execution_binding(planning_kind):
|
|
result = scanner.apply_trufflehog_diagnostics(
|
|
{'findings': [], 'errors': []}, diagnostic, 1, source,
|
|
)
|
|
finally:
|
|
scanner._client_scan_manifest.reset(manifest_token)
|
|
|
|
assert result['skipped'] == skipped
|
|
assert result['error_class'] == error_class
|
|
assert result['retryable'] is False
|
|
assert result['errors'] == []
|