fix(adagents): resolve-and-validate gate against DNS-based SSRF (#757, partial) (#760)

Closes the common case where a public-looking hostname's authoritative
DNS points at a private address — e.g. `metadata.example.com` →
`169.254.169.254`. `_check_safe_host` is purely string-level and would
let this pass; the new `_dns_validate_host` resolves the hostname once
via `socket.getaddrinfo` (in an executor) and gates every returned
address through the same IP-block check.

Plumbed in front of every outbound HTTP request in adagents discovery:

- `_fetch_adagents_url` (publisher hop, authoritative_location hop,
  conditional-refresh hop)
- `_fetch_ads_txt_managerdomains` (MANAGERDOMAIN fallback fetch — fails
  closed to "no MANAGERDOMAIN" rather than surfacing a validation error,
  matching the existing best-effort semantics of that path)

IP literals short-circuit (already gated by `_check_safe_host`). RFC
2606 / 6761 reserved domains (`.example`, `.test`, `.invalid`,
`.localhost`, `example.com`/`.net`/`.org`) also short-circuit so tests
using these names don't need DNS mocks.

Test infrastructure adds an autouse `_stub_getaddrinfo` fixture in
test_adagents.py that returns a benign public IP (`8.8.8.8`) for every
host by default. Tests that want to exercise the DNS gate
(`test_resolved_private_ip_rejected_before_connect`,
`test_resolved_dns_failure_surfaces_as_validation_error`,
`test_resolved_mixed_public_and_private_is_rejected`) override via
`monkeypatch.setattr(socket, "getaddrinfo", ...)`.

Residual rebinding window (still open after this PR): a determined
attacker controlling the authoritative DNS can return a public IP on
this lookup and a private IP on httpx's connect lookup milliseconds
later. Closing that window requires intercepting httpx's network
backend to pin the connection IP — left tracked on #757 as the
remaining work after this resolve-and-validate gate ships.

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bokelley

src/adcp/adagents.py

@@ -8,9 +8,11 @@
 for sales agents to verify they are authorized for specific properties.
 """
 
+import asyncio
 import ipaddress
 import json
 import re
+import socket
 from dataclasses import dataclass, field
 from typing import Any, Literal
 from urllib.parse import urlparse
@@ -199,6 +201,61 @@ def _check_safe_host(hostname: str, context: str) -> None:
         raise AdagentsValidationError(f"{context} must not target a private/reserved address")
 
 
+# Reserved TLDs (RFC 6761) and second-level domains (RFC 2606) that never
+# resolve in the public DNS. Tests use these consistently; skipping the
+# resolve gate on them saves every test from having to mock
+# socket.getaddrinfo while still applying the gate to real-world hostnames.
+_NON_RESOLVING_SUFFIXES: tuple[str, ...] = (
+    ".example",
+    ".test",
+    ".invalid",
+    ".localhost",
+    ".example.com",
+    ".example.net",
+    ".example.org",
+)
+_NON_RESOLVING_HOSTS: frozenset[str] = frozenset({"example.com", "example.net", "example.org"})
+
+
+async def _dns_validate_host(host: str, port: int) -> None:
+    """Resolve a hostname once and gate every returned address.
+
+    Closes the common SSRF path where a public-looking hostname's DNS
+    points at a private address (e.g. ``metadata.example.com`` →
+    ``169.254.169.254``). ``_check_safe_host`` only sees the string;
+    this helper sees what the resolver returns.
+
+    Residual rebinding window: a determined attacker controlling the
+    authoritative DNS can still return a public IP on this lookup and
+    a private IP on httpx's connect lookup milliseconds later. Closing
+    that window requires intercepting httpx's network backend to pin
+    the connection IP — tracked separately.
+    """
+    if not host:
+        return
+    try:
+        ipaddress.ip_address(host)
+        return  # IP literal — already gated by _check_safe_host
+    except ValueError:
+        pass
+    if host in _NON_RESOLVING_HOSTS or host.endswith(_NON_RESOLVING_SUFFIXES):
+        return
+    loop = asyncio.get_running_loop()
+    try:
+        # Run via executor (not loop.getaddrinfo) so tests can mock
+        # socket.getaddrinfo at the C-level entry point — patching the
+        # concrete event loop's bound method doesn't intercept reliably.
+        infos = await loop.run_in_executor(
+            None, lambda: socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
+        )
+    except socket.gaierror as e:
+        raise AdagentsValidationError(f"DNS resolution failed for {host!r}: {e}") from e
+    for info in infos:
+        addr = info[4][0]
+        if isinstance(addr, str):
+            _check_safe_host(addr, "resolved address")
+
+
 def _validate_publisher_domain(domain: str) -> str:
     """Validate and sanitize publisher domain for security.
 
@@ -636,6 +693,10 @@ async def _fetch_ads_txt_managerdomains(
     """
     url = f"https://{publisher_domain}/ads.txt"
     headers = {"User-Agent": user_agent, "Accept": "text/plain"}
+    try:
+        await _dns_validate_host(publisher_domain, 443)
+    except AdagentsValidationError:
+        return []
     try:
         if client is not None:
             response = await client.get(
@@ -970,6 +1031,11 @@ async def _fetch_adagents_url(
         if cache_entry.last_modified:
             headers["If-Modified-Since"] = cache_entry.last_modified
 
+    parsed = urlparse(url)
+    await _dns_validate_host(
+        parsed.hostname or "", parsed.port or (443 if parsed.scheme == "https" else 80)
+    )
+
     try:
         if client is not None:
             body, status_code, response_headers = await _stream_capped(

tests/test_adagents.py

@@ -3,6 +3,7 @@
 """Tests for adagents.json validation functionality."""
 
 import json
+import socket
 import unittest.mock
 from unittest.mock import AsyncMock, MagicMock
 
@@ -26,6 +27,27 @@
 )
 
 
+@pytest.fixture(autouse=True)
+def _stub_getaddrinfo(monkeypatch):
+    """Stub socket.getaddrinfo to a benign public IP for every test.
+
+    The DNS pre-check in `_dns_validate_host` calls
+    `socket.getaddrinfo` (via an executor) for every outbound URL whose
+    host isn't a reserved RFC 2606 / 6761 name. Without this stub, tests
+    that use arbitrary public-looking hostnames (e.g.
+    `cdn.other-domain.com`) either hit live DNS or fail in CI
+    environments without resolution.
+
+    Tests that specifically want to exercise the DNS gate override this
+    fixture or call `monkeypatch.setattr` on the same target.
+    """
+
+    def _fake_resolve(host, port, *args, **kwargs):
+        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", port))]
+
+    monkeypatch.setattr(socket, "getaddrinfo", _fake_resolve)
+
+
 def _make_stream_response(
     *,
     status_code: int,
@@ -756,6 +778,57 @@ async def test_rejects_dot_local_redirect(self):
         with pytest.raises(AdagentsValidationError, match="localhost"):
             await fetch_adagents("example.com", client=mock_client)
 
+    @pytest.mark.asyncio
+    async def test_resolved_private_ip_rejected_before_connect(self, monkeypatch):
+        # A public-looking hostname whose DNS points at a private address
+        # must be rejected by the resolve-and-validate pre-check, not
+        # silently connected to. Closes the string-level gap left by
+        # _check_safe_host alone.
+        from adcp.adagents import fetch_adagents
+
+        def _resolve_to_private(host, port, *args, **kwargs):
+            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("169.254.169.254", port))]
+
+        monkeypatch.setattr(socket, "getaddrinfo", _resolve_to_private)
+
+        # Use a hostname that isn't in the RFC 2606 / 6761 reserved list
+        # so the resolve pre-check actually runs.
+        with pytest.raises(AdagentsValidationError, match="private/reserved"):
+            await fetch_adagents("metadata.realhost.org")
+
+    @pytest.mark.asyncio
+    async def test_resolved_dns_failure_surfaces_as_validation_error(self, monkeypatch):
+        # A DNS gaierror (NXDOMAIN, transient SERVFAIL) becomes a clear
+        # AdagentsValidationError rather than bubbling raw socket errors
+        # through the SDK boundary.
+        from adcp.adagents import fetch_adagents
+
+        def _resolve_fails(host, port, *args, **kwargs):
+            raise socket.gaierror(-2, "Name or service not known")
+
+        monkeypatch.setattr(socket, "getaddrinfo", _resolve_fails)
+
+        with pytest.raises(AdagentsValidationError, match="DNS resolution failed"):
+            await fetch_adagents("nonexistent.realhost.org")
+
+    @pytest.mark.asyncio
+    async def test_resolved_mixed_public_and_private_is_rejected(self, monkeypatch):
+        # If a hostname resolves to a list of addresses where ANY one is
+        # private, the SDK must reject — defending against split-horizon
+        # DNS that returns both a public and a private address.
+        from adcp.adagents import fetch_adagents
+
+        def _resolve_mixed(host, port, *args, **kwargs):
+            return [
+                (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("8.8.8.8", port)),
+                (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.5", port)),
+            ]
+
+        monkeypatch.setattr(socket, "getaddrinfo", _resolve_mixed)
+
+        with pytest.raises(AdagentsValidationError, match="private/reserved"):
+            await fetch_adagents("split-horizon.realhost.org")
+
     @pytest.mark.asyncio
     async def test_redirect_uses_fresh_client(self):
         """Redirect hops should not reuse the caller's client."""