Merge pull request #2905 from Jiangrong-W/harness-fix/net-egress-cowagent-browser-navigate-no-internal-url-guard

fix(browser): block link-local / cloud-metadata navigation (SSRF guard)
2026-07-17 11:07:11 +08:00 · 2026-06-19 20:36:28 +08:00
parent a5aaecc48d 033480eef1
commit 75e3110e8c
2 changed files with 246 additions and 0 deletions
--- a/agent/tools/browser/browser_tool.py
+++ b/agent/tools/browser/browser_tool.py
@@ -15,15 +15,24 @@ Launch modes (configured under `tools.browser` in config.json):
  - fresh: Set `persistent` to false to fall back to a clean context every run.
 """

+import ipaddress
 import json
 import os
+import socket
 from typing import Dict, Any, Optional
+from urllib.parse import urlparse

 from agent.tools.base_tool import BaseTool, ToolResult
 from agent.tools.browser.browser_service import BrowserService
 from common.log import logger


+# Cloud-metadata endpoints worth blocking even though they are not link-local.
+# (169.254.169.254 — AWS/GCP/Azure IMDS — is already covered by is_link_local;
+# fd00:ec2::254 is the AWS IPv6 IMDS address.)
+_CLOUD_METADATA_IPS = frozenset({ipaddress.ip_address("fd00:ec2::254")})
+
+
 class BrowserTool(BaseTool):
    """Single tool exposing all browser actions via an 'action' parameter."""

@@ -121,6 +130,61 @@ class BrowserTool(BaseTool):
        BrowserTool._shared_service = self._service
        return self._service

+    def _allow_private_targets(self) -> bool:
+        """Whether the link-local / cloud-metadata guard is disabled.
+
+        Defaults to False (guard active). Loopback and RFC1918/LAN targets are
+        always reachable so local dev servers work out of the box; this opt-out
+        only lifts the remaining block on link-local / cloud-metadata targets,
+        for an operator who deliberately needs them, by setting
+        ``allow_private_targets: true`` under ``tools.browser`` in config.json.
+        """
+        return bool(self.config.get("allow_private_targets", False))
+
+    @staticmethod
+    def _validate_url_safe(url: str) -> None:
+        """Reject URLs that target link-local / cloud-metadata addresses (SSRF guard).
+
+        Resolves the hostname to its IP address(es) and blocks any that are
+        link-local (169.254.0.0/16 — which includes the 169.254.169.254
+        cloud-metadata endpoint — and IPv6 fe80::/10) or a known IPv6
+        cloud-metadata address. Also rejects URLs with no host, non-HTTP(S)
+        schemes, or hosts that fail DNS resolution.
+
+        Loopback and RFC1918/LAN targets are intentionally left reachable:
+        unlike the vision/web_fetch tools, the browser legitimately opens local
+        pages (a dev server on ``localhost`` / ``127.0.0.1`` / a LAN IP), so a
+        blanket "block all internal" policy would break that core workflow.
+
+        Raises:
+            ValueError: if the URL targets a disallowed address.
+        """
+        parsed = urlparse(url)
+        if parsed.scheme not in ("http", "https"):
+            raise ValueError(f"Unsupported URL scheme: {parsed.scheme}")
+
+        hostname = parsed.hostname
+        if not hostname:
+            raise ValueError("URL has no hostname")
+
+        try:
+            # Resolve all addresses for the hostname.
+            addr_infos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
+        except socket.gaierror:
+            raise ValueError(f"Cannot resolve hostname: {hostname}")
+
+        for family, _, _, _, sockaddr in addr_infos:
+            ip_str = sockaddr[0]
+            ip = ipaddress.ip_address(ip_str)
+            # Block only the high-risk targets — link-local (incl. the
+            # 169.254.169.254 cloud-metadata endpoint) and the IPv6 metadata
+            # address. Loopback and RFC1918/LAN stay reachable for local dev.
+            if ip.is_link_local or ip in _CLOUD_METADATA_IPS:
+                raise ValueError(
+                    f"URL resolves to a link-local / cloud-metadata address "
+                    f"({ip_str}), request blocked for security"
+                )
+
    def execute(self, args: Dict[str, Any]) -> ToolResult:
        action = args.get("action", "").strip().lower()
        if not action:
@@ -148,6 +212,16 @@ class BrowserTool(BaseTool):
        # Only auto-prepend https:// for bare hosts; preserve file://, about:, data:, etc.
        if "://" not in url and not url.startswith(("about:", "data:")):
            url = "https://" + url
+        # SSRF guard: for http(s) targets, reject hosts that resolve to
+        # link-local / cloud-metadata addresses before the browser navigates
+        # (and then auto-snapshots the page back to the model). Loopback and
+        # RFC1918/LAN are allowed so local dev servers work. Non-HTTP schemes
+        # (about:/data:/file:/chrome:) are not network-egress targets here.
+        if url.split(":", 1)[0].lower() in ("http", "https") and not self._allow_private_targets():
+            try:
+                self._validate_url_safe(url)
+            except ValueError as e:
+                return ToolResult.fail(f"Error: {e}")
        timeout = args.get("timeout", 30000)
        service = self._get_service()
        result = service.navigate(url, timeout=timeout)
--- a/tests/test_security_ssrf_browser_navigate.py
+++ b/tests/test_security_ssrf_browser_navigate.py
@@ -0,0 +1,172 @@
+# encoding:utf-8
+"""
+Regression tests for browser-navigate SSRF protection.
+
+The browser tool navigates to a model-supplied URL via Playwright
+``page.goto`` and then auto-snapshots the page back to the model. Without a
+guard, a model (including one under prompt injection) can point it at the
+cloud-metadata endpoint (169.254.169.254) and read the credentials back
+through the snapshot.
+
+Unlike the vision / web_fetch tools, the browser legitimately needs local
+pages — a dev server on ``localhost`` / ``127.0.0.1`` / a LAN IP. So the guard
+is deliberately narrow: it blocks only **link-local** addresses
+(169.254.0.0/16, which includes the metadata endpoint, plus IPv6 fe80::/10) and
+the IPv6 cloud-metadata address, while leaving loopback and RFC1918/LAN
+reachable.
+
+These tests ensure ``BrowserTool``:
+  - blocks link-local / cloud-metadata targets *before* the navigation reaches
+    the browser service,
+  - still lets loopback, RFC1918/LAN and public URLs through to the (stubbed)
+    service,
+  - preserves the documented non-HTTP scheme behaviour (about:/data:), and
+  - honours an explicit opt-out (allow_private_targets).
+
+No real browser / Playwright / network is used: the BrowserService that the
+tool would create is replaced with a stub, and DNS resolution is mocked.
+"""
+import os
+import sys
+import types
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+# Stub 'requests' if not installed so sibling tool imports don't fail.
+if "requests" not in sys.modules:
+    _requests_stub = types.ModuleType("requests")
+    _requests_stub.get = lambda *a, **k: None
+    sys.modules["requests"] = _requests_stub
+
+
+def _gai(ip_str):
+    """Build a socket.getaddrinfo return value for a single IPv4 address."""
+    return [(2, 1, 6, "", (ip_str, 0))]
+
+
+class _StubService:
+    """Stand-in for BrowserService that records navigation attempts."""
+
+    def __init__(self):
+        self.navigated = []
+
+    def navigate(self, url, timeout=30000):
+        self.navigated.append(url)
+        return {"url": url, "title": "page", "status": 200}
+
+    def snapshot(self, selector=None):
+        return "Page: page  (http://page/)\nInteractive elements: 0\n---\ncontent"
+
+
+class TestBrowserNavigateSSRF(unittest.TestCase):
+    """Browser navigate blocks link-local/metadata but keeps local dev reachable."""
+
+    def setUp(self):
+        from agent.tools.browser.browser_tool import BrowserTool
+        self.tool = BrowserTool()
+        self.stub = _StubService()
+        # Force the tool to use our stub instead of a real BrowserService.
+        self.tool._service = self.stub
+        patcher = patch.object(BrowserTool, "_get_service", return_value=self.stub)
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
+    # --- Link-local / cloud-metadata: rejected before any service call ---
+
+    def test_cloud_metadata_literal_blocked(self):
+        result = self.tool.execute(
+            {"action": "navigate", "url": "http://169.254.169.254/latest/meta-data/"}
+        )
+        self.assertEqual(result.status, "error")
+        self.assertIn("blocked for security", str(result.result))
+        self.assertEqual(self.stub.navigated, [])
+
+    def test_link_local_literal_blocked(self):
+        result = self.tool.execute({"action": "navigate", "url": "http://169.254.1.1/x"})
+        self.assertEqual(result.status, "error")
+        self.assertIn("blocked for security", str(result.result))
+        self.assertEqual(self.stub.navigated, [])
+
+    def test_ipv6_metadata_literal_blocked(self):
+        result = self.tool.execute(
+            {"action": "navigate", "url": "http://[fd00:ec2::254]/latest/"}
+        )
+        self.assertEqual(result.status, "error")
+        self.assertIn("blocked for security", str(result.result))
+        self.assertEqual(self.stub.navigated, [])
+
+    def test_metadata_hostname_blocked(self):
+        # A hostname that resolves to the metadata endpoint is blocked too.
+        with patch("socket.getaddrinfo", return_value=_gai("169.254.169.254")):
+            result = self.tool.execute(
+                {"action": "navigate", "url": "http://metadata.internal/latest/meta-data/"}
+            )
+        self.assertEqual(result.status, "error")
+        self.assertIn("blocked for security", str(result.result))
+        self.assertEqual(self.stub.navigated, [])
+
+    # --- Local dev targets stay reachable (the maintainer's core workflow) ---
+
+    def test_loopback_literal_allowed(self):
+        result = self.tool.execute({"action": "navigate", "url": "http://127.0.0.1:3000/"})
+        self.assertEqual(result.status, "success")
+        self.assertEqual(self.stub.navigated, ["http://127.0.0.1:3000/"])
+
+    def test_ipv6_loopback_literal_allowed(self):
+        result = self.tool.execute({"action": "navigate", "url": "http://[::1]:5173/"})
+        self.assertEqual(result.status, "success")
+        self.assertEqual(self.stub.navigated, ["http://[::1]:5173/"])
+
+    def test_localhost_bare_allowed(self):
+        """A bare 'localhost' (no scheme) gets https:// prepended, then allowed."""
+        with patch("socket.getaddrinfo", return_value=_gai("127.0.0.1")):
+            result = self.tool.execute({"action": "navigate", "url": "localhost:3000"})
+        self.assertEqual(result.status, "success")
+        self.assertEqual(self.stub.navigated, ["https://localhost:3000"])
+
+    def test_rfc1918_10_hostname_allowed(self):
+        with patch("socket.getaddrinfo", return_value=_gai("10.1.2.3")):
+            result = self.tool.execute({"action": "navigate", "url": "http://dev.lan/app"})
+        self.assertEqual(result.status, "success")
+        self.assertEqual(self.stub.navigated, ["http://dev.lan/app"])
+
+    def test_rfc1918_192_168_hostname_allowed(self):
+        with patch("socket.getaddrinfo", return_value=_gai("192.168.0.5")):
+            result = self.tool.execute({"action": "navigate", "url": "http://router.lan/admin"})
+        self.assertEqual(result.status, "success")
+        self.assertEqual(self.stub.navigated, ["http://router.lan/admin"])
+
+    # --- Public URL is allowed through to the (stubbed) service ---
+
+    def test_public_url_allowed(self):
+        with patch("socket.getaddrinfo", return_value=_gai("93.184.216.34")):
+            result = self.tool.execute({"action": "navigate", "url": "http://example.com/page"})
+        self.assertEqual(result.status, "success")
+        self.assertEqual(self.stub.navigated, ["http://example.com/page"])
+
+    # --- Documented non-HTTP scheme behaviour preserved (not an egress path) ---
+
+    def test_about_blank_not_blocked(self):
+        result = self.tool.execute({"action": "navigate", "url": "about:blank"})
+        self.assertEqual(result.status, "success")
+        self.assertEqual(self.stub.navigated, ["about:blank"])
+
+    # --- Explicit opt-out lets an operator re-enable metadata/link-local ---
+
+    def test_opt_out_allows_metadata(self):
+        from agent.tools.browser.browser_tool import BrowserTool
+        tool = BrowserTool({"allow_private_targets": True})
+        stub = _StubService()
+        tool._service = stub
+        with patch.object(BrowserTool, "_get_service", return_value=stub):
+            result = tool.execute(
+                {"action": "navigate", "url": "http://169.254.169.254/latest/meta-data/"}
+            )
+        self.assertEqual(result.status, "success")
+        self.assertEqual(stub.navigated, ["http://169.254.169.254/latest/meta-data/"])
+
+
+if __name__ == "__main__":
+    unittest.main()