Merge remote-tracking branch 'upstream/master'

# Please enter a commit message to explain why this merge is necessary,
# especially if it merges an updated upstream into a topic branch.
#
# Lines starting with '#' will be ignored, and an empty message aborts
# the commit.

tests/test_custom_provider.py

@@ -88,15 +88,15 @@ class TestResolveCustomCredentials(unittest.TestCase):
         set_conf({
             "bot_type": "custom:abc12345",
             "custom_providers": [
-                {"id": "sf001", "name": "siliconflow", "api_key": "sf-key",
-                 "api_base": "https://api.siliconflow.cn/v1", "model": "deepseek-ai/DeepSeek-V3"},
-                {"id": "abc12345", "name": "qiniu", "api_key": "qn-key",
-                 "api_base": "https://api.qnaigc.com/v1", "model": "deepseek-v3"},
+                {"id": "sf001", "name": "provider-a", "api_key": "key-a",
+                 "api_base": "https://api.example.com/v1", "model": "model-a"},
+                {"id": "abc12345", "name": "provider-b", "api_key": "key-b",
+                 "api_base": "https://api.example.org/v1", "model": "model-b"},
             ],
         })
         self.assertEqual(
             self.resolve(),
-            ("qn-key", "https://api.qnaigc.com/v1", "deepseek-v3"),
+            ("key-b", "https://api.example.org/v1", "model-b"),
         )
 
     def test_id_not_found_falls_back_to_legacy(self):
