diff --git a/nanobot/agent/tools/mcp.py b/nanobot/agent/tools/mcp.py index bd47d77f..7d35eb4c 100644 --- a/nanobot/agent/tools/mcp.py +++ b/nanobot/agent/tools/mcp.py @@ -23,7 +23,12 @@ from nanobot.bus.events import ( RUNTIME_CONTROL_MCP_RELOAD, InboundMessage, ) -from nanobot.security.network import validate_url_target +from nanobot.security.network import ( + PinnedDNSAsyncTransport, + pin_resolved_url_dns, + resolve_url_target, + validate_url_target, +) # Transient connection errors that warrant a single retry. # These typically happen when an MCP server restarts or a network @@ -175,11 +180,15 @@ 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) + if not ok: + return False try: - reader, writer = await asyncio.wait_for( - asyncio.open_connection(host, port), - timeout=timeout, - ) + with pin_resolved_url_dns(url, resolved_ips): + reader, writer = await asyncio.wait_for( + asyncio.open_connection(host, port), + timeout=timeout, + ) writer.close() with suppress(OSError, asyncio.TimeoutError): await asyncio.wait_for(writer.wait_closed(), timeout=0.2) @@ -876,6 +885,7 @@ async def connect_mcp_servers( follow_redirects=True, timeout=timeout, auth=auth, + transport=PinnedDNSAsyncTransport(), ) read, write = await server_stack.enter_async_context( @@ -893,6 +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(), ) ) read, write, _ = await server_stack.enter_async_context( diff --git a/nanobot/agent/tools/web.py b/nanobot/agent/tools/web.py index aa5645ac..74f17918 100644 --- a/nanobot/agent/tools/web.py +++ b/nanobot/agent/tools/web.py @@ -111,6 +111,13 @@ def _validate_url_safe(url: str) -> tuple[bool, str]: return validate_url_target(url) +def _resolve_url_safe(url: str) -> tuple[bool, str, tuple[str, ...]]: + """Validate URL and return the resolved IPs to pin during the request.""" + from nanobot.security.network import resolve_url_target + + return resolve_url_target(url) + + async def _get_with_safe_redirects( client: httpx.AsyncClient, url: str, @@ -119,11 +126,14 @@ async def _get_with_safe_redirects( """GET a URL while validating every redirect target before requesting it.""" current_url = url for _ in range(MAX_REDIRECTS + 1): - is_valid, error_msg = _validate_url_safe(current_url) + is_valid, error_msg, resolved_ips = _resolve_url_safe(current_url) if not is_valid: return None, f"Redirect blocked: {error_msg}" - response = await client.get(current_url, headers=headers, follow_redirects=False) + from nanobot.security.network import pin_resolved_url_dns + + with pin_resolved_url_dns(current_url, resolved_ips): + response = await client.get(current_url, headers=headers, follow_redirects=False) is_redirect = 300 <= response.status_code < 400 if not is_redirect: return response, None @@ -152,17 +162,20 @@ async def _stream_with_safe_redirects( """Open a streamed response while validating every redirect target first.""" current_url = url for _ in range(MAX_REDIRECTS + 1): - is_valid, error_msg = _validate_url_safe(current_url) + is_valid, error_msg, resolved_ips = _resolve_url_safe(current_url) if not is_valid: return None, None, f"Redirect blocked: {error_msg}" + from nanobot.security.network import pin_resolved_url_dns + stream = client.stream( "GET", current_url, headers=headers, follow_redirects=False, ) - response = await stream.__aenter__() + with pin_resolved_url_dns(current_url, resolved_ips): + response = await stream.__aenter__() is_redirect = 300 <= response.status_code < 400 if not is_redirect: return response, stream, None diff --git a/nanobot/security/network.py b/nanobot/security/network.py index e6861f94..7d4d6214 100644 --- a/nanobot/security/network.py +++ b/nanobot/security/network.py @@ -5,9 +5,11 @@ from __future__ import annotations import ipaddress import re import socket -from contextlib import suppress +from contextlib import contextmanager, suppress from urllib.parse import urlparse +import httpx + _BLOCKED_NETWORKS = [ ipaddress.ip_network("0.0.0.0/8"), ipaddress.ip_network("10.0.0.0/8"), @@ -58,7 +60,7 @@ def _is_private(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: return any(normalized in net for net in _BLOCKED_NETWORKS) -def validate_url_target(url: str, *, allow_loopback: bool = False) -> tuple[bool, str]: +def resolve_url_target(url: str, *, allow_loopback: bool = False) -> tuple[bool, str, tuple[str, ...]]: """Validate a URL is safe to fetch: scheme, hostname, and resolved IPs. ``allow_loopback`` is intentionally narrow: it only permits literal @@ -66,26 +68,27 @@ def validate_url_target(url: str, *, allow_loopback: bool = False) -> tuple[bool loopback. It does not allow RFC1918, link-local, metadata, or public DNS names that happen to resolve to loopback. - Returns (ok, error_message). When ok is True, error_message is empty. + Returns (ok, error_message, resolved_ips). When ok is True, + resolved_ips contains the public IPs that were validated for this URL. """ try: p = urlparse(url) except Exception as e: - return False, str(e) + return False, str(e), () if p.scheme not in ("http", "https"): - return False, f"Only http/https allowed, got '{p.scheme or 'none'}'" + return False, f"Only http/https allowed, got '{p.scheme or 'none'}'", () if not p.netloc: - return False, "Missing domain" + return False, "Missing domain", () hostname = p.hostname if not hostname: - return False, "Missing hostname" + return False, "Missing hostname", () try: infos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM) except socket.gaierror: - return False, f"Cannot resolve hostname: {hostname}" + return False, f"Cannot resolve hostname: {hostname}", () addrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = [] for info in infos: @@ -95,12 +98,70 @@ def validate_url_target(url: str, *, allow_loopback: bool = False) -> tuple[bool continue addrs.append(addr) if allow_loopback and _is_allowed_loopback_target(hostname, addrs): - return True, "" + return True, "", tuple(dict.fromkeys(str(_normalize_addr(addr)) for addr in addrs)) for addr in addrs: if _is_private(addr): - return False, f"Blocked: {hostname} resolves to private/internal address {addr}" + return False, f"Blocked: {hostname} resolves to private/internal address {addr}", () - return True, "" + return True, "", tuple(dict.fromkeys(str(_normalize_addr(addr)) for addr in addrs)) + + +def validate_url_target(url: str, *, allow_loopback: bool = False) -> tuple[bool, str]: + """Validate a URL is safe to fetch: scheme, hostname, and resolved IPs.""" + ok, error, _ = resolve_url_target(url, allow_loopback=allow_loopback) + return ok, error + + +@contextmanager +def pin_resolved_url_dns(url: str, resolved_ips: tuple[str, ...]): + """Pin DNS lookups for the URL hostname to previously validated IPs.""" + try: + hostname = urlparse(url).hostname + except Exception: + hostname = None + if not hostname or not resolved_ips: + yield + return + + pinned_host = hostname.rstrip(".").lower() + original_getaddrinfo = socket.getaddrinfo + + def _getaddrinfo(host, port, family=0, type=0, proto=0, flags=0): # noqa: A002 + if str(host).rstrip(".").lower() != pinned_host: + return original_getaddrinfo(host, port, family, type, proto, flags) + infos = [] + for ip in resolved_ips: + addr = ipaddress.ip_address(ip) + addr_family = socket.AF_INET6 if addr.version == 6 else socket.AF_INET + if family not in (0, socket.AF_UNSPEC, addr_family): + continue + sockaddr = (ip, port or 0, 0, 0) if addr_family == socket.AF_INET6 else (ip, port or 0) + infos.append((addr_family, type or socket.SOCK_STREAM, proto, "", sockaddr)) + return infos + + socket.getaddrinfo = _getaddrinfo + try: + yield + finally: + socket.getaddrinfo = original_getaddrinfo + + +class PinnedDNSAsyncTransport(httpx.AsyncBaseTransport): + """HTTPX transport that pins each request to the IPs validated for its URL.""" + + def __init__(self) -> None: + 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) + if not ok: + raise httpx.RequestError(error, request=request) + with pin_resolved_url_dns(url, resolved_ips): + return await self._inner.handle_async_request(request) + + async def aclose(self) -> None: + await self._inner.aclose() def validate_resolved_url(url: str) -> tuple[bool, str]: diff --git a/tests/security/test_security_network.py b/tests/security/test_security_network.py index 024293ba..9584fb20 100644 --- a/tests/security/test_security_network.py +++ b/tests/security/test_security_network.py @@ -11,6 +11,8 @@ import pytest from nanobot.security.network import ( configure_ssrf_whitelist, contains_internal_url, + pin_resolved_url_dns, + resolve_url_target, validate_url_target, ) @@ -157,6 +159,25 @@ def test_allows_public_ip(): assert ok, f"Should allow public IP, got: {err}" +def test_resolve_url_target_returns_validated_public_ips(): + with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve("example.com", ["93.184.216.34"])): + ok, err, resolved_ips = resolve_url_target("http://example.com/page") + + assert ok, err + assert resolved_ips == ("93.184.216.34",) + + +def test_pin_resolved_url_dns_prevents_second_resolution_rebind(): + def _rebinding_resolver(hostname, port, family=0, type_=0): + return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", ("169.254.169.254", 0))] + + with patch("nanobot.security.network.socket.getaddrinfo", _rebinding_resolver): + with pin_resolved_url_dns("http://example.com/page", ("93.184.216.34",)): + infos = socket.getaddrinfo("example.com", 80, socket.AF_UNSPEC, socket.SOCK_STREAM) + + assert infos[0][4][0] == "93.184.216.34" + + def test_allows_normal_https(): with patch("nanobot.security.network.socket.getaddrinfo", _fake_resolve("github.com", ["140.82.121.3"])): ok, err = validate_url_target("https://github.com/HKUDS/nanobot") diff --git a/tests/tools/test_mcp_tool.py b/tests/tools/test_mcp_tool.py index b14558ba..3b8763a1 100644 --- a/tests/tools/test_mcp_tool.py +++ b/tests/tools/test_mcp_tool.py @@ -804,6 +804,7 @@ 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.httpx, "AsyncClient", _async_client_with_mock_transport) monkeypatch.setattr(sys.modules["mcp.client.sse"], "sse_client", _fake_sse_client) monkeypatch.setattr( diff --git a/tests/tools/test_web_fetch_security.py b/tests/tools/test_web_fetch_security.py index 6fb1d0f6..dfda4b92 100644 --- a/tests/tools/test_web_fetch_security.py +++ b/tests/tools/test_web_fetch_security.py @@ -10,7 +10,7 @@ import httpx import pytest from nanobot.agent.tools import web as web_module -from nanobot.agent.tools.web import WebFetchTool +from nanobot.agent.tools.web import WebFetchTool, _get_with_safe_redirects from nanobot.config.schema import WebFetchConfig from nanobot.security.workspace_access import ( bind_workspace_scope, @@ -97,6 +97,30 @@ async def test_web_fetch_result_contains_untrusted_flag(): assert "[External content" in data.get("text", "") +@pytest.mark.asyncio +async def test_safe_redirect_request_pins_validated_dns(monkeypatch): + calls: list[str] = [] + + def _rebinding_resolver(hostname, port, family=0, type_=0): + calls.append(hostname) + ip = "93.184.216.34" if len(calls) == 1 else "169.254.169.254" + return [(socket.AF_INET, socket.SOCK_STREAM, 0, "", (ip, 0))] + + class FakeClient: + async def get(self, url, headers=None, follow_redirects=False): + infos = socket.getaddrinfo("attacker.example", 443, socket.AF_UNSPEC, socket.SOCK_STREAM) + assert infos[0][4][0] == "93.184.216.34" + return httpx.Response(200, request=httpx.Request("GET", url)) + + monkeypatch.setattr("nanobot.security.network.socket.getaddrinfo", _rebinding_resolver) + + response, error = await _get_with_safe_redirects(FakeClient(), "https://attacker.example/") + + assert error is None + assert response is not None + assert calls == ["attacker.example"] + + @pytest.mark.asyncio async def test_web_fetch_can_skip_jina_and_use_custom_user_agent(monkeypatch): tool = WebFetchTool(