diff --git a/nanobot/agent/tools/mcp.py b/nanobot/agent/tools/mcp.py index 7d35eb4c..99baf91f 100644 --- a/nanobot/agent/tools/mcp.py +++ b/nanobot/agent/tools/mcp.py @@ -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( diff --git a/nanobot/security/network.py b/nanobot/security/network.py index 7d4d6214..a17919a4 100644 --- a/nanobot/security/network.py +++ b/nanobot/security/network.py @@ -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): diff --git a/tests/tools/test_mcp_tool.py b/tests/tools/test_mcp_tool.py index 3b8763a1..819550b7 100644 --- a/tests/tools/test_mcp_tool.py +++ b/tests/tools/test_mcp_tool.py @@ -712,8 +712,8 @@ async def test_connect_mcp_servers_logs_stdio_pollution_hint( @pytest.mark.parametrize( "config", [ - MCPServerConfig(url="http://127.0.0.1:9/sse"), - MCPServerConfig(type="streamableHttp", url="http://127.0.0.1:9/mcp"), + MCPServerConfig(url="http://169.254.169.254/sse"), + MCPServerConfig(type="streamableHttp", url="http://169.254.169.254/mcp"), ], ) async def test_connect_mcp_servers_rejects_unsafe_http_urls_before_probe( @@ -762,7 +762,7 @@ async def test_connect_mcp_servers_http_clients_reject_unsafe_redirect_targets( sent_urls: list[str] = [] used_transports: list[str] = [] - def _validate(url: str) -> tuple[bool, str]: + def _validate(url: str, **_kwargs: object) -> tuple[bool, str]: checked_urls.append(url) if url == "http://127.0.0.1/private": return False, "loopback blocked" @@ -804,7 +804,11 @@ async def test_connect_mcp_servers_http_clients_reject_unsafe_redirect_targets( monkeypatch.setattr(mcp_mod, "validate_url_target", _validate) monkeypatch.setattr(mcp_mod, "_probe_http_url", _reachable) - monkeypatch.setattr(mcp_mod, "PinnedDNSAsyncTransport", lambda: httpx.MockTransport(_handler)) + monkeypatch.setattr( + mcp_mod, + "PinnedDNSAsyncTransport", + lambda **_kwargs: httpx.MockTransport(_handler), + ) monkeypatch.setattr(mcp_mod.httpx, "AsyncClient", _async_client_with_mock_transport) monkeypatch.setattr(sys.modules["mcp.client.sse"], "sse_client", _fake_sse_client) monkeypatch.setattr(