mirror of
https://github.com/zhayujie/chatgpt-on-wechat.git
synced 2026-07-17 11:07:11 +08:00
fix(browser): block link-local / cloud-metadata navigation (SSRF guard)
The browser tool navigates to a model-supplied URL via Playwright page.goto and then auto-snapshots the page back to the model. The http(s) navigate path performed no filtering, so a model tool call (including under prompt injection) could point the agent-driven browser 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 a blanket "block all internal" policy is the wrong default here. The guard is therefore narrow: it resolves the hostname and rejects only link-local addresses (169.254.0.0/16, which includes the 169.254.169.254 metadata endpoint, and IPv6 fe80::/10) plus the AWS IPv6 IMDS address (fd00:ec2::254), before navigation. Loopback and RFC1918/LAN stay reachable so local dev works out of the box. Only http/https targets are validated; the documented non-HTTP scheme handling (about:/data:/file:) is unchanged. An operator who deliberately needs the link-local/metadata target can opt out with tools.browser.allow_private_targets = true. tests/test_security_ssrf_browser_navigate.py asserts link-local/metadata targets are blocked while loopback, RFC1918/LAN and public targets navigate through (browser service and DNS are stubbed; no real browser/network). Signed-off-by: christop <825583681@qq.com>
This commit is contained in:
172
tests/test_security_ssrf_browser_navigate.py
Normal file
172
tests/test_security_ssrf_browser_navigate.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user