From b3d3a3e6c35496551e47a6681d12a0afc385f728 Mon Sep 17 00:00:00 2001 From: chengyongru <2755839590@qq.com> Date: Mon, 27 Jul 2026 01:11:31 +0800 Subject: [PATCH] fix(image): delegate DNS to explicit proxy --- .agent/security.md | 4 +- docs/image-generation.md | 2 +- nanobot/config/schema.py | 2 +- nanobot/providers/image_generation.py | 5 ++- nanobot/security/network.py | 30 ++++++++++++-- .../test_image_generation_security.py | 14 +++---- tests/security/test_security_network.py | 41 +++++++++++++++++++ 7 files changed, 83 insertions(+), 15 deletions(-) diff --git a/.agent/security.md b/.agent/security.md index ca961266..91f2f41b 100644 --- a/.agent/security.md +++ b/.agent/security.md @@ -14,9 +14,9 @@ Shell execution (`ExecTool`, `agent/tools/shell.py`) also respects `restrict_to_ ## SSRF Protection -All outbound HTTP requests from agent tools must pass through `validate_url_target` (`security/network.py`). By default it blocks loopback, RFC1918 private addresses, CGNAT ranges, link-local ranges, and cloud metadata endpoints (including `169.254.169.254`). +All outbound HTTP requests from agent tools must pass through the shared URL guards in `security/network.py` (`validate_url_target` or `resolve_url_target`). By default they block loopback, RFC1918 private addresses, CGNAT ranges, link-local ranges, and cloud metadata endpoints (including `169.254.169.254`). -The only escape hatch is `configure_ssrf_whitelist(cidrs)`, which reads from `config.tools.ssrf_whitelist` at load time. +For direct requests, the only escape hatch is `configure_ssrf_whitelist(cidrs)`, which reads from `config.tools.ssrf_whitelist` at load time. An explicitly configured `providers..proxy` is a separate user-authorized trust boundary for provider requests and provider-returned image URL downloads. Those downloads still reject malformed URLs and locally identifiable private/internal targets on every redirect, but hostnames unavailable to local DNS are delegated to the trusted proxy. The user-selected proxy owns final DNS resolution and network egress policy. HTTP/SSE MCP transports are part of this boundary: validate configured MCP URLs before probing or constructing clients, and validate each outgoing HTTP request before redirects are followed. Local/private HTTP MCP endpoints are allowed only through the explicit SSRF whitelist. Stdio MCP servers are not part of the HTTP SSRF path. diff --git a/docs/image-generation.md b/docs/image-generation.md index bf727411..763bd775 100644 --- a/docs/image-generation.md +++ b/docs/image-generation.md @@ -72,7 +72,7 @@ Provider settings reuse normal provider config fields: | `providers..extraBody` | Extra JSON fields merged into provider request bodies | | `providers..proxy` | Explicit trusted HTTP proxy for provider requests and returned image URL downloads | -For providers that return image URLs, direct downloads use DNS pinning. When an explicit provider `proxy` is configured, nanobot validates the initial URL and every redirect locally, then relies on that trusted proxy for final DNS resolution and network egress. Process-wide proxy environment variables are not used for these downloads. +For providers that return image URLs, direct downloads use DNS pinning. When an explicit provider `proxy` is configured, nanobot rejects malformed URLs and locally identifiable private/internal targets on the initial URL and every redirect. Hostnames unavailable to local DNS are delegated to that trusted proxy, which owns final DNS resolution and network egress. Process-wide proxy environment variables are not used for these downloads. Both camelCase and snake_case config keys are accepted, but docs use camelCase to match `config.json`. diff --git a/nanobot/config/schema.py b/nanobot/config/schema.py index a36ded81..8b51404e 100644 --- a/nanobot/config/schema.py +++ b/nanobot/config/schema.py @@ -199,7 +199,7 @@ class ProviderConfig(Base): extra_headers: dict[str, str] | None = None # Custom headers (e.g. APP-Code for AiHubMix) extra_body: dict[str, Any] | None = None # Extra provider request fields; shape depends on provider/API surface extra_query: dict[str, str] | None = None # Extra query params (e.g. api-version for Azure-style gateways) - proxy: str | None = None # OpenAI-compatible/Codex HTTP proxy URL + proxy: str | None = None # Explicit HTTP proxy; image downloads trust its DNS and egress thinking_style: str | None = None # Thinking/reasoning style for custom providers # Valid values mirror the keys of _THINKING_STYLE_MAP in diff --git a/nanobot/providers/image_generation.py b/nanobot/providers/image_generation.py index d3ad017e..06170bac 100644 --- a/nanobot/providers/image_generation.py +++ b/nanobot/providers/image_generation.py @@ -161,7 +161,10 @@ async def _download_image_data_url( current_url = url for _ in range(_IMAGE_DOWNLOAD_MAX_REDIRECTS + 1): if proxy: - ok, error, _ = resolve_url_target(current_url) + ok, error, _ = resolve_url_target( + current_url, + trust_remote_dns=True, + ) if not ok: raise ImageGenerationError( f"blocked unsafe generated image URL: {error}" diff --git a/nanobot/security/network.py b/nanobot/security/network.py index 23daf980..dba5e14a 100644 --- a/nanobot/security/network.py +++ b/nanobot/security/network.py @@ -74,7 +74,12 @@ def _is_private(addr: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: return any(normalized in net for net in _BLOCKED_NETWORKS) -def resolve_url_target(url: str, *, allow_loopback: bool = False) -> tuple[bool, str, tuple[str, ...]]: +def resolve_url_target( + url: str, + *, + allow_loopback: bool = False, + trust_remote_dns: 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 @@ -82,8 +87,14 @@ def resolve_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. + ``trust_remote_dns`` accepts ordinary hostnames unavailable to local DNS. + This is only safe when a user-configured trusted proxy owns final DNS + resolution and network egress. Localhost names and private/internal IP + literals remain blocked. + Returns (ok, error_message, resolved_ips). When ok is True, - resolved_ips contains the public IPs that were validated for this URL. + resolved_ips contains the public IPs that were validated for this URL, or + is empty when an unresolved hostname is delegated to a trusted proxy. """ try: p = urlparse(url) @@ -102,7 +113,20 @@ def resolve_url_target(url: str, *, allow_loopback: bool = False) -> tuple[bool, try: infos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM) except socket.gaierror: - return False, f"Cannot resolve hostname: {hostname}", () + if not trust_remote_dns: + return False, f"Cannot resolve hostname: {hostname}", () + + normalized_hostname = hostname.rstrip(".").lower() + if normalized_hostname == "localhost" or normalized_hostname.endswith(".localhost"): + return False, f"Blocked local/internal hostname: {hostname}", () + + try: + literal_addr = ipaddress.ip_address(normalized_hostname) + except ValueError: + return True, "", () + if _is_private(literal_addr): + return False, f"Blocked private/internal address: {literal_addr}", () + return True, "", (str(_normalize_addr(literal_addr)),) addrs: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = [] for info in infos: diff --git a/tests/providers/test_image_generation_security.py b/tests/providers/test_image_generation_security.py index 48bc2e7f..5f0ca608 100644 --- a/tests/providers/test_image_generation_security.py +++ b/tests/providers/test_image_generation_security.py @@ -139,13 +139,13 @@ class _StreamContext: @pytest.mark.asyncio -async def test_generated_image_download_uses_explicit_provider_proxy( +async def test_generated_image_download_delegates_unresolved_host_to_provider_proxy( monkeypatch, ) -> None: - monkeypatch.setattr( - "nanobot.security.network.socket.getaddrinfo", - _resolve_public, - ) + def fail_local_dns(host: str, port: int | None, *args, **kwargs): + raise socket.gaierror(f"cannot resolve {host}") + + monkeypatch.setattr("nanobot.security.network.socket.getaddrinfo", fail_local_dns) captured: dict[str, object] = {} class FakeAsyncClient: @@ -167,12 +167,12 @@ async def test_generated_image_download_uses_explicit_provider_proxy( proxy = "http://127.0.0.1:23458" result = await _download_image_data_url( - "https://cdn.example/image.png", + "https://proxy-only.example/image.png", proxy=proxy, ) assert result.startswith("data:image/png;base64,") - assert captured["request"] == ("GET", "https://cdn.example/image.png") + assert captured["request"] == ("GET", "https://proxy-only.example/image.png") assert captured["kwargs"] == { "follow_redirects": False, "timeout": image_generation._DEFAULT_TIMEOUT_S, diff --git a/tests/security/test_security_network.py b/tests/security/test_security_network.py index 9b71022f..25b64631 100644 --- a/tests/security/test_security_network.py +++ b/tests/security/test_security_network.py @@ -195,6 +195,47 @@ def test_resolve_url_target_returns_validated_public_ips(): assert resolved_ips == ("93.184.216.34",) +@pytest.mark.parametrize( + ("trust_remote_dns", "expected_ok"), + [(False, False), (True, True)], +) +def test_resolve_url_target_only_delegates_dns_to_trusted_proxy( + trust_remote_dns: bool, + expected_ok: bool, +): + with patch( + "nanobot.security.network.socket.getaddrinfo", + side_effect=socket.gaierror("local DNS unavailable"), + ): + ok, err, resolved_ips = resolve_url_target( + "https://proxy-only.example/image.png", + trust_remote_dns=trust_remote_dns, + ) + + assert ok is expected_ok, err + assert resolved_ips == () + + +@pytest.mark.parametrize( + "url", + [ + "http://localhost/secret", + "http://service.localhost/secret", + "http://127.0.0.1/secret", + "http://169.254.169.254/latest", + "http://[::1]/secret", + ], +) +def test_resolve_url_target_does_not_delegate_local_targets(url: str): + with patch( + "nanobot.security.network.socket.getaddrinfo", + side_effect=socket.gaierror("local DNS unavailable"), + ): + ok, _, _ = resolve_url_target(url, trust_remote_dns=True) + + assert not ok + + 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))]