diff --git a/src/serving/api/alerts/escalation.py b/src/serving/api/alerts/escalation.py index a499b9b..4fc9627 100644 --- a/src/serving/api/alerts/escalation.py +++ b/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 diff --git a/src/serving/api/egress_guard.py b/src/serving/api/egress_guard.py index 3aae0e4..ee28c2a 100644 --- a/src/serving/api/egress_guard.py +++ b/src/serving/api/egress_guard.py @@ -7,9 +7,19 @@ 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 @@ -17,7 +27,8 @@ 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) diff --git a/src/serving/api/webhook_dispatcher.py b/src/serving/api/webhook_dispatcher.py index d4f349b..6755386 100644 --- a/src/serving/api/webhook_dispatcher.py +++ b/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 diff --git a/tests/unit/test_egress_guard.py b/tests/unit/test_egress_guard.py index 3d397ef..02c50d3 100644 --- a/tests/unit/test_egress_guard.py +++ b/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