fix: harden README audit edge cases

scripts/audit_readme.py

@@ -14,7 +14,7 @@ from enum import StrEnum
 from itertools import islice
 from pathlib import Path
 from typing import Any, Callable, Iterable
-from urllib.parse import urlsplit
+from urllib.parse import quote, urlsplit
 
 from github import Auth, Github, GithubException
 
@@ -71,7 +71,8 @@ _REPORT_SECTION_ORDER = {
     for kind in kinds
 }
 _MARKDOWN_CONTROLS = frozenset(string.punctuation)
-_URL_RE = re.compile(r"(?<!<)https?://[^\s<>]+")
+_URL_RE = re.compile(r"https?://\S+")
+_URL_SAFE_CHARACTERS = ":/?#[]@!$&'()*+,;=%"
 
 
 @dataclass(frozen=True)
@@ -134,15 +135,25 @@ def _escape_markdown(text: str) -> str:
     )
 
 
+def _render_url(url: str) -> str:
+    encoded_url = quote(url, safe=_URL_SAFE_CHARACTERS)
+    return f"<{encoded_url}>"
+
+
 def _render_text(text: str) -> str:
     text = " ".join(text.split())
-
-    def angle_link(match: re.Match[str]) -> str:
+    parts = []
+    offset = 0
+    for match in _URL_RE.finditer(text):
+        parts.append(_escape_markdown(text[offset : match.start()]))
         value = match.group(0)
         url = value.rstrip(".,;:!?")
-        return f"<{url}>{value[len(url):]}"
+        parts.append(_render_url(url))
+        parts.append(_escape_markdown(value[len(url) :]))
+        offset = match.end()
 
-    return _URL_RE.sub(angle_link, text)
+    parts.append(_escape_markdown(text[offset:]))
+    return "".join(parts)
 
 
 def render_report(findings: Iterable[Finding], *, checked_at: datetime) -> str:
@@ -178,14 +189,15 @@ def render_report(findings: Iterable[Finding], *, checked_at: datetime) -> str:
                 f"- **Entry:** {_escape_markdown(finding.entry_name)}; "
                 f"**README line:** {finding.line_number}; "
                 f"**Section:** {_escape_markdown(finding.section)}; "
-                f"**Checked URL:** <{finding.url}>; "
+                f"**Checked URL:** {_render_url(finding.url)}; "
                 f"**Evidence:** {_render_text(finding.evidence)}; "
                 f"**Manual suggestion:** {_render_text(finding.suggestion)}"
             )
             for index, candidate in enumerate(_sorted_candidates(finding), start=1):
                 lines.append(
-                    f"  - **Candidate {index}:** {candidate.full_name} — "
-                    f"<{candidate.url}> — {candidate.stars} stars"
+                    f"  - **Candidate {index}:** "
+                    f"{_escape_markdown(candidate.full_name)} — "
+                    f"{_render_url(candidate.url)} — {candidate.stars} stars"
                 )
         lines.append("")
     return "\n".join(lines)

scripts/url_probe.py

@@ -18,6 +18,7 @@ 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)
+NAT64_WELL_KNOWN_NETWORK = ipaddress.IPv6Network("64:ff9b::/96")
 
 
 class Outcome(StrEnum):
@@ -74,6 +75,38 @@ class _PinnedHTTPConnection(http.client.HTTPConnection):
     """HTTP connection using an already-validated address as its TCP peer."""
 
 
+def _embedded_ipv4_addresses(
+    address: ipaddress.IPv4Address | ipaddress.IPv6Address,
+) -> tuple[ipaddress.IPv4Address, ...]:
+    if not isinstance(address, ipaddress.IPv6Address):
+        return ()
+
+    embedded = [
+        candidate
+        for candidate in (address.ipv4_mapped, address.sixtofour)
+        if candidate is not None
+    ]
+    if address.teredo is not None:
+        embedded.extend(address.teredo)
+    if address in NAT64_WELL_KNOWN_NETWORK:
+        embedded.append(ipaddress.IPv4Address(int(address) & 0xFFFFFFFF))
+    return tuple(embedded)
+
+
+def _is_public_address(
+    address: ipaddress.IPv4Address | ipaddress.IPv6Address,
+) -> bool:
+    embedded_addresses_are_public = all(
+        embedded.is_global and not embedded.is_multicast
+        for embedded in _embedded_ipv4_addresses(address)
+    )
+    return (
+        address.is_global
+        and not address.is_multicast
+        and embedded_addresses_are_public
+    )
+
+
 def validate_public_url(url: str) -> tuple[str, int, str]:
     """Validate an HTTP(S) URL and return hostname, effective port, and pinned IP."""
     try:
