fix(common-http): pin validated DNS answers at connect time

components/src/dynamo/common/http/README.md

@@ -11,9 +11,22 @@ uses aiohttp.
 `AiohttpClient` is the single supported backend. Its connector queues
 pending connections in `O(1)`, so latency stays close to the offered
 rate when one request fans out to many URLs (e.g. 100 image fetches),
-and it exposes a `TCPConnector(resolver=...)` DNS hook that can pin
-validated DNS answers for a connect-time SSRF backstop (the default
-client here uses the stock resolver; the pinning lands as a follow-up). See the
+and it exposes a `TCPConnector(resolver=...)` DNS hook that pins the
+validated DNS answers as a connect-time SSRF backstop against DNS
+rebinding — the default client wires a `BlocklistResolver` (see
+`_ssrf_resolver.py`) keyed to the `DYN_MM_ALLOW_INTERNAL` env baseline.
+
+> [!IMPORTANT]
+> The backstop governs **direct** connections only. With an egress proxy
+> configured (`HTTP_PROXY` / `HTTPS_PROXY` — the session runs `trust_env=True`),
+> the connector dials the proxy and the *proxy* resolves the origin, out of this
+> resolver's sight. The configured proxy is therefore exempt from filtering, so
+> a proxy on a private address keeps working; enforcement for proxied fetches
+> has to happen at the proxy or network layer. Note aiohttp never calls a
+> resolver for an IP literal, so literal blocked addresses are `validate_url`'s
+> job rather than the backstop's.
+
+See the
 [NeMo Gym aiohttp vs httpx note](https://docs.nvidia.com/nemo/gym/latest/infrastructure/engineering-notes/aiohttp-vs-httpx.html)
 for the fan-out latency comparison.
 

components/src/dynamo/common/http/__init__.py

@@ -11,9 +11,14 @@
 
 ``DYN_HTTP_BACKEND`` accepts only ``aiohttp``; any other value logs a warning
 and uses aiohttp. aiohttp scales well under fan-out and exposes a
-``TCPConnector(resolver=...)`` hook that a policy-aware resolver can use to pin
-validated DNS answers (the connect-time SSRF backstop lands on top of this
-capability; it is not wired in the default client here).
+``TCPConnector(resolver=...)`` hook. The default client wires a
+``BlocklistResolver`` (``_ssrf_resolver.py``) into it that pins validated DNS
+answers at connect time — the SSRF backstop against DNS rebinding, keyed to the
+``DYN_MM_ALLOW_INTERNAL`` env baseline. It governs **direct** connections: with
+an egress proxy configured the proxy resolves the origin, so the configured
+proxy is exempt from filtering and enforcement for proxied fetches belongs at
+the proxy or network layer. IP literals never reach a resolver in aiohttp and
+stay ``validate_url``'s job.
 """
 
 from __future__ import annotations

components/src/dynamo/common/http/_ssrf_resolver.py

@@ -0,0 +1,91 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Connect-time SSRF backstop for the aiohttp backend.
+
+``validate_url`` checks a hostname's resolved IPs, but the client re-resolves at
+connect, so a DNS-rebinding server can return a public IP on the check and an
+internal one at connect. aiohttp's ``TCPConnector(resolver=...)`` hook lets us
+resolve + filter once and hand the connector the validated addresses to dial,
+while the hostname is still used for TLS SNI / certificate verification — the
+same mechanism the Rust frontend uses on reqwest's ``dns_resolver``
+(``lib/llm/src/preprocessor/media/loader.rs``). Reuses
+:func:`url_validator.is_blocked_ip`.
+
+Scope: this governs **direct** connections. When an egress proxy is configured,
+the proxy resolves the origin, so SSRF must be enforced at the proxy / network
+layer instead (true of the Rust path as well).
+"""
+
+from __future__ import annotations
+
+import socket
+
+from aiohttp.abc import AbstractResolver, ResolveResult
+from aiohttp.helpers import proxies_from_env
+from aiohttp.resolver import DefaultResolver
+
+from .url_validator import describe_media_source, is_blocked_ip
+
+
+class SsrfBlockedAddress(OSError):
+    """Raised at connect time when every resolved IP is in a blocked range."""
+
+
+def _env_proxy_hosts() -> frozenset[str]:
+    """Hosts of any egress proxy configured in the environment.
+
+    The session runs with ``trust_env=True``, so when a proxy is configured the
+    connector dials *the proxy*, and it is the proxy's own address that reaches
+    this resolver -- the origin is resolved by the proxy, out of our sight. A
+    corporate proxy on a private address would therefore be filtered here and
+    take out every fetch: measured, HTTP_PROXY at a private host turns each one
+    into ClientConnectorDNSError. Exempt the configured proxy so proxied
+    deployments keep working, and see the module docstring for what that means
+    for enforcement.
+    """
+    return frozenset(
+        info.proxy.host
+        for info in proxies_from_env().values()
+        if info.proxy.host is not None
+    )
+
+
+class BlocklistResolver(AbstractResolver):
+    """aiohttp resolver that drops blocked IPs before the connector dials.
+
+    Returns the full set of non-blocked addresses (not just the first) so
+    aiohttp keeps its normal multi-address / Happy-Eyeballs fallback.
+
+    Note aiohttp short-circuits IP literals in ``TCPConnector._resolve_host``
+    and never calls a resolver for them, so literal blocked addresses are
+    ``validate_url``'s job, not this one. This covers the hostname case, which
+    is the one that rebinds.
+    """
+
+    def __init__(self, *, allow_private_ips: bool) -> None:
+        self._inner = DefaultResolver()
+        self._allow_private_ips = allow_private_ips
+        self._proxy_hosts = _env_proxy_hosts()
+
+    async def resolve(
+        self,
+        host: str,
+        port: int = 0,
+        family: socket.AddressFamily = socket.AF_INET,
+    ) -> list[ResolveResult]:
+        hosts = await self._inner.resolve(host, port, family)
+        if self._allow_private_ips or host in self._proxy_hosts:
+            return hosts
+        allowed = [h for h in hosts if not is_blocked_ip(h["host"])]
+        if not allowed:
+            # ``host`` is caller-supplied and unbounded; this text reaches an
+            # error response and a log line through the facade's connection
+            # error, so bound it the way every other message on this path is.
+            raise SsrfBlockedAddress(
+                f"host {describe_media_source(host)} resolves only to blocked IPs"
+            )
+        return allowed
+
+    async def close(self) -> None:
+        await self._inner.close()

components/src/dynamo/common/http/aiohttp_client.py

@@ -17,14 +17,19 @@
 import aiohttp
 from yarl import URL
 
+from ._ssrf_resolver import BlocklistResolver
 from .base import (
     HttpClient,
     HttpConnectionError,
     HttpStatusError,
     HttpTimeoutError,
     collect_capped,
 )
-from .url_validator import describe_error_detail, describe_media_source
+from .url_validator import (
+    UrlValidationPolicy,
+    describe_error_detail,
+    describe_media_source,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -58,14 +63,22 @@ def _effective_timeout(self, timeout: float) -> aiohttp.ClientTimeout:
         )
 
     def _build_session(self) -> aiohttp.ClientSession:
-        connector = aiohttp.TCPConnector(
+        connector_kwargs = dict(
             limit=self._config.max_connections,
             # Single-origin fan-out is the whole point; capping per-host
             # would defeat it. Hard-coded rather than env-tunable.
             limit_per_host=0,
             keepalive_timeout=self._config.keepalive_timeout,
             enable_cleanup_closed=True,
         )
+        # Connect-time SSRF backstop against DNS rebinding (see _ssrf_resolver).
+        # Deliberately keyed to the env baseline, not the per-call fetch policy:
+        # this shared connector must not let a per-request allow_private_ips=True
+        # loosen the SSRF backstop. DYN_MM_ALLOW_INTERNAL is the deployment knob.
+        connector_kwargs["resolver"] = BlocklistResolver(
+            allow_private_ips=UrlValidationPolicy.from_env().allow_private_ips
+        )
+        connector = aiohttp.TCPConnector(**connector_kwargs)
         return aiohttp.ClientSession(connector=connector, trust_env=True)
 
     async def _get_session(self) -> aiohttp.ClientSession:

components/src/dynamo/common/tests/http/test_ssrf_resolver.py

@@ -0,0 +1,108 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Unit tests for the aiohttp connect-time SSRF resolver (``_ssrf_resolver``).
+
+Exercise the DNS-rebinding case deterministically: the resolver is fed a mix of
+public and blocked answers (as if a rebinding server flipped between check and
+connect) and must drop the blocked ones, keep the rest, and fail closed when
+none remain. No network.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from dynamo.common.http._ssrf_resolver import BlocklistResolver, SsrfBlockedAddress
+
+pytestmark = [
+    pytest.mark.unit,
+    pytest.mark.pre_merge,
+    pytest.mark.gpu_0,
+]
+
+
+class _FakeInner:
+    """Stand-in for aiohttp's DefaultResolver returning canned entries."""
+
+    def __init__(self, ips: list[str]) -> None:
+        self._ips = ips
+
+    async def resolve(self, host, port=0, family=0):
+        return [{"hostname": host, "host": ip, "port": port} for ip in self._ips]
+
+    async def close(self):
+        pass
+
+
+def _resolver_with(ips: list[str]) -> BlocklistResolver:
+    r = BlocklistResolver(allow_private_ips=False)
+    r._inner = _FakeInner(ips)
+    return r
+
+
+async def test_resolver_drops_blocked_and_keeps_the_rest() -> None:
+    # Rebinding answer (public + metadata IP): the blocked one is dropped, and
+    # every non-blocked address survives for multi-address fallback.
+    resolver = _resolver_with(["93.184.216.34", "169.254.169.254", "8.8.8.8"])
+    out = await resolver.resolve("evil.example.com")
+    assert [h["host"] for h in out] == ["93.184.216.34", "8.8.8.8"]
+
+
+async def test_resolver_fails_closed_when_only_blocked() -> None:
+    resolver = _resolver_with(["169.254.169.254"])
+    with pytest.raises(SsrfBlockedAddress):
+        await resolver.resolve("evil.example.com")
+
+
+async def test_allow_private_ips_bypasses_filtering() -> None:
+    # The env baseline (DYN_MM_ALLOW_INTERNAL=1) opts the deployment into
+    # internal targets, so the resolver returns every answer unfiltered.
+    resolver = BlocklistResolver(allow_private_ips=True)
+    resolver._inner = _FakeInner(["169.254.169.254", "10.0.0.5"])
+    out = await resolver.resolve("internal.svc")
+    assert [h["host"] for h in out] == ["169.254.169.254", "10.0.0.5"]
+
+
+async def test_configured_egress_proxy_is_not_filtered(monkeypatch) -> None:
+    """A corporate proxy on a private address must not be filtered out.
+
+    The session runs with ``trust_env=True``, so with a proxy configured the
+    connector dials the proxy and it is the proxy's own address that reaches
+    this resolver. Filtering it turns every fetch into a connection error --
+    measured before the exemption, HTTP_PROXY at a private host failed them all.
+    """
+    # aiohttp's proxies_from_env() prefers the lowercase name, and CI images can
+    # carry one already -- clear both so this test controls what is configured.
+    monkeypatch.delenv("http_proxy", raising=False)
+    monkeypatch.setenv("HTTP_PROXY", "http://proxy.internal:3128")
+
+    resolver = BlocklistResolver(allow_private_ips=False)
+    resolver._inner = _FakeInner(["10.1.2.3"])
+
+    hosts = await resolver.resolve("proxy.internal")
+
+    assert [h["host"] for h in hosts] == ["10.1.2.3"]
+
+
+async def test_a_non_proxy_host_is_still_filtered(monkeypatch) -> None:
+    """Control: the exemption is for the configured proxy only."""
+    monkeypatch.delenv("http_proxy", raising=False)
+    monkeypatch.setenv("HTTP_PROXY", "http://proxy.internal:3128")
+
+    resolver = BlocklistResolver(allow_private_ips=False)
+    resolver._inner = _FakeInner(["10.1.2.3"])
+
+    with pytest.raises(SsrfBlockedAddress):
+        await resolver.resolve("origin.example.com")
+
+
+async def test_blocked_message_bounds_the_hostname() -> None:
+    """The host is caller-supplied; this text reaches a response and a log."""
+    resolver = _resolver_with(["169.254.169.254"])
+
+    with pytest.raises(SsrfBlockedAddress) as excinfo:
+        await resolver.resolve("h" * 200_000 + ".example.com")
+
+    assert len(str(excinfo.value)) < 500
+    assert "h" * 200 not in str(excinfo.value)