fix(egress): pin resolved public IP to close SSRF DNS-rebinding TOCTOU (S-1)

src/serving/api/alerts/escalation.py

@@ -8,7 +8,11 @@
 import httpx
 import structlog
 
-from src.serving.api.egress_guard import UnsafeEgressURLError, validate_public_url
+from src.serving.api.egress_guard import (
+    UnsafeEgressURLError,
+    pinned_transport,
+    resolve_public_ip,
+)
 from src.serving.api.webhook_dispatcher import _event_body, _signature
 from src.serving.control_plane import get_control_plane_store
 
@@ -263,10 +267,12 @@ async def deliver(
     error: str | None = None
 
     target_url = webhook_url or alert.webhook_url
-    # Re-validate at delivery time (DNS rebinding): a name public at registration
-    # could now resolve to an internal address. (audit_28_06_26.md #2)
+    # Resolve + validate at delivery time (DNS rebinding): a name public at
+    # registration could now resolve to an internal address. Resolve once and pin
+    # the connection to that IP so httpx does not re-resolve between the check and
+    # the connect (rebinding TOCTOU, audit S-1). (audit_28_06_26.md #2)
     try:
-        await asyncio.to_thread(validate_public_url, target_url)
+        pinned_ip = await asyncio.to_thread(resolve_public_ip, target_url)
     except UnsafeEgressURLError as exc:
         error = f"unsafe egress URL: {exc}"
         store.log_alert_delivery(
@@ -297,7 +303,9 @@ async def deliver(
             "attempts": 0,
         }
 
-    async with httpx.AsyncClient(timeout=5.0) as client:
+    async with httpx.AsyncClient(
+        timeout=5.0, transport=pinned_transport(target_url, pinned_ip)
+    ) as client:
         for attempt in range(1, 4):
             attempts = attempt
             error = None

src/serving/api/egress_guard.py

@@ -7,17 +7,28 @@
 
 This module resolves the host and rejects any URL that is not an http(s) target
 resolving exclusively to public unicast addresses. It is applied both at
-registration time (reject early, 4xx) and immediately before each delivery
-(narrowing the DNS-rebinding window — a name that resolved public at creation
-could later point at an internal IP).
+registration time (reject early, 4xx) and immediately before each delivery.
+
+**DNS-rebinding TOCTOU (audit S-1).** A validate-then-``post`` pair is racy: the
+guard resolves the host and approves it, then ``httpx`` re-resolves the *same
+hostname* to open the socket, so a low-TTL name that flips public→internal
+between the two lookups slips past the guard and is fetched. Re-validating
+before each delivery only narrows the window. The fix here removes the second
+lookup entirely: :func:`resolve_public_ip` resolves and validates the host
+**once** and returns the approved public IP, and :func:`pinned_transport` pins
+the connection to *that IP literal* — the transport never resolves the hostname
+again. The original ``Host`` header and TLS SNI / certificate hostname are
+preserved (via the ``sni_hostname`` request extension) so routing and TLS
+verification still key on the hostname, not the IP.
 
 ``AGENTFLOW_EGRESS_ALLOWED_HOSTS`` is an opt-in escape hatch (default empty, so
 production keeps the full guard): a comma-separated allowlist of exact
 hostnames a controlled deployment trusts even though they resolve to a
 private/loopback address — e.g. the ``host.docker.internal`` gateway the e2e
 compose stack delivers to, or the ``127.0.0.1`` relay the staging kind cluster
 stands up. It never relaxes the guard for tenant traffic, only for hosts the
-operator explicitly listed.
+operator explicitly listed. Allowlisted hosts are *not* pinned (they are trusted
+by name); every other host is.
 """
 
 from __future__ import annotations
@@ -27,6 +38,8 @@
 import socket
 from urllib.parse import urlsplit
 
+import httpx
+
 _ALLOWED_SCHEMES = {"http", "https"}
 _ALLOWLIST_ENV = "AGENTFLOW_EGRESS_ALLOWED_HOSTS"
 
@@ -67,15 +80,12 @@ def _ip_is_public(ip: str) -> bool:
     )
 
 
-def validate_public_url(url: str) -> None:
-    """Raise :class:`UnsafeEgressURLError` unless ``url`` is an http(s) URL whose
-    host resolves *only* to public unicast addresses.
-
-    Resolution is synchronous (``socket.getaddrinfo``); call it via
-    ``asyncio.to_thread`` on the event loop. IP-literal hosts resolve to
-    themselves, so loopback/private/link-local literals are rejected without any
-    network DNS.
-    """
+def _split_egress_url(url: str) -> tuple[str, str, int, bool]:
+    """Parse ``url`` and enforce the scheme/host rules shared by both entry
+    points. Return ``(scheme, host, port, allowlisted)``; raise
+    :class:`UnsafeEgressURLError` on a non-http(s) scheme or a missing host.
+    ``allowlisted`` is ``True`` when the operator opted this exact host out of
+    the public-address check."""
     parts = urlsplit(url)
     scheme = parts.scheme.lower()
     if scheme not in _ALLOWED_SCHEMES:
@@ -87,8 +97,22 @@ def validate_public_url(url: str) -> None:
         # Operator explicitly trusts this host (controlled test/relay target).
         # The scheme check above still applies; only the public-address check
         # is waived.
-        return
+        return scheme, host, parts.port or (443 if scheme == "https" else 80), True
     port = parts.port or (443 if scheme == "https" else 80)
+    return scheme, host, port, False
+
+
+def _resolve_public_ips(host: str, port: int) -> set[str]:
+    """Resolve ``host`` and return the set of resolved addresses, having verified
+    that **every** one is a public unicast address. Raise
+    :class:`UnsafeEgressURLError` if the host does not resolve or any resolved
+    address is non-public (so a host answering with both a public and an internal
+    IP is rejected, not partially trusted).
+
+    Resolution is synchronous (``socket.getaddrinfo``); call it off the event
+    loop via ``asyncio.to_thread``. IP-literal hosts resolve to themselves, so
+    loopback/private/link-local literals are rejected without any network DNS.
+    """
     try:
         infos = socket.getaddrinfo(host, port, proto=socket.IPPROTO_TCP)
     except (socket.gaierror, UnicodeError) as exc:
@@ -99,3 +123,73 @@ def validate_public_url(url: str) -> None:
     for ip in resolved:
         if not _ip_is_public(ip):
             raise UnsafeEgressURLError(f"host {host} resolves to non-public address {ip}")
+    return resolved
+
+
+def validate_public_url(url: str) -> None:
+    """Raise :class:`UnsafeEgressURLError` unless ``url`` is an http(s) URL whose
+    host resolves *only* to public unicast addresses.
+
+    Registration-time gate (reject obviously-unsafe targets early with a 4xx).
+    Delivery-time code should call :func:`resolve_public_ip` instead, which
+    additionally pins the connection and closes the rebinding TOCTOU.
+    """
+    _scheme, host, port, allowlisted = _split_egress_url(url)
+    if allowlisted:
+        return
+    _resolve_public_ips(host, port)
+
+
+def resolve_public_ip(url: str) -> str | None:
+    """Resolve ``url``'s host to a single validated public IP and return it, so
+    the caller can pin the connection to that literal (closing the DNS-rebinding
+    TOCTOU — no second name resolution happens between this check and the
+    connect).
+
+    Return ``None`` for an operator-allowlisted host: those are trusted by name
+    and connected as-is, without pinning. Raise :class:`UnsafeEgressURLError` on
+    a bad scheme / missing host / unresolvable host / any non-public resolved
+    address — exactly the rejections :func:`validate_public_url` makes.
+
+    Synchronous DNS; call via ``asyncio.to_thread`` on the event loop.
+    """
+    _scheme, host, port, allowlisted = _split_egress_url(url)
+    if allowlisted:
+        return None
+    ips = _resolve_public_ips(host, port)
+    # Deterministic pick; every address in the set was verified public above, so
+    # any of them is a safe pin target.
+    return sorted(ips)[0]
+
+
+class _PinnedIPTransport(httpx.AsyncHTTPTransport):
+    """httpx transport that rewrites each request's host to a pre-validated
+    public IP literal before connecting, so httpx performs **no** second DNS
+    resolution (the rebinding TOCTOU close). The original ``Host`` header — set
+    by httpx at request-build time from the hostname URL — is left untouched, and
+    the hostname is carried into the TLS handshake as ``sni_hostname`` so SNI and
+    certificate verification still key on the hostname rather than the IP."""
+
+    def __init__(self, hostname: str, pinned_ip: str) -> None:
+        super().__init__()
+        self._hostname = hostname
+        self._pinned_ip = pinned_ip
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        request.extensions = {**request.extensions, "sni_hostname": self._hostname}
+        request.url = request.url.copy_with(host=self._pinned_ip)
+        return await super().handle_async_request(request)
+
+
+def pinned_transport(url: str, pinned_ip: str | None) -> httpx.AsyncBaseTransport | None:
+    """Build the transport for delivering to ``url``.
+
+    Given the ``pinned_ip`` from :func:`resolve_public_ip`, return a transport
+    that connects to that IP literal while preserving the hostname's ``Host``
+    header and TLS identity. Return ``None`` when ``pinned_ip`` is ``None`` (an
+    allowlisted host) so the caller uses httpx's default transport unchanged.
+    """
+    if pinned_ip is None:
+        return None
+    host = urlsplit(url).hostname or ""
+    return _PinnedIPTransport(host, pinned_ip)

src/serving/api/webhook_dispatcher.py

@@ -19,7 +19,11 @@
 from fastapi import FastAPI
 from pydantic import BaseModel, Field
 
-from src.serving.api.egress_guard import UnsafeEgressURLError, validate_public_url
+from src.serving.api.egress_guard import (
+    UnsafeEgressURLError,
+    pinned_transport,
+    resolve_public_ip,
+)
 from src.serving.api.metrics import WEBHOOK_SETTLE_VIOLATIONS
 from src.serving.backends import BackendExecutionError
 from src.serving.control_plane import get_control_plane_store
@@ -529,12 +533,14 @@ async def _deliver_body(
         status_code: int | None = None
         error: str | None = None
 
-        # Re-validate at delivery time too (not only at registration): a hostname
-        # that resolved to a public IP when the webhook was created could now
-        # point at an internal address (DNS rebinding). Fail the delivery instead
-        # of fetching an internal target. (audit_28_06_26.md #2)
+        # Resolve + validate at delivery time (not only at registration): a
+        # hostname that resolved to a public IP when the webhook was created
+        # could now point at an internal address (DNS rebinding). Resolve once
+        # here and pin the connection to that IP so httpx does not re-resolve
+        # between the check and the connect (rebinding TOCTOU, audit S-1). Fail
+        # the delivery instead of fetching an internal target. (audit_28_06_26.md #2)
         try:
-            await asyncio.to_thread(validate_public_url, webhook.url)
+            pinned_ip = await asyncio.to_thread(resolve_public_ip, webhook.url)
         except UnsafeEgressURLError as exc:
             error = f"unsafe egress URL: {exc}"
             store.log_webhook_delivery(
@@ -558,7 +564,9 @@ async def _deliver_body(
                 "attempts": 0,
             }
 
-        async with httpx.AsyncClient(timeout=5.0) as client:
+        async with httpx.AsyncClient(
+            timeout=5.0, transport=pinned_transport(webhook.url, pinned_ip)
+        ) as client:
             for attempt in range(1, 4):
                 attempts = attempt
                 error = None

tests/unit/test_egress_guard.py

@@ -7,11 +7,19 @@
 
 from __future__ import annotations
 
+import asyncio
 import socket
 
+import httpx
 import pytest
 
-from src.serving.api.egress_guard import UnsafeEgressURLError, validate_public_url
+from src.serving.api.egress_guard import (
+    UnsafeEgressURLError,
+    _PinnedIPTransport,
+    pinned_transport,
+    resolve_public_ip,
+    validate_public_url,
+)
 
 
 @pytest.mark.parametrize(
@@ -161,3 +169,88 @@ def test_empty_allowlist_preserves_loopback_rejection(monkeypatch: pytest.Monkey
     monkeypatch.setenv("AGENTFLOW_EGRESS_ALLOWED_HOSTS", "")
     with pytest.raises(UnsafeEgressURLError):
         validate_public_url("http://127.0.0.1/x")
+
+
+# --- IP pinning: DNS-rebinding TOCTOU close (audit S-1) -----------------------
+# resolve_public_ip resolves+validates once and returns the approved public IP;
+# pinned_transport connects to that literal so httpx never re-resolves the
+# hostname between the check and the connect. The Host header and TLS SNI stay
+# on the hostname.
+
+
+def test_resolve_public_ip_returns_validated_ip(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        socket,
+        "getaddrinfo",
+        lambda *a, **k: [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 80))],
+    )
+    assert resolve_public_ip("http://example.com/hook") == "93.184.216.34"
+
+
+def test_resolve_public_ip_rejects_private(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        socket,
+        "getaddrinfo",
+        lambda *a, **k: [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.1.2.3", 80))],
+    )
+    with pytest.raises(UnsafeEgressURLError):
+        resolve_public_ip("http://internal.example.com/")
+
+
+def test_resolve_public_ip_allowlisted_is_not_pinned(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Allowlisted hosts are trusted by name and connected as-is: resolve returns
+    # None (no resolution) and the factory yields the default transport.
+    monkeypatch.setenv("AGENTFLOW_EGRESS_ALLOWED_HOSTS", "host.docker.internal")
+
+    def _no_resolution(*_a: object, **_k: object) -> list[object]:
+        raise AssertionError("allowlisted host must not be resolved")
+
+    monkeypatch.setattr(socket, "getaddrinfo", _no_resolution)
+    assert resolve_public_ip("http://host.docker.internal:9000/cb") is None
+    assert pinned_transport("http://host.docker.internal:9000/cb", None) is None
+
+
+def test_pinned_transport_connects_to_validated_ip_not_rebound(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Rebinding shape: the first resolution (the guard) returns a public IP; a
+    # later one would flip to cloud-metadata. Pinning must connect to the
+    # validated public IP and never re-resolve, so the rebind never takes effect.
+    calls = {"n": 0}
+
+    def _flipping(host: str, *_a: object, **_k: object) -> list[object]:
+        calls["n"] += 1
+        ip = "93.184.216.34" if calls["n"] == 1 else "169.254.169.254"
+        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, 80))]
+
+    monkeypatch.setattr(socket, "getaddrinfo", _flipping)
+
+    pinned_ip = resolve_public_ip("http://rebind.example.com/hook")
+    assert pinned_ip == "93.184.216.34"
+    assert calls["n"] == 1  # resolved exactly once, by the guard
+
+    transport = pinned_transport("http://rebind.example.com/hook", pinned_ip)
+    assert isinstance(transport, _PinnedIPTransport)
+
+    captured: dict[str, object] = {}
+
+    async def _fake_super(self: object, request: httpx.Request) -> httpx.Response:
+        captured["host"] = request.url.host
+        captured["host_header"] = request.headers.get("host")
+        captured["sni"] = request.extensions.get("sni_hostname")
+        return httpx.Response(200, request=request)
+
+    monkeypatch.setattr(httpx.AsyncHTTPTransport, "handle_async_request", _fake_super)
+
+    request = httpx.Request("POST", "http://rebind.example.com/hook", content=b"x")
+    response = asyncio.run(transport.handle_async_request(request))
+
+    assert response.status_code == 200
+    # The socket target is the validated public IP literal — no second lookup.
+    assert captured["host"] == "93.184.216.34"
+    # Host header + TLS identity still key on the hostname.
+    assert captured["host_header"] == "rebind.example.com"
+    assert captured["sni"] == "rebind.example.com"
+    # getaddrinfo was consulted exactly once (the guard); the transport never
+    # re-resolved, so the rebound metadata address was never reachable.
+    assert calls["n"] == 1