mirror of
https://github.com/wilsonfreitas/awesome-quant.git
synced 2026-11-02 04:16:10 +00:00
329 lines
13 KiB
Python
329 lines
13 KiB
Python
import http.client
|
|
import socket
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from scripts.url_probe import (
|
|
Outcome,
|
|
RawResponse,
|
|
UnsafeUrlError,
|
|
probe_url,
|
|
request_once,
|
|
validate_public_url,
|
|
)
|
|
|
|
|
|
PUBLIC_URL = "https://example.test/start"
|
|
|
|
|
|
def public_validation(url: str):
|
|
return "example.test", 443, "8.8.8.8"
|
|
|
|
|
|
class SequenceRequester:
|
|
def __init__(self, responses):
|
|
self.responses = iter(responses)
|
|
self.urls = []
|
|
|
|
def __call__(self, url, timeout):
|
|
self.urls.append((url, timeout))
|
|
result = next(self.responses)
|
|
if isinstance(result, BaseException):
|
|
raise result
|
|
return result
|
|
|
|
|
|
class UrlProbeTests(unittest.TestCase):
|
|
def test_changed_301_canonical_target_is_a_permanent_redirect(self):
|
|
requester = SequenceRequester(
|
|
[
|
|
RawResponse(301, "https://www.example.test/guide/"),
|
|
RawResponse(200, None),
|
|
]
|
|
)
|
|
|
|
with patch("scripts.url_probe.validate_public_url", side_effect=public_validation):
|
|
observation = probe_url(PUBLIC_URL, requester=requester, sleeper=lambda _: None)
|
|
|
|
self.assertEqual(observation.outcome, Outcome.PERMANENT_REDIRECT)
|
|
self.assertEqual(observation.final_url, "https://www.example.test/guide/")
|
|
self.assertEqual(observation.status, 200)
|
|
self.assertEqual(observation.attempts, 1)
|
|
|
|
def test_retryable_http_status_uses_all_three_attempts(self):
|
|
requester = SequenceRequester([RawResponse(503, None)] * 3)
|
|
delays = []
|
|
|
|
with patch("scripts.url_probe.validate_public_url", side_effect=public_validation):
|
|
observation = probe_url(
|
|
PUBLIC_URL, requester=requester, sleeper=delays.append
|
|
)
|
|
|
|
self.assertEqual(observation.outcome, Outcome.TRANSIENT)
|
|
self.assertEqual(observation.status, 503)
|
|
self.assertEqual(observation.attempts, 3)
|
|
self.assertEqual(delays, [0, 1, 2])
|
|
|
|
def test_terminal_statuses_are_classified_without_retry(self):
|
|
cases = [
|
|
(404, Outcome.DEAD),
|
|
(410, Outcome.DEAD),
|
|
(401, Outcome.RESTRICTED),
|
|
(403, Outcome.RESTRICTED),
|
|
(407, Outcome.RESTRICTED),
|
|
(429, Outcome.RESTRICTED),
|
|
(451, Outcome.RESTRICTED),
|
|
(418, Outcome.HTTP_ERROR),
|
|
]
|
|
|
|
with patch("scripts.url_probe.validate_public_url", side_effect=public_validation):
|
|
for status, expected in cases:
|
|
with self.subTest(status=status):
|
|
observation = probe_url(
|
|
PUBLIC_URL,
|
|
requester=SequenceRequester([RawResponse(status, None)]),
|
|
sleeper=lambda _: None,
|
|
)
|
|
self.assertEqual(observation.outcome, expected)
|
|
self.assertEqual(observation.attempts, 1)
|
|
|
|
def test_temporary_redirect_to_success_is_ok(self):
|
|
requester = SequenceRequester(
|
|
[RawResponse(302, "/next"), RawResponse(200, None)]
|
|
)
|
|
|
|
with patch("scripts.url_probe.validate_public_url", side_effect=public_validation):
|
|
observation = probe_url(PUBLIC_URL, requester=requester, sleeper=lambda _: None)
|
|
|
|
self.assertEqual(observation.outcome, Outcome.OK)
|
|
self.assertEqual(observation.final_url, "https://example.test/next")
|
|
|
|
def test_private_dns_answer_is_rejected(self):
|
|
answers = [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 443))]
|
|
|
|
with patch("scripts.url_probe.socket.getaddrinfo", return_value=answers):
|
|
with self.assertRaisesRegex(
|
|
UnsafeUrlError, "resolved address is not globally routable"
|
|
):
|
|
validate_public_url(PUBLIC_URL)
|
|
|
|
def test_dns_resolution_failures_are_retried_as_transient(self):
|
|
delays = []
|
|
|
|
with patch(
|
|
"scripts.url_probe.socket.getaddrinfo",
|
|
side_effect=socket.gaierror("no dns"),
|
|
) as getaddrinfo:
|
|
observation = probe_url(PUBLIC_URL, sleeper=delays.append)
|
|
|
|
self.assertEqual(observation.outcome, Outcome.TRANSIENT)
|
|
self.assertEqual(observation.attempts, 3)
|
|
self.assertEqual(observation.error, "transport error: gaierror")
|
|
self.assertEqual(delays, [0, 1, 2])
|
|
self.assertEqual(getaddrinfo.call_count, 3)
|
|
|
|
def test_default_requester_revalidation_failure_is_an_http_error(self):
|
|
public_answer = [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", 443))
|
|
]
|
|
private_answer = [
|
|
(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 443))
|
|
]
|
|
|
|
with patch(
|
|
"scripts.url_probe.socket.getaddrinfo",
|
|
side_effect=[public_answer, private_answer],
|
|
) as getaddrinfo:
|
|
observation = probe_url(PUBLIC_URL, sleeper=lambda _: None)
|
|
|
|
self.assertEqual(observation.outcome, Outcome.HTTP_ERROR)
|
|
self.assertEqual(observation.attempts, 1)
|
|
self.assertEqual(
|
|
observation.error,
|
|
"unsafe URL: resolved address is not globally routable",
|
|
)
|
|
self.assertEqual(getaddrinfo.call_count, 2)
|
|
|
|
def test_missing_redirect_location_is_an_http_error(self):
|
|
with patch("scripts.url_probe.validate_public_url", side_effect=public_validation):
|
|
observation = probe_url(
|
|
PUBLIC_URL,
|
|
requester=SequenceRequester([RawResponse(302, None)]),
|
|
sleeper=lambda _: None,
|
|
)
|
|
|
|
self.assertEqual(observation.outcome, Outcome.HTTP_ERROR)
|
|
self.assertEqual(observation.error, "missing redirect location")
|
|
|
|
def test_redirect_loop_and_limit_are_http_errors(self):
|
|
with patch("scripts.url_probe.validate_public_url", side_effect=public_validation):
|
|
loop = probe_url(
|
|
PUBLIC_URL,
|
|
requester=SequenceRequester([RawResponse(302, "/start")]),
|
|
sleeper=lambda _: None,
|
|
)
|
|
overflow = probe_url(
|
|
PUBLIC_URL,
|
|
requester=SequenceRequester([RawResponse(302, f"/hop-{index}") for index in range(6)]),
|
|
sleeper=lambda _: None,
|
|
)
|
|
|
|
self.assertEqual(loop.outcome, Outcome.HTTP_ERROR)
|
|
self.assertEqual(loop.error, "redirect loop")
|
|
self.assertEqual(overflow.outcome, Outcome.HTTP_ERROR)
|
|
self.assertEqual(overflow.error, "redirect limit exceeded")
|
|
|
|
def test_transport_failures_use_all_three_attempts(self):
|
|
requester = SequenceRequester(
|
|
[socket.gaierror("no dns"), OSError("reset"), TimeoutError("late")]
|
|
)
|
|
delays = []
|
|
|
|
with patch("scripts.url_probe.validate_public_url", side_effect=public_validation):
|
|
observation = probe_url(
|
|
PUBLIC_URL, requester=requester, sleeper=delays.append
|
|
)
|
|
|
|
self.assertEqual(observation.outcome, Outcome.TRANSIENT)
|
|
self.assertEqual(observation.attempts, 3)
|
|
self.assertEqual(delays, [0, 1, 2])
|
|
self.assertEqual(observation.error, "transport error: TimeoutError")
|
|
|
|
def test_invalid_request_url_is_an_http_error_without_retry(self):
|
|
requester = SequenceRequester(
|
|
[http.client.InvalidURL("URL can't contain control characters")]
|
|
)
|
|
delays = []
|
|
|
|
with patch("scripts.url_probe.validate_public_url", side_effect=public_validation):
|
|
observation = probe_url(
|
|
"https://example.test/a path",
|
|
requester=requester,
|
|
sleeper=delays.append,
|
|
)
|
|
|
|
self.assertEqual(observation.outcome, Outcome.HTTP_ERROR)
|
|
self.assertEqual(observation.attempts, 1)
|
|
self.assertEqual(observation.error, "invalid request URL")
|
|
self.assertEqual(delays, [0])
|
|
self.assertEqual(len(requester.urls), 1)
|
|
|
|
def test_malformed_redirect_location_is_an_http_error_without_retry(self):
|
|
requester = SequenceRequester([RawResponse(302, "http://[::1")])
|
|
delays = []
|
|
|
|
with patch("scripts.url_probe.validate_public_url", side_effect=public_validation):
|
|
observation = probe_url(
|
|
PUBLIC_URL,
|
|
requester=requester,
|
|
sleeper=delays.append,
|
|
)
|
|
|
|
self.assertEqual(observation.outcome, Outcome.HTTP_ERROR)
|
|
self.assertEqual(observation.status, 302)
|
|
self.assertEqual(observation.attempts, 1)
|
|
self.assertEqual(observation.error, "invalid redirect location")
|
|
self.assertEqual(delays, [0])
|
|
self.assertEqual(len(requester.urls), 1)
|
|
|
|
def test_multicast_resolved_addresses_are_rejected(self):
|
|
for family, address in (
|
|
(socket.AF_INET, "224.0.0.1"),
|
|
(socket.AF_INET6, "ff0e::1"),
|
|
):
|
|
with self.subTest(address=address):
|
|
answers = [(family, socket.SOCK_STREAM, 6, "", (address, 443))]
|
|
with patch("scripts.url_probe.socket.getaddrinfo", return_value=answers):
|
|
with self.assertRaisesRegex(
|
|
UnsafeUrlError, "resolved address is not globally routable"
|
|
):
|
|
validate_public_url(PUBLIC_URL)
|
|
|
|
def test_non_public_embedded_ipv4_addresses_are_rejected(self):
|
|
for address in ("64:ff9b::127.0.0.1", "2002:7f00:1::"):
|
|
with self.subTest(address=address):
|
|
answers = [
|
|
(socket.AF_INET6, socket.SOCK_STREAM, 6, "", (address, 443))
|
|
]
|
|
with patch("scripts.url_probe.socket.getaddrinfo", return_value=answers):
|
|
with self.assertRaisesRegex(
|
|
UnsafeUrlError, "resolved address is not globally routable"
|
|
):
|
|
validate_public_url(PUBLIC_URL)
|
|
|
|
def test_public_unicast_and_public_nat64_addresses_are_allowed(self):
|
|
for family, address, expected in (
|
|
(socket.AF_INET, "8.8.8.8", "8.8.8.8"),
|
|
(socket.AF_INET6, "2606:4700:4700::1111", "2606:4700:4700::1111"),
|
|
(socket.AF_INET6, "64:ff9b::8.8.8.8", "64:ff9b::808:808"),
|
|
):
|
|
with self.subTest(address=address):
|
|
answers = [(family, socket.SOCK_STREAM, 6, "", (address, 443))]
|
|
with patch("scripts.url_probe.socket.getaddrinfo", return_value=answers):
|
|
self.assertEqual(
|
|
validate_public_url(PUBLIC_URL),
|
|
("example.test", 443, expected),
|
|
)
|
|
|
|
def test_request_once_falls_back_to_get_for_unsupported_head(self):
|
|
calls = []
|
|
|
|
class Response:
|
|
def __init__(self, status):
|
|
self.status = status
|
|
|
|
def getheader(self, name):
|
|
return "/redirect" if name == "Location" else None
|
|
|
|
def close(self):
|
|
calls.append("response-close")
|
|
|
|
class Connection:
|
|
def __init__(self, *args, **kwargs):
|
|
calls.append(("connect", args, kwargs))
|
|
|
|
def request(self, method, target, headers):
|
|
calls.append((method, target, headers))
|
|
|
|
def getresponse(self):
|
|
return Response(405 if len([call for call in calls if isinstance(call, tuple) and call[0] in {"HEAD", "GET"}]) == 1 else 200)
|
|
|
|
def close(self):
|
|
calls.append("connection-close")
|
|
|
|
with (
|
|
patch("scripts.url_probe.validate_public_url", return_value=("example.test", 443, "8.8.8.8")),
|
|
patch("scripts.url_probe._PinnedHTTPSConnection", Connection),
|
|
):
|
|
response = request_once("https://example.test/path?x=1", timeout=7)
|
|
|
|
self.assertEqual(response, RawResponse(200, "/redirect"))
|
|
self.assertEqual([call[0] for call in calls if isinstance(call, tuple) and call[0] in {"HEAD", "GET"}], ["HEAD", "GET"])
|
|
request = next(call for call in calls if isinstance(call, tuple) and call[0] == "GET")
|
|
self.assertEqual(request[1], "/path?x=1")
|
|
self.assertEqual(request[2]["Host"], "example.test")
|
|
self.assertEqual(request[2]["User-Agent"], "awesome-quant-url-audit/1.0")
|
|
|
|
def test_public_redirect_to_private_target_is_an_http_error(self):
|
|
def validate(url):
|
|
if url == "https://example.test/private":
|
|
raise UnsafeUrlError("resolved address is not globally routable")
|
|
return public_validation(url)
|
|
|
|
with patch("scripts.url_probe.validate_public_url", side_effect=validate):
|
|
observation = probe_url(
|
|
PUBLIC_URL,
|
|
requester=SequenceRequester([RawResponse(302, "/private")]),
|
|
sleeper=lambda _: None,
|
|
)
|
|
|
|
self.assertEqual(observation.outcome, Outcome.HTTP_ERROR)
|
|
self.assertEqual(
|
|
observation.error,
|
|
"unsafe redirect target: resolved address is not globally routable",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|