ext/openworker
2
0
Fork 0
mirror of https://github.com/andrewyng/openworker.git synced 2026-09-05 00:56:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #415 from coderdailyone/security/pin-web-fetch-connections

Browse Source 
security: pin fetched connections to the vetted address (close DNS rebinding)
This commit is contained in:
Rohit Prasad
2026-08-07 20:36:25 -07:00
committed by GitHub
parent 41d4c54d31 18ac388162
commit 8861da136c
3 changed files with 145 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coworker/web/fetch.py
+3 -2
View File
@@ -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
coworker/web/guard.py
+71 -20
View File
@@ -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})")
tests/test_url_address_guard.py
+71 -3
View File
@@ -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):