wehub-resource-sync
173 lines
7.3 KiB
Python
173 lines
7.3 KiB
Python
# 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()
|