fix: allow local mcp urls with pinned dns

This commit is contained in:
hamb1y
2026-07-07 15:40:53 +08:00
committed by Xubin Ren
parent 73bf299a59
commit 4353f4680b
3 changed files with 15 additions and 10 deletions
+4 -4
View File
@@ -180,7 +180,7 @@ async def _probe_http_url(url: str, timeout: float = 3.0) -> bool:
port = parsed.port
if not port:
port = 443 if parsed.scheme == "https" else 80
ok, _, resolved_ips = resolve_url_target(url)
ok, _, resolved_ips = resolve_url_target(url, allow_loopback=True)
if not ok:
return False
try:
@@ -218,7 +218,7 @@ def _redact_url(url: str) -> str:
async def _validate_mcp_request_url(request: httpx.Request) -> None:
"""Validate each outgoing MCP HTTP request, including redirect targets."""
ok, error = validate_url_target(str(request.url))
ok, error = validate_url_target(str(request.url), allow_loopback=True)
if not ok:
raise httpx.RequestError(
f"Blocked unsafe MCP URL {_redact_url(str(request.url))} ({error})",
@@ -885,7 +885,7 @@ async def connect_mcp_servers(
follow_redirects=True,
timeout=timeout,
auth=auth,
transport=PinnedDNSAsyncTransport(),
transport=PinnedDNSAsyncTransport(allow_loopback=True),
)
read, write = await server_stack.enter_async_context(
@@ -903,7 +903,7 @@ async def connect_mcp_servers(
event_hooks={"request": [_validate_mcp_request_url]},
follow_redirects=True,
timeout=httpx.Timeout(30.0, connect=10.0),
transport=PinnedDNSAsyncTransport(),
transport=PinnedDNSAsyncTransport(allow_loopback=True),
)
)
read, write, _ = await server_stack.enter_async_context(
+3 -2
View File
@@ -149,12 +149,13 @@ def pin_resolved_url_dns(url: str, resolved_ips: tuple[str, ...]):
class PinnedDNSAsyncTransport(httpx.AsyncBaseTransport):
"""HTTPX transport that pins each request to the IPs validated for its URL."""
def __init__(self) -> None:
def __init__(self, *, allow_loopback: bool = False) -> None:
self._allow_loopback = allow_loopback
self._inner = httpx.AsyncHTTPTransport()
async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
url = str(request.url)
ok, error, resolved_ips = resolve_url_target(url)
ok, error, resolved_ips = resolve_url_target(url, allow_loopback=self._allow_loopback)
if not ok:
raise httpx.RequestError(error, request=request)
with pin_resolved_url_dns(url, resolved_ips):