fix(security): validate integration api_call URLs with the outbound SSRF guard (#5145)

execute_api_call — reachable by the LLM through the api_call agent
tool — joined the integration's user-configured base_url with an
LLM-controlled path and requested it with no IP validation, so a
base_url (or a hostname resolving) into the metadata range
(169.254.169.254) was fetched server-side with the integration's auth
headers attached.

Run check_outbound_url on the joined URL before connecting, matching
the gallery endpoint, embeddings, CardDAV, and reminder webhook
surfaces. Link-local/metadata is always rejected;
INTEGRATION_API_BLOCK_PRIVATE_IPS=true also blocks RFC-1918/loopback.
Private stays allowed by default because LAN integrations
(Home Assistant, Miniflux, ntfy) are the primary use case.

The truncation-test helpers stub the guard open because their
api.example.com fixture host does not resolve and the guard fails
closed on DNS errors.

Fixes #5143

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

src/integrations.py

@@ -394,6 +394,22 @@ async def execute_api_call(
         return {"error": "Path must not contain a fragment", "exit_code": 1}
 
     url = _join_integration_url(base_url, path)
+
+    # SSRF guard — same check used by the gallery endpoint, embeddings,
+    # CardDAV, and the reminder webhook sender. Link-local / metadata
+    # addresses (169.254.x.x — the cloud credential-exfil vector) are always
+    # rejected; INTEGRATION_API_BLOCK_PRIVATE_IPS=true also blocks RFC-1918 /
+    # 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
+    block_private = os.getenv(
+        "INTEGRATION_API_BLOCK_PRIVATE_IPS", "false"
+    ).lower() == "true"
+    ok, reason = check_outbound_url(url, block_private=block_private)
+    if not ok:
+        return {"error": f"URL rejected: {reason}", "exit_code": 1}
+
     method = method.upper()
 
     # Build headers

tests/test_integration_api_call_ssrf.py

@@ -0,0 +1,99 @@
+"""Regression: execute_api_call must run the outbound SSRF guard.
+
+The api_call agent tool lets the LLM drive HTTP requests against a
+user-configured integration base_url. Before this guard, a base_url (or a
+hostname resolving) to the cloud metadata range was requested server-side
+with the integration's auth headers attached. execute_api_call now validates
+the joined URL with src.url_safety.check_outbound_url before connecting:
+link-local/metadata is always rejected; RFC-1918/loopback only when
+INTEGRATION_API_BLOCK_PRIVATE_IPS=true (LAN integrations are the primary
+use case, so private stays allowed by default).
+"""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from src import integrations
+
+
+def _integration(base_url):
+    return {
+        "id": "test_integ",
+        "name": "TestInteg",
+        "enabled": True,
+        "base_url": base_url,
+        "auth_type": "bearer",
+        "api_key": "secret-token",
+        "auth_header": "",
+        "auth_param": "",
+        "description": "",
+        "preset": "",
+    }
+
+
+async def _call(base_url, path="/items"):
+    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)
+
+    with (
+        patch.object(integrations, "_find_integration",
+                     return_value=_integration(base_url)),
+        patch("httpx.AsyncClient", return_value=client),
+    ):
+        result = await integrations.execute_api_call("test_integ", "GET", path)
+    return result, client
+
+
+@pytest.mark.asyncio
+async def test_metadata_ip_base_url_is_rejected_without_requesting():
+    result, client = await _call("http://169.254.169.254")
+
+    assert result["exit_code"] == 1
+    assert "rejected" in result["error"].lower()
+    client.request.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_hostname_resolving_to_metadata_ip_is_rejected(monkeypatch):
+    """DNS-based variant: an innocuous-looking hostname that resolves into
+    the link-local range must be caught by the resolver check."""
+    monkeypatch.setattr("src.url_safety._default_resolver",
+                        lambda host: ["169.254.169.254"])
+    result, client = await _call("http://internal.attacker.example")
+
+    assert result["exit_code"] == 1
+    assert "rejected" in result["error"].lower()
+    client.request.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_public_ip_base_url_still_requests():
+    # Public literal — no DNS involved.
+    result, client = await _call("http://93.184.216.34")
+
+    assert result.get("exit_code") == 0
+    client.request.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_private_base_url_allowed_by_default_blocked_with_knob(monkeypatch):
+    # Local-first default: LAN integrations (Home Assistant etc.) must work.
+    monkeypatch.delenv("INTEGRATION_API_BLOCK_PRIVATE_IPS", raising=False)
+    result, client = await _call("http://192.168.1.50")
+    assert result.get("exit_code") == 0
+    client.request.assert_called_once()
+
+    # Locked-down deployments opt in to a full private/loopback block.
+    monkeypatch.setenv("INTEGRATION_API_BLOCK_PRIVATE_IPS", "true")
+    result, client = await _call("http://192.168.1.50")
+    assert result["exit_code"] == 1
+    assert "rejected" in result["error"].lower()
+    client.request.assert_not_called()

tests/test_integrations_api_call_truncation.py

@@ -83,6 +83,9 @@ 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")),
     ):
         return await integrations.execute_api_call("test_integ", "GET", "/items")
 
@@ -98,6 +101,9 @@ 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")),
     ):
         result = await integrations.execute_api_call("test_integ", "GET", path)
     return result, mock_client