diff --git a/agent/tools/utils/url_safety.py b/agent/tools/utils/url_safety.py index 59e6bddb..62a89723 100644 --- a/agent/tools/utils/url_safety.py +++ b/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}") diff --git a/tests/test_security_ssrf_path_traversal.py b/tests/test_security_ssrf_path_traversal.py index 6c2dbb09..bafb54e1 100644 --- a/tests/test_security_ssrf_path_traversal.py +++ b/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: diff --git a/tests/test_security_ssrf_web_fetch.py b/tests/test_security_ssrf_web_fetch.py index c4c4d319..6da6e0f8 100644 --- a/tests/test_security_ssrf_web_fetch.py +++ b/tests/test_security_ssrf_web_fetch.py @@ -74,12 +74,24 @@ def _fake_ok_response(body=b"internalsec 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()