fix(common-http): exempt the egress proxy only on proxy connections

benchmarks/multimodal/http/sweep.py

@@ -9,6 +9,11 @@
 one table for each pair, and one grid for each request rate that lists all of
 its pairs.
 
+The media server runs on localhost, and the fetch path refuses private
+addresses by default. So the sweep sets DYN_MM_ALLOW_INTERNAL=1 when the
+variable is not set. If every request of a pair fails, the sweep exits with
+status 1.
+
 Usage:
   python -m benchmarks.multimodal.http.sweep \\
       --server-processing-time-means-ms 50,100,200,300,400 \\
@@ -21,6 +26,7 @@
 
 import argparse
 import asyncio
+import os
 import resource
 import sys
 
@@ -101,6 +107,7 @@ async def _run_sweep(args: argparse.Namespace) -> int:
     peak_inflight = int(max(rates) * (max(means) / 1000.0)) + args.requests // 10
     _check_ulimit(peak_inflight)
 
+    failed_pairs: list[str] = []
     for rate in rates:
         print_batch_header(request_rate=rate, requests=args.requests)
         grid_rows: list[tuple[float, Summary]] = []
@@ -115,15 +122,30 @@ async def _run_sweep(args: argparse.Namespace) -> int:
             summary = summarize(result)
             print_iteration(mean_ms, summary)
             grid_rows.append((mean_ms, summary))
+            # A pair with no success measured nothing: its latencies print as 0.
+            if summary.n and not summary.outcomes.get("success"):
+                outcomes = ", ".join(
+                    f"{name}={count}"
+                    for name, count in sorted(summary.outcomes.items())
+                )
+                failed_pairs.append(
+                    f"request_rate={rate:g} mean_ms={mean_ms:g}: {outcomes}"
+                )
 
         print_grid(grid_rows)
         print()
 
-    return 0
+    for pair in failed_pairs:
+        print(f"[sweep] every request failed at {pair}", file=sys.stderr)
+    return 1 if failed_pairs else 0
 
 
 def main(argv: list[str] | None = None) -> int:
     args = parse_args(argv)
+    # The media server runs on localhost. The client refuses private addresses
+    # unless the deployment allows them, and a per-request policy cannot loosen
+    # that. Only here, not in _run_sweep: tests call that in a shared process.
+    os.environ.setdefault("DYN_MM_ALLOW_INTERNAL", "1")
     return asyncio.run(_run_sweep(args))
 
 

benchmarks/multimodal/http/tests/test_http_sweep_runner.py

@@ -1,25 +1,33 @@
 # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 
