From 18ac388162080c5144ce4fda19e1381c76a5691c Mon Sep 17 00:00:00 2001 From: coderdailyone Date: Fri, 31 Jul 2026 10:16:09 +0100 Subject: [PATCH] security: pin fetched connections to the vetted address (close DNS rebinding) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit guard.get_checked previously resolved a name for the address check and then let the client resolve it again to connect, so a ~0-TTL record could flip to 127.0.0.1 (or the metadata endpoint) between the two — the exact gap the module docstring called out as "not covered". Every hop now connects to the address that passed its check: the request URL carries the vetted IP, Host keeps the original name (and explicit port) for virtual hosting, and the sni_hostname extension keeps the TLS handshake — and certificate verification — against the name. Literal-IP URLs are unchanged. Redirect Locations are resolved against the logical URL, and the final logical URL rides resp.extensions["logical_url"] so web_fetch reports the name, not the pinned address. Verified against a local TLS server whose cert carries only DNS:pinned.test: the pinned request (URL=127.0.0.1, sni_hostname=pinned.test) verifies and the server sees Host: pinned.test; the same request without the extension fails with "IP address mismatch" — httpcore does verify certificates against sni_hostname, not the URL host. Co-Authored-By: Claude Fable 5 --- coworker/web/fetch.py | 5 +- coworker/web/guard.py | 91 +++++++++++++++++++++++++-------- tests/test_url_address_guard.py | 74 +++++++++++++++++++++++++-- 3 files changed, 145 insertions(+), 25 deletions(-) diff --git a/coworker/web/fetch.py b/coworker/web/fetch.py index f58291da..b7199038 100644 --- a/coworker/web/fetch.py +++ b/coworker/web/fetch.py @@ -87,7 +87,7 @@ def make_web_fetch_tool() -> Callable[..., Any]: import httpx # follow_redirects=False: guard.get_checked walks the chain so every hop is - # address-checked, not just the URL the model first supplied. + # address-checked and pinned, not just the URL the model first supplied. with httpx.Client( follow_redirects=False, timeout=20.0, @@ -97,7 +97,8 @@ def make_web_fetch_tool() -> Callable[..., Any]: resp.raise_for_status() ctype = resp.headers.get("content-type", "") body = resp.text - final_url = str(resp.url) + # resp.url names the pinned address; the guard stashes the logical URL. + final_url = resp.extensions.get("logical_url", url) except PermissionError as exc: # blocked address (loopback, private, metadata) return {"error": str(exc)} except Exception as exc: # network / HTTP / TLS diff --git a/coworker/web/guard.py b/coworker/web/guard.py index 549d7841..e1c2117b 100644 --- a/coworker/web/guard.py +++ b/coworker/web/guard.py @@ -14,9 +14,12 @@ metadata endpoint at 169.254.169.254), and the reserved/multicast blocks. Every hop is checked, not just the first: `follow_redirects=True` otherwise lets a public URL 302 straight to loopback, which is the standard way this filter is bypassed. -Not covered: DNS rebinding. The name is resolved here and resolved again by the client when -it connects, so a record with a ~0 TTL can change between the two. Closing that needs -connection-level IP pinning; the hop check is the cheap 90% and is stated as such. +DNS rebinding is closed by connection-level pinning: `get_checked` rewrites each hop so the +client connects to the exact address that passed the check (name in Host and SNI, so virtual +hosting and certificate verification still see the name). A record with a ~0 TTL that flips +to 127.0.0.1 between the check and the connect therefore changes nothing — the client never +resolves the name itself. `check_url` alone (browser_open_url's pre-check) still carries the +resolve-twice gap, because the browser owns its own connections and cannot be pinned from here. """ from __future__ import annotations @@ -24,7 +27,7 @@ from __future__ import annotations import ipaddress import socket from typing import Optional -from urllib.parse import urlsplit +from urllib.parse import urljoin, urlsplit, urlunsplit MAX_REDIRECTS = 5 @@ -50,18 +53,20 @@ def _blocked_reason(ip: ipaddress._BaseAddress) -> Optional[str]: return None -def check_url(url: str) -> Optional[str]: - """None if the URL may be fetched, else a human-readable refusal reason. +def _vet(url: str) -> tuple[Optional[str], Optional[str]]: + """(refusal reason, address to pin the connection to). - Resolves the host and rejects when *any* answer lands in a blocked range, so a name - with both a public and a private A record cannot be used to slip through. + The reason is None when the URL may be fetched. The address is None for literal-IP + URLs (the URL already names the connection target) and the first resolved answer + otherwise — valid to pin because a refusal is returned when *any* answer lands in a + blocked range, so a name with both a public and a private A record cannot slip through. """ parts = urlsplit(url) if parts.scheme not in ("http", "https"): - return "url must start with http:// or https://" + return "url must start with http:// or https://", None host = parts.hostname if not host: - return "url has no host" + return "url has no host", None # A literal address needs no lookup. try: @@ -70,14 +75,15 @@ def check_url(url: str) -> Optional[str]: literal = None if literal is not None: reason = _blocked_reason(literal) - return f"refusing to fetch {host}: {reason}" if reason else None + return (f"refusing to fetch {host}: {reason}" if reason else None), None try: infos = socket.getaddrinfo(host, parts.port or (443 if parts.scheme == "https" else 80), proto=socket.IPPROTO_TCP) except OSError as exc: - return f"could not resolve {host}: {exc}" + return f"could not resolve {host}: {exc}", None + pin: Optional[str] = None for info in infos: raw = info[4][0] try: @@ -90,27 +96,72 @@ def check_url(url: str) -> Optional[str]: ip = mapped reason = _blocked_reason(ip) if reason: - return f"refusing to fetch {host} ({ip}): {reason}" - return None + return f"refusing to fetch {host} ({ip}): {reason}", None + if pin is None: + pin = raw + return None, pin + + +def check_url(url: str) -> Optional[str]: + """None if the URL may be fetched, else a human-readable refusal reason. + + Resolves the host and rejects when *any* answer lands in a blocked range, so a name + with both a public and a private A record cannot be used to slip through. + """ + return _vet(url)[0] + + +def _pinned(url: str, ip: str) -> tuple[str, dict, dict]: + """Rewrite `url` so the client connects to `ip` while presenting the original name. + + Returns (request_url, headers, extensions): the URL carries the vetted address so the + client never resolves the name itself, Host carries the name (and any explicit port) + for virtual hosting, and `sni_hostname` keeps the TLS handshake — including certificate + verification — against the name rather than the address. + """ + parts = urlsplit(url) + host = parts.hostname + addr = f"[{ip}]" if ":" in ip else ip + userinfo, _, _ = parts.netloc.rpartition("@") + netloc = (f"{userinfo}@" if userinfo else "") + addr + host_header = host + if parts.port is not None: + netloc += f":{parts.port}" + host_header += f":{parts.port}" + request_url = urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment)) + extensions = {"sni_hostname": host} if parts.scheme == "https" else {} + return request_url, {"Host": host_header}, extensions def get_checked(client, url: str, *, max_redirects: int = MAX_REDIRECTS): - """GET `url`, validating the address before every hop. + """GET `url`, validating and pinning the address before every hop. `client` must be built with `follow_redirects=False`; redirects are walked here so each - Location is checked. Returns the final response. Raises `PermissionError` when a hop is - refused, `RuntimeError` when the redirect budget is exhausted. + Location is checked. Every hop connects to the exact address that passed its check (see + `_pinned`), so a rebinding name cannot swap targets between check and connect. Returns + the final response, with the final *logical* URL — the name, not the pinned address — + stashed as `resp.extensions["logical_url"]` for callers that display it. Raises + `PermissionError` when a hop is refused, `RuntimeError` when the budget is exhausted. """ seen = url for _ in range(max_redirects + 1): - reason = check_url(seen) + reason, pin = _vet(seen) if reason: raise PermissionError(reason) - resp = client.get(seen) + if pin is None: + resp = client.get(seen) + else: + request_url, headers, extensions = _pinned(seen, pin) + resp = client.get(request_url, headers=headers, extensions=extensions) if resp.status_code not in (301, 302, 303, 307, 308): + ext = getattr(resp, "extensions", None) + if isinstance(ext, dict): + ext["logical_url"] = seen return resp location = resp.headers.get("location") if not location: return resp - seen = str(resp.url.join(location)) + # Resolved against the logical URL, not resp.url — the latter names the pinned + # address, and a relative Location must stay on the original host. + seen = urljoin(seen, location) raise RuntimeError(f"too many redirects (>{max_redirects})") diff --git a/tests/test_url_address_guard.py b/tests/test_url_address_guard.py index 33111899..360f3703 100644 --- a/tests/test_url_address_guard.py +++ b/tests/test_url_address_guard.py @@ -123,9 +123,11 @@ class _Client: def __init__(self, script): self.script = script self.requested = [] + self.calls = [] - def get(self, url): + def get(self, url, headers=None, extensions=None): self.requested.append(url) + self.calls.append({"url": url, "headers": headers or {}, "extensions": extensions or {}}) return self.script.pop(0) @@ -134,7 +136,7 @@ def test_redirect_into_loopback_is_blocked_before_the_second_request(monkeypatch client = _Client([_Resp(302, location="http://127.0.0.1:11434/api/tags")]) with pytest.raises(PermissionError, match="loopback"): guard.get_checked(client, "https://example.com/start") - assert client.requested == ["https://example.com/start"], ( + assert client.requested == ["https://93.184.216.34/start"], ( "the redirect target must never be requested" ) @@ -144,7 +146,7 @@ def test_allowed_redirect_chain_is_followed(monkeypatch): client = _Client([_Resp(302, location="https://example.com/b"), _Resp(200)]) resp = guard.get_checked(client, "https://example.com/a") assert resp.status_code == 200 - assert client.requested == ["https://example.com/a", "https://example.com/b"] + assert client.requested == ["https://93.184.216.34/a", "https://93.184.216.34/b"] def test_redirect_loop_is_bounded(monkeypatch): @@ -154,6 +156,72 @@ def test_redirect_loop_is_bounded(monkeypatch): guard.get_checked(client, "https://example.com/loop") +# -- pinning (DNS rebinding) -------------------------------------------------- + +def test_connection_is_pinned_to_the_vetted_address(monkeypatch): + """The client must be told to connect to the address that was checked, with the + original name in Host and SNI — never left to resolve the name a second time.""" + _resolves_to(monkeypatch, "93.184.216.34") + client = _Client([_Resp(200)]) + guard.get_checked(client, "https://example.com/docs") + call = client.calls[0] + assert call["url"] == "https://93.184.216.34/docs" + assert call["headers"]["Host"] == "example.com" + assert call["extensions"]["sni_hostname"] == "example.com" + + +def test_rebinding_after_the_check_cannot_reach_loopback(monkeypatch): + """A ~0-TTL record that flips to 127.0.0.1 between check and connect must not + matter: the connection goes to the address that passed the check.""" + answers = iter(["93.184.216.34", "127.0.0.1"]) + + def flipping(*a, **k): + ip = next(answers, "127.0.0.1") + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, 80))] + + monkeypatch.setattr(guard.socket, "getaddrinfo", flipping) + client = _Client([_Resp(200)]) + guard.get_checked(client, "http://rebind.example.com/") + assert client.requested == ["http://93.184.216.34/"], ( + "the second resolution must never influence where the client connects" + ) + + +def test_pinned_host_header_preserves_an_explicit_port(monkeypatch): + _resolves_to(monkeypatch, "93.184.216.34") + client = _Client([_Resp(200)]) + guard.get_checked(client, "http://example.com:8080/x") + call = client.calls[0] + assert call["url"] == "http://93.184.216.34:8080/x" + assert call["headers"]["Host"] == "example.com:8080" + assert "sni_hostname" not in call["extensions"], "plain http has no TLS handshake" + + +def test_ipv6_answers_are_pinned_with_brackets(monkeypatch): + _resolves_to(monkeypatch, "2606:2800:220:1:248:1893:25c8:1946") + client = _Client([_Resp(200)]) + guard.get_checked(client, "https://example.com/") + assert client.requested == ["https://[2606:2800:220:1:248:1893:25c8:1946]/"] + + +def test_literal_address_urls_are_fetched_unchanged(): + client = _Client([_Resp(200)]) + guard.get_checked(client, "https://93.184.216.34/x") + call = client.calls[0] + assert call["url"] == "https://93.184.216.34/x" + assert "Host" not in call["headers"], "a literal needs no name-based Host override" + + +def test_logical_url_is_reported_not_the_pinned_address(monkeypatch): + """Callers show the final URL to the model; it must be the name, not the address.""" + _resolves_to(monkeypatch, "93.184.216.34") + resp = _Resp(200) + resp.extensions = {} + client = _Client([_Resp(302, location="https://example.com/b"), resp]) + out = guard.get_checked(client, "https://example.com/a") + assert out.extensions["logical_url"] == "https://example.com/b" + + # -- the tool ----------------------------------------------------------------- def test_web_fetch_returns_the_refusal_as_a_tool_error(monkeypatch):