Merge pull request #272 from openqj/feat-platform-browser

feat: platform browser multi-tab, navigation allowlist, close cleanup, egress filter, allowlist worker wiring

cloud/playwright_browser.py

@@ -1,7 +1,10 @@
 """Headless Chromium automation for Codex tasks, via Playwright.
 
-One Browser object owns one page for the lifetime of an MCP
-process. It is deliberately small: navigate, snapshot (page text),
+One Browser object owns a list of pages for the lifetime of an
+MCP process (multi-tab: new/list/switch/close tools; the
+single-page tools act on the current page, and each new_page
+gets a fresh Playwright context, so cookies are isolated
+between pages). It is deliberately small: navigate, snapshot (page text),
 click, fill, evaluate, screenshot, close. Anything fancier the
 agent can do with the shell tool, which has the same sandbox
 privileges as this browser (tenant Docker container, egress
@@ -48,9 +51,65 @@ def _validate_basename(name, what):
     return name
 
 
+def _allowed_hosts():
+    """BROWSER_ALLOWED_HOSTS, or None when unset/empty.
+
+    Comma list, case-insensitive. An entry with a leading dot
+    (e.g. .python.org) matches the bare domain and any subdomain;
+    a plain entry matches exactly. localhost/127.0.0.1 are NOT
+    special-cased: they are allowed only when listed explicitly.
+    Unset (or empty) means no allowlist check at all -- the old
+    behaviour, where only the egress proxy whitelist applies.
+    Read on every navigate so tests and env changes take effect
+    without recreating the Browser."""
+    raw = os.environ.get("BROWSER_ALLOWED_HOSTS")
+    if raw is None:
+        return None
+    hosts = [entry.strip().lower() for entry in raw.split(",")]
+    hosts = [entry for entry in hosts if entry]
+    return hosts or None
+
+
+def _check_allowed_url(url):
+    """Raise BrowserError before launch when the allowlist is set
+    and the URL is not allowed. file:// is always allowed; every
+    other scheme except http(s) is blocked when the allowlist is
+    set, because it has no host to match."""
+    allowed = _allowed_hosts()
+    if allowed is None:
+        return
+    from urllib.parse import urlparse
+    parsed = urlparse(str(url))
+    scheme = (parsed.scheme or "").lower()
+    if scheme == "file":
+        return
+    if scheme not in ("http", "https"):
+        raise BrowserError(
+            "host allowlist blocks non-http(s) URL %r when "
+            "BROWSER_ALLOWED_HOSTS is set" % (str(url),))
+    host = (parsed.hostname or "").lower()
+    if not host:
+        raise BrowserError(
+            "host allowlist blocks URL with no host %r" % (str(url),))
+    for entry in allowed:
+        if entry.startswith("."):
+            if host == entry[1:] or host.endswith(entry):
+                return
+        elif host == entry:
+            return
+    raise BrowserError(
+        "host %r is not in BROWSER_ALLOWED_HOSTS (%s); "
+        "add it (or a leading-dot suffix entry) to allow this "
+        "domain" % (host, ", ".join(allowed)))
+
+
 class Browser:
-    """A single Chromium page. ``browser_factory`` is injectable:
-    tests pass a fake returning a fake sync_playwright object."""
+    """Chromium pages (multi-tab). ``browser_factory`` is
+    injectable: tests pass a fake returning a fake
+    sync_playwright object. Single-page methods act on the
+    current page; new_page/switch_page/close_page manage the
+    page list, and each new_page gets a fresh Playwright
+    context, so cookies are isolated between pages."""
 
     def __init__(self, cwd=None, proxy=None, browser_factory=None):
         self.cwd = os.path.abspath(cwd or os.getcwd())
@@ -60,6 +119,8 @@ class Browser:
         self._pw = None
         self._browser = None
         self._page = None
+        self._pages = []
+        self._current_index = 0
         self._current_url = ""
         self._last_title = ""
         # ref snapshot state (playwright-mcp style): ref -> css path
@@ -110,6 +171,8 @@ class Browser:
         self._pw = pw
         self._browser = browser
         self._page = page
+        self._pages = [page]
+        self._current_index = 0
 
     def _info(self):
         try:
@@ -128,6 +191,9 @@ class Browser:
     def navigate(self, url):
         if not url or not str(url).strip():
             raise BrowserError("url is required")
+        # Before _ensure on purpose: a blocked host must fail
+        # without launching Chromium at all.
+        _check_allowed_url(url)
         self._ensure()
         try:
             self._page.goto(str(url), timeout=NAV_TIMEOUT_MS,
@@ -141,6 +207,146 @@ class Browser:
         result["ok"] = True
         return result
 
+    def _parse_page_index(self, index):
+        """Shared switch/close index validation. Raises
+        BrowserError with the page count, never IndexError."""
+        if isinstance(index, bool):
+            raise BrowserError(
+                "page index must be an integer, got %r" % (index,))
+        try:
+            idx = int(index)
+        except (TypeError, ValueError):
+            raise BrowserError(
+                "page index must be an integer, got %r" % (index,))
+        if isinstance(index, float) and idx != index:
+            raise BrowserError(
+                "page index must be an integer, got %r" % (index,))
+        if not self._pages:
+            raise BrowserError(
+                "no pages open: call browser_new_page or navigate "
+                "first")
+        if not 0 <= idx < len(self._pages):
+            raise BrowserError(
+                "no such page: index %d (%d page(s) open, valid "
+                "0-%d)" % (idx, len(self._pages),
+                            len(self._pages) - 1))
+        return idx
+
+    def _stop_screencast(self):
+        if self._cdp is not None:
+            try:
+                self._cdp.send("Page.stopScreencast")
+            except Exception:  # noqa: BLE001 - best-effort
+                pass
+            self._cdp = None
+            self._cdp_page = None
+
+    def _sync_current_url(self):
+        try:
+            self._current_url = self._page.url or ""
+        except Exception:  # noqa: BLE001 - url is best-effort
+            self._current_url = ""
+
+    def new_page(self):
+        """Open a new page and make it current. Every new_page
+        gets a fresh Playwright context (browser.new_page()),
+        so cookies and storage are isolated from other pages.
+        Single-page tools then act on this page until a switch."""
+        self._ensure()
+        try:
+            page = self._browser.new_page(
+                viewport={"width": 1280, "height": 800})
+        except Exception as exc:  # noqa: BLE001
+            raise BrowserError("new_page failed: %s" % exc)
+        self._stop_screencast()
+        self._pages.append(page)
+        self._current_index = len(self._pages) - 1
+        self._page = page
+        self._ref_paths = {}
+        self._sync_current_url()
+        result = self._info()
+        result["ok"] = True
+        result["index"] = self._current_index
+        result["count"] = len(self._pages)
+        return result
+
+    def list_pages(self):
+        """List all open pages without launching a browser when
+        none is open. Titles/URLs are best-effort per page."""
+        if not self._pages:
+            return {"pages": [], "current": None, "count": 0}
+        pages = []
+        for i, page in enumerate(self._pages):
+            try:
+                url = page.url or ""
+            except Exception:  # noqa: BLE001 - best-effort
+                url = ""
+            try:
+                title = page.title() or ""
+            except Exception:  # noqa: BLE001 - best-effort
+                title = ""
+            pages.append({"index": i, "url": url, "title": title,
+                          "current": i == self._current_index})
+        return {"pages": pages, "current": self._current_index,
+                "count": len(pages)}
+
+    def switch_page(self, index):
+        """Make page ``index`` current. Single-page tools then
+        act on it; refs from the previous page are dropped,
+        because they map to that page's DOM."""
+        idx = self._parse_page_index(index)
+        if idx != self._current_index:
+            self._stop_screencast()
+            self._current_index = idx
+            self._page = self._pages[idx]
+            self._ref_paths = {}
+            self._sync_current_url()
+        result = self._info()
+        result["ok"] = True
+        result["index"] = self._current_index
+        result["count"] = len(self._pages)
+        return result
+
+    def close_page(self, index=None):
+        """Close one page. Closing the last page is an error --
+        use browser_close for the whole browser. Closing the
+        current page switches to the previous page (index-1, or
+        0 when the first page was closed); closing a page before
+        the current one keeps the same page current (its index
+        shifts down by one)."""
+        if not self._pages:
+            raise BrowserError(
+                "no pages open: call browser_new_page or navigate "
+                "first")
+        if len(self._pages) == 1:
+            raise BrowserError(
+                "cannot close the last page (index 0): use "
+                "browser_close to close the whole browser")
+        idx = (self._current_index if index is None
+               else self._parse_page_index(index))
+        page = self._pages[idx]
+        try:
+            page.close()
+        except Exception as exc:  # noqa: BLE001
+            raise BrowserError("close_page failed: %s" % exc)
+        del self._pages[idx]
+        if self._cdp_page is page:
+            self._cdp = None
+            self._cdp_page = None
+        if idx < self._current_index:
+            self._current_index -= 1
+        elif idx == self._current_index:
+            self._current_index = idx - 1 if idx > 0 else 0
+        self._page = self._pages[self._current_index]
+        self._ref_paths = {}
+        self._sync_current_url()
+        result = self._info()
+        result["ok"] = True
+        result["closed_index"] = idx
+        result["index"] = self._current_index
+        result["count"] = len(self._pages)
+        return result
+
     def snapshot(self):
         self._ensure()
         try:
