From 444358924043a009f4fc99a8ed7180e945794c42 Mon Sep 17 00:00:00 2001 From: ashvinctrl Date: Fri, 24 Jul 2026 11:18:28 +0530 Subject: [PATCH] fix(integrations): pin api_call to the SSRF-validated IP execute_api_call runs check_outbound_url on the target, but that guard only resolves the host to answer (ok, reason) and hands back no address. The request right after it opened a plain httpx.AsyncClient, which resolves the host again at connect time. A base_url host on a low TTL can pass the guard as a public IP and then flip to 169.254.169.254 for the connect, so the call lands on cloud metadata with the integration's stored auth headers attached. Resolve once, remember the IPs the guard actually validated, and pin the client's socket to that set through a small AnyIO-backed transport. SNI and the Host header still come from the URL, so TLS and vhost routing are unchanged; connect-time fallback stays inside the approved address set over one shared deadline. This is the same pinning the webhook sender and web-fetch paths already do -- api_call was the last outbound path that skipped it. Fixes #5513 --- src/integrations.py | 163 ++++++++++++- tests/test_integration_api_call_ssrf.py | 219 ++++++++++++++++++ .../test_integrations_api_call_truncation.py | 14 +- 3 files changed, 387 insertions(+), 9 deletions(-) diff --git a/src/integrations.py b/src/integrations.py index aa6c4982e..5f94330d2 100644 --- a/src/integrations.py +++ b/src/integrations.py @@ -1,11 +1,14 @@ +import ipaddress import json import os +import time 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,140 @@ 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): + """Network backend that connects only to the pre-validated IPs, in order. + + Every address here came out of the single SSRF resolution, so moving to the + next one after a connect failure is not re-resolution — it's ordinary + multi-address fallback restricted to the set the guard already approved. + httpcore takes TLS SNI and the ``Host`` header from the request URL rather + than the connect host, so pinning the socket destination leaves certificate + validation and vhost routing pointed at the original hostname. + """ + + def __init__(self, ips: List[ipaddress._BaseAddress]): + self._ips = [str(ip) for ip in ips] + self._real = httpcore.AnyIOBackend() + + async def connect_tcp(self, host, port, timeout=None, local_address=None, + socket_options=None): + # One shared connect budget: each attempt gets the time left until the + # original deadline, so N dead addresses can't stretch the connect + # phase to N * timeout. + deadline = None if timeout is None else time.monotonic() + timeout + last_exc: Optional[Exception] = None + for ip in self._ips: + remaining = None if deadline is None else max(0.0, deadline - time.monotonic()) + try: + return await self._real.connect_tcp( + ip, port, remaining, local_address, socket_options + ) + except (httpcore.ConnectError, httpcore.ConnectTimeout) as exc: + last_exc = exc + if deadline is not None and time.monotonic() >= deadline: + break + raise last_exc + + 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 the pre-resolved IP(s). + + Kept local, mirroring the per-module pinned transports web fetch and + webhook delivery already carry, rather than coupling api_call to the + webhook subsystem. The request URL passes through unchanged, so SNI and the + ``Host`` header stay the original hostname; only the socket destination is + pinned, which is what closes the rebinding window. + """ + + def __init__(self, ips: List[ipaddress._BaseAddress]): + self._pinned_ips = list(ips) + self._pool = httpcore.AsyncConnectionPool( + # Reuse the CA trust the default httpx client would build (certifi + # plus SSL_CERT_FILE / SSL_CERT_DIR when trust_env is set) so + # swapping in this transport doesn't quietly change which chains + # verify. ssl.create_default_context() would use system roots. + ssl_context=httpx.create_ssl_context(), + http1=True, + http2=False, + network_backend=_PinnedAsyncBackend(ips), + ) + + 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 _validated_ips(raw_ips: List[str]) -> List[ipaddress._BaseAddress]: + """Return every entry that parses as an IP address, order preserved. + + check_outbound_url only reports ok when *all* of these classify as safe, so + the whole list is guard-approved and any of them is a legitimate connect + target. Skipping unparseable entries mirrors how the guard walks the same + resolver output. + """ + ips: List[ipaddress._BaseAddress] = [] + for raw in raw_ips: + if not isinstance(raw, str): + continue + try: + ips.append(ipaddress.ip_address(raw.split("%")[0])) # strip IPv6 zone id + except ValueError: + continue + return ips + + async def execute_api_call( integration_id: str, method: str, @@ -409,13 +546,31 @@ 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 and remember the IPs the guard validated so + # the request below can be pinned to them. check_outbound_url only reports + # (ok, reason); a plain httpx client re-resolves the host at connect time, + # which reopens 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. + 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_ips = _validated_ips(resolved_ips) + if not pinned_ips: + return {"error": "URL rejected: host did not resolve to a usable address", "exit_code": 1} method = method.upper() @@ -455,7 +610,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_ips) + ) 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 53dc671c5..3b74b569c 100644 --- a/tests/test_integration_api_call_ssrf.py +++ b/tests/test_integration_api_call_ssrf.py @@ -9,8 +9,13 @@ INTEGRATION_API_BLOCK_PRIVATE_IPS=true (LAN integrations are the primary use case, so private stays allowed by default). """ +import asyncio +import ipaddress +import ssl from unittest.mock import AsyncMock, MagicMock, patch +import httpcore +import httpx import pytest from src import integrations @@ -97,3 +102,217 @@ 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(ip) for ip in transport._pinned_ips] == ["93.184.216.34"] + + +@pytest.mark.asyncio +async def test_pin_carries_the_whole_validated_ip_set(monkeypatch): + """When a host resolves to several records the transport keeps all of them + (check_outbound_url validated every one), in resolver order, so it can fall + back past a dead first address instead of failing the whole call.""" + 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(ip) for ip in transport._pinned_ips] == ["93.184.216.34", "198.51.100.7"] + + +class _FakeStream: + """Stand-in for the connected socket the real backend returns.""" + + +class _RecordingBackend: + """Fake httpcore backend: connect_tcp fails for the addresses in `dead` + and succeeds for the rest, recording the order it was asked to connect.""" + + def __init__(self, dead): + self.dead = set(dead) + self.attempts = [] + + async def connect_tcp(self, host, port, timeout=None, local_address=None, + socket_options=None): + self.attempts.append((host, timeout)) + if host in self.dead: + raise httpcore.ConnectError(f"connection refused: {host}") + return _FakeStream() + + +def _pinned_backend(ips, dead): + """A _PinnedAsyncBackend whose underlying connect is the recording fake.""" + backend = integrations._PinnedAsyncBackend(ips) + backend._real = _RecordingBackend(dead) + return backend + + +@pytest.mark.asyncio +async def test_connect_falls_back_from_dead_first_to_live_second(): + """first-dead / second-live: the pinned backend must try the next validated + address when the first refuses, rather than surfacing the failure. It also + ignores the `host` httpcore passes (the original hostname) and connects to + the pinned IPs, which is what keeps TLS SNI / Host on the real hostname.""" + ips = [ipaddress.ip_address("203.0.113.10"), ipaddress.ip_address("198.51.100.7")] + backend = _pinned_backend(ips, dead={"203.0.113.10"}) + + stream = await backend.connect_tcp("original.hostname.example", 443, timeout=5.0) + + assert isinstance(stream, _FakeStream) + # Tried the dead address first, then the live one — never the hostname. + assert [host for host, _ in backend._real.attempts] == ["203.0.113.10", "198.51.100.7"] + # Fallback shared one budget: the second attempt got the time left, not a fresh 5s. + assert backend._real.attempts[1][1] <= 5.0 + + +@pytest.mark.asyncio +async def test_connect_raises_when_every_validated_address_is_dead(): + ips = [ipaddress.ip_address("203.0.113.10"), ipaddress.ip_address("198.51.100.7")] + backend = _pinned_backend(ips, dead={"203.0.113.10", "198.51.100.7"}) + + with pytest.raises(httpcore.ConnectError): + await backend.connect_tcp("original.hostname.example", 443, timeout=5.0) + assert [host for host, _ in backend._real.attempts] == ["203.0.113.10", "198.51.100.7"] + + +@pytest.mark.asyncio +async def test_pinned_transport_reuses_httpx_ca_trust(monkeypatch): + """TLS trust must come from the same builder the default httpx client uses + (certifi + SSL_CERT_FILE / SSL_CERT_DIR via trust_env), not from + ssl.create_default_context()'s system roots — otherwise chains that verified + under the old default client can silently stop verifying.""" + sentinel = ssl.create_default_context() + calls = [] + + def _fake_create(*args, **kwargs): + calls.append(kwargs) + return sentinel + + monkeypatch.setattr(httpx, "create_ssl_context", _fake_create) + transport = integrations._PinnedAsyncTransport([ipaddress.ip_address("93.184.216.34")]) + try: + assert calls, "transport did not build its context via httpx.create_ssl_context" + assert transport._pool._ssl_context is sentinel + finally: + await transport.aclose() + + +@pytest.mark.asyncio +async def test_real_socket_falls_back_from_dead_first_to_live_second(): + """End-to-end over real loopback sockets: pin [127.0.0.2 (nothing + listening), 127.0.0.1 (live)], and the request must succeed by falling back + to the second address while the Host header stays the original hostname — + i.e. only the socket destination moved, vhost/SNI routing did not.""" + captured = {} + + async def handle(reader, writer): + request = await reader.read(4096) + for line in request.split(b"\r\n"): + if line.lower().startswith(b"host:"): + captured["host"] = line.split(b":", 1)[1].strip().decode() + writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nhi") + await writer.drain() + writer.close() + + server = await asyncio.start_server(handle, "127.0.0.1", 0) + port = server.sockets[0].getsockname()[1] + async with server: + await server.start_serving() + transport = integrations._PinnedAsyncTransport( + [ipaddress.ip_address("127.0.0.2"), ipaddress.ip_address("127.0.0.1")] + ) + try: + async with httpx.AsyncClient(transport=transport) as client: + resp = await client.get(f"http://pinned.example:{port}/health") + finally: + await transport.aclose() + + assert resp.status_code == 200 + assert resp.text == "hi" + assert captured.get("host") == f"pinned.example:{port}" + + +@pytest.mark.asyncio +async def test_ip_literal_base_url_still_pins_and_is_not_rejected(): + """A base_url that is already an IP has nothing to rebind, but it must not + trip the "did not resolve" guard either. + + check_outbound_url resolves even a literal (getaddrinfo returns the address + itself), so the captured list is populated and the pin is a no-op rather + than a rejection. Uses the real resolver on purpose — no monkeypatch — so + this would catch the fail-closed branch firing on a literal. + """ + result, transport, client = await _call_capturing_transport( + "http://93.184.216.34") + + assert result.get("exit_code") == 0 + assert isinstance(transport, integrations._PinnedAsyncTransport) + assert [str(ip) for ip in transport._pinned_ips] == ["93.184.216.34"] + + +@pytest.mark.asyncio +async def test_ipv6_base_url_pins_every_validated_address(monkeypatch): + """IPv6 goes down the same path as v4. + + Resolution is stubbed rather than using a literal so this doesn't depend on + the runner having IPv6 configured. + """ + v6 = "2606:2800:220:1:248:1893:25c8:1946" + monkeypatch.setattr("src.url_safety._default_resolver", lambda host: [v6]) + result, transport, client = await _call_capturing_transport("http://v6.example") + + assert result.get("exit_code") == 0 + assert isinstance(transport, integrations._PinnedAsyncTransport) + assert [str(ip) for ip in transport._pinned_ips] == [v6] + + +def test_validated_ips_strips_zone_id_and_drops_junk(): + """getaddrinfo can hand back a scoped v6 address like 'fe80::1%eth0'.""" + got = integrations._validated_ips( + ["93.184.216.34", "fe80::1%eth0", "not-an-ip", None, "2001:db8::5"] + ) + assert [str(ip) for ip in got] == ["93.184.216.34", "fe80::1", "2001:db8::5"] diff --git a/tests/test_integrations_api_call_truncation.py b/tests/test_integrations_api_call_truncation.py index bf1ec7d05..a0ad61b4a 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. Point the resolver at a public + # address instead of stubbing the guard open, so the real check (and + # the connect-IP pinning that reads its result) still runs. + 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. Point the resolver at a public + # address instead of stubbing the guard open, so the real check (and + # the connect-IP pinning that reads its result) still runs. + 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