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

components/src/dynamo/common/configuration/groups/http_args.py

@@ -146,7 +146,7 @@ def add_arguments(self, parser: argparse.ArgumentParser) -> None:
             default=100,
             arg_type=int,
             dest="max_connections",
-            help="Total pool size cap (httpx Limits.max_connections / aiohttp TCPConnector.limit).",
+            help="Pool size cap per connect-time policy (aiohttp TCPConnector.limit). The aiohttp client keeps one pool per connect-time policy outcome, at most two, so a deployment that sets DYN_MM_ALLOW_INTERNAL=1 and also issues stricter per-request policies can reach twice this value in total.",
         )
         add_argument(
             g,

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

@@ -11,9 +11,32 @@ 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`). The connector may return private addresses only when the
+`DYN_MM_ALLOW_INTERNAL` deployment baseline **and** the request policy both
+allow it, so neither side alone can weaken the other. The client keeps one
+session per outcome and closes each resolver itself, because aiohttp closes
+only a resolver it created.
+
+> [!IMPORTANT]
+> The backstop governs **direct** connections only. When a proxy applies
+> (`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, so the check cannot govern the destination.
+>
+> A policy-protected fetch that a proxy would carry therefore **fails closed**.
+> Set `DYN_MM_TRUST_EGRESS_PROXY=1` to assert that the proxy enforces
+> destination policy itself, and the fetch proceeds. The gate asks aiohttp
+> which proxy applies to that specific URL, so `NO_PROXY` is honored and a
+> fetch that goes direct is never refused. It does not apply when
+> `DYN_MM_ALLOW_INTERNAL=1`, which already permits private destinations.
+>
+> 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,16 @@
 
 ``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. The
+connector may return private addresses only when the ``DYN_MM_ALLOW_INTERNAL``
+deployment baseline and the request policy both allow it. It governs
+**direct** connections: when a proxy applies, the proxy resolves the origin,
+so a policy-protected fetch that would be proxied fails closed unless
+``DYN_MM_TRUST_EGRESS_PROXY=1`` asserts that the proxy enforces destination
+policy. ``NO_PROXY`` is honored, so a fetch that goes direct is never refused.
+IP literals never reach a resolver in aiohttp and stay ``validate_url``'s job.
 """
 
 from __future__ import annotations
@@ -32,6 +39,7 @@
 from .aiohttp_client import AiohttpClient
 from .base import (
     HttpClient,
+    HttpConfigurationError,
     HttpConnectionError,
     HttpError,
     HttpStatusError,
@@ -97,6 +105,7 @@ async def close_http_client() -> None:
     "AiohttpClient",
     "HttpError",
     "HttpTimeoutError",
+    "HttpConfigurationError",
     "HttpConnectionError",
     "HttpStatusError",
     "HttpConfigBase",

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

@@ -0,0 +1,98 @@
+# 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 errno
+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.
+            #
+            # Two-argument OSError on purpose: aiohttp wraps this in
+            # ClientConnectorDNSError and renders ``strerror``, so the
+            # one-argument form reaches the caller as a bare "[None]" and the
+            # reason is lost exactly where an operator would read it.
+            raise SsrfBlockedAddress(
+                errno.EHOSTUNREACH,
+                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

@@ -12,19 +12,27 @@
 
 import asyncio
 import logging
+import os
 from typing import Optional
 
 import aiohttp
+from aiohttp.helpers import get_env_proxy_for_url
 from yarl import URL
 
+from ._ssrf_resolver import BlocklistResolver
 from .base import (
     HttpClient,
+    HttpConfigurationError,
     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__)
 
@@ -35,13 +43,23 @@
 # bounds how far past the limit a single read can carry.
 _READ_CHUNK = 64 * 1024
 
+# Set to "1" to assert that the configured egress proxy enforces destination
+# policy itself. Spelled like DYN_MM_ALLOW_INTERNAL, which it sits beside.
+DYN_MM_TRUST_EGRESS_PROXY = "DYN_MM_TRUST_EGRESS_PROXY"
+
 
 class AiohttpClient(HttpClient):
     """aiohttp-backed concrete client."""
 
     def __init__(self, config=None) -> None:
         super().__init__(config)
-        self._session: Optional[aiohttp.ClientSession] = None
+        # Keyed by whether the connector may return private addresses. At most
+        # two entries, so a request that asks for a stricter policy than the
+        # deployment baseline gets a connector that actually enforces it.
+        self._sessions: dict[bool, aiohttp.ClientSession] = {}
+        # aiohttp marks an injected resolver as externally owned, so closing a
+        # session never closes it. Hold each one and close it ourselves.
+        self._resolvers: dict[bool, BlocklistResolver] = {}
 
     def _effective_timeout(self, timeout: float) -> aiohttp.ClientTimeout:
         # The override caps ``total`` (whole request). ``sock_connect``
@@ -57,21 +75,77 @@ def _effective_timeout(self, timeout: float) -> aiohttp.ClientTimeout:
             total=total, sock_connect=self._config.connect_timeout
         )
 
-    def _build_session(self) -> aiohttp.ClientSession:
+    @staticmethod
+    def _connect_allows_private(policy: Optional[UrlValidationPolicy]) -> bool:
+        """Whether the connector may return private addresses for this fetch.
+
+        The intersection of the deployment baseline and the request policy.
+        Taking the baseline alone lets ``DYN_MM_ALLOW_INTERNAL=1`` override a
+        caller that explicitly asked for ``allow_private_ips=False``, and
+        taking the request alone lets any caller opt out of the deployment
+        setting. Only when both allow it does the backstop stand down.
+        """
+        env_allows = UrlValidationPolicy.from_env().allow_private_ips
+        if policy is None:
+            return env_allows
+        return env_allows and policy.allow_private_ips
+
+    @staticmethod
+    async def _require_trusted_egress_proxy(url: str) -> None:
+        """Refuse a protected fetch that a proxy would put out of our reach.
+
+        When a proxy applies, the connector dials the proxy and the proxy
+        resolves the origin, so the origin hostname never reaches the
+        connect-time check. Measured: the resolver sees only the proxy host,
+        while the proxy receives the absolute URI. Documenting that boundary
+        does not enforce it, so fail closed unless an operator asserts that
+        the proxy enforces destination policy itself.
+
+        Asks aiohttp which proxy applies to *this* URL rather than whether any
+        proxy variable is set, so ``NO_PROXY`` is honored and a fetch that
+        would go direct is not refused.
+        """
+        if os.getenv(DYN_MM_TRUST_EGRESS_PROXY, "").strip() == "1":
+            return
+        try:
+            # aiohttp runs this same helper through asyncio.to_thread because
+            # it does proxy-bypass discovery and .netrc file reads. Match that
+            # rather than repeating the blocking work on the event loop.
+            await asyncio.to_thread(get_env_proxy_for_url, URL(url))
+        except LookupError:
+            # No proxy for this URL, so aiohttp dials the origin and the
+            # connect-time check governs it.
+            return
+        raise HttpConfigurationError(
+            f"{describe_media_source(url)} would be fetched through an egress "
+            "proxy, so the connect-time address check cannot govern the "
+            f"origin; set {DYN_MM_TRUST_EGRESS_PROXY}=1 to assert that the "
+            "proxy enforces destination policy, or unset the proxy"
+        )
+
+    def _build_session(self, allow_private_ips: bool) -> aiohttp.ClientSession:
+        resolver = BlocklistResolver(allow_private_ips=allow_private_ips)
+        self._resolvers[allow_private_ips] = resolver
         connector = aiohttp.TCPConnector(
             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). See _connect_allows_private for how the flag is
+            # derived. DYN_MM_ALLOW_INTERNAL is the deployment knob.
+            resolver=resolver,
         )
         return aiohttp.ClientSession(connector=connector, trust_env=True)
 
-    async def _get_session(self) -> aiohttp.ClientSession:
+    async def _get_session(self, allow_private_ips: bool) -> aiohttp.ClientSession:
         async with self._lock:
-            if self._session is None or self._session.closed:
-                self._session = self._build_session()
+            session = self._sessions.get(allow_private_ips)
+            if session is None or session.closed:
+                session = self._build_session(allow_private_ips)
+                self._sessions[allow_private_ips] = session
                 logger.info(
                     "aiohttp backend initialized: limit=%d, limit_per_host=0, "
                     "keepalive_timeout=%.1fs%s",
@@ -81,12 +155,22 @@ async def _get_session(self) -> aiohttp.ClientSession:
                     if self._config.per_call_timeout_override is not None
                     else "; total timeout set per-request",
                 )
-        return self._session
+        return session
 
     async def _fetch_simple(
-        self, url: str, timeout: float, *, max_bytes: Optional[int] = None
+        self,
+        url: str,
+        timeout: float,
+        *,
+        max_bytes: Optional[int] = None,
+        policy: Optional[UrlValidationPolicy] = None,
     ) -> bytes:
-        session = await self._get_session()
+        allow_private = self._connect_allows_private(policy)
+        # Only when the check is meant to bite. If private destinations are
+        # already permitted for this fetch, the proxy gate protects nothing.
+        if not allow_private:
+            await self._require_trusted_egress_proxy(url)
+        session = await self._get_session(allow_private)
         client_timeout = self._effective_timeout(timeout)
         try:
             async with session.get(
@@ -116,9 +200,19 @@ async def _fetch_simple(
             ) from e
 
     async def _fetch_body_or_redirect(
-        self, url: str, timeout: float, *, max_bytes: Optional[int] = None
+        self,
+        url: str,
+        timeout: float,
+        *,
+        max_bytes: Optional[int] = None,
+        policy: Optional[UrlValidationPolicy] = None,
     ) -> tuple[bytes | None, str | None]:
-        session = await self._get_session()
+        allow_private = self._connect_allows_private(policy)
+        # Only when the check is meant to bite. If private destinations are
+        # already permitted for this fetch, the proxy gate protects nothing.
+        if not allow_private:
+            await self._require_trusted_egress_proxy(url)
+        session = await self._get_session(allow_private)
         client_timeout = self._effective_timeout(timeout)
         try:
             async with session.get(
@@ -165,6 +259,13 @@ async def _fetch_body_or_redirect(
 
     async def close(self) -> None:
         async with self._lock:
-            if self._session is not None and not self._session.closed:
-                await self._session.close()
-            self._session = None
+            for session in self._sessions.values():
+                if not session.closed:
+                    await session.close()
+            self._sessions = {}
+            # TCPConnector sets _resolver_owner=False for an injected resolver
+            # and only closes the one it made itself, so closing the session
+            # above leaves ours open.
+            for resolver in self._resolvers.values():
+                await resolver.close()
+            self._resolvers = {}