adcontextprotocol
diff --git a/‎src/adcp/signing/ip_pinned_transport.py‎
Lines changed: 10 additions & 6 deletions b/‎src/adcp/signing/ip_pinned_transport.py‎
Lines changed: 10 additions & 6 deletions
diff --git a/‎src/adcp/webhook_sender.py‎
Lines changed: 34 additions & 13 deletions b/‎src/adcp/webhook_sender.py‎
Lines changed: 34 additions & 13 deletions
diff --git a/‎tests/conformance/signing/test_jwks.py‎
Lines changed: 1 addition & 6 deletions b/‎tests/conformance/signing/test_jwks.py‎
Lines changed: 1 addition & 6 deletions
diff --git a/‎tests/conformance/signing/test_webhook_sender_e2e.py‎
Lines changed: 160 additions & 0 deletions b/‎tests/conformance/signing/test_webhook_sender_e2e.py‎
Lines changed: 160 additions & 0 deletions
@@ -295,11 +295,14 @@ def build_ip_pinned_transport(
     """Resolve ``uri`` once and return a transport pinned to the validated IP.
 
     Raises :class:`SSRFValidationError` if the URI's scheme isn't
-    ``http``/``https``, the port is not in the allowlist, the host
-    doesn't resolve, or every resolved IP is in a blocked range.
+    ``http``/``https``, ``allowed_ports`` is set and the URI's port is
+    outside it, the host doesn't resolve, or every resolved IP is in a
+    blocked range.
 
-    ``allowed_ports`` defaults to
-    :data:`adcp.signing.jwks.DEFAULT_ALLOWED_PORTS` (`{443, 8443}`).
+    ``allowed_ports`` defaults to ``None`` (no port filter — AdCP
+    doesn't constrain webhook ports). Hardened deployments pass
+    :data:`adcp.signing.jwks.DEFAULT_ALLOWED_PORTS` (`{443, 8443}`)
+    or a custom set.
 
     Typical use inside a fetcher::
 
@@ -328,8 +331,9 @@ def build_async_ip_pinned_transport(
     function itself is not awaitable. The returned transport plugs
     into :class:`httpx.AsyncClient`.
 
-    ``allowed_ports`` defaults to
-    :data:`adcp.signing.jwks.DEFAULT_ALLOWED_PORTS` (`{443, 8443}`).
+    ``allowed_ports`` defaults to ``None`` (no port filter); see
+    :func:`build_ip_pinned_transport` for the hardening kwarg
+    semantics.
     """
     hostname, resolved_ip, _port = resolve_and_validate_host(
         uri,
 
@@ -44,7 +44,10 @@
     load_private_key_pem,
     private_key_from_jwk,
 )
-from adcp.signing.ip_pinned_transport import build_async_ip_pinned_transport
+from adcp.signing.ip_pinned_transport import (
+    AsyncIpPinnedTransport,
+    build_async_ip_pinned_transport,
+)
 from adcp.signing.webhook_signer import sign_webhook
 from adcp.types import GeneratedTaskStatus
 from adcp.types.generated_poc.core.async_response_data import AdcpAsyncResponseData
@@ -505,7 +508,26 @@ async def _send_bytes(
         with mTLS to a known buyer set, or an ASGI transport for testing),
         the sender trusts the operator's transport completely. Pin-and-bind
         is skipped; the operator's transport owns SSRF.
+
+        On the owned-client path, SSRF validation runs **before** signing
+        so a hostile URL is rejected without first generating an
+        Ed25519/ES256 signature over the body. That signature would
+        otherwise sit in process memory until the SSRF rejection —
+        anything that snapshots locals on exception (faulthandler,
+        custom logging) could capture it. Validate first, sign second.
         """
+        # Build the pinned transport up-front for the owned-client path.
+        # This runs SSRF + port validation against the URL before any
+        # signing happens; a hostile URL raises SSRFValidationError here
+        # and the body never gets signed.
+        transport: AsyncIpPinnedTransport | None = None
+        if self._owns_client:
+            transport = build_async_ip_pinned_transport(
+                url,
+                allow_private=self._allow_private_destinations,
+                allowed_ports=self._allowed_destination_ports,
+            )
+
         base_headers = {"Content-Type": "application/json"}
         signed = sign_webhook(
             method="POST",
@@ -529,27 +551,26 @@ async def _send_bytes(
             for k, v in extra_headers.items():
                 headers[k] = v
 
-        if self._owns_client:
-            # Per-request pinned transport. Building one client per delivery
-            # is the security-correct choice: keeping a pinned transport
-            # alive across deliveries to the same hostname would defeat the
-            # rebinding defense (the IP would be frozen at first delivery).
-            transport = build_async_ip_pinned_transport(
-                url,
-                allow_private=self._allow_private_destinations,
-                allowed_ports=self._allowed_destination_ports,
-            )
+        if transport is not None:
+            # Owned-client path. ``trust_env=False`` prevents httpx from
+            # routing the request through ``HTTPS_PROXY`` / ``HTTP_PROXY``
+            # env vars — every other pinned-transport callsite in the
+            # codebase sets this for the same reason (default_jwks_fetcher,
+            # async_default_jwks_fetcher, revocation_fetcher). Without it,
+            # an attacker who controls process env can route the signed
+            # webhook through their endpoint, defeating the IP pin entirely.
             async with httpx.AsyncClient(
                 transport=transport,
                 timeout=self._timeout,
                 follow_redirects=False,
+                trust_env=False,
             ) as client:
                 response = await client.post(url, content=body, headers=headers)
         else:
             # Operator-supplied client — they own the SSRF guarantees on
             # their transport (proxy allowlist, mTLS, etc.). Reachable as
-            # None after aclose(); a runtime check beats an assert that
-            # python -O strips to silently NoneType.post().
+            # None after aclose(); explicit raise survives ``python -O``
+            # which would strip an assert.
             if self._client is None:
                 raise RuntimeError(
                     "WebhookSender's operator-supplied client was already "
 
@@ -101,12 +101,7 @@ def test_ssrf_caps_resolved_address_scan() -> None:
 
 
 # ---- Port allowlist (opt-in operator hardening) ----
-#
-# AdCP doesn't constrain webhook ports in the spec, so the default validator
-# imposes no port filter. Adopters who want a hardening posture pass
-# ``allowed_ports=DEFAULT_ALLOWED_PORTS`` (or a custom set); rejection of
-# non-standard ports closes a smuggle vector for buyers bouncing traffic to
-# internal services on the same routable IP.
+# Rationale lives in adcp.signing.jwks.DEFAULT_ALLOWED_PORTS docstring.
 
 
 @pytest.mark.parametrize(
 
@@ -487,3 +487,163 @@ async def echo(request: Request, op_id: str) -> JSONResponse:
 
     assert result.ok
     assert captured_payloads[0]["operation_id"] == "op_abc123"
+
+
+@pytest.mark.asyncio
+async def test_owned_client_default_allows_non_standard_ports() -> None:
+    """The default ``WebhookSender`` (no operator client, no
+    ``allowed_destination_ports``) accepts AdCP-spec-compliant buyers on
+    non-standard ports — :9443 (Tomcat default), :4443 (Spring Boot
+    default), path-routed multi-tenant gateways. The IP-range check still
+    applies; we just don't impose a port filter unless the operator
+    explicitly opts in.
+
+    Sender-level positive analog of test_ssrf_default_imposes_no_port_filter
+    in test_jwks.py — confirms the permissive default reaches the actual
+    delivery path, not just the underlying validator."""
+    from unittest.mock import patch
+
+    captured: list[tuple[str, int]] = []
+    app = FastAPI()
+
+    @app.post("/webhooks/adcp")
+    async def echo(_request: Request) -> JSONResponse:
+        return JSONResponse({"ok": True}, status_code=200)
+
+    asgi_transport = httpx.ASGITransport(app=app)
+
+    # Stub the pinned-transport build so we don't open a real socket to
+    # a public IP; capture that the build was attempted for the
+    # non-standard port and route the actual POST through ASGI.
+    def fake_build(uri: str, **_kwargs: Any) -> Any:
+        from urllib.parse import urlparse
+
+        parsed = urlparse(uri)
+        captured.append((parsed.hostname or "", parsed.port or 443))
+        return asgi_transport
+
+    sender = WebhookSender.from_jwk(
+        {**WEBHOOK_JWK, "d": WEBHOOK_JWK["_private_d_for_test_only"]},
+    )
+    with patch(
+        "adcp.webhook_sender.build_async_ip_pinned_transport",
+        side_effect=fake_build,
+    ):
+        async with sender:
+            result = await sender.send_mcp(
+                url="http://test:9443/webhooks/adcp",
+                task_id="task_nonstd",
+                task_type="create_media_buy",
+                status="completed",
+            )
+
+    assert result.ok
+    assert captured == [("test", 9443)]
+
+
+@pytest.mark.asyncio
+async def test_operator_supplied_client_bypasses_ssrf_guard() -> None:
+    """When the operator passes their own httpx client (vetted egress
+    proxy, ASGI test transport, etc.), the framework trusts them
+    completely — pin-and-bind is skipped and the SSRF range check does
+    NOT fire. The operator owns SSRF on their transport.
+
+    Named regression test for the documented contract; without this, a
+    future refactor that mistakenly applies pin-and-bind to both
+    branches breaks ASGI-based unit tests and any vetted-proxy
+    deployments that route via private networks."""
+    app = FastAPI()
+
+    @app.post("/webhooks/adcp")
+    async def echo(_request: Request) -> JSONResponse:
+        return JSONResponse({"ok": True}, status_code=200)
+
+    transport = httpx.ASGITransport(app=app)
+    # base_url is loopback-equivalent. With the SSRF guard active this
+    # would raise SSRFValidationError; under the operator-trust contract
+    # it must succeed.
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        sender = WebhookSender.from_jwk(
+            {**WEBHOOK_JWK, "d": WEBHOOK_JWK["_private_d_for_test_only"]},
+            client=client,
+        )
+        result = await sender.send_mcp(
+            url="http://test/webhooks/adcp",
+            task_id="task_op_trust",
+            task_type="create_media_buy",
+            status="completed",
+        )
+
+    assert result.ok
+
+
+@pytest.mark.asyncio
+async def test_owned_client_ignores_https_proxy_env() -> None:
+    """``HTTPS_PROXY`` / ``HTTP_PROXY`` env vars MUST NOT defeat the IP
+    pin on the owned-client path. httpx's default ``trust_env=True``
+    routes requests through proxy env vars, which would bypass the
+    AsyncIpPinnedTransport's network_backend entirely — an attacker who
+    controls process env (sidecar config, dotenv, malicious cluster
+    egress policy) could otherwise pivot to receiving the signed
+    webhook body.
+
+    The sender constructs its per-request ``httpx.AsyncClient`` with
+    ``trust_env=False`` to close this. Regression guard: if a future
+    refactor drops the kwarg, this test catches it by setting a proxy
+    env var that points at an unreachable address; the pinned transport
+    must still route via the resolved IP and reach the test ASGI app
+    (which we can't directly observe under a real socket, so we assert
+    the proxy var is ignored by checking the constructor config)."""
+    import os
+    from unittest.mock import MagicMock, patch
+
+    sender = WebhookSender.from_jwk(
+        {**WEBHOOK_JWK, "d": WEBHOOK_JWK["_private_d_for_test_only"]},
+    )
+
+    captured_kwargs: dict[str, Any] = {}
+
+    class _FakeAsyncClient:
+        def __init__(self, **kwargs: Any) -> None:
+            # Capture kwargs only from the per-request construction in
+            # _send_bytes; __aenter__'s eager _get_client() also flows
+            # through here but its kwargs don't affect the per-request
+            # delivery path. The last writer wins; the per-request call
+            # is the one we care about.
+            captured_kwargs.update(kwargs)
+            self._response = MagicMock(status_code=200, headers={}, content=b"{}")
+
+        async def __aenter__(self) -> _FakeAsyncClient:
+            return self
+
+        async def __aexit__(self, *_args: Any) -> None:
+            return None
+
+        async def aclose(self) -> None:
+            return None
+
+        async def post(self, *_args: Any, **_kwargs: Any) -> Any:
+            return self._response
+
+    # Set HTTPS_PROXY pointing at an unreachable address. With
+    # trust_env=True (the httpx default), this would override the
+    # transport. The sender MUST pass trust_env=False to ignore it.
+    with patch.dict(os.environ, {"HTTPS_PROXY": "http://attacker.invalid:9999"}):
+        with patch(
+            "adcp.webhook_sender.build_async_ip_pinned_transport",
+            return_value=MagicMock(),  # transport itself isn't used here
+        ):
+            with patch("adcp.webhook_sender.httpx.AsyncClient", _FakeAsyncClient):
+                async with sender:
+                    await sender.send_mcp(
+                        url="https://buyer.example.com/webhooks/adcp",
+                        task_id="task_proxy",
+                        task_type="create_media_buy",
+                        status="completed",
+                    )
+
+    assert captured_kwargs.get("trust_env") is False, (
+        "WebhookSender's per-request httpx.AsyncClient must construct with "
+        "trust_env=False — otherwise HTTPS_PROXY env vars defeat the IP pin"
+    )
+    assert captured_kwargs.get("follow_redirects") is False