diff --git a/scripts/url_probe.py b/scripts/url_probe.py
new file mode 100644
index 0000000..fcb90e0
--- /dev/null
+++ b/scripts/url_probe.py
@@ -0,0 +1,353 @@
+"""Safe, retrying HTTP URL classification for README audit tooling."""
+
+from __future__ import annotations
+
+import http.client
+import ipaddress
+import socket
+import ssl
+from dataclasses import dataclass
+from enum import StrEnum
+from time import sleep
+from typing import Callable
+from urllib.parse import SplitResult, urljoin, urlsplit, urlunsplit
+
+
+USER_AGENT = "awesome-quant-url-audit/1.0"
+RETRYABLE_STATUSES = {408, 425, 500, 502, 503, 504}
+REDIRECT_STATUSES = {300, 301, 302, 303, 307, 308}
+PERMANENT_REDIRECT_STATUSES = {301, 308}
+TRANSPORT_ERRORS = (OSError, ssl.SSLError, socket.gaierror, TimeoutError)
+
+
+class Outcome(StrEnum):
+    OK = "ok"
+    DEAD = "dead"
+    RESTRICTED = "restricted"
+    TRANSIENT = "transient"
+    HTTP_ERROR = "http_error"
+    PERMANENT_REDIRECT = "permanent_redirect"
+
+
+@dataclass(frozen=True)
+class RawResponse:
+    status: int
+    location: str | None
+
+
+@dataclass(frozen=True)
+class UrlObservation:
+    requested_url: str
+    final_url: str
+    status: int | None
+    outcome: Outcome
+    attempts: int
+    redirect_chain: tuple[str, ...] = ()
+    error: str = ""
+
+
+class UnsafeUrlError(ValueError):
+    """Raised before connecting to a URL that is not publicly routable."""
+
+
+class _PinnedHTTPSConnection(http.client.HTTPSConnection):
+    """HTTPS connection whose TCP peer is pinned while SNI remains the hostname."""
+
+    def __init__(self, host: str, port: int, server_hostname: str, timeout: float):
+        super().__init__(host, port, timeout=timeout, context=ssl.create_default_context())
+        self._server_hostname = server_hostname
+
+    def connect(self) -> None:
+        self.sock = self._create_connection(
+            (self.host, self.port), self.timeout, self.source_address
+        )
+        if self._tunnel_host:
+            self._tunnel()
+        self.sock = self._context.wrap_socket(self.sock, server_hostname=self._server_hostname)
+
+
+class _PinnedHTTPConnection(http.client.HTTPConnection):
+    """HTTP connection using an already-validated address as its TCP peer."""
+
+
+def validate_public_url(url: str) -> tuple[str, int, str]:
+    """Validate an HTTP(S) URL and return hostname, effective port, and pinned IP."""
+    try:
+        parsed = urlsplit(url)
+    except ValueError as error:
+        raise UnsafeUrlError("invalid URL") from error
+
+    scheme = parsed.scheme.lower()
+    if scheme not in {"http", "https"}:
+        raise UnsafeUrlError("unsupported URL scheme")
+    if parsed.username is not None or parsed.password is not None:
+        raise UnsafeUrlError("URL credentials are not allowed")
+
+    try:
+        hostname = parsed.hostname
+        explicit_port = parsed.port
+    except ValueError as error:
+        raise UnsafeUrlError("invalid port") from error
+    if not hostname:
+        raise UnsafeUrlError("missing hostname")
+    if explicit_port == 0:
+        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
+
+    addresses: list[str] = []
+    for _family, _socktype, _protocol, _canonname, sockaddr in results:
+        try:
+            address = ipaddress.ip_address(sockaddr[0])
+        except (IndexError, ValueError) as error:
+            raise UnsafeUrlError("invalid resolved address") from error
+        if not address.is_global:
+            raise UnsafeUrlError("resolved address is not globally routable")
+        addresses.append(str(address))
+    if not addresses:
+        raise UnsafeUrlError("hostname did not resolve")
+
+    return hostname, port, min(addresses)
+
+
+def _host_header(hostname: str, port: int, scheme: str) -> str:
+    bracketed_host = f"[{hostname}]" if ":" in hostname else hostname
+    default_port = 443 if scheme == "https" else 80
+    return bracketed_host if port == default_port else f"{bracketed_host}:{port}"
+
+
+def _request_target(parsed: SplitResult) -> str:
+    target = parsed.path or "/"
+    return f"{target}?{parsed.query}" if parsed.query else target
+
+
+def _single_request(
+    method: str,
+    parsed: SplitResult,
+    hostname: str,
+    port: int,
+    address: str,
+    timeout: float,
+) -> RawResponse:
+    if parsed.scheme.lower() == "https":
+        connection = _PinnedHTTPSConnection(address, port, hostname, timeout)
+    else:
+        connection = _PinnedHTTPConnection(address, port, timeout=timeout)
+
+    response = None
+    try:
+        connection.request(
+            method,
+            _request_target(parsed),
+            headers={
+                "Host": _host_header(hostname, port, parsed.scheme.lower()),
+                "User-Agent": USER_AGENT,
+            },
+        )
+        response = connection.getresponse()
+        return RawResponse(response.status, response.getheader("Location"))
+    finally:
+        if response is not None:
+            response.close()
+        connection.close()
+
+
+def request_once(url: str, timeout: float = 10) -> RawResponse:
+    """Perform a pinned HEAD request, falling back to GET when HEAD is unsupported."""
+    hostname, port, address = validate_public_url(url)
+    parsed = urlsplit(url)
+    response = _single_request("HEAD", parsed, hostname, port, address, timeout)
+    if response.status in {405, 501}:
+        response = _single_request("GET", parsed, hostname, port, address, timeout)
+    return response
+
+
+def _canonical_url(url: str) -> str:
+    parsed = urlsplit(url)
+    scheme = parsed.scheme.lower()
+    hostname = (parsed.hostname or "").lower()
+    port = parsed.port
+    default_port = 443 if scheme == "https" else 80
+    host = f"[{hostname}]" if ":" in hostname else hostname
+    netloc = host if port in {None, default_port} else f"{host}:{port}"
+    path = parsed.path.rstrip("/")
+    return urlunsplit((scheme, netloc, path, parsed.query, ""))
+
+
+def _http_error(
+    requested_url: str,
+    final_url: str,
+    status: int | None,
+    attempts: int,
+    redirects: list[str],
+    error: str,
+) -> UrlObservation:
+    return UrlObservation(
+        requested_url,
+        final_url,
+        status,
+        Outcome.HTTP_ERROR,
+        attempts,
+        tuple(redirects),
+        error,
+    )
+
+
+def _classify_terminal(
+    requested_url: str,
+    final_url: str,
+    response: RawResponse,
+    attempts: int,
+    redirects: list[str],
+    saw_meaningful_permanent_redirect: bool,
+) -> UrlObservation:
+    if response.status in {404, 410}:
+        outcome = Outcome.DEAD
+        error = ""
+    elif response.status in {401, 403, 407, 429, 451}:
+        outcome = Outcome.RESTRICTED
+        error = ""
+    elif 200 <= response.status <= 399:
+        outcome = (
+            Outcome.PERMANENT_REDIRECT
+            if saw_meaningful_permanent_redirect
+            else Outcome.OK
+        )
+        error = ""
+    else:
+        outcome = Outcome.HTTP_ERROR
+        error = f"HTTP status {response.status}"
+    return UrlObservation(
+        requested_url,
+        final_url,
+        response.status,
+        outcome,
+        attempts,
+        tuple(redirects),
+        error,
+    )
+
+
+def probe_url(
+    requested_url: str,
+    *,
+    requester: Callable[[str, float], RawResponse] = request_once,
+    sleeper: Callable[[float], None] = sleep,
+    timeout: float = 10,
+) -> UrlObservation:
+    """Classify a public HTTP(S) URL without allowing DNS rebinding requests."""
+    last_transport_error = ""
+    for attempt, delay in enumerate((0, 1, 2), start=1):
+        sleeper(delay)
+        current_url = requested_url
+        redirects: list[str] = []
+        seen: set[str] = set()
+        saw_meaningful_permanent_redirect = False
+        try:
+            while True:
+                try:
+                    validate_public_url(current_url)
+                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}",
+                    )
+
+                current_canonical = _canonical_url(current_url)
+                seen.add(current_canonical)
+
+                response = requester(current_url, timeout)
+                if response.status not in REDIRECT_STATUSES:
+                    if response.status in RETRYABLE_STATUSES:
+                        if attempt == 3:
+                            return UrlObservation(
+                                requested_url,
+                                current_url,
+                                response.status,
+                                Outcome.TRANSIENT,
+                                attempt,
+                                tuple(redirects),
+                                f"retry limit exceeded: HTTP {response.status}",
+                            )
+                        break
+                    return _classify_terminal(
+                        requested_url,
+                        current_url,
+                        response,
+                        attempt,
+                        redirects,
+                        saw_meaningful_permanent_redirect,
+                    )
+
+                if not response.location:
+                    return _http_error(
+                        requested_url,
+                        current_url,
+                        response.status,
+                        attempt,
+                        redirects,
+                        "missing redirect location",
+                    )
+                if len(redirects) >= 5:
+                    return _http_error(
+                        requested_url,
+                        current_url,
+                        response.status,
+                        attempt,
+                        redirects,
+                        "redirect limit exceeded",
+                    )
+
+                target_url = urljoin(current_url, response.location)
+                try:
+                    validate_public_url(target_url)
+                except UnsafeUrlError as error:
+                    return _http_error(
+                        requested_url,
+                        current_url,
+                        response.status,
+                        attempt,
+                        redirects,
+                        f"unsafe redirect target: {error}",
+                    )
+                target_canonical = _canonical_url(target_url)
+                if target_canonical in seen:
+                    return _http_error(
+                        requested_url,
+                        current_url,
+                        response.status,
+                        attempt,
+                        redirects,
+                        "redirect loop",
+                    )
+                if (
+                    response.status in PERMANENT_REDIRECT_STATUSES
+                    and target_canonical != current_canonical
+                ):
+                    saw_meaningful_permanent_redirect = True
+                redirects.append(target_url)
+                seen.add(target_canonical)
+                current_url = target_url
+        except TRANSPORT_ERRORS as error:
+            last_transport_error = type(error).__name__
+            if attempt == 3:
+                return UrlObservation(
+                    requested_url,
+                    requested_url,
+                    None,
+                    Outcome.TRANSIENT,
+                    attempt,
+                    (),
+                    f"transport error: {last_transport_error}",
+                )
+
+    raise AssertionError("retry loop exhausted unexpectedly")
diff --git a/tests/test_url_probe.py b/tests/test_url_probe.py
new file mode 100644
index 0000000..0a66e9d
--- /dev/null
+++ b/tests/test_url_probe.py
@@ -0,0 +1,214 @@
+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_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_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()