From a3f9180d5e28af63b7ca7b5030b4dee6794fc574 Mon Sep 17 00:00:00 2001 From: ashvinctrl Date: Tue, 14 Jul 2026 13:02:59 +0530 Subject: [PATCH] fix(security): pin integration api_call to the SSRF-validated IP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit execute_api_call validated the outbound URL with check_outbound_url (#5145) but then opened a plain httpx client that re-resolves the host at connect time. A base_url host that answers with a public IP for the guard and rebinds to 169.254.169.254 for the connect slips past the check and reaches cloud metadata with the integration's auth headers attached — the same DNS-rebinding TOCTOU already closed for web fetch (#704) and webhook delivery (#5147). Resolve the host once through the guard's own resolver, remember the validated IPs, and pin the request connection to one of them via a local async transport (SNI and Host stay the original hostname). The per-module transport mirrors the ones web fetch and webhook delivery already carry. --- src/integrations.py | 142 +++++++++++++++++- tests/test_integration_api_call_ssrf.py | 59 ++++++++ .../test_integrations_api_call_truncation.py | 14 +- 3 files changed, 206 insertions(+), 9 deletions(-) diff --git a/src/integrations.py b/src/integrations.py index aa6c4982e3..f3aa1819c1 100644 --- a/src/integrations.py +++ b/src/integrations.py @@ -1,11 +1,14 @@ +import ipaddress import json import os +import ssl import uuid import logging import re from typing import Dict, List, Optional, Any from urllib.parse import urljoin, urlparse, urlunparse +import httpcore import httpx from fastapi import HTTPException @@ -354,6 +357,117 @@ def _find_integration(identifier: str) -> Optional[Dict[str, Any]]: return None +# httpcore raises its own exception hierarchy; map the ones a simple request can +# surface back to their httpx equivalents so the caller's `except httpx.*` blocks +# below behave exactly as they did with the default transport. +_HTTPCORE_TO_HTTPX_EXC = { + httpcore.ConnectError: httpx.ConnectError, + httpcore.ConnectTimeout: httpx.ConnectTimeout, + httpcore.NetworkError: httpx.NetworkError, + httpcore.PoolTimeout: httpx.PoolTimeout, + httpcore.ProtocolError: httpx.ProtocolError, + httpcore.ReadError: httpx.ReadError, + httpcore.ReadTimeout: httpx.ReadTimeout, + httpcore.RemoteProtocolError: httpx.RemoteProtocolError, + httpcore.TimeoutException: httpx.TimeoutException, + httpcore.WriteError: httpx.WriteError, + httpcore.WriteTimeout: httpx.WriteTimeout, +} + + +class _PinnedAsyncBackend(httpcore.AsyncNetworkBackend): + """Async network backend that routes every TCP connect to a fixed IP. + + httpcore derives TLS SNI and the ``Host`` header from the request URL, not + from the connect host, so pinning only the socket destination keeps + certificate validation and vhost routing pointed at the original hostname. + """ + + def __init__(self, ip: ipaddress._BaseAddress): + self._ip = str(ip) + self._real = httpcore.AnyIOBackend() + + async def connect_tcp(self, host, port, timeout=None, local_address=None, + socket_options=None): + return await self._real.connect_tcp( + self._ip, port, timeout, local_address, socket_options + ) + + async def connect_unix_socket(self, path, timeout=None, socket_options=None): + return await self._real.connect_unix_socket(path, timeout, socket_options) + + async def sleep(self, seconds: float) -> None: + return await self._real.sleep(seconds) + + +class _PinnedAsyncTransport(httpx.AsyncBaseTransport): + """httpx transport that pins the TCP connect to a pre-resolved IP. + + Kept local to mirror the per-module pinned transports web fetch (#704) and + webhook delivery (#5147) already carry, rather than coupling api_call to the + webhook subsystem. The request URL is passed through unchanged, so SNI / the + ``Host`` header stay the original hostname; only the socket destination is + pinned, closing the DNS-rebinding TOCTOU between the SSRF check and connect. + """ + + def __init__(self, ip: ipaddress._BaseAddress): + self._pinned_ip = ip + self._pool = httpcore.AsyncConnectionPool( + ssl_context=ssl.create_default_context(), + http1=True, + http2=False, + network_backend=_PinnedAsyncBackend(ip), + ) + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + core_req = httpcore.Request( + method=request.method, + url=httpcore.URL( + scheme=request.url.raw_scheme, + host=request.url.raw_host, + port=request.url.port, + target=request.url.raw_path, + ), + headers=request.headers.raw, + content=request.stream, + extensions=request.extensions, + ) + try: + core_resp = await self._pool.handle_async_request(core_req) + content = b"".join([chunk async for chunk in core_resp.aiter_stream()]) + await core_resp.aclose() + except Exception as exc: + mapped = _HTTPCORE_TO_HTTPX_EXC.get(type(exc)) + if mapped is not None: + raise mapped(str(exc)) from exc + raise + return httpx.Response( + status_code=core_resp.status, + headers=core_resp.headers, + content=content, + extensions=core_resp.extensions, + ) + + async def aclose(self) -> None: + await self._pool.aclose() + + +def _first_resolved_ip(raw_ips: List[str]) -> Optional[ipaddress._BaseAddress]: + """Return the first entry that parses as an IP address. + + Matches how ``check_outbound_url`` walks its resolver output (it skips + unparseable results), so the pinned IP is one the guard actually validated. + """ + for raw in raw_ips: + if not isinstance(raw, str): + continue + try: + return ipaddress.ip_address(raw.split("%")[0]) # strip IPv6 zone id + except ValueError: + continue + return None + + async def execute_api_call( integration_id: str, method: str, @@ -409,13 +523,33 @@ async def execute_api_call( # loopback for locked-down deployments. Private stays allowed by default # because LAN integrations (Home Assistant, Miniflux, ntfy) are the # primary use case. - from src.url_safety import check_outbound_url + from src.url_safety import check_outbound_url, _default_resolver block_private = os.getenv( "INTEGRATION_API_BLOCK_PRIVATE_IPS", "false" ).lower() == "true" - ok, reason = check_outbound_url(url, block_private=block_private) + # Resolve the host exactly once, remembering the IPs the guard validated so + # the request below can be pinned to one of them. check_outbound_url only + # reports (ok, reason); a plain httpx client re-resolves the host at connect + # time, reopening a DNS-rebinding TOCTOU — a base_url host that answers with + # a public IP for the guard and then flips to 169.254.169.254 for the connect + # would reach cloud metadata with the integration's auth headers attached. + # Pinning the connection to the validated IP closes that window (as webhook + # delivery does since #5147). + resolved_ips: List[str] = [] + + def _recording_resolver(host: str) -> List[str]: + ips = _default_resolver(host) + resolved_ips[:] = ips + return ips + + ok, reason = check_outbound_url( + url, block_private=block_private, resolver=_recording_resolver + ) if not ok: return {"error": f"URL rejected: {reason}", "exit_code": 1} + pinned_ip = _first_resolved_ip(resolved_ips) + if pinned_ip is None: + return {"error": "URL rejected: host did not resolve to a usable address", "exit_code": 1} method = method.upper() @@ -455,7 +589,9 @@ async def execute_api_call( auth = httpx.BasicAuth(parts[0], parts[1]) try: - async with httpx.AsyncClient(timeout=30.0) as client: + async with httpx.AsyncClient( + timeout=30.0, transport=_PinnedAsyncTransport(pinned_ip) + ) as client: response = await client.request( method, url, diff --git a/tests/test_integration_api_call_ssrf.py b/tests/test_integration_api_call_ssrf.py index 53dc671c58..61ebb8f918 100644 --- a/tests/test_integration_api_call_ssrf.py +++ b/tests/test_integration_api_call_ssrf.py @@ -97,3 +97,62 @@ async def test_private_base_url_allowed_by_default_blocked_with_knob(monkeypatch assert result["exit_code"] == 1 assert "rejected" in result["error"].lower() client.request.assert_not_called() + + +async def _call_capturing_transport(base_url, path="/items"): + """Drive execute_api_call and return (result, transport) where transport is + the object passed to httpx.AsyncClient(transport=...).""" + resp = MagicMock() + resp.status_code = 200 + resp.headers = {"content-type": "application/json"} + resp.json.return_value = {"ok": True} + resp.text = '{"ok": true}' + + client = AsyncMock() + client.__aenter__ = AsyncMock(return_value=client) + client.__aexit__ = AsyncMock(return_value=None) + client.request = AsyncMock(return_value=resp) + + captured = {} + + def _fake_async_client(*args, **kwargs): + captured.update(kwargs) + return client + + with ( + patch.object(integrations, "_find_integration", + return_value=_integration(base_url)), + patch("httpx.AsyncClient", side_effect=_fake_async_client), + ): + result = await integrations.execute_api_call("test_integ", "GET", path) + return result, captured.get("transport"), client + + +@pytest.mark.asyncio +async def test_connection_is_pinned_to_the_validated_ip(monkeypatch): + """DNS-rebinding defense: the guard resolves the host once to a benign + public IP, and the request must be pinned to *that* IP so a host that + rebinds to the metadata range at connect time can't be reached with the + integration's auth headers. Static resolution passing the guard is not + enough — a plain client would re-resolve at connect.""" + monkeypatch.setattr("src.url_safety._default_resolver", + lambda host: ["93.184.216.34"]) + result, transport, client = await _call_capturing_transport( + "http://rebinding.attacker.example") + + assert result.get("exit_code") == 0 + client.request.assert_called_once() + assert isinstance(transport, integrations._PinnedAsyncTransport) + assert str(transport._pinned_ip) == "93.184.216.34" + + +@pytest.mark.asyncio +async def test_pin_uses_a_validated_ip_when_host_has_several(monkeypatch): + """When a host resolves to multiple records the pin targets the first one + the guard walked (and therefore validated).""" + monkeypatch.setattr("src.url_safety._default_resolver", + lambda host: ["93.184.216.34", "198.51.100.7"]) + result, transport, _ = await _call_capturing_transport("http://multi.example") + + assert result.get("exit_code") == 0 + assert str(transport._pinned_ip) == "93.184.216.34" diff --git a/tests/test_integrations_api_call_truncation.py b/tests/test_integrations_api_call_truncation.py index bf1ec7d05e..3102cea788 100644 --- a/tests/test_integrations_api_call_truncation.py +++ b/tests/test_integrations_api_call_truncation.py @@ -83,9 +83,10 @@ async def _call(json_data, status=200): with ( patch.object(integrations, "_find_integration", return_value=DUMMY_INTEGRATION), patch("httpx.AsyncClient", return_value=mock_client), - # api.example.com doesn't resolve; the SSRF guard would fail closed. - # These tests are about truncation, so stub the guard open. - patch("src.url_safety.check_outbound_url", return_value=(True, "ok")), + # api.example.com doesn't resolve here; feed the SSRF guard a public IP + # so it passes (and the connection pins to it). These tests are about + # truncation, not the guard. + patch("src.url_safety._default_resolver", lambda host: ["93.184.216.34"]), ): return await integrations.execute_api_call("test_integ", "GET", "/items") @@ -101,9 +102,10 @@ async def _call_with_integration(integration, path="/items"): with ( patch.object(integrations, "_find_integration", return_value=integration), patch("httpx.AsyncClient", return_value=mock_client), - # api.example.com doesn't resolve; the SSRF guard would fail closed. - # These tests are about URL joining, so stub the guard open. - patch("src.url_safety.check_outbound_url", return_value=(True, "ok")), + # api.example.com doesn't resolve here; feed the SSRF guard a public IP + # so it passes (and the connection pins to it). These tests are about + # URL joining, not the guard. + patch("src.url_safety._default_resolver", lambda host: ["93.184.216.34"]), ): result = await integrations.execute_api_call("test_integ", "GET", path) return result, mock_client