-"""Regression tests: the sweep harness must not attribute a result to a backend.
-
-The harness used to select a backend by writing ``DYN_HTTP_BACKEND``, and it
-labeled each result with the name that the caller passed. The facade now logs
-a warning for any value other than ``aiohttp`` and uses aiohttp. A run
-requested as ``httpx`` therefore used ``AiohttpClient`` but printed under an
-``httpx`` column, so both columns of the table measured aiohttp. There is one
-backend now, so the harness takes no selector and prints no backend label.
+"""Regression tests for the sweep harness.
+
+The harness must not attribute a result to a backend. It used to select a
+backend by writing ``DYN_HTTP_BACKEND``, and it labeled each result with the
+name that the caller passed. The facade now logs a warning for any value other
+than ``aiohttp`` and uses aiohttp. A run requested as ``httpx`` therefore used
+``AiohttpClient`` but printed under an ``httpx`` column, so both columns of the
+table measured aiohttp. There is one backend now, so the harness takes no
+selector and prints no backend label.
+
+The harness must also reach its own media server, and it must not exit 0 when
+it measured nothing.
 """
 
 from __future__ import annotations
 
 import contextlib
+import http.server
 import os
+import socket
+import threading
 
 import pytest
 
 from benchmarks.multimodal.http import sweep
-from benchmarks.multimodal.http.runner import run_one
+from benchmarks.multimodal.http.runner import RunResult, run_one
+from dynamo.common.http import _ssrf_resolver
 
 # Leave these tests unmarked. The root conftest.py then adds ``pre_merge``,
 # ``gpu_0`` and ``defaulted``, and the dynamo-runtime pipeline runs tests with
@@ -90,3 +98,139 @@ async def test_run_one_does_not_clobber_an_operator_set_backend(monkeypatch) ->
     monkeypatch.setenv("DYN_HTTP_BACKEND", "aiohttp")
     await run_one([], timeout=1.0, request_rate=100.0)
     assert os.environ["DYN_HTTP_BACKEND"] == "aiohttp"
+
+
+@pytest.mark.unit
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("labels", "expected_exit"),
+    [
+        (["HttpConnectionError"] * 3, 1),
+        (["HttpConnectionError", "HttpConnectionError", "success"], 0),
+    ],
+    ids=["every-request-failed", "one-succeeded"],
+)
+async def test_sweep_exit_status_flags_a_pair_that_measured_nothing(
+    monkeypatch, capsys, labels, expected_exit
+) -> None:
+    """A pair in which every request failed prints zero latencies, so the sweep
+    must not exit 0. One success is a measurement, and the exit stays 0."""
+
+    async def canned_run_one(urls, timeout, request_rate):
+        samples = [(0.01, label) for label in labels]
+        return RunResult(n=len(samples), wall_s=0.1, samples=samples)
+
+    @contextlib.contextmanager
+    def fake_media_server(**kwargs):
+        yield "http://media.invalid/test"
+
+    monkeypatch.setattr(sweep, "run_one", canned_run_one)
+    monkeypatch.setattr(sweep, "local_media_server", fake_media_server)
+    args = sweep.parse_args(
+        [
+            "--server-processing-time-means-ms",
+            "10",
+            "--request-rate",
+            "5",
+            "--requests",
+            "3",
+        ]
+    )
+
+    assert await sweep._run_sweep(args) == expected_exit
+    err = capsys.readouterr().err
+    failed = "every request failed at request_rate=5 mean_ms=10: HttpConnectionError=3"
+    assert (failed in err) is (expected_exit == 1)
+
+
+class _LoopbackResolver:
+    def __init__(self, *args, **kwargs) -> None:
+        pass
+
+    async def resolve(self, host, port=0, family=socket.AF_INET):
+        return [
+            {
+                "hostname": host,
+                "host": "127.0.0.1",
+                "port": port,
+                "family": socket.AF_INET,
+                "proto": 0,
+                "flags": 0,
+            }
+        ]
+
+    async def close(self) -> None:
+        pass
+
+
+@pytest.mark.integration
+@pytest.mark.timeout(30)
+def test_main_reaches_the_local_media_server(monkeypatch) -> None:
+    """The media server runs on localhost, which the fetch path refuses by
+    default. The sweep used to fail every request there and still exit 0.
+
+    This server answers on loopback under a host name, so each fetch passes
+    the client's connect-time address check, as the real sweep's localhost URL
+    does. An IP literal would skip that check, and the test would pass without
+    the fix.
+    """
+    hits: list[str] = []
+
+    class _Handler(http.server.BaseHTTPRequestHandler):
+        def do_GET(self) -> None:  # noqa: N802 - http.server API
+            hits.append(self.path)
+            body = b"image"
+            self.send_response(200)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, *args) -> None:
+            pass
+
+    @contextlib.contextmanager
+    def fake_media_server(**kwargs):
+        yield f"http://media.sweep.test:{server.server_port}/test"
+
+    monkeypatch.setattr(sweep, "local_media_server", fake_media_server)
+    monkeypatch.setattr(_ssrf_resolver, "DefaultResolver", _LoopbackResolver)
+    for name in (
+        "http_proxy",
+        "https_proxy",
+        "all_proxy",
+        "no_proxy",
+        "HTTP_PROXY",
+        "HTTPS_PROXY",
+        "ALL_PROXY",
+        "NO_PROXY",
+    ):
+        monkeypatch.delenv(name, raising=False)
+    # Set, then delete: monkeypatch then restores the variable as it was before
+    # the test, which also removes the value that main() writes.
+    monkeypatch.setenv("DYN_MM_ALLOW_INTERNAL", "")
+    monkeypatch.delenv("DYN_MM_ALLOW_INTERNAL")
+    # Start the server last, so that no setup step can fail and leave it
+    # running.
+    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        exit_status = sweep.main(
+            [
+                "--server-processing-time-means-ms",
+                "1",
+                "--request-rate",
+                "50",
+                "--requests",
+                "3",
+                "--timeout",
+                "5",
+            ]
+        )
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+    assert len(hits) == 3
+    assert exit_status == 0

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="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.",
+            help="Pool size cap per connect-time policy (aiohttp TCPConnector.limit). The aiohttp client keeps one pool per connect-time policy outcome, at most three: one that may reach private addresses (DYN_MM_ALLOW_INTERNAL=1), one for strict direct fetches, and one for strict fetches through a trusted egress proxy (DYN_MM_TRUST_EGRESS_PROXY=1). The total can therefore reach three times this value.",
         )
         add_argument(
             g,

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

@@ -33,6 +33,10 @@ only a resolver it created.
 > fetch that goes direct is never refused. It does not apply when
 > `DYN_MM_ALLOW_INTERNAL=1`, which already permits private destinations.
 >
+> A fetch through a trusted proxy runs on its own session. Only that session
+> resolves the configured proxy unfiltered. A direct fetch to the proxy's host
+> gets the same check as any other origin.
+>
 > Note aiohttp never calls a resolver for an IP literal, so literal blocked
 > addresses are `validate_url`'s job rather than the backstop's.
 

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

@@ -33,8 +33,8 @@ 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.
+def _env_proxy_endpoints() -> frozenset[tuple[str, int]]:
+    """``(host, port)`` of each 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
@@ -44,11 +44,15 @@ def _env_proxy_hosts() -> frozenset[str]:
     into ClientConnectorDNSError. Exempt the configured proxy so proxied
     deployments keep working, and see the module docstring for what that means
     for enforcement.
