Add SSRF protection for connector and MCP URLs, and HMAC-hash API keys

Outbound requests to user-configured connector and MCP URLs now block private, loopback, and link-local targets by default; ALLOW_PRIVATE_CONNECTOR_URLS opts back in for trusted single-tenant deploys. API keys are stored as a keyed HMAC-SHA256 peppered with the server secret. Resolves the CodeQL SSRF and weak-hashing alerts.

Author: sanmaxdev

.env.example

@@ -47,6 +47,10 @@ S3_SECRET_KEY=
 RUN_MAX_STEPS=25
 RUN_MAX_TOKENS=200000
 RUN_MAX_WALLCLOCK_SEC=600
+# Block outbound connector/MCP requests to private, loopback, and link-local
+# addresses (SSRF protection). Set true only on trusted single-tenant deploys
+# that must reach internal endpoints.
+ALLOW_PRIVATE_CONNECTOR_URLS=false
 # When true, agent runs execute inline instead of in the background (tests/dev).
 AGENT_RUN_INLINE=false
 # When true, document ingestion runs inline instead of via a Celery task (tests/dev).

apps/api/legate/agent/builtin_tools.py

@@ -11,6 +11,7 @@
 import httpx
 
 from legate.agent.tooling import ToolContext, ToolRegistry, ToolResult, ToolSpec
+from legate.security.ssrf import SSRFError, validate_public_url
 
 _MAX_BODY_CHARS = 8_000
 
@@ -80,6 +81,10 @@ async def http_request(args: dict, ctx: ToolContext) -> ToolResult:
         return ToolResult(ok=False, error="only read-only GET requests are permitted")
     if not (url.startswith("http://") or url.startswith("https://")):
         return ToolResult(ok=False, error="url must be an absolute http(s) URL")
+    try:
+        validate_public_url(url)
+    except SSRFError as exc:
+        return ToolResult(ok=False, error=str(exc))
 
     try:
         if ctx.http_client is not None:

apps/api/legate/config.py

@@ -72,6 +72,10 @@ def cors_origin_list(self) -> list[str]:
     run_max_steps: int = 25
     run_max_tokens: int = 200_000
     run_max_wallclock_sec: int = 600
+    # Outbound requests to connector / MCP URLs block private, loopback, and
+    # link-local targets to prevent SSRF. Trusted single-tenant deployments that
+    # need internal endpoints can opt back in.
+    allow_private_connector_urls: bool = False
 
     # When true, POST /agents/{id}/run executes the agent loop inline before
     # responding instead of scheduling it in the background. Used by tests for

apps/api/legate/connectors/http.py

@@ -12,6 +12,7 @@
 from legate.agent.tooling import ToolContext, ToolResult, ToolSpec
 from legate.connectors.base import BaseConnector, ConnectionResult, ConnectorConfigError
 from legate.connectors.runtime import new_async_client
+from legate.security.ssrf import SSRFError, validate_public_url
 
 _MAX_BODY_CHARS = 8_000
 _METHODS = {"GET", "POST", "PUT", "PATCH", "DELETE"}
@@ -82,6 +83,10 @@ async def test_connection(self, config: dict, secrets: dict) -> ConnectionResult
         target = config.get("test_url") or config.get("base_url")
         if not target:
             return ConnectionResult(ok=True, message="no test_url/base_url configured to probe")
+        try:
+            validate_public_url(target)
+        except SSRFError as exc:
+            return ConnectionResult(ok=False, message=str(exc))
         try:
             async with new_async_client() as client:
                 resp = await client.get(target, headers=self._auth_headers(config, secrets))