@@ -106,7 +139,7 @@ def validate_public_url(url: str) -> tuple[str, int, str]:
             address = ipaddress.ip_address(sockaddr[0])
         except (IndexError, ValueError) as error:
             raise UnsafeUrlError("invalid resolved address") from error
-        if not address.is_global:
+        if not _is_public_address(address):
             raise UnsafeUrlError("resolved address is not globally routable")
         addresses.append(str(address))
     if not addresses:
@@ -267,7 +300,17 @@ def probe_url(
                 current_canonical = _canonical_url(current_url)
                 seen.add(current_canonical)
 
-                response = requester(current_url, timeout)
+                try:
+                    response = requester(current_url, timeout)
+                except http.client.InvalidURL:
+                    return _http_error(
+                        requested_url,
+                        current_url,
+                        None,
+                        attempt,
+                        redirects,
+                        "invalid request URL",
+                    )
                 if response.status not in REDIRECT_STATUSES:
                     if response.status in RETRYABLE_STATUSES:
                         if attempt == 3:
@@ -309,7 +352,17 @@ def probe_url(
                         "redirect limit exceeded",
                     )
 
-                target_url = urljoin(current_url, response.location)
+                try:
+                    target_url = urljoin(current_url, response.location)
+                except ValueError:
+                    return _http_error(
+                        requested_url,
+                        current_url,
+                        response.status,
+                        attempt,
+                        redirects,
+                        "invalid redirect location",
+                    )
                 try:
                     validate_public_url(target_url)
                 except UnsafeUrlError as error:

tests/test_audit_readme.py

@@ -432,35 +432,35 @@ Entries are reported for manual review; this automation did not modify README.md
 
 ## Confirmed dead links
 
-- **Entry:** A\_B \[tool\]; **README line:** 10; **Section:** Alpha \& Beta; **Checked URL:** <https://dead.example/a>; **Evidence:** Confirmed HTTP 404 response.; **Manual suggestion:** Manually verify, then remove or replace this link.
-- **Entry:** Zed; **README line:** 11; **Section:** Alpha \& Beta; **Checked URL:** <https://dead.example/z>; **Evidence:** Confirmed HTTP 410 response.; **Manual suggestion:** Manually verify, then remove or replace this link.
+- **Entry:** A\_B \[tool\]; **README line:** 10; **Section:** Alpha \& Beta; **Checked URL:** <https://dead.example/a>; **Evidence:** Confirmed HTTP 404 response\.; **Manual suggestion:** Manually verify\, then remove or replace this link\.
+- **Entry:** Zed; **README line:** 11; **Section:** Alpha \& Beta; **Checked URL:** <https://dead.example/z>; **Evidence:** Confirmed HTTP 410 response\.; **Manual suggestion:** Manually verify\, then remove or replace this link\.
 
 ## Repeated transient failures
 
-- **Entry:** Retry; **README line:** 20; **Section:** Transient; **Checked URL:** <https://retry.example>; **Evidence:** Transient failure after 3 attempts; last status 503.; **Manual suggestion:** Retry this link on the next audit run.
+- **Entry:** Retry; **README line:** 20; **Section:** Transient; **Checked URL:** <https://retry.example>; **Evidence:** Transient failure after 3 attempts\; last status 503\.; **Manual suggestion:** Retry this link on the next audit run\.
 
 ## Access-restricted links
 
-- **Entry:** Restricted; **README line:** 30; **Section:** Access; **Checked URL:** <https://restricted.example>; **Evidence:** Automated access was restricted by HTTP 403.; **Manual suggestion:** Perform manual browser verification before changing the link.
+- **Entry:** Restricted; **README line:** 30; **Section:** Access; **Checked URL:** <https://restricted.example>; **Evidence:** Automated access was restricted by HTTP 403\.; **Manual suggestion:** Perform manual browser verification before changing the link\.
 
 ## Other HTTP failures
 
-- **Entry:** Broken; **README line:** 40; **Section:** HTTP; **Checked URL:** <https://error.example>; **Evidence:** HTTP probe failed; last error unsafe URL.; **Manual suggestion:** Perform manual diagnosis before changing the link.
+- **Entry:** Broken; **README line:** 40; **Section:** HTTP; **Checked URL:** <https://error.example>; **Evidence:** HTTP probe failed\; last error unsafe URL\.; **Manual suggestion:** Perform manual diagnosis before changing the link\.
 
 ## Permanent redirects
 
-- **Entry:** Moved; **README line:** 50; **Section:** Redirects; **Checked URL:** <https://old.example/path>; **Evidence:** The link permanently redirects to <https://new.example/path>.; **Manual suggestion:** Update the link to <https://new.example/path>.
+- **Entry:** Moved; **README line:** 50; **Section:** Redirects; **Checked URL:** <https://old.example/path>; **Evidence:** The link permanently redirects to <https://new.example/path>\.; **Manual suggestion:** Update the link to <https://new.example/path>\.
 
 ## GitHub repository findings
 
-- **Entry:** Archive; **README line:** 60; **Section:** Repositories; **Checked URL:** <https://github.com/owner/archive>; **Evidence:** GitHub marks the repository as archived.; **Manual suggestion:** Manually verify whether to replace it or move it to the historical section.
+- **Entry:** Archive; **README line:** 60; **Section:** Repositories; **Checked URL:** <https://github.com/owner/archive>; **Evidence:** GitHub marks the repository as archived\.; **Manual suggestion:** Manually verify whether to replace it or move it to the historical section\.
 
 ## Candidate replacements
 
-- **Entry:** Replacement \*Tool\* \~\~Gone\~\~ \&copy\;; **README line:** 80; **Section:** Zeta \[Section\] \#1 \| Top; **Checked URL:** <https://dead.example/tool>; **Evidence:** The original primary URL returned confirmed HTTP 404.; **Manual suggestion:** Manual verification of these candidate repositories is required.
-  - **Candidate 1:** alpha/tool — <https://github.com/alpha/tool> — 9 stars
-  - **Candidate 2:** beta/tool — <https://github.com/beta/tool> — 9 stars
-  - **Candidate 3:** zeta/tool — <https://github.com/zeta/tool> — 2 stars
+- **Entry:** Replacement \*Tool\* \~\~Gone\~\~ \&copy\;; **README line:** 80; **Section:** Zeta \[Section\] \#1 \| Top; **Checked URL:** <https://dead.example/tool>; **Evidence:** The original primary URL returned confirmed HTTP 404\.; **Manual suggestion:** Manual verification of these candidate repositories is required\.
+  - **Candidate 1:** alpha\/tool — <https://github.com/alpha/tool> — 9 stars
+  - **Candidate 2:** beta\/tool — <https://github.com/beta/tool> — 9 stars
+  - **Candidate 3:** zeta\/tool — <https://github.com/zeta/tool> — 2 stars
 """
 
         self.assertEqual(render_report(findings, checked_at=checked_at), expected)
@@ -487,6 +487,67 @@ No findings.
 """,
         )
 