@@ -105,8 +105,8 @@ class TestResolveCustomCredentials(unittest.TestCase):
             "custom_api_key": "legacy-key",
             "custom_api_base": "https://legacy.example.com/v1",
             "custom_providers": [
-                {"id": "sf001", "name": "siliconflow", "api_key": "sf-key",
-                 "api_base": "https://api.siliconflow.cn/v1"},
+                {"id": "sf001", "name": "provider-a", "api_key": "key-a",
+                 "api_base": "https://api.example.com/v1"},
             ],
         })
         self.assertEqual(

tests/test_custom_provider_handlers.py

@@ -74,8 +74,8 @@ class TestSetCustomProvider(unittest.TestCase):
 
     def test_create_provider_does_not_hijack_bot_type(self):
         """Creating a provider without make_active must not change bot_type."""
-        res = self.h.call(action="set_custom_provider", name="siliconflow",
-                          api_base="https://api.siliconflow.cn/v1", api_key="sf-key")
+        res = self.h.call(action="set_custom_provider", name="my-provider",
+                          api_base="https://api.example.com/v1", api_key="key-a")
         self.assertEqual(res["status"], "success")
         self.assertTrue(res["created"])
         self.assertIn("id", res)
@@ -85,13 +85,13 @@ class TestSetCustomProvider(unittest.TestCase):
         providers = config_module.conf().get("custom_providers")
         self.assertEqual(len(providers), 1)
         self.assertEqual(providers[0]["id"], res["id"])
-        self.assertEqual(providers[0]["name"], "siliconflow")
+        self.assertEqual(providers[0]["name"], "my-provider")
         self.assertEqual(self.h.bridge_resets, 1)
 
     def test_create_with_make_active_switches_bot_type(self):
         """Creating a provider with make_active=true must switch bot_type."""
-        res = self.h.call(action="set_custom_provider", name="siliconflow",
-                          api_base="https://api.siliconflow.cn/v1", api_key="sf-key",
+        res = self.h.call(action="set_custom_provider", name="my-provider",
+                          api_base="https://api.example.com/v1", api_key="key-a",
                           make_active=True)
         self.assertEqual(res["status"], "success")
         bot_type = config_module.conf().get("bot_type")

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()

tests/test_security_ssrf_web_fetch.py

@@ -0,0 +1,188 @@
+# encoding:utf-8
+"""
+Regression tests for web_fetch SSRF protection.
+
+The web_fetch tool fetches model-supplied URLs. Without a guard, a model
+(including one under prompt injection) can point it at loopback, RFC1918,
+link-local or cloud-metadata (169.254.169.254) endpoints, or use a public
+URL that 3xx-redirects into such a target. These tests ensure web_fetch
+refuses the request instead of connecting to the internal address.
+
+No real network is used: DNS resolution and ``requests.get`` are stubbed.
+"""
+import os
+import sys
+import types
+import unittest
+from unittest.mock import patch, MagicMock
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+# Stub 'requests' if not installed so the module can be imported for testing.
+if "requests" not in sys.modules:
+    _requests_stub = types.ModuleType("requests")
+    _requests_stub.get = lambda *a, **k: None
+
+    class _Exc(Exception):
+        pass
+
+    _requests_stub.Timeout = type("Timeout", (_Exc,), {})
+    _requests_stub.ConnectionError = type("ConnectionError", (_Exc,), {})
+    _requests_stub.HTTPError = type("HTTPError", (_Exc,), {})
+    _requests_stub.Response = object
+    _compat = types.SimpleNamespace(urljoin=__import__("urllib.parse", fromlist=["urljoin"]).urljoin)
+    _requests_stub.compat = _compat
+    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 _FakeRedirect:
+    """Minimal stand-in for a requests redirect Response."""
+
+    def __init__(self, location):
+        self.is_redirect = True
+        self.is_permanent_redirect = False
+        self.headers = {"Location": location}
+        self.closed = False
+
+    def close(self):
+        self.closed = True
+
+
+def _fake_ok_response(body=b"<html><head><title>internal</title></head><body>secret</body></html>"):
+    """A well-formed non-redirect response.
+
+    Returned by the mocked ``requests.get`` so that on UNPATCHED code the
+    fetch path runs to completion and the test fails specifically on the
+    ``assert_not_called`` guard (proving a request reached the internal
+    target), rather than on an incidental error.
+    """
+    resp = MagicMock()
+    resp.is_redirect = False
+    resp.is_permanent_redirect = False
+    resp.status_code = 200
+    resp.headers = {"Content-Type": "text/html; charset=utf-8"}
+    resp.content = body
+    resp.text = body.decode("utf-8")
+    resp.apparent_encoding = "utf-8"
+    resp.raise_for_status = lambda: None
+    return resp
+
+
+class TestWebFetchSSRF(unittest.TestCase):
+    """web_fetch must refuse internal targets and never connect to them."""
+
+    def setUp(self):
+        from agent.tools.web_fetch.web_fetch import WebFetch
+        self.tool = WebFetch()
+
+    # --- Literal internal IPs: rejected before any socket call ---
+
+    def test_loopback_literal_blocked(self):
+        """http://127.0.0.1:<port>/x must be refused, no request issued."""
+        with patch("requests.get", return_value=_fake_ok_response()) as mock_get:
+            result = self.tool.execute({"url": "http://127.0.0.1:8080/canary"})
+        self.assertEqual(result.status, "error")
+        self.assertIn("non-public", result.result)
+        mock_get.assert_not_called()
+
+    def test_cloud_metadata_literal_blocked(self):
+        """http://169.254.169.254/latest/meta-data/ must be refused."""
+        with patch("requests.get", return_value=_fake_ok_response()) as mock_get:
+            result = self.tool.execute(
+                {"url": "http://169.254.169.254/latest/meta-data/"}
+            )
+        self.assertEqual(result.status, "error")
+        self.assertIn("non-public", result.result)
+        mock_get.assert_not_called()
+
+    def test_ipv6_loopback_literal_blocked(self):
+        """http://[::1]/x must be refused."""
+        with patch("requests.get", return_value=_fake_ok_response()) as mock_get:
+            result = self.tool.execute({"url": "http://[::1]/canary"})
+        self.assertEqual(result.status, "error")
+        self.assertIn("non-public", result.result)
+        mock_get.assert_not_called()
+
+    # --- RFC1918 host resolved via DNS: rejected after resolution ---
+
+    def test_rfc1918_hostname_blocked(self):
+        """A hostname that resolves to 10.x.x.x must be refused, no request."""
+        with patch("socket.getaddrinfo", return_value=_gai("10.1.2.3")), \
+                patch("requests.get", return_value=_fake_ok_response()) as mock_get:
+            result = self.tool.execute({"url": "http://internal.corp/secret"})
+        self.assertEqual(result.status, "error")
+        self.assertIn("non-public", result.result)
+        mock_get.assert_not_called()
+
+    def test_192_168_hostname_blocked(self):
+        """A hostname that resolves to 192.168.x.x must be refused."""
+        with patch("socket.getaddrinfo", return_value=_gai("192.168.0.5")), \
+                patch("requests.get", return_value=_fake_ok_response()) as mock_get:
+            result = self.tool.execute({"url": "http://router.local/admin"})
+        self.assertEqual(result.status, "error")
+        self.assertIn("non-public", result.result)
+        mock_get.assert_not_called()
+
+    # --- Redirect bounce: public entry URL 302 -> loopback ---
+
+    def test_public_to_loopback_redirect_blocked(self):
+        """A public URL that redirects to a loopback target must be refused.
+
+        The first hop resolves to a public IP and returns a 302 pointing at
+        127.0.0.1; the guard must re-validate the redirect target and refuse
+        instead of fetching the internal address.
+        """
+        redirect = _FakeRedirect("http://127.0.0.1:8080/canary")
+
+        def fake_getaddrinfo(host, *a, **k):
+            # Public entry host resolves to a public IP; the loopback literal
+            # echoes back (as the real getaddrinfo does for an IP literal).
+            if host == "evil.example.com":
+                return _gai("93.184.216.34")
+            return _gai(host)
+
+        with patch("socket.getaddrinfo", side_effect=fake_getaddrinfo), \
+                patch("requests.get", return_value=redirect) as mock_get:
+            result = self.tool.execute({"url": "http://evil.example.com/start"})
+
+        self.assertEqual(result.status, "error")
+        self.assertIn("non-public", result.result)
+        # The first (public) hop is issued exactly once; the loopback hop is
+        # rejected by the guard BEFORE a second requests.get to the internal
+        # target is made.
+        self.assertEqual(mock_get.call_count, 1)
+        first_call_url = mock_get.call_args[0][0]
+        self.assertEqual(first_call_url, "http://evil.example.com/start")
+        # The follow-up request to the internal target was never issued.
+        for call in mock_get.call_args_list:
+            self.assertNotIn("127.0.0.1", call[0][0])
+
+    # --- Sanity: a public URL is allowed to proceed to the fetch path ---
+
+    def test_public_url_allowed_through_guard(self):
+        """A public URL passes the guard and a (mocked) request is issued."""
+        ok = MagicMock()
+        ok.is_redirect = False
+        ok.is_permanent_redirect = False
+        ok.headers = {"Content-Type": "text/html; charset=utf-8"}
+        ok.content = b"<html><head><title>Hi</title></head><body>ok</body></html>"
+        ok.text = "<html><head><title>Hi</title></head><body>ok</body></html>"
+        ok.apparent_encoding = "utf-8"
+        ok.raise_for_status = lambda: None
+
+        with patch("socket.getaddrinfo", return_value=_gai("93.184.216.34")), \
+                patch("requests.get", return_value=ok) as mock_get:
+            result = self.tool.execute({"url": "http://example.com/page"})
+
+        self.assertEqual(result.status, "success")
+        mock_get.assert_called_once()
+        self.assertEqual(mock_get.call_args[0][0], "http://example.com/page")
+
+
+if __name__ == "__main__":
+    unittest.main()