"""Worker wiring for web search MCP: tenant seeding + egress filter. No Docker, no network: _tenant_dirs runs against temp dirs, and the egress check only reads the repo filter file. """ import os import re import sys import tempfile import unittest from pathlib import Path try: from cloud.server import init_db from cloud.worker import Worker except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.server import init_db from cloud.worker import Worker REPO_ROOT = Path(__file__).resolve().parents[2] class WorkerWebsearchTestCase(unittest.TestCase): 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, **kwargs): kwargs.setdefault("env", {"XCM_API_KEY": "test-placeholder"}) kwargs.setdefault("tenants_root", os.path.join(self.tmp.name, "tenants")) kwargs.setdefault("tenant_template_home", self.template_home) kwargs.setdefault("tenant_template_repo", self.template_repo) kwargs.setdefault("tenant_isolation", True) return Worker(self.db_path, "/fake/host.py", **kwargs) def test_flag_off_is_old_behaviour(self): worker = self._worker() self.assertFalse(worker.websearch_mcp) dirs = worker._tenant_dirs("projA") config = open(os.path.join(dirs["codex_home"], "config.toml")).read() self.assertNotIn("[mcp_servers.websearch]", config) self.assertFalse(os.path.exists( os.path.join(dirs["codex_home"], "websearch_mcp.py"))) def test_flag_on_seeds_container_path_and_scripts(self): worker = self._worker(websearch_mcp=True) dirs = worker._tenant_dirs("projA") home = dirs["codex_home"] config = open(os.path.join(home, "config.toml")).read() self.assertIn("[mcp_servers.websearch]", config) # Container path, not the host temp path: this config is # read inside the task container, where the tenant home is # mounted at /var/lib/codex-app-server/codex-home. self.assertIn("/var/lib/codex-app-server/codex-home/" "websearch_mcp.py", config) self.assertNotIn(self.tmp.name, config) self.assertTrue(os.path.exists( os.path.join(home, "websearch.py"))) self.assertTrue(os.path.exists( os.path.join(home, "websearch_mcp.py"))) # The copied MCP script is the real one (not a stub). script = open(os.path.join(home, "websearch_mcp.py")).read() self.assertIn("web_search", script) self.assertIn("web_fetch", script) def test_old_tenant_merges_on_next_task(self): worker = self._worker() dirs = worker._tenant_dirs("projA") config_path = os.path.join(dirs["codex_home"], "config.toml") self.assertNotIn("[mcp_servers.websearch]", open(config_path).read()) # Flag flips on later (deploy enables it): next task's # _tenant_dirs must merge, not require a new project. worker2 = self._worker(websearch_mcp=True) worker2._tenant_dirs("projA") config = open(config_path).read() self.assertIn("[mcp_servers.websearch]", config) self.assertIn("xcmapi.org", config) self.assertEqual(config.count("[mcp_servers.websearch]"), 1) # Third call stays idempotent. worker2._tenant_dirs("projA") config = open(config_path).read() self.assertEqual(config.count("[mcp_servers.websearch]"), 1) class DockerEnvTestCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) def tearDown(self): self.tmp.cleanup() def _docker_context(self): worker = Worker(self.db_path, "/fake/host.py", env={"XCM_API_KEY": "test-placeholder"}, tenant_isolation=False, egress_proxy_port=None) return worker._docker_task_context({"project_id": "projA"}) def test_searxng_url_passes_into_container_env(self): from unittest import mock with mock.patch.dict(os.environ, {"SEARXNG_URL": "http://searx.diante.me"}): ok, _, _, env, err = self._docker_context() self.assertTrue(ok) self.assertIsNone(err) self.assertEqual(env.get("SEARXNG_URL"), "http://searx.diante.me") def test_no_search_env_means_no_extra_env(self): from unittest import mock with mock.patch.dict(os.environ, {}, clear=False): for k in ("SEARXNG_URL", "PERPLEXITY_API_KEY", "EXA_API_KEY", "TAVILY_API_KEY", "BRAVE_API_KEY"): os.environ.pop(k, None) ok, _, _, env, err = self._docker_context() self.assertTrue(ok) self.assertIsNone(err) self.assertIsNone(env) class EgressFilterTestCase(unittest.TestCase): def test_search_domains_present_in_basic_form(self): lines = open(REPO_ROOT / "deploy" / "egress" / "tinyproxy-filter").read().splitlines() entries = [ln for ln in lines if ln and not ln.startswith("#")] # No extended-regex syntax in any entry: + ? | and bare # parens are literal in BASIC and would break the allowlist. for entry in entries: self.assertNotIn("+", entry) self.assertNotIn("?", entry) self.assertNotIn("|", entry) self.assertNotIn("(", entry.replace("\\(", "")) self.assertNotIn(")", entry.replace("\\)", "")) expected = [ r"^\(.*\.\)*perplexity\.ai$", r"^\(.*\.\)*exa\.ai$", r"^\(.*\.\)*tavily\.com$", r"^\(.*\.\)*search\.brave\.com$", r"^\(.*\.\)*duckduckgo\.com$", r"^searx\.diante\.me$", ] 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()