sashatrask
298 lines
14 KiB
Python
298 lines
14 KiB
Python
import io
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import time
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
import requests
|
|
from urllib3.response import HTTPResponse
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'app'))
|
|
import scanner
|
|
|
|
|
|
@pytest.fixture
|
|
def transport(monkeypatch):
|
|
state = SimpleNamespace(replies=[], calls=[], responses=[])
|
|
|
|
def send(adapter, request, **kwargs):
|
|
state.calls.append((request, kwargs))
|
|
assert state.replies, 'unexpected offline HTTP request'
|
|
reply = state.replies.pop(0)
|
|
if isinstance(reply, Exception):
|
|
raise reply
|
|
status, headers, body = reply
|
|
response = requests.Response()
|
|
response.status_code = status
|
|
response.headers.update(headers)
|
|
response.url = request.url
|
|
response.request = request
|
|
response.raw = HTTPResponse(io.BytesIO(body), preload_content=False)
|
|
state.responses.append(response)
|
|
return response
|
|
|
|
monkeypatch.setattr(requests.adapters.HTTPAdapter, 'send', send)
|
|
for key in ('http_proxy', 'https_proxy', 'all_proxy', 'HTTP_PROXY', 'HTTPS_PROXY', 'ALL_PROXY'):
|
|
monkeypatch.setenv(key, 'http://ambient.invalid:8080')
|
|
monkeypatch.setenv('NO_PROXY', '')
|
|
monkeypatch.setenv('no_proxy', '')
|
|
monkeypatch.setenv('NETRC', str(Path(__file__).with_name('missing-netrc')))
|
|
return state
|
|
|
|
|
|
@pytest.fixture
|
|
def proxy_file(tmp_path, monkeypatch):
|
|
path = tmp_path / 'routing-proxies.txt'
|
|
path.write_text('http://configured.invalid:8080\n', encoding='ascii')
|
|
monkeypatch.setattr(scanner.scan_config, 'api_proxy_enabled', True)
|
|
monkeypatch.setattr(scanner.scan_config, 'api_proxy_file', str(path))
|
|
monkeypatch.setattr(scanner.scan_config, 'api_proxy_max_retries', 2)
|
|
monkeypatch.setattr(scanner.scan_config, 'api_proxy_retry_delay', 0)
|
|
monkeypatch.setattr(scanner.scan_config, 'api_proxy_timeout', 99)
|
|
for name, value in (('path', None), ('mtime', None), ('entries', []), ('index', 0)):
|
|
monkeypatch.setattr(scanner, '_api_proxy_cache_' + name, value)
|
|
return path
|
|
|
|
|
|
@pytest.mark.parametrize('scheme', ['http', 'https'])
|
|
@pytest.mark.parametrize('stream', [False, True])
|
|
def test_direct_ignores_ambient_and_caller_proxies_across_redirects(transport, proxy_file, scheme, stream):
|
|
transport.replies = [
|
|
(302, {'Location': 'https://second.invalid/payload'}, b''),
|
|
(307, {'Location': 'http://third.invalid/payload'}, b''),
|
|
(200, {}, b'complete artifact'),
|
|
]
|
|
environment = dict(os.environ)
|
|
with mock.patch.object(scanner, 'next_api_proxy', side_effect=AssertionError('direct selected proxy')):
|
|
response = scanner.api_request(
|
|
'GET', scheme + '://first.invalid/payload', use_proxy=False,
|
|
timeout=(3, 7), stream=stream,
|
|
proxies={'https://second.invalid': 'http://caller.invalid:8080'},
|
|
)
|
|
assert len(response.history) == 2
|
|
if stream:
|
|
assert not response._content_consumed
|
|
assert not response.raw.closed
|
|
assert b''.join(response.iter_content(3)) == b'complete artifact'
|
|
else:
|
|
assert response.content == b'complete artifact'
|
|
response.close()
|
|
for request, options in transport.calls:
|
|
assert requests.utils.select_proxy(request.url, options['proxies']) is None
|
|
assert options['proxies'] == {'no_proxy': '*'}
|
|
assert options['timeout'] == (3, 7)
|
|
assert 'Proxy-Authorization' not in request.headers
|
|
assert dict(os.environ) == environment
|
|
|
|
|
|
@pytest.mark.parametrize('verify,expected', [(None, 'fixture-ca.pem'), (True, 'fixture-ca.pem'), (False, False), ('explicit.pem', 'explicit.pem')])
|
|
def test_direct_preserves_requests_certificate_merge(transport, monkeypatch, verify, expected):
|
|
monkeypatch.setenv('REQUESTS_CA_BUNDLE', 'fixture-ca.pem')
|
|
transport.replies = [(200, {}, b'ok')]
|
|
with scanner._direct_request('GET', 'https://fixture.invalid', verify=verify, cert=('cert.pem', 'key.pem')):
|
|
pass
|
|
options = transport.calls[0][1]
|
|
assert options['verify'] == expected
|
|
assert options['cert'] == ('cert.pem', 'key.pem')
|
|
|
|
|
|
@pytest.mark.parametrize('missing', [False, True])
|
|
def test_configured_proxy_file_failure_never_falls_back_direct(transport, proxy_file, missing):
|
|
if missing:
|
|
proxy_file.unlink()
|
|
else:
|
|
proxy_file.write_text('# empty\n', encoding='ascii')
|
|
with pytest.raises(scanner.ApiRequestError, match='no valid proxies'):
|
|
scanner.api_request('GET', 'https://fixture.invalid')
|
|
assert not transport.calls
|
|
transport.replies = [(200, {}, b'direct')]
|
|
with scanner.api_request('GET', 'https://fixture.invalid', use_proxy=False) as response:
|
|
assert response.content == b'direct'
|
|
|
|
|
|
def test_proxy_transport_failure_retries_file_proxy_without_direct_fallback(transport, proxy_file):
|
|
transport.replies = [requests.exceptions.ProxyError('offline fixture')] * 2
|
|
with pytest.raises(scanner.ApiRequestError, match='after 2 attempt'):
|
|
scanner.api_request('GET', 'https://fixture.invalid')
|
|
assert len(transport.calls) == 2
|
|
for request, options in transport.calls:
|
|
assert requests.utils.select_proxy(request.url, options['proxies']) == 'http://configured.invalid:8080'
|
|
|
|
|
|
def test_metadata_file_proxy_survives_worker_no_proxy_and_redirects(transport, proxy_file, monkeypatch):
|
|
monkeypatch.setenv('NO_PROXY', '*')
|
|
monkeypatch.setenv('no_proxy', '*')
|
|
transport.replies = [
|
|
(302, {'Location': 'https://second.invalid/metadata'}, b''),
|
|
(200, {}, b'{}'),
|
|
]
|
|
with scanner.api_request('GET', 'https://first.invalid/metadata') as response:
|
|
assert response.content == b'{}'
|
|
assert len(transport.calls) == 2
|
|
for request, options in transport.calls:
|
|
assert requests.utils.select_proxy(request.url, options['proxies']) == 'http://configured.invalid:8080'
|
|
|
|
|
|
@pytest.mark.parametrize('timeout,expected', [
|
|
(30, (5, 30)),
|
|
(15, (5, 15)),
|
|
(3, (3, 3)),
|
|
((2, 30), (2, 30)),
|
|
((12, 7), (5, 7)),
|
|
([9, 15], (5, 15)),
|
|
((None, 30), (5, 30)),
|
|
((2, None), (2, None)),
|
|
(None, (5, 5)),
|
|
])
|
|
def test_proxy_connect_cap_keeps_caller_read_timeout(transport, proxy_file, monkeypatch, timeout, expected):
|
|
monkeypatch.setattr(scanner.scan_config, 'api_proxy_timeout', 5)
|
|
transport.replies = [(200, {}, b'{}')]
|
|
with scanner.api_request('GET', 'https://fixture.invalid/metadata', timeout=timeout):
|
|
pass
|
|
request, options = transport.calls[0]
|
|
assert options['timeout'] == expected
|
|
assert requests.utils.select_proxy(request.url, options['proxies']) == 'http://configured.invalid:8080'
|
|
|
|
|
|
def test_proxy_retry_budget_is_shared_and_clamps_read_timeout(transport, proxy_file, monkeypatch):
|
|
clock = [0.0]
|
|
waits = []
|
|
calls = []
|
|
monkeypatch.setattr(scanner.scan_config, 'api_proxy_timeout', 5)
|
|
monkeypatch.setattr(scanner.time, 'monotonic', lambda: clock[0])
|
|
|
|
def request(*args, **kwargs):
|
|
calls.append(kwargs)
|
|
clock[0] += 7
|
|
raise requests.exceptions.ReadTimeout('offline fixture')
|
|
|
|
def wait(seconds):
|
|
waits.append(seconds)
|
|
clock[0] += seconds
|
|
|
|
monkeypatch.setattr(scanner.requests, 'request', request)
|
|
monkeypatch.setattr(scanner, '_wait_or_raise_scan_slot_fatal', wait)
|
|
with pytest.raises(scanner.ApiRequestError, match='deadline expired during retry'):
|
|
scanner.api_request(
|
|
'GET', 'https://fixture.invalid/metadata', timeout=30,
|
|
max_retries=3, retry_delay=5, deadline=20,
|
|
)
|
|
assert [call['timeout'] for call in calls] == [(5, 20), (5, 8)]
|
|
assert all(call['proxies']['https'] == 'http://configured.invalid:8080' for call in calls)
|
|
assert waits == [5]
|
|
|
|
|
|
def test_direct_retries_deadline_and_response_cleanup(transport, proxy_file):
|
|
transport.replies = [(503, {}, b''), requests.exceptions.ConnectionError('offline'), (200, {}, b'ok')]
|
|
with mock.patch.object(scanner, 'next_api_proxy', side_effect=AssertionError('direct selected proxy')):
|
|
response = scanner.api_request(
|
|
'GET', 'https://fixture.invalid', use_proxy=False, stream=True,
|
|
timeout=(2, 4), max_retries=3, retry_delay=0, deadline=time.monotonic() + 1,
|
|
)
|
|
assert transport.responses[0].raw.closed
|
|
assert not response.raw.closed
|
|
response.close()
|
|
assert len(transport.calls) == 3
|
|
assert all(0 < value <= 1 for _, options in transport.calls for value in options['timeout'])
|
|
|
|
|
|
def test_direct_cancellation_closes_response_and_expired_deadline_sends_nothing(transport):
|
|
transport.replies = [(200, {}, b'ok')]
|
|
with mock.patch.object(scanner, '_raise_if_scan_slot_fatal', side_effect=[None, scanner.ScanSlotFatalError('offline')]), \
|
|
mock.patch.object(scanner._scan_slot_fatal_event, 'is_set', return_value=True):
|
|
with pytest.raises(scanner.ScanSlotFatalError):
|
|
scanner.api_request('GET', 'https://fixture.invalid', use_proxy=False, stream=True)
|
|
assert transport.responses[0].raw.closed
|
|
with pytest.raises(scanner.ApiRequestError, match='deadline expired'):
|
|
scanner.api_request('GET', 'https://fixture.invalid', use_proxy=False, deadline=time.monotonic() - 1)
|
|
assert len(transport.calls) == 1
|
|
|
|
|
|
@pytest.mark.parametrize('blob', [False, True])
|
|
def test_registry_metadata_uses_file_proxy_but_blob_and_401_retry_are_direct(transport, proxy_file, monkeypatch, blob):
|
|
challenge = 'Bearer realm="https://auth.docker.io/token",service="registry.docker.io"'
|
|
transport.replies = [
|
|
(401, {'WWW-Authenticate': challenge}, b''),
|
|
(200, {}, b'{"token":"synthetic-token"}'),
|
|
(200, {}, b'{"schemaVersion":2}'),
|
|
]
|
|
monkeypatch.setattr(scanner, 'docker_token_manager', scanner.DockerTokenManager())
|
|
monkeypatch.setattr(scanner, '_require_docker_blob_url', lambda *a, **k: None)
|
|
deadline = time.monotonic() + 10
|
|
digest = 'sha256:' + 'a' * 64
|
|
if blob:
|
|
response, _ = scanner._docker_registry_blob_response('fixture/repo', digest, None, deadline)
|
|
assert not response.raw.closed
|
|
response.close()
|
|
else:
|
|
scanner.docker_registry_manifest('fixture/repo', digest, deadline=deadline)
|
|
assert len(transport.calls) == 3
|
|
for index, (request, options) in enumerate(transport.calls):
|
|
selected = requests.utils.select_proxy(request.url, options['proxies'])
|
|
assert selected == (None if blob and index != 1 else 'http://configured.invalid:8080')
|
|
assert options['stream'] is True
|
|
|
|
|
|
def test_github_pool_forwards_content_optout_but_metadata_keeps_proxy(transport, proxy_file):
|
|
transport.replies = [(200, {}, b'{"size":2,"encoding":"base64","content":"e30="}'), (200, {}, b'[]')]
|
|
pool = scanner.GitHubTokenPool(token='synthetic-token')
|
|
assert scanner.github_content_bytes({'url': 'https://fixture.invalid/content'}, pool) == b'{}'
|
|
with pool.request('GET', 'https://fixture.invalid/commits'):
|
|
pass
|
|
assert [requests.utils.select_proxy(req.url, options['proxies']) for req, options in transport.calls] == [None, 'http://configured.invalid:8080']
|
|
|
|
|
|
def test_ci_download_direct_redirects_strip_credentials(transport, proxy_file, tmp_path, monkeypatch):
|
|
transport.replies = [
|
|
(302, {'Location': 'https://other.invalid/artifact'}, b''),
|
|
(200, {}, b'zip fixture'),
|
|
]
|
|
monkeypatch.setattr(scanner, 'require_private_directory', lambda *a, **k: None)
|
|
monkeypatch.setattr(scanner, 'reject_reparse_components', lambda *a: None)
|
|
monkeypatch.setattr(scanner, 'harden_private_file', lambda *a: None)
|
|
monkeypatch.setattr(scanner, 'private_file_ready', lambda *a: True)
|
|
monkeypatch.setattr(scanner, 'durable_replace', os.replace)
|
|
destination = tmp_path / 'artifact'
|
|
outcome = scanner.download_to_file(
|
|
'https://fixture.invalid/artifact', str(destination), timeout=7,
|
|
headers={'Authorization': 'synthetic-token', 'Private-Token': 'synthetic-token'},
|
|
)
|
|
assert outcome.ok and destination.read_bytes() == b'zip fixture'
|
|
assert len(transport.calls) == 2
|
|
for request, options in transport.calls:
|
|
assert requests.utils.select_proxy(request.url, options['proxies']) is None
|
|
assert options['timeout'] == 7
|
|
assert 'Authorization' not in transport.calls[1][0].headers
|
|
assert 'Private-Token' not in transport.calls[1][0].headers
|
|
assert all(response.raw.closed for response in transport.responses)
|
|
|
|
|
|
@pytest.mark.parametrize('wrapper', [scanner.github_api_get, scanner.gitlab_api_get])
|
|
def test_ci_metadata_wrappers_keep_configured_proxy_even_when_streamed(transport, proxy_file, wrapper):
|
|
transport.replies = [(200, {}, b'{}')]
|
|
with wrapper('https://fixture.invalid/metadata', stream=True):
|
|
pass
|
|
request, options = transport.calls[0]
|
|
assert requests.utils.select_proxy(request.url, options['proxies']) == 'http://configured.invalid:8080'
|
|
assert options['stream'] is True
|
|
|
|
|
|
def test_pypi_bulk_index_is_direct_without_touching_a_database(transport, proxy_file, tmp_path, monkeypatch):
|
|
path = tmp_path / 'mock-index'
|
|
path.touch()
|
|
connection = mock.Mock()
|
|
connection.execute.return_value.fetchone.side_effect = [None, (1,)]
|
|
monkeypatch.setattr(scanner, '_pypi_index_path', lambda: str(path))
|
|
monkeypatch.setattr(scanner.sqlite3, 'connect', mock.Mock(return_value=connection))
|
|
monkeypatch.setattr(scanner, 'harden_private_file', lambda *a: None)
|
|
transport.replies = [(200, {}, b'<a href="/simple/fixture/">fixture</a>')]
|
|
assert scanner.load_pypi_project_index(request_timeout=7) == str(path)
|
|
request, options = transport.calls[0]
|
|
assert requests.utils.select_proxy(request.url, options['proxies']) is None
|
|
assert options['timeout'] == 7 and options['stream'] is True
|
|
assert transport.responses[0].raw.closed
|
|
connection.close.assert_called_once()
|