Merge pull request #272 from openqj/feat-platform-browser
feat: platform browser multi-tab, navigation allowlist, close cleanup, egress filter, allowlist worker wiring
This commit is contained in:
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()
|
||||
@@ -5,6 +5,7 @@ egress check only reads the repo filter file.
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
@@ -160,6 +161,282 @@ class EgressFilterTestCase(unittest.TestCase):
|
||||
for want in expected:
|
||||
self.assertIn(want, entries, "missing filter entry: %s" % want)
|
||||
|
||||
def test_platform_browser_programming_domains(self):
|
||||
lines = open(REPO_ROOT / "deploy" / "egress" /
|
||||
"tinyproxy-filter").read().splitlines()
|
||||
entries = [ln for ln in lines
|
||||
if ln and not ln.startswith("#")]
|
||||
expected = [
|
||||
r"^\(.*\.\)*python\.org$",
|
||||
r"^\(.*\.\)*nodejs\.org$",
|
||||
r"^\(.*\.\)*go\.dev$",
|
||||
r"^\(.*\.\)*pkg\.go\.dev$",
|
||||
r"^\(.*\.\)*rust-lang\.org$",
|
||||
r"^\(.*\.\)*developer\.mozilla\.org$",
|
||||
r"^\(.*\.\)*crates\.io$",
|
||||
r"^\(.*\.\)*stackoverflow\.com$",
|
||||
r"^\(.*\.\)*cve\.mitre\.org$",
|
||||
r"^\(.*\.\)*nvd\.nist\.gov$",
|
||||
]
|
||||
for want in expected:
|
||||
self.assertIn(want, entries, "missing filter entry: %s" % want)
|
||||
# Already-present programming domains are NOT duplicated.
|
||||
for already in (r"^\(.*\.\)*github\.com$",
|
||||
r"^\(.*\.\)*pypi\.org$",
|
||||
r"^\(.*\.\)*npmjs\.org$"):
|
||||
self.assertEqual(entries.count(already), 1,
|
||||
"duplicated filter entry: %s" % already)
|
||||
# Control-plane guard: searx stays exact-only, and no
|
||||
# diante.me wildcard may appear (it would open api.diante.me).
|
||||
self.assertIn(r"^searx\.diante\.me$", entries)
|
||||
self.assertNotIn(r"^\(.*\.\)*diante\.me$", entries)
|
||||
|
||||
def test_platform_browser_filter_matching_semantics(self):
|
||||
# Translate POSIX BASIC group syntax to Python regex and
|
||||
# check the subdomain anchoring the comments promise.
|
||||
lines = open(REPO_ROOT / "deploy" / "egress" /
|
||||
"tinyproxy-filter").read().splitlines()
|
||||
entries = [ln for ln in lines
|
||||
if ln and not ln.startswith("#")]
|
||||
|
||||
def matches(domain):
|
||||
for entry in entries:
|
||||
py = entry.replace("\\(", "(").replace("\\)", ")")
|
||||
if re.match(py, domain):
|
||||
return True
|
||||
return False
|
||||
|
||||
# Subdomain coverage via the apex rules.
|
||||
self.assertTrue(matches("docs.python.org"))
|
||||
self.assertTrue(matches("docs.rust-lang.org"))
|
||||
self.assertTrue(matches("developer.mozilla.org"))
|
||||
self.assertTrue(matches("pkg.go.dev"))
|
||||
# Anchoring: lookalikes and suffix tricks must not match.
|
||||
self.assertFalse(matches("evilpython.org"))
|
||||
self.assertFalse(matches("python.org.evil.com"))
|
||||
self.assertFalse(matches("notstackoverflow.com"))
|
||||
# Fail-closed: control plane and out-of-scope categories
|
||||
# (shopping, social, mail, video) open no rule.
|
||||
self.assertFalse(matches("api.diante.me"))
|
||||
self.assertFalse(matches("www.amazon.com"))
|
||||
self.assertFalse(matches("www.youtube.com"))
|
||||
self.assertFalse(matches("mail.google.com"))
|
||||
|
||||
|
||||
class PlaywrightAllowlistWiringTestCase(unittest.TestCase):
|
||||
"""BROWSER_ALLOWED_HOSTS: worker env -> MCP env wiring.
|
||||
|
||||
No Docker, no network, no real Chromium: _docker_task_context
|
||||
runs against temp tenant dirs with a fake executor whose
|
||||
ensure_network always succeeds, and the MCP-side enforcement
|
||||
is checked through _allowed_hosts/_check_allowed_url (the same
|
||||
entry points Browser.navigate uses), not a launched browser.
|
||||
Values here are test placeholders, never real credentials.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||||
init_db(self.db_path)
|
||||
self.template_home = os.path.join(self.tmp.name, "tpl-home")
|
||||
os.makedirs(self.template_home)
|
||||
open(os.path.join(self.template_home, "config.toml"),
|
||||
"w").write('[model_provider.xcmapi]\n'
|
||||
'base_url="https://xcmapi.org/v1"\n')
|
||||
self.template_repo = os.path.join(self.tmp.name, "tpl-repo")
|
||||
os.makedirs(os.path.join(self.template_repo, ".git"))
|
||||
open(os.path.join(self.template_repo, "sample.py"),
|
||||
"w").write("def add(a, b):\n return a + b\n")
|
||||
|
||||
def tearDown(self):
|
||||
self.tmp.cleanup()
|
||||
|
||||
def _worker(self):
|
||||
from types import SimpleNamespace
|
||||
fake_executor = SimpleNamespace(
|
||||
ensure_network=lambda network: True)
|
||||
return Worker(self.db_path, "/fake/host.py",
|
||||
env={"XCM_API_KEY": "test-placeholder"},
|
||||
tenants_root=os.path.join(self.tmp.name,
|
||||
"tenants"),
|
||||
tenant_template_home=self.template_home,
|
||||
tenant_template_repo=self.template_repo,
|
||||
tenant_isolation=True, playwright_mcp=True,
|
||||
executor=fake_executor)
|
||||
|
||||
def _context(self, worker):
|
||||
return worker._docker_task_context({"project_id": "projA"})
|
||||
|
||||
def _config(self, worker):
|
||||
home = worker._tenant_dirs("projA")["codex_home"]
|
||||
return open(os.path.join(home, "config.toml")).read()
|
||||
|
||||
def _env_block(self, config):
|
||||
import json
|
||||
block = {}
|
||||
in_block = False
|
||||
for line in config.splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped == "[mcp_servers.playwright.env]":
|
||||
in_block = True
|
||||
continue
|
||||
if in_block:
|
||||
if stripped.startswith("["):
|
||||
break
|
||||
if "=" in stripped:
|
||||
key, raw = stripped.split("=", 1)
|
||||
block[key.strip()] = json.loads(raw.strip())
|
||||
return block
|
||||
|
||||
def _drop_ambient(self):
|
||||
# mock.patch.dict only overlays: drop any search/allowlist
|
||||
# keys the ambient environment may carry so each case sees
|
||||
# exactly the env it declares (pops are restored on exit).
|
||||
for key in ("SEARXNG_URL", "SEARXNG_ENGINES",
|
||||
"PERPLEXITY_API_KEY", "EXA_API_KEY",
|
||||
"TAVILY_API_KEY", "BRAVE_API_KEY"):
|
||||
os.environ.pop(key, None)
|
||||
|
||||
def test_switch_off_passes_no_variable(self):
|
||||
from unittest import mock
|
||||
worker = self._worker()
|
||||
with mock.patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
|
||||
self._drop_ambient()
|
||||
ok, _, _, env, err = self._context(worker)
|
||||
self.assertTrue(ok)
|
||||
self.assertIsNone(err)
|
||||
block = self._env_block(self._config(worker))
|
||||
self.assertEqual(block.get("PLAYWRIGHT_BROWSERS_PATH"),
|
||||
"/opt/ms-playwright")
|
||||
self.assertNotIn("BROWSER_ALLOWED_HOSTS", block)
|
||||
# The allowlist never reaches the container env either:
|
||||
# the config.toml MCP env block is the only channel.
|
||||
self.assertIsNone(env)
|
||||
|
||||
def test_switch_on_reaches_mcp_env(self):
|
||||
from unittest import mock
|
||||
worker = self._worker()
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{"BROWSER_ALLOWED_HOSTS": "example.com, .python.org"}):
|
||||
self._drop_ambient()
|
||||
ok, _, _, env, err = self._context(worker)
|
||||
self.assertTrue(ok)
|
||||
self.assertIsNone(err)
|
||||
block = self._env_block(self._config(worker))
|
||||
# Verbatim, not normalised here: normalisation happens
|
||||
# at read time in _allowed_hosts.
|
||||
self.assertEqual(block.get("BROWSER_ALLOWED_HOSTS"),
|
||||
"example.com, .python.org")
|
||||
self.assertEqual(block.get("PLAYWRIGHT_BROWSERS_PATH"),
|
||||
"/opt/ms-playwright")
|
||||
self.assertIsNone(env)
|
||||
|
||||
def test_empty_value_is_never_written(self):
|
||||
from unittest import mock
|
||||
worker = self._worker()
|
||||
with mock.patch.dict(os.environ, {"BROWSER_ALLOWED_HOSTS": ""}):
|
||||
ok, _, _, _, err = self._context(worker)
|
||||
self.assertTrue(ok)
|
||||
self.assertIsNone(err)
|
||||
block = self._env_block(self._config(worker))
|
||||
self.assertNotIn("BROWSER_ALLOWED_HOSTS", block)
|
||||
|
||||
def test_old_tenant_merge_is_idempotent(self):
|
||||
from unittest import mock
|
||||
worker = self._worker()
|
||||
# Old tenant: seeded before the wiring existed, with a
|
||||
# stale env block and a section after it.
|
||||
home = worker._tenant_dirs("projA")["codex_home"]
|
||||
config_path = os.path.join(home, "config.toml")
|
||||
with open(config_path, "a", encoding="utf-8") as fh:
|
||||
fh.write("\n[mcp_servers.playwright.env]\n"
|
||||
'BROWSER_ALLOWED_HOSTS = "stale.example"\n'
|
||||
"\n[other]\nkey = 1\n")
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{"BROWSER_ALLOWED_HOSTS": "example.com"}):
|
||||
ok, _, _, _, err = self._context(worker)
|
||||
self.assertTrue(ok)
|
||||
self.assertIsNone(err)
|
||||
config = open(config_path).read()
|
||||
self.assertEqual(config.count("[mcp_servers.playwright.env]"), 1)
|
||||
self.assertIn("[other]\nkey = 1\n", config)
|
||||
self.assertEqual(
|
||||
self._env_block(config).get("BROWSER_ALLOWED_HOSTS"),
|
||||
"example.com")
|
||||
# Second prep with the same env rewrites nothing.
|
||||
before = open(config_path, "rb").read()
|
||||
mtime_before = os.stat(config_path).st_mtime_ns
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{"BROWSER_ALLOWED_HOSTS": "example.com"}):
|
||||
ok, _, _, _, err = self._context(worker)
|
||||
self.assertTrue(ok)
|
||||
self.assertIsNone(err)
|
||||
self.assertEqual(open(config_path, "rb").read(), before)
|
||||
self.assertEqual(os.stat(config_path).st_mtime_ns, mtime_before)
|
||||
|
||||
def test_ensure_playwright_mcp_env_noop_when_identical(self):
|
||||
from cloud.playwright_browser import ensure_playwright_mcp_env
|
||||
home = os.path.join(self.tmp.name, "home")
|
||||
os.makedirs(home)
|
||||
config_path = os.path.join(home, "config.toml")
|
||||
open(config_path, "w").write(
|
||||
'[model_provider.xcmapi]\n'
|
||||
'base_url="https://xcmapi.org/v1"\n')
|
||||
env = {"PLAYWRIGHT_BROWSERS_PATH": "/opt/ms-playwright",
|
||||
"BROWSER_ALLOWED_HOSTS": "example.com"}
|
||||
self.assertTrue(ensure_playwright_mcp_env(home, env))
|
||||
before = open(config_path, "rb").read()
|
||||
mtime_before = os.stat(config_path).st_mtime_ns
|
||||
self.assertFalse(ensure_playwright_mcp_env(home, env))
|
||||
self.assertEqual(open(config_path, "rb").read(), before)
|
||||
self.assertEqual(os.stat(config_path).st_mtime_ns, mtime_before)
|
||||
# A changed value still rewrites, in place, once.
|
||||
env2 = dict(env, BROWSER_ALLOWED_HOSTS="other.example")
|
||||
self.assertTrue(ensure_playwright_mcp_env(home, env2))
|
||||
config = open(config_path).read()
|
||||
self.assertEqual(config.count("[mcp_servers.playwright.env]"), 1)
|
||||
self.assertEqual(
|
||||
self._env_block(config).get("BROWSER_ALLOWED_HOSTS"),
|
||||
"other.example")
|
||||
|
||||
def test_wiring_semantics_allowlist_only_after_wiring(self):
|
||||
from unittest import mock
|
||||
from cloud.playwright_browser import (
|
||||
BrowserError, _allowed_hosts, _check_allowed_url)
|
||||
worker = self._worker()
|
||||
# Switch off: the MCP env has no variable, so no limit.
|
||||
with mock.patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
|
||||
self._drop_ambient()
|
||||
ok, _, _, _, err = self._context(worker)
|
||||
self.assertTrue(ok)
|
||||
self.assertIsNone(err)
|
||||
block = self._env_block(self._config(worker))
|
||||
self.assertNotIn("BROWSER_ALLOWED_HOSTS", block)
|
||||
with mock.patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("BROWSER_ALLOWED_HOSTS", None)
|
||||
self.assertIsNone(_allowed_hosts())
|
||||
_check_allowed_url("https://anything.example")
|
||||
# Switch on: the written block, applied as MCP process
|
||||
# env, makes the allowlist enforce.
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{"BROWSER_ALLOWED_HOSTS": "example.com, .python.org"}):
|
||||
ok, _, _, _, err = self._context(worker)
|
||||
self.assertTrue(ok)
|
||||
block = self._env_block(self._config(worker))
|
||||
self.assertIn("BROWSER_ALLOWED_HOSTS", block)
|
||||
with mock.patch.dict(os.environ, block):
|
||||
self.assertEqual(_allowed_hosts(),
|
||||
["example.com", ".python.org"])
|
||||
_check_allowed_url("https://docs.python.org")
|
||||
with self.assertRaises(BrowserError):
|
||||
_check_allowed_url("https://evil.com")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user