+    def test_redirect_url_cannot_escape_angle_link_and_create_a_mention(self):
+        source_url = "https://redirect.example/old"
+        readme = (
+            "## Trading & Backtesting\n"
+            f"- [Redirect]({source_url}) - `Python` - Entry.\n"
+        )
+        findings = self.audit(
+            readme,
+            FakeGithubClient(),
+            {
+                source_url: observation(
+                    source_url,
+                    Outcome.PERMANENT_REDIRECT,
+                    final_url="https://redirect.example/new>@octocat",
+                )
+            },
+        )
+
+        report = render_report(
+            findings,
+            checked_at=datetime(2026, 9, 2, 12, 45, tzinfo=timezone.utc),
+        )
+
+        self.assertIn("<https://redirect.example/new%3E@octocat>", report)
+        self.assertNotIn(">@octocat", report)
+
+    def test_checked_and_candidate_urls_are_rendered_safely(self):
+        finding = Finding(
+            FindingKind.CANDIDATES,
+            "Section",
+            "Entry",
+            10,
+            "https://checked.example/a b>@checked",
+            "Candidate replacement found.",
+            "Verify manually.",
+            (
+                Candidate(
+                    "@octocat/repo",
+                    "https://candidate.example/a b>@candidate",
+                    1,
+                ),
+            ),
+        )
+
+        report = render_report(
+            [finding],
+            checked_at=datetime(2026, 9, 2, 12, 45, tzinfo=timezone.utc),
+        )
+
+        self.assertIn(
+            "**Checked URL:** <https://checked.example/a%20b%3E@checked>",
+            report,
+        )
+        self.assertIn(
+            r"**Candidate 1:** \@octocat\/repo — "
+            "<https://candidate.example/a%20b%3E@candidate>",
+            report,
+        )
+        self.assertNotIn(">@checked", report)
+        self.assertNotIn(">@candidate", report)
+
     def test_sync_tracking_issue_follows_the_full_lifecycle_matrix(self):
         findings = [SimpleNamespace()]
         new_body = f"{TRACKING_MARKER}\nnew report"

tests/test_url_probe.py

@@ -1,3 +1,4 @@
+import http.client
 import socket
 import unittest
 from unittest.mock import patch
@@ -188,6 +189,82 @@ class UrlProbeTests(unittest.TestCase):
         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 = []