@@ -101,6 +106,10 @@ async def execute(
         url = self._resolve_url(config, str(args.get("url", "")))
         if not url:
             return ToolResult(ok=False, error="url is required")
+        try:
+            validate_public_url(url)
+        except SSRFError as exc:
+            return ToolResult(ok=False, error=str(exc))
 
         headers = {**self._auth_headers(config, secrets), **(args.get("headers") or {})}
         try:

apps/api/legate/connectors/webhook.py

@@ -17,6 +17,7 @@
 from legate.agent.tooling import ToolContext, ToolResult, ToolSpec
 from legate.connectors.base import BaseConnector, ConnectionResult, ConnectorConfigError
 from legate.connectors.runtime import new_async_client
+from legate.security.ssrf import SSRFError, validate_public_url
 
 _DEFAULT_SIGNATURE_HEADER = "X-Legate-Signature"
 
@@ -82,6 +83,10 @@ async def execute(
         url = config.get("url")
         if not url:
             return ToolResult(ok=False, error="no url configured")
+        try:
+            validate_public_url(url)
+        except SSRFError as exc:
+            return ToolResult(ok=False, error=str(exc))
 
         payload = args.get("payload")
         if payload is None:

apps/api/legate/mcp/client.py

@@ -16,6 +16,7 @@
 import httpx
 
 from legate.connectors.runtime import new_async_client
+from legate.security.ssrf import SSRFError, validate_public_url
 
 _PROTOCOL_VERSION = "2025-06-18"
 _CLIENT_INFO = {"name": "legate-agent", "version": "0.1"}
@@ -100,8 +101,16 @@ async def _open_session(client: httpx.AsyncClient, url: str, token: str | None)
     return session_id
 
 
+def _guard(url: str) -> str:
+    try:
+        return validate_public_url(url)
+    except SSRFError as exc:
+        raise MCPError(str(exc)) from exc
+
+
 async def list_tools(url: str, token: str | None = None) -> list[dict]:
     """Discover the tools a server offers. Returns raw MCP tool definitions."""
+    url = _guard(url)
     async with new_async_client() as client:
         session = await _open_session(client, url, token)
         try:
@@ -119,6 +128,7 @@ async def list_tools(url: str, token: str | None = None) -> list[dict]:
 
 async def call_tool(url: str, token: str | None, name: str, arguments: dict) -> dict:
     """Invoke a tool on the server and return the raw MCP result."""
+    url = _guard(url)
     async with new_async_client() as client:
         session = await _open_session(client, url, token)
         try:

apps/api/legate/security/api_keys.py

@@ -1,15 +1,21 @@
 """API key generation and hashing.
 
-A key looks like ``lg_<43 url-safe chars>``. Only its SHA-256 digest and a short
-non-secret prefix are stored; the plaintext is shown to the caller exactly once.
+A key looks like ``lg_<43 url-safe chars>`` (a 256-bit random token). Only a keyed
+HMAC-SHA256 digest and a short non-secret prefix are stored; the plaintext is
+shown to the caller exactly once. The HMAC is peppered with the server secret, so
+a database leak alone cannot be used to recover or verify keys, and it stays
+deterministic so lookups by digest still work.
 """
 
 from __future__ import annotations
 
 import hashlib
+import hmac
 import secrets
 from dataclasses import dataclass
 
+from legate.config import get_settings
+
 KEY_PREFIX = "lg_"
 
 
@@ -21,7 +27,8 @@ class GeneratedApiKey:
 
 
 def hash_api_key(plaintext: str) -> str:
-    return hashlib.sha256(plaintext.encode("utf-8")).hexdigest()
+    pepper = get_settings().legate_secret_key.encode("utf-8")
+    return hmac.new(pepper, plaintext.encode("utf-8"), hashlib.sha256).hexdigest()
 
 
 def generate_api_key() -> GeneratedApiKey:

apps/api/legate/security/ssrf.py

@@ -0,0 +1,72 @@
+"""SSRF guard for outbound requests to user-configured URLs.
+
+Connectors and MCP servers point Legate at URLs the user provides, which is the
+feature, but on a shared deployment a malicious URL could target internal
+services (cloud metadata at 169.254.169.254, localhost admin ports, private
+network hosts). :func:`validate_public_url` resolves the host and rejects any URL
+that lands on a private, loopback, link-local, or otherwise non-public address.
+
+Trusted single-tenant deployments that legitimately need internal endpoints can
+set ``ALLOW_PRIVATE_CONNECTOR_URLS=true`` to bypass the check.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import socket
+from urllib.parse import urlparse
+
+from legate.config import Settings, get_settings
+
+_BLOCKED_ATTRS = (
+    "is_private",
+    "is_loopback",
+    "is_link_local",
+    "is_reserved",
+    "is_multicast",
+    "is_unspecified",
+)
+
+
+class SSRFError(ValueError):
+    """Raised when a URL is not allowed for outbound requests."""
+
+
+def _is_blocked(ip: str) -> bool:
+    addr = ipaddress.ip_address(ip)
+    return any(getattr(addr, attr) for attr in _BLOCKED_ATTRS)
+
+
+def _resolve(host: str) -> list[str]:
+    try:
+        infos = socket.getaddrinfo(host, None)
+    except socket.gaierror as exc:
+        raise SSRFError(f"could not resolve host: {host}") from exc
+    return [str(info[4][0]) for info in infos]
+
+
+def validate_public_url(url: str, *, settings: Settings | None = None) -> str:
+    """Return the URL if it is safe to fetch, otherwise raise :class:`SSRFError`."""
+    settings = settings or get_settings()
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise SSRFError("url must use http or https")
+    host = parsed.hostname
+    if not host:
+        raise SSRFError("url has no host")
+
+    if settings.allow_private_connector_urls:
+        return url
+
+    # A literal IP needs no DNS; a hostname is resolved and every address checked
+    # so a name that maps to a private range cannot slip through.
+    try:
+        ipaddress.ip_address(host)
+        candidates = [host]
+    except ValueError:
+        candidates = _resolve(host)
+
+    for ip in candidates:
+        if _is_blocked(ip):
+            raise SSRFError(f"url resolves to a blocked (non-public) address: {ip}")
+    return url

apps/api/tests/conftest.py

@@ -14,6 +14,9 @@
 os.environ.setdefault("DATABASE_URL", "sqlite+pysqlite:///:memory:")
 os.environ.setdefault("LEGATE_SECRET_KEY", "test-secret-key-do-not-use-in-prod")
 os.environ.setdefault("LEGATE_ENV", "test")
+# Tests exercise connectors against mock transports on placeholder hosts, so the
+# SSRF guard is disabled here; ssrf-specific tests pass an explicit settings.
+os.environ.setdefault("ALLOW_PRIVATE_CONNECTOR_URLS", "true")
 # Credential vault master key (base64 of 32 bytes).
 os.environ.setdefault("LEGATE_MASTER_KEY", base64.b64encode(b"0" * 32).decode())
 # Agent + workflow runs execute inline so they complete deterministically.

apps/api/tests/test_ssrf.py

@@ -0,0 +1,50 @@
+"""Tests for the SSRF guard used by connectors and the MCP client."""
+
+from __future__ import annotations
+
+import pytest
+from legate.config import Settings
+from legate.security.ssrf import SSRFError, validate_public_url
+
+# Guard active (blocks private targets).
+STRICT = Settings(allow_private_connector_urls=False)
+# Guard disabled (trusted single-tenant opt-in).
+RELAXED = Settings(allow_private_connector_urls=True)
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "http://127.0.0.1/x",  # loopback
+        "https://10.0.0.5/admin",  # private
+        "http://192.168.1.10",  # private
+        "http://169.254.169.254/latest/meta-data",  # cloud metadata (link-local)
+        "http://[::1]/",  # IPv6 loopback
+        "http://0.0.0.0",  # unspecified
+    ],
+)
+def test_blocks_internal_targets(url: str) -> None:
+    with pytest.raises(SSRFError):
+        validate_public_url(url, settings=STRICT)
+
+
+@pytest.mark.parametrize("url", ["https://8.8.8.8", "https://93.184.216.34/path"])
+def test_allows_public_ip(url: str) -> None:
+    assert validate_public_url(url, settings=STRICT) == url
+
+
+@pytest.mark.parametrize("url", ["ftp://example.com", "file:///etc/passwd", "gopher://x"])
+def test_rejects_non_http_schemes(url: str) -> None:
+    with pytest.raises(SSRFError):
+        validate_public_url(url, settings=STRICT)
+
+
+def test_rejects_missing_host() -> None:
+    with pytest.raises(SSRFError):
+        validate_public_url("http://", settings=STRICT)
+
+
+def test_opt_in_bypasses_the_guard() -> None:
+    # With the flag on, an internal URL is allowed (and no DNS is attempted).
+    assert validate_public_url("http://127.0.0.1:9000/rpc", settings=RELAXED)
+    assert validate_public_url("http://internal-host.local/rpc", settings=RELAXED)