diff --git a/scripts/url_probe.py b/scripts/url_probe.py index fcb90e0..7425b74 100644 --- a/scripts/url_probe.py +++ b/scripts/url_probe.py @@ -94,10 +94,7 @@ def validate_public_url(url: str) -> tuple[str, int, str]: raise UnsafeUrlError("port zero is not allowed") port = explicit_port or (443 if scheme == "https" else 80) - try: - results = socket.getaddrinfo(hostname, port, type=socket.SOCK_STREAM) - except OSError as error: - raise UnsafeUrlError("hostname did not resolve") from error + results = socket.getaddrinfo(hostname, port, type=socket.SOCK_STREAM) addresses: list[str] = [] for _family, _socktype, _protocol, _canonname, sockaddr in results: @@ -349,5 +346,15 @@ def probe_url( (), f"transport error: {last_transport_error}", ) + except UnsafeUrlError as error: + prefix = "unsafe URL" if current_url == requested_url else "unsafe redirect target" + return _http_error( + requested_url, + current_url, + None, + attempt, + redirects, + f"{prefix}: {error}", + ) raise AssertionError("retry loop exhausted unexpectedly") diff --git a/tests/test_url_probe.py b/tests/test_url_probe.py index 0a66e9d..712293a 100644 --- a/tests/test_url_probe.py +++ b/tests/test_url_probe.py @@ -106,6 +106,43 @@ class UrlProbeTests(unittest.TestCase): ): 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(