@@ -482,8 +688,17 @@ class Browser:
                 pass
             self._cdp = None
             self._cdp_page = None
-        for closer in (lambda: self._page and self._page.close(),
-                       lambda: self._browser and self._browser.close(),
+        for page in list(self._pages):
+            try:
+                page.close()
+            except Exception as exc:  # noqa: BLE001 - close is best-effort
+                errors.append(str(exc))
+        if self._page is not None and self._page not in self._pages:
+            try:
+                self._page.close()
+            except Exception as exc:  # noqa: BLE001 - close is best-effort
+                errors.append(str(exc))
+        for closer in (lambda: self._browser and self._browser.close(),
                        lambda: self._pw and self._pw.stop()):
             try:
                 closer()
@@ -492,6 +707,19 @@ class Browser:
         self._page = None
         self._browser = None
         self._pw = None
+        self._pages = []
+        self._current_index = 0
+        self._ref_paths = {}
+        # Live-view files are per-browser, not per-shot: closing
+        # the browser must not leave a stale browser-live.png /
+        # browser-live.json for a watcher to mistake for a live
+        # page. Failures are ignored on purpose (close is
+        # best-effort; a missing file is the normal case).
+        for name in ("browser-live.png", "browser-live.json"):
+            try:
+                os.remove(os.path.join(self.cwd, name))
+            except OSError:
+                pass
         return {"closed": True, "errors": errors}
 
 
@@ -576,7 +804,9 @@ def ensure_playwright_mcp_env(codex_home_dir, env):
     proxy vars and PLAYWRIGHT_BROWSERS_PATH must live in
     config.toml. Idempotent in-place replacement of the env
     subsection; other sections untouched. Returns True when the
-    file was written.
+    file was written, False without rewriting when the resulting
+    file would be identical to the existing one (so a second
+    task prep for an unchanged tenant leaves the file alone).
     """
     if not env:
         return False
@@ -589,6 +819,7 @@ def ensure_playwright_mcp_env(codex_home_dir, env):
                 existing = fh.read()
     except OSError:
         return False
+    original = existing
     start = existing.find("[mcp_servers.playwright.env]")
     if start >= 0:
         end = existing.find("\n[", start + 1)
@@ -598,12 +829,23 @@ def ensure_playwright_mcp_env(codex_home_dir, env):
     lines = ["\n[mcp_servers.playwright.env]\n"]
     for key in sorted(env):
         lines.append("%s = %s\n" % (key, json.dumps(str(env[key]))))
+    # Canonical join: the old write (stripped body + lines[0]'s
+    # leading blank line) grew one blank line per call, so a
+    # second prep never produced identical bytes. Normalise
+    # instead: body without trailing newlines, exactly one blank
+    # line, then the block; an empty body keeps a single leading
+    # blank line. This converges in one rewrite and is stable
+    # afterwards, which the no-op check below relies on.
+    block = "".join(lines).lstrip("\n")
+    if existing.strip():
+        new_text = existing.rstrip("\n") + "\n\n" + block
+    else:
+        new_text = "\n" + block
+    if new_text == original:
+        return False
     try:
         with open(config_path, "w", encoding="utf-8") as fh:
-            fh.write(existing)
-            if existing and not existing.endswith("\n"):
-                fh.write("\n")
-            fh.writelines(lines)
+            fh.write(new_text)
     except OSError:
         return False
     return True

cloud/playwright_mcp.py

@@ -16,7 +16,11 @@ proxy vars and PLAYWRIGHT_BROWSERS_PATH arrive via the MCP env
 block in config.toml, never via tool arguments.
 
 One Browser instance lives for the whole MCP process: navigate
-once, then snapshot/click/fill/screenshot against the same page.
+once, then snapshot/click/fill/screenshot against the current
+page. browser_new_page / browser_list_pages /
+browser_switch_page / browser_close_page manage multiple pages;
+each new page gets a fresh Playwright context, so cookies are
+isolated between pages.
 """
 
 import json
@@ -217,6 +221,57 @@ _TOOLS = [
             },
         },
     },
