feat: platform browser multi-tab, navigation allowlist, close cleanup, egress filter, allowlist worker wiring
- Multi-tab: browser_new_page / browser_list_pages / browser_switch_page / browser_close_page in playwright_mcp.py; each new page gets a fresh Playwright context, so cookies and storage are isolated between pages. Single-page tools act on the current page; switching to a missing index is an error, not a crash; closing the last page is an error (use browser_close). - Navigation allowlist: BROWSER_ALLOWED_HOSTS (comma list, leading-dot suffix match, case/port-insensitive) checked in _check_allowed_url before launching Chromium; unset (or empty) means no check, the old behaviour. file:// always allowed; localhost/127.0.0.1 have no special treatment. - Close cleanup: browser_close closes all pages and best-effort removes cwd browser-live.png / browser-live.json; custom screenshot files are untouched. list_pages needs no launch. - Egress filter: 10 programming-source Tinyproxy BASIC rules (python.org, nodejs.org, go.dev, pkg.go.dev, rust-lang.org, developer.mozilla.org, crates.io, stackoverflow.com, cve.mitre.org, nvd.nist.gov); already-present github.com / pypi.org / npmjs.org not duplicated; searx.diante.me stays exact-only, no diante.me wildcard. - Allowlist worker wiring: _docker_task_context now passes a set, non-empty BROWSER_ALLOWED_HOSTS from the worker env into [mcp_servers.playwright.env] in the tenant config, so the MCP process actually sees it (container -e alone does not reach MCP subprocesses, 2026-10-30). Unset/empty writes nothing and never overwrites; the value never goes into logs; it is not added to the websearch env block or the container env. ensure_playwright_mcp_env now returns False without rewriting when the result would be identical (canonical join: one blank line before the block), so a second prep leaves an unchanged tenant config alone; other sections untouched. - Skill + docs: browser-automation SKILL.md multi-tab/allowlist section; docs/platform-browser.md comparison doc (new file). - Tests (mock/fake only, no network, no real Chromium, no real credentials): cloud.tests.test_playwright + cloud.tests.test_websearch_worker, 48 tests OK, including 6 new wiring tests (switch off passes no variable, switch on reaches MCP env verbatim, empty never written, old-tenant merge idempotent + no-rewrite, ensure no-op when identical, allowlist enforces only after wiring). - NOT done (see PR body): no persistent profile, no credential vault, no approval card, no CAPTCHA bypass, in-page link navigation does not pass the allowlist, no real-Chromium live verification, host tinyproxy filter not reloaded.
This commit is contained in:
host committed
1 parent
27f48fe1ed
commit
da7a9950fb
8 files changed
+1050
-12
No files matched your search
@@ -455,5 +455,293 @@ class LineReaderTest(unittest.TestCase):
|
||||
pass
|
||||
|
||||
|
||||
class _FakeBrowserHandleMulti:
|
||||
"""Distinct fake page per new_page (the old single-page fake
|
||||
returns the same page every time, which cannot tell pages
|
||||
apart)."""
|
||||
|
||||
def __init__(self):
|
||||
self.pages = []
|
||||
|
||||
def new_page(self, viewport=None):
|
||||
page = _FakePage()
|
||||
page.url = "https://page%d.example.com/" % len(self.pages)
|
||||
self.pages.append(page)
|
||||
return page
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
|
||||
class _FakePWMulti:
|
||||
def __init__(self):
|
||||
self.handle = _FakeBrowserHandleMulti()
|
||||
self.launch_kwargs = None
|
||||
self.launch_count = 0
|
||||
|
||||
def start(self):
|
||||
return self
|
||||
|
||||
@property
|
||||
def chromium(self):
|
||||
outer = self
|
||||
|
||||
class _Chromium:
|
||||
def launch(self, **kwargs):
|
||||
outer.launch_kwargs = kwargs
|
||||
outer.launch_count += 1
|
||||
return outer.handle
|
||||
|
||||
return _Chromium()
|
||||
|
||||
def stop(self):
|
||||
pass
|
||||
|
||||
|
||||
class _EnvGuard:
|
||||
"""Save/restore BROWSER_ALLOWED_HOSTS around a test case."""
|
||||
|
||||
def setUp(self):
|
||||
self._old_allowlist = os.environ.get("BROWSER_ALLOWED_HOSTS")
|
||||
|
||||
def tearDown(self):
|
||||
if self._old_allowlist is None:
|
||||
os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
|
||||
else:
|
||||
os.environ["BROWSER_ALLOWED_HOSTS"] = self._old_allowlist
|
||||
|
||||
|
||||
class MultiPageTest(_EnvGuard, unittest.TestCase):
|
||||
def setUp(self):
|
||||
_EnvGuard.setUp(self)
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.factory = _FakePWMulti()
|
||||
self.browser = Browser(cwd=self._tmp.name,
|
||||
proxy="http://proxy:8888",
|
||||
browser_factory=lambda: self.factory)
|
||||
|
||||
def tearDown(self):
|
||||
self._tmp.cleanup()
|
||||
_EnvGuard.tearDown(self)
|
||||
|
||||
def test_new_list_switch_current_page_semantics(self):
|
||||
self.browser.navigate("https://a.example.com")
|
||||
self.assertEqual(self.browser.list_pages()["count"], 1)
|
||||
out = self.browser.new_page()
|
||||
self.assertEqual(out["index"], 1)
|
||||
self.assertEqual(out["count"], 2)
|
||||
self.browser.navigate("https://b.example.com")
|
||||
self.browser.click("button.x")
|
||||
page0, page1 = self.factory.handle.pages
|
||||
self.assertIn(("goto", "https://b.example.com"), page1.calls)
|
||||
self.assertNotIn(("goto", "https://b.example.com"), page0.calls)
|
||||
self.assertIn(("click", "button.x"), page1.calls)
|
||||
self.assertNotIn(("click", "button.x"), page0.calls)
|
||||
switched = self.browser.switch_page(0)
|
||||
self.assertEqual(switched["index"], 0)
|
||||
self.browser.click("a.y")
|
||||
self.assertIn(("click", "a.y"), page0.calls)
|
||||
self.assertNotIn(("click", "a.y"), page1.calls)
|
||||
listed = self.browser.list_pages()
|
||||
self.assertEqual(listed["count"], 2)
|
||||
self.assertEqual(listed["current"], 0)
|
||||
self.assertEqual([p["url"] for p in listed["pages"]],
|
||||
["https://a.example.com",
|
||||
"https://b.example.com"])
|
||||
self.assertEqual([p["current"] for p in listed["pages"]],
|
||||
[True, False])
|
||||
|
||||
def test_switch_missing_page_errors_and_keeps_current(self):
|
||||
self.browser.navigate("https://a.example.com")
|
||||
self.browser.new_page()
|
||||
for bad in (99, -1, "x"):
|
||||
with self.assertRaises(BrowserError) as ctx:
|
||||
self.browser.switch_page(bad)
|
||||
if bad != "x":
|
||||
self.assertIn("no such page", str(ctx.exception))
|
||||
self.assertEqual(self.browser.list_pages()["current"], 1)
|
||||
|
||||
def test_close_current_switches_to_previous(self):
|
||||
self.browser.navigate("https://a.example.com")
|
||||
self.browser.new_page()
|
||||
self.browser.new_page()
|
||||
out = self.browser.close_page()
|
||||
self.assertEqual(out["closed_index"], 2)
|
||||
self.assertEqual(out["index"], 1)
|
||||
self.assertEqual(out["count"], 2)
|
||||
self.assertIs(self.browser._page,
|
||||
self.factory.handle.pages[1])
|
||||
self.assertEqual(self.browser.list_pages()["current"], 1)
|
||||
|
||||
def test_close_noncurrent_keeps_same_page(self):
|
||||
self.browser.navigate("https://a.example.com")
|
||||
self.browser.new_page()
|
||||
self.browser.new_page()
|
||||
self.browser.switch_page(2)
|
||||
page2 = self.factory.handle.pages[2]
|
||||
out = self.browser.close_page(0)
|
||||
self.assertEqual(out["count"], 2)
|
||||
self.assertEqual(out["index"], 1)
|
||||
self.assertIs(self.browser._page, page2)
|
||||
|
||||
def test_close_last_page_errors_and_keeps_page(self):
|
||||
self.browser.navigate("https://a.example.com")
|
||||
with self.assertRaises(BrowserError) as ctx:
|
||||
self.browser.close_page()
|
||||
self.assertIn("last page", str(ctx.exception))
|
||||
with self.assertRaises(BrowserError):
|
||||
self.browser.close_page(0)
|
||||
self.assertEqual(self.browser.list_pages()["count"], 1)
|
||||
# Still usable after the refused close.
|
||||
self.browser.navigate("https://b.example.com")
|
||||
self.assertEqual(self.browser._current_url,
|
||||
"https://b.example.com")
|
||||
|
||||
def test_close_removes_live_files_and_list_needs_no_launch(self):
|
||||
self.browser.navigate("https://a.example.com")
|
||||
self.browser.live_shot()
|
||||
self.assertTrue(os.path.exists(os.path.join(
|
||||
self._tmp.name, "browser-live.png")))
|
||||
self.assertTrue(os.path.exists(os.path.join(
|
||||
self._tmp.name, "browser-live.json")))
|
||||
launches = self.factory.launch_count
|
||||
self.assertTrue(self.browser.close()["closed"])
|
||||
self.assertFalse(os.path.exists(os.path.join(
|
||||
self._tmp.name, "browser-live.png")))
|
||||
self.assertFalse(os.path.exists(os.path.join(
|
||||
self._tmp.name, "browser-live.json")))
|
||||
self.assertTrue(self.browser.close()["closed"])
|
||||
self.assertEqual(self.browser.list_pages(),
|
||||
{"pages": [], "current": None, "count": 0})
|
||||
self.assertEqual(self.factory.launch_count, launches)
|
||||
self.assertIsNone(self.browser._pw)
|
||||
|
||||
|
||||
class AllowlistTest(_EnvGuard, unittest.TestCase):
|
||||
def setUp(self):
|
||||
_EnvGuard.setUp(self)
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.factory = _FakePWMulti()
|
||||
self.browser = Browser(cwd=self._tmp.name,
|
||||
proxy="http://proxy:8888",
|
||||
browser_factory=lambda: self.factory)
|
||||
|
||||
def tearDown(self):
|
||||
self._tmp.cleanup()
|
||||
_EnvGuard.tearDown(self)
|
||||
|
||||
def test_unset_allowlist_behaves_as_before(self):
|
||||
os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
|
||||
out = self.browser.navigate("https://anything.example")
|
||||
self.assertTrue(out["ok"])
|
||||
self.assertEqual(self.factory.launch_count, 1)
|
||||
|
||||
def test_exact_and_suffix_match(self):
|
||||
os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com, .python.org"
|
||||
self.assertTrue(self.browser.navigate(
|
||||
"https://example.com")["ok"])
|
||||
self.assertTrue(self.browser.navigate(
|
||||
"https://docs.python.org")["ok"])
|
||||
self.assertTrue(self.browser.navigate(
|
||||
"https://python.org")["ok"])
|
||||
|
||||
def test_port_and_case_insensitive(self):
|
||||
os.environ["BROWSER_ALLOWED_HOSTS"] = "Example.COM"
|
||||
out = self.browser.navigate("https://EXAMPLE.com:8443/path")
|
||||
self.assertTrue(out["ok"])
|
||||
|
||||
def test_miss_blocked_before_launch(self):
|
||||
os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
|
||||
with self.assertRaises(BrowserError) as ctx:
|
||||
self.browser.navigate("https://evil.com")
|
||||
self.assertIn("evil.com", str(ctx.exception))
|
||||
self.assertEqual(self.factory.launch_count, 0)
|
||||
self.assertIsNone(self.browser._pw)
|
||||
self.assertIsNone(self.browser._page)
|
||||
|
||||
def test_localhost_blocked_unless_listed(self):
|
||||
os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
|
||||
with self.assertRaises(BrowserError):
|
||||
self.browser.navigate("http://localhost:8000/")
|
||||
self.assertEqual(self.factory.launch_count, 0)
|
||||
os.environ["BROWSER_ALLOWED_HOSTS"] = "localhost"
|
||||
out = self.browser.navigate("http://localhost:8000/")
|
||||
self.assertTrue(out["ok"])
|
||||
self.assertEqual(self.factory.launch_count, 1)
|
||||
|
||||
def test_file_allowed_when_allowlist_set(self):
|
||||
os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
|
||||
out = self.browser.navigate("file:///tmp/page.html")
|
||||
self.assertTrue(out["ok"])
|
||||
|
||||
def test_non_http_blocked_when_allowlist_set(self):
|
||||
os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com"
|
||||
with self.assertRaises(BrowserError):
|
||||
self.browser.navigate("about:blank")
|
||||
self.assertEqual(self.factory.launch_count, 0)
|
||||
|
||||
|
||||
class MCPMultiPageTest(_EnvGuard, unittest.TestCase):
|
||||
def setUp(self):
|
||||
_EnvGuard.setUp(self)
|
||||
os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.factory = _FakePWMulti()
|
||||
self.browser = Browser(cwd=self._tmp.name,
|
||||
proxy="http://proxy:8888",
|
||||
browser_factory=lambda: self.factory)
|
||||
|
||||
def tearDown(self):
|
||||
self._tmp.cleanup()
|
||||
_EnvGuard.tearDown(self)
|
||||
|
||||
def _call(self, name, args=None):
|
||||
resp = handle_request(
|
||||
{"jsonrpc": "2.0", "id": 9, "method": "tools/call",
|
||||
"params": {"name": name, "arguments": args or {}}},
|
||||
browser=self.browser)
|
||||
return resp
|
||||
|
||||
def _call_result(self, name, args=None):
|
||||
resp = self._call(name, args)
|
||||
self.assertFalse(resp["result"]["isError"],
|
||||
resp["result"]["content"][0]["text"])
|
||||
return json.loads(resp["result"]["content"][0]["text"])
|
||||
|
||||
def test_tools_list_includes_new_page_tools(self):
|
||||
listed = handle_request({"jsonrpc": "2.0", "id": 9,
|
||||
"method": "tools/list"})
|
||||
names = [t["name"] for t in listed["result"]["tools"]]
|
||||
for name in ("browser_new_page", "browser_list_pages",
|
||||
"browser_switch_page", "browser_close_page"):
|
||||
self.assertIn(name, names)
|
||||
|
||||
def test_mcp_dispatch_new_list_switch_close(self):
|
||||
self._call_result("browser_navigate",
|
||||
{"url": "https://a.example.com"})
|
||||
self.assertEqual(self._call_result("browser_new_page")["count"],
|
||||
2)
|
||||
listed = self._call_result("browser_list_pages")
|
||||
self.assertEqual(listed["count"], 2)
|
||||
self.assertEqual(listed["current"], 1)
|
||||
self.assertEqual(self._call_result("browser_switch_page",
|
||||
{"index": 0})["index"], 0)
|
||||
closed = self._call_result("browser_close_page")
|
||||
self.assertEqual(closed["count"], 1)
|
||||
self.assertEqual(closed["index"], 0)
|
||||
|
||||
def test_mcp_switch_missing_and_close_last_are_errors_not_crash(self):
|
||||
self._call_result("browser_navigate",
|
||||
{"url": "https://a.example.com"})
|
||||
resp = self._call("browser_switch_page", {"index": 99})
|
||||
self.assertTrue(resp["result"]["isError"])
|
||||
self.assertIn("no such page",
|
||||
resp["result"]["content"][0]["text"])
|
||||
resp = self._call("browser_close_page")
|
||||
self.assertTrue(resp["result"]["isError"])
|
||||
self.assertIn("last page",
|
||||
resp["result"]["content"][0]["text"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user