fix(tools): make web SSRF protection opt-in, disabled by default

agent/tools/utils/url_safety.py

@@ -1,18 +1,41 @@
 """
 Shared SSRF guard utilities for tools that fetch model-supplied URLs.
 
-A URL is only considered safe when it uses an http/https scheme, has a
-hostname, that hostname resolves, and every resolved address is a public
-(internet-routable) address. Loopback, private (RFC1918 / ULA), link-local
-(incl. the 169.254.169.254 cloud-metadata endpoint) and otherwise reserved
-addresses are rejected, for both IPv4 and IPv6.
+SSRF protection is OPT-IN and disabled by default, because legitimate use
+cases (local dev servers, LAN services, proxy fake-ip resolution) need to
+reach non-public addresses. Enable it by setting the config option
+``web_security_ssrf_protection: true`` (or env ``WEB_SECURITY_SSRF_PROTECTION``).
+
+When enabled, a URL is only considered safe when it uses an http/https
+scheme, has a hostname, that hostname resolves, and every resolved address
+is a public (internet-routable) address. Loopback, private (RFC1918 / ULA),
+link-local (incl. the 169.254.169.254 cloud-metadata endpoint) and otherwise
+reserved addresses are rejected, for both IPv4 and IPv6.
 """
 
 import ipaddress
+import os
 import socket
 from urllib.parse import urlparse
 
 
+def _ssrf_protection_enabled() -> bool:
+    """Return True only when SSRF protection is explicitly turned on.
+
+    Disabled by default. Reads the env var first, then falls back to the
+    global config; any failure to read config is treated as "disabled" so
+    the guard never breaks normal fetching.
+    """
+    env = os.getenv("WEB_SECURITY_SSRF_PROTECTION")
+    if env is not None:
+        return env.strip().lower() in ("1", "true", "yes", "on")
+    try:
+        from config import conf
+        return bool(conf().get("web_security_ssrf_protection", False))
+    except Exception:
+        return False
+
+
 def _is_blocked_ip(ip: "ipaddress._BaseAddress") -> bool:
     """Return True if the address is not safe to connect to (non-public)."""
     return (
@@ -28,8 +51,11 @@ def _is_blocked_ip(ip: "ipaddress._BaseAddress") -> bool:
 def assert_public_ip(ip_str: str) -> None:
     """Raise ValueError if the given literal IP is a non-public address.
 
-    Used to re-validate the concrete address a redirect resolved to.
+    No-op when SSRF protection is disabled (the default). Used to re-validate
+    the concrete address a redirect resolved to.
     """
+    if not _ssrf_protection_enabled():
+        return
     ip = ipaddress.ip_address(ip_str)
     if _is_blocked_ip(ip):
         raise ValueError(
@@ -41,13 +67,17 @@ def assert_public_ip(ip_str: str) -> None:
 def validate_url_safe(url: str) -> None:
     """Reject URLs that target private/loopback/link-local addresses (SSRF guard).
 
-    Resolves the hostname to its IP address(es) and blocks any that fall
+    No-op when SSRF protection is disabled (the default). When enabled,
+    resolves the hostname to its IP address(es) and blocks any that fall
     into non-public ranges. Also rejects URLs with no host, non-HTTP(S)
     schemes, or hosts that fail DNS resolution.
 
     Raises:
         ValueError: if the URL targets a disallowed address.
     """
+    if not _ssrf_protection_enabled():
+        return
+
     parsed = urlparse(url)
     if parsed.scheme not in ("http", "https"):
         raise ValueError(f"Unsupported URL scheme: {parsed.scheme}")

tests/test_security_ssrf_path_traversal.py

@@ -25,12 +25,24 @@ if "requests" not in sys.modules:
 # =============================================================================
 
 class TestVisionSSRFValidation(unittest.TestCase):
-    """Test that _validate_url_safe blocks internal/private URLs."""
+    """Test that _validate_url_safe blocks internal/private URLs.
+
+    SSRF protection is opt-in (disabled by default); enable it via env for
+    the duration of these tests.
+    """
 
     def setUp(self):
+        self._prev_ssrf_env = os.environ.get("WEB_SECURITY_SSRF_PROTECTION")
+        os.environ["WEB_SECURITY_SSRF_PROTECTION"] = "true"
         from agent.tools.vision.vision import Vision
         self.validate = Vision._validate_url_safe
 
+    def tearDown(self):
+        if self._prev_ssrf_env is None:
+            os.environ.pop("WEB_SECURITY_SSRF_PROTECTION", None)
+        else:
+            os.environ["WEB_SECURITY_SSRF_PROTECTION"] = self._prev_ssrf_env
+
     def test_loopback_ipv4_blocked(self):
         """127.0.0.1 must be rejected."""
         with self.assertRaises(ValueError) as ctx:

tests/test_security_ssrf_web_fetch.py

@@ -74,12 +74,24 @@ def _fake_ok_response(body=b"<html><head><title>internal</title></head><body>sec
 
 
 class TestWebFetchSSRF(unittest.TestCase):
-    """web_fetch must refuse internal targets and never connect to them."""
+    """web_fetch must refuse internal targets and never connect to them.
+
+    SSRF protection is opt-in (disabled by default), so these tests enable it
+    via the WEB_SECURITY_SSRF_PROTECTION env var for the duration of the test.
+    """
 
     def setUp(self):
+        self._prev_ssrf_env = os.environ.get("WEB_SECURITY_SSRF_PROTECTION")
+        os.environ["WEB_SECURITY_SSRF_PROTECTION"] = "true"
         from agent.tools.web_fetch.web_fetch import WebFetch
         self.tool = WebFetch()
 
+    def tearDown(self):
+        if self._prev_ssrf_env is None:
+            os.environ.pop("WEB_SECURITY_SSRF_PROTECTION", None)
+        else:
+            os.environ["WEB_SECURITY_SSRF_PROTECTION"] = self._prev_ssrf_env
+
     # --- Literal internal IPs: rejected before any socket call ---
 
     def test_loopback_literal_blocked(self):
@@ -184,5 +196,29 @@ class TestWebFetchSSRF(unittest.TestCase):
         self.assertEqual(mock_get.call_args[0][0], "http://example.com/page")
 
 
+class TestWebFetchSSRFDisabledByDefault(unittest.TestCase):
+    """With protection disabled (default), local/internal targets are reachable."""
+
+    def setUp(self):
+        self._prev_ssrf_env = os.environ.get("WEB_SECURITY_SSRF_PROTECTION")
+        os.environ.pop("WEB_SECURITY_SSRF_PROTECTION", None)
+        from agent.tools.web_fetch.web_fetch import WebFetch
+        self.tool = WebFetch()
+
+    def tearDown(self):
+        if self._prev_ssrf_env is None:
+            os.environ.pop("WEB_SECURITY_SSRF_PROTECTION", None)
+        else:
+            os.environ["WEB_SECURITY_SSRF_PROTECTION"] = self._prev_ssrf_env
+
+    def test_loopback_allowed_when_disabled(self):
+        """http://127.0.0.1/x must be fetched when protection is off (default)."""
+        with patch("socket.getaddrinfo", return_value=_gai("127.0.0.1")), \
+                patch("requests.get", return_value=_fake_ok_response()) as mock_get:
+            result = self.tool.execute({"url": "http://127.0.0.1:8080/local"})
+        self.assertEqual(result.status, "success")
+        mock_get.assert_called_once()
+
+
 if __name__ == "__main__":
     unittest.main()