+    {
+        "name": "browser_new_page",
+        "description": (
+            "Open a new page (tab) and make it current. Every "
+            "new page gets a fresh Playwright context, so "
+            "cookies and storage are isolated from other pages. "
+            "Later navigate/snapshot/click/fill/screenshot act "
+            "on this page until browser_switch_page."
+        ),
+        "inputSchema": {"type": "object", "properties": {}},
+    },
+    {
+        "name": "browser_list_pages",
+        "description": (
+            "List all open pages with index, URL, title and "
+            "which one is current. Does not launch a browser "
+            "when none is open."
+        ),
+        "inputSchema": {"type": "object", "properties": {}},
+    },
+    {
+        "name": "browser_switch_page",
+        "description": (
+            "Make page ``index`` (from browser_list_pages) "
+            "current. Later single-page tools act on it. "
+            "Switching to a missing index is an error, not a "
+            "crash; refs from the previous page are dropped."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "index": {"type": "integer"},
+            },
+            "required": ["index"],
+        },
+    },
+    {
+        "name": "browser_close_page",
+        "description": (
+            "Close one page (``index``, or the current page "
+            "when omitted). Closing the last page is an error; "
+            "use browser_close for the whole browser. Closing "
+            "the current page switches to the previous page."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "index": {"type": "integer"},
+            },
+        },
+    },
     {
         "name": "browser_close",
         "description": "Close the browser page. Safe to call twice.",
@@ -337,6 +392,14 @@ def handle_request(request, browser=None):
             elif name == "browser_screenshot":
                 result = target.screenshot(
                     args.get("output_filename"))
+            elif name == "browser_new_page":
+                result = target.new_page()
+            elif name == "browser_list_pages":
+                result = target.list_pages()
+            elif name == "browser_switch_page":
+                result = target.switch_page(args.get("index"))
+            elif name == "browser_close_page":
+                result = target.close_page(args.get("index"))
             elif name == "browser_close":
                 result = target.close()
             else:

cloud/skills/browser-automation/SKILL.md

@@ -60,3 +60,38 @@ snapshot_refs 再点）；`browser_mouse_move` / `browser_key_press` /
 5. 不要用浏览器做静态抓取能做的事；不要用 evaluate 绕过
    click / fill 的正常路径去改页面状态，除非 snapshot 已证明
    正常路径走不通。
+
+## 多页与域名白名单（本期新增）
+
+多页工具：`browser_new_page`（开新页并切到它，每次
+new_page 都是全新 Playwright context，cookie 与存储天然
+隔离，适合同时登录两个账号或对照两个页面）、
+`browser_list_pages`（列出全部页：index、URL、标题、
+哪一页是当前页；浏览器未启动时不启动，只返回空列表）、
+`browser_switch_page`（按 index 切到指定页，之后
+navigate / snapshot / click / fill / screenshot 都针对该页；
+切到不存在的 index 会明确报错，不会崩掉 MCP，当前页不变；
+切换后旧页的 snapshot_refs 全部失效，要重新 snapshot_refs）、
+`browser_close_page`（关一页，缺省关当前页；关当前页后
+自动切到上一页；关最后一页会明确报错，改用
+`browser_close` 关整个浏览器）。
+
+域名白名单：环境变量 `BROWSER_ALLOWED_HOSTS`（逗号列表，
+如 `example.com,.python.org`）。未设置时行为完全不变，
+只靠 egress 代理白名单；设置后 navigate 在启动浏览器之前
+就检查：`file://` 一律放行，`http(s)` 的 host 必须命中列表
+（前导点条目做后缀匹配：`.python.org` 同时放行
+`python.org` 与 `docs.python.org`，不放行
+`notpython.org`），大小写与端口不影响匹配，未命中直接
+报错且不启动 Chromium。`localhost` / `127.0.0.1` 没有
+特殊待遇：本地开发服务器要自己把它列进变量，否则会被挡。
+
+清理：`browser_close` 会关掉全部页，并尽力删除工作区里的
+`browser-live.png` 与 `browser-live.json`（删不掉也不报错），
+所以 close 之后不要再拿旧 live 文件当“当前画面”汇报；
+单个截图（`browser_screenshot` 的 `shot-*.png`）不受影响。
+
+不在本轨范围：下载没有落盘清单（下载仍只落工作区文件，
+无统一清单可查）；没有持久 profile（每次 MCP 进程结束
+cookie 即丢）；没有登录保险库（账号密码仍不进浏览器工具）。
+需要这些时不要假装已有，先如实说明缺口。

cloud/tests/test_playwright.py

@@ -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()

cloud/tests/test_websearch_worker.py

@@ -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()

cloud/worker.py

@@ -3240,6 +3240,18 @@ class Worker:
                 playwright_env = dict(proxy_env or {})
                 playwright_env["PLAYWRIGHT_BROWSERS_PATH"] = (
                     "/opt/ms-playwright")
+                # 导航白名单透传：worker 环境里的
+                # BROWSER_ALLOWED_HOSTS 只在已设置且非空时写进
+                # MCP env；未设置时绝不写空值（空值会覆盖租户
+                # 自己的配置，且 _allowed_hosts 对空串本就当未
+                # 设置——写它没有意义）。只写 playwright 的 env
+                # 块，不进 websearch env、不进容器 -e：MCP 子进
+                # 程只读 config.toml 里的 env（2026-10-30 实测
+                # 容器 -e 到不了 MCP 进程）。值绝不进日志：下
+                # 面失败分支只记异常，不记 env 内容。
+                allowed_hosts = os.environ.get("BROWSER_ALLOWED_HOSTS")
+                if allowed_hosts:
+                    playwright_env["BROWSER_ALLOWED_HOSTS"] = allowed_hosts
                 home = self._tenant_dirs(
                     task.get("project_id"))["codex_home"]
                 _ensure_playwright_mcp_env(home, playwright_env)

deploy/egress/tinyproxy-filter

@@ -45,3 +45,23 @@
 ^\(.*\.\)*weather\.com\.cn$
 ^\(.*\.\)*meteoblue\.com$
 ^\(.*\.\)*accuweather\.com$
+# Platform browser programming sources (2026-11-01, feat-platform-browser).
+# Only programming sources: official docs, GitHub, package registries,
+# Stack Overflow, CVE/NVD. Deliberately NOT included: shopping, social,
+# mail, general news, video sites. Same BASIC form as above.
+# Already present above, not duplicated here: github.com, pypi.org,
+# npmjs.org (plus pythonhosted.org for pip). Notes on coverage:
+# docs.python.org is covered by the python.org line (subdomain rule);
+# docs.rust-lang.org by rust-lang.org; nodejs.org docs live on
+# nodejs.org itself. developer.mozilla.org is scoped to that host
+# only, NOT a mozilla.org wildcard, so no other Mozilla host opens.
+^\(.*\.\)*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$

docs/platform-browser.md

@@ -0,0 +1,101 @@
+# 平台浏览器：与 Muse browser 的对照
+
+本轨目标：把编程任务平台的浏览器能力复刻到与 Muse browser
+可对照的程度——只服务编程任务（验证自己的 Web 应用、读官方
+文档、查包与 CVE），不复刻个人 agent 的通用上网能力。
+
+范围边界先说清：
+
+- 搜索（browser.search / browser.open / browser.find 那一路）
+  由另一轨负责，本轨只消费、不重建。
+- 多标签页已由本轨另一路完成：`browser_new_page` / `browser_list_pages` / `browser_switch_page` / `browser_close_page`，导航白名单见下文补充。
+- 本轨不部署：filter 改动只落在仓库文件里，主机 tinyproxy
+  未 reload；真实站点未 live 验证（见文末“未验证”一节）。
+
+## 逐项对照
+
+| 能力 | Muse browser | 本轨状态 |
+| --- | --- | --- |
+| Live Chromium（真浏览器，非文本抓取） | 有 | 已有：`cloud/playwright_browser.py`，单页 headless Chromium，1280x800 视口 |
+| 导航 | 有 | 已有：`browser_navigate`；代理显式传 launch，域名不在白名单时报代理 403，不编造页面 |
+| 快照（无障碍式清单） | 有 | 已有：`browser_snapshot`，及 `browser_snapshot_refs`（ref -> 元素清单，`browser_click_ref` 按 ref 点击） |
+| 点击 / 填写 | 有 | 已有：`browser_click`、`browser_click_at`（坐标，有视口边界检查）、`browser_fill`、`browser_type_text` |
+| 键盘 / 鼠标 | 有 | 已有：`browser_key_press`、`browser_mouse_move` |
+| 截图 / 实时画面 | 有 | 已有：`browser_screenshot`、`browser_live_shot`（固定 live 文件 + JSON sidecar，供 console 轮询）；截图只写任务工作区（`cwd`，文件名 basename 校验），不写共享目录 |
+| JS 求值 | 有 | 已有：`browser_evaluate` |
+| MCP 工具面 | 有 | 已有：`cloud/playwright_mcp.py`，stdio MCP，14 个工具（navigate / snapshot / snapshot_refs / click / click_at / click_ref / fill / type_text / key_press / mouse_move / evaluate / screenshot / live_shot / close） |
+| 观看 + 接管 | 有（观看与接管） | 已有：`BrowserPanel.tsx` 是观看 + 接管面板，不是纯观看面板；`holder` 驱动——用户接管（holder=user）时 agent 工具报错、把控制权交还用户，交还后 agent 续跑前先快照确认用户改动 |
+| Worker 播种 | — | 已有：worker `--playwright-mcp` 把 MCP 文件与配置播种进租户 codex-home |
+| 租户工作区隔离 | — | 已有：浏览器跑在任务沙箱内，截图与下载落任务工作区 |
+| 代理 403 如实报错 | — | 已有：域名不在 Tinyproxy 白名单时工具 plain 报错 403，不静默失败、不编造内容 |
+| 持久 profile / 跨回合登录态 | 有 | 缺口（本轨未做）：每次任务的浏览器是 fresh 会话，登录态不跨回合保留 |
+| 登录凭证保险库 | 有 | 缺口 + 本轨不做：不存任何已存凭证；见“本轨不做” |
+| 服务端审批卡片（表单提交 / 下载 / 上传的 Allow / Ask / Deny） | 有 | 缺口（本轨未做）：本轨只有 holder 接管阻塞（用户接管时 agent 工具报错停下），没有提交/下载/上传的审批流 |
+| 验证码 / 机器人检查 | 有处理流程 | 缺口（本轨未做绕过）：被挡就报 blocked 停下，不绕过、不伪装 |
+| 下载落盘清单 | 有 | 缺口（本轨未做）：除截图落任务工作区外，无统一的下载物清单面 |
+| 多标签页 | 有 | 已有（本轨补）：`browser_new_page` / `browser_list_pages` / `browser_switch_page` / `browser_close_page`；新页自带全新 Playwright context，cookie/存储隔离 |
+| 真实站点 live 验证 | 有 | 未做：本轨只改 egress 白名单与文档，未在主机 tinyproxy 上 reload、未对真实站点跑 live 验证 |
+
+## 本轨不做（明确排除）
+
+- 个人购物下单。
+- 个人邮箱 / 社交登录。
+- 验证码绕过、机器人检查绕过。
+- 已存凭证保险库（任何真实密码、cookie、token 绝不写进
+  代码、测试、文档、回执）。
+
+## 本轨改动：egress 白名单
+
+文件：`deploy/egress/tinyproxy-filter`（Tinyproxy POSIX BASIC
+正则写法，一行一条，`^\(.*\.\)*域名$` 形式；不要用 extended
+锚定写法，2026-10-26 已实测 extended 写法会把允许域名也挡成
+403；文件保持 644 可读）。
+
+本轨新增 10 条（编程来源限定）：
+
+- `^\(.*\.\)*python\.org$`（含 docs.python.org 子域名规则覆盖）
+- `^\(.*\.\)*nodejs\.org$`
+- `^\(.*\.\)*go\.dev$`
+- `^\(.*\.\)*pkg\.go\.dev$`
+- `^\(.*\.\)*rust-lang\.org$`（含 docs.rust-lang.org）
+- `^\(.*\.\)*developer\.mozilla\.org$`（只放行该 host，不放行
+  mozilla.org 通配）
+- `^\(.*\.\)*crates\.io$`
+- `^\(.*\.\)*stackoverflow\.com$`
+- `^\(.*\.\)*cve\.mitre\.org$`
+- `^\(.*\.\)*nvd\.nist\.gov$`
+
+已在白名单、无需重复加：`github.com`、`pypi.org`、`npmjs.org`
+（另有 `pythonhosted.org` 供 pip）。
+
+刻意不动：`^searx\.diante\.me$` 精确行不动；未加任何
+`diante.me` 通配（那会放行 control plane api.diante.me）；
+未加购物、社交、邮箱、通用新闻、视频站。
+
+对应测试：`cloud/tests/test_websearch_worker.py` 的
+`EgressFilterTestCase` 新增 2 项（见“验证”一节）——断言新规则
+存在、已存在规则不重复、searx 精确行仍在且无 diante.me 通配、
+以及用 BASIC 转 extended 后的实际匹配语义（子域名放行、
+形似域名与后缀 trick 不放行、control plane 与购物/视频/邮箱
+域名不放行）。
+
+## 本轨补充：多标签页、导航白名单、关闭清理
+
+- 多标签页：`browser_new_page`（开新页并切为当前页）、`browser_list_pages`（未启动时不启动、返回空列表）、`browser_switch_page`（按 index 切页，不存在时报明确错误、当前页不变、旧页 refs 清空）、`browser_close_page`（缺省关当前页，关后自动切上一页；关最后一页报错“use browser_close”）。现有单页方法全部作用于当前页、行为不变。
+- 导航白名单：`BROWSER_ALLOWED_HOSTS`（逗号列表，前导点后缀匹配，大小写与端口不影响匹配）。未设置时行为不变；设置后在启动 Chromium 之前检查，未命中直接报错且不启动；`file://` 一律放行；`localhost/127.0.0.1` 无特殊待遇、须显式列出（本地开发服务器要自己加 localhost）。注意缺口：白名单只卡 `navigate` 入口，页内链接点击导致的跳转不经过该检查，只能靠 Tinyproxy 代理 403 兜底；接线：worker 在容器启动前把 worker 环境里的 `BROWSER_ALLOWED_HOSTS`（已设置且非空时）写进租户 config 的 `[mcp_servers.playwright.env]`，MCP 进程由此拿到它；未设置时不写该键、白名单不生效（见未验证一节）。
+- 关闭清理：`browser_close()` 关掉全部页并尽力删除 cwd 下的 `browser-live.png` 与 `browser-live.json`（删失败不报错）；自定义文件名的截图文件不在清理范围内。
+
+## 安全硬线
+
+- 任何真实密码、cookie、token 绝不写进代码、测试、文档、回执。
+- 浏览器只在任务沙箱与 egress 白名单内跑；白名单外域名一律
+  代理 403，如实报错。
+- 截图与下载落任务工作区，不写共享目录。
+
+## 未验证（诚实边界）
+
+- filter 未在主机 tinyproxy 上 reload（本轨不部署、不碰主机）。
+- 真实站点未 live 验证：上述 10 条域名在主机生效后的实际
+  可达性、以及浏览器在真实站点上的端到端回合，仍待部署后
+  在有 egress 的环境里实测；本轨的测试只验证 filter 文件本身
+  的语法与匹配语义，不冒充生产可用。