+
+    aiohttp resolves a proxy at its URL's ``raw_host`` and ``port``, which is
+    the scheme default when the URL names none, so another port on the proxy's
+    host is an origin and stays filtered.
     """
     return frozenset(
-        info.proxy.host
+        (info.proxy.raw_host, info.proxy.port)
         for info in proxies_from_env().values()
-        if info.proxy.host is not None
+        if info.proxy.raw_host is not None and info.proxy.port is not None
     )
 
 
@@ -64,10 +68,14 @@ class BlocklistResolver(AbstractResolver):
     is the one that rebinds.
     """
 
-    def __init__(self, *, allow_private_ips: bool) -> None:
+    def __init__(self, *, allow_private_ips: bool, via_proxy: bool = False) -> None:
         self._inner = DefaultResolver()
         self._allow_private_ips = allow_private_ips
-        self._proxy_hosts = _env_proxy_hosts()
+        # Only the resolver of a connector that dials a proxy exempts one. A
+        # direct fetch to the proxy's host and port is an origin like any
+        # other, and aiohttp caches answers per connector, keyed by
+        # (host, port) alone, so the two must not share a connector.
+        self._proxy_endpoints = _env_proxy_endpoints() if via_proxy else frozenset()
 
     async def resolve(
         self,
@@ -76,7 +84,7 @@ async def resolve(
         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:
+        if self._allow_private_ips or (host, port) in self._proxy_endpoints:
             return hosts
         allowed = [h for h in hosts if not is_blocked_ip(h["host"])]
         if not allowed: