From 125b436ab3334fc4b204019f01de58f52466899e Mon Sep 17 00:00:00 2001 From: host Date: Sun, 1 Nov 2026 09:02:25 +0000 Subject: [PATCH] browser: click-path allowlist, safe tenant script upgrade, allowlist activation plan - click/click_ref: pre-click allowlist gate on resolvable anchor href (same check as navigate; miss refuses before the click executes); click/click_ref/click_at: post-click landing URL check. click_at has no pre-click gate by design (coordinates name no element); JS-driven navigation, evaluate, meta refresh and popups are documented as unintercepable with the Tinyproxy 403 backstop. - Tenant script upgrade is now safe: only proven-unmodified scripts (manifest match, 2a569ac known-shipped SHA-256, missing, or current) auto-merge; tenant-modified scripts are never overwritten, are reported as skipped, and get one deduped audit record. The manifest never learns a tenant hash (else the next upgrade would mistake a modified file for untouched). Idempotent reruns. - Activation candidate = deployed Tinyproxy programming set encoded with leading-dot entries (plain entries would regress docs.python.org); npmjs.org per the deployed filter, not npmjs.com. Host activation itself is NOT performed this round: only the plan, validation tests and rollback are delivered. Tests: 28 new (click 13, upgrade 8, activation 7), all fake/no-network; 98 browser-related tests pass including prior suites. (cherry picked from commit e4a93f5748ec815f163d0bc096ff4dc4b8c651f5) --- cloud/playwright_browser.py | 283 ++++++++++++++-- cloud/playwright_mcp.py | 18 +- cloud/skills/browser-automation/SKILL.md | 33 +- cloud/tests/test_playwright_activation.py | 209 ++++++++++++ .../tests/test_playwright_allowlist_click.py | 309 ++++++++++++++++++ cloud/tests/test_playwright_upgrade.py | 216 ++++++++++++ cloud/worker.py | 13 +- docs/platform-browser.md | 102 +++++- 8 files changed, 1150 insertions(+), 33 deletions(-) create mode 100644 cloud/tests/test_playwright_activation.py create mode 100644 cloud/tests/test_playwright_allowlist_click.py create mode 100644 cloud/tests/test_playwright_upgrade.py diff --git a/cloud/playwright_browser.py b/cloud/playwright_browser.py index 0078b7b..6c59582 100644 --- a/cloud/playwright_browser.py +++ b/cloud/playwright_browser.py @@ -103,6 +103,23 @@ def _check_allowed_url(url): "domain" % (host, ", ".join(allowed))) +def _check_navigated_url(url): + """Post-navigation / click-target check: same allowlist as + _check_allowed_url, but only for http(s) URLs. Anything else + is not a page navigation (file:// is always allowed anyway, + about:blank / data: / mailto: / javascript: never navigate + the page to a host), so it is skipped here instead of + raising a misleading non-http(s) error. Unset allowlist + means no check at all, same as _check_allowed_url.""" + if _allowed_hosts() is None: + return + from urllib.parse import urlparse + scheme = (urlparse(str(url)).scheme or "").lower() + if scheme not in ("http", "https"): + return + _check_allowed_url(url) + + class Browser: """Chromium pages (multi-tab). ``browser_factory`` is injectable: tests pass a fake returning a fake @@ -361,15 +378,84 @@ class Browser: result["truncated"] = truncated return result + def _click_href(self, selector): + """Best-effort anchor href for a click target, or None. + + Reads the first matching element's href (the same element + page.click would click). Returns the href resolved + against the current page URL, or None when the target has + no href, is not an anchor, or the locator API cannot + answer (fake pages, non-anchor selectors). A None here + does NOT mean allowed: it means the target is unknown + before the click, so only the post-click URL check can + catch it. Fragment-only hrefs resolve to the current + page and are returned as None for the same reason: they + never navigate to another host.""" + try: + locator = self._page.locator(str(selector)) + except Exception: # noqa: BLE001 - unknown target + return None + target = getattr(locator, "first", locator) + get_attr = getattr(target, "get_attribute", None) + if get_attr is None: + return None + try: + href = get_attr("href") + except Exception: # noqa: BLE001 - unknown target + return None + if not href or not isinstance(href, str): + return None + href = href.strip() + if not href or href.startswith("#"): + return None + from urllib.parse import urljoin + try: + return urljoin(self._current_url or "", href) + except Exception: # noqa: BLE001 - unknown target + return None + + def _check_click_target(self, selector): + """Pre-click allowlist gate for click/click_ref. + + When the allowlist is unset this is a no-op, so click + behaviour is byte-identical to before. When set and the + target href resolves to an http(s) URL, the same check + as navigate runs BEFORE the click: a miss raises and + the click is never executed. Unknown targets (no href, + button/JS handler) skip this gate and are caught, if at + all, by _check_navigated_current after the click.""" + if _allowed_hosts() is None: + return + target = self._click_href(selector) + if target is not None: + _check_navigated_url(target) + + def _check_navigated_current(self): + """Post-click allowlist detection (not prevention). + + The click already happened, so this can only report a + blocked landing host, never undo the navigation. It + exists because some click targets cannot be resolved + before the click (coordinates, JS handlers, forms): for + those, a raised error here is still better than silently + continuing on a blocked host, and the egress proxy 403 + is the layer that actually stops the page load when the + allowlist equals the proxy whitelist.""" + if _allowed_hosts() is None or not self._current_url: + return + _check_navigated_url(self._current_url) + def click(self, selector): if not selector or not str(selector).strip(): raise BrowserError("selector is required") self._ensure() + self._check_click_target(selector) try: self._page.click(str(selector), timeout=10000) self._current_url = self._page.url except Exception as exc: # noqa: BLE001 raise BrowserError("click failed: %s" % exc) + self._check_navigated_current() result = self._info() result["ok"] = True return result @@ -499,11 +585,16 @@ class Browser: "unknown ref %r: run browser_snapshot_refs first" % (ref,)) self._ensure() path = self._ref_paths[ref] + # The ref path is a CSS path, so the same href gate as + # click applies: an anchor ref to a blocked host is + # refused before the click, never executed. + self._check_click_target(path) try: self._page.locator(path).first.click(timeout=10000) self._current_url = self._page.url except Exception as exc: # noqa: BLE001 raise BrowserError("click_ref failed: %s" % exc) + self._check_navigated_current() result = self._info() result["ok"] = True result["ref"] = ref @@ -619,11 +710,21 @@ class Browser: raise BrowserError( "click_at out of viewport 1280x800: %s,%s" % (x, y)) self._ensure() + # Deliberately NO pre-click gate here: coordinates do not + # identify an element, so there is no href to check + # before the click. Probing elementFromPoint would be + # viewport/scroll-dependent, would miss JS-driven + # navigations (the common click_at case: buttons, canvas, + # SPA controls), and would run page JS on every user + # click. What click_at does get is the post-click URL + # check below, plus the egress proxy 403 backstop for + # any host outside the proxy whitelist. try: self._page.mouse.click(fx, fy) self._current_url = self._page.url except Exception as exc: # noqa: BLE001 raise BrowserError("click_at failed: %s" % exc) + self._check_navigated_current() result = self._info() result["ok"] = True result["x"] = fx @@ -725,49 +826,183 @@ class Browser: # ------------------------------------------------------- codex-home glue +PLAYWRIGHT_SCRIPT_NAMES = ("playwright_browser.py", "playwright_mcp.py", + "playwright_remote.py") + +# SHA-256 of the three MCP scripts as shipped at origin/main +# 2a569ac (PR #272), verified against the git blobs. A tenant +# script matching one of these hashes is PROVEN unmodified +# (it is exactly what a previous seeding wrote), so the safe +# upgrade below may replace it even when no manifest exists +# (legacy tenants seeded before manifests were written). +# A tenant script matching neither a manifest entry nor a hash +# here is treated as tenant-modified and is NEVER overwritten. +KNOWN_SHIPPED_SCRIPT_HASHES = { + "playwright_browser.py": + "0c36a286da3f8733fef4fec2dd18dec003fa68f54385fd4dc8da4b38e7571828", + "playwright_mcp.py": + "8455e67c2eb2ec6eb07bc9645292e2323b587e01d5a75fd8a985bbcd04a11198", + "playwright_remote.py": + "15a8026a02dfed6c630ea1614f75d54764a851df9d977c2380e15d0ed4f8a173", +} + +_MANIFEST_NAME = "playwright_scripts_manifest.json" +_AUDIT_NAME = "playwright_scripts_audit.jsonl" + + +def _sha256_bytes(data): + import hashlib + return hashlib.sha256(data).hexdigest() + + +def _read_script_manifest(home): + """name -> sha256 last written by seeding, or {}. + + The manifest is this helper's own metadata file in the + tenant codex-home (not tenant data): it records what the + last seeding wrote, so a later upgrade can tell an + untouched script from a tenant-modified one. Missing or + corrupt manifests read as empty -- the caller then falls + back to KNOWN_SHIPPED_SCRIPT_HASHES, never to overwriting + an unknown file.""" + try: + with open(os.path.join(home, _MANIFEST_NAME), + encoding="utf-8") as fh: + data = json.load(fh) + except Exception: # noqa: BLE001 - missing/corrupt = no proof + return {} + if not isinstance(data, dict): + return {} + return {k: v for k, v in data.items() + if isinstance(k, str) and isinstance(v, str)} + + +def _write_script_manifest(home, hashes): + """Best-effort manifest write; failures never block seeding.""" + try: + with open(os.path.join(home, _MANIFEST_NAME), "w", + encoding="utf-8") as fh: + json.dump({k: hashes[k] for k in sorted(hashes)}, + fh, ensure_ascii=False, sort_keys=True) + fh.write("\n") + except OSError: + pass + + +def _audit_script_skipped(home, name, dst_hash, src_hash): + """Append one audit record for a skipped modified script. + + Idempotent: a record with the same script + tenant hash is + never appended twice, so re-running an upgrade does not + grow the log. The audit file is append-only metadata; the + skipped script itself is left byte-identical.""" + path = os.path.join(home, _AUDIT_NAME) + try: + with open(path, encoding="utf-8") as fh: + for line in fh: + try: + rec = json.loads(line) + except ValueError: + continue + if (isinstance(rec, dict) + and rec.get("script") == name + and rec.get("dst_sha256") == dst_hash + and rec.get("action") == "skipped_modified"): + return path + except OSError: + pass + record = {"ts": int(time.time()), + "action": "skipped_modified", + "script": name, + "dst_sha256": dst_hash, + "src_sha256": src_hash, + "note": "tenant-modified script left untouched; " + "manual review required before upgrade"} + try: + with open(path, "a", encoding="utf-8") as fh: + fh.write(json.dumps(record, ensure_ascii=False) + "\n") + except OSError: + pass + return path + + def ensure_codex_home_playwright(codex_home_dir, command_args=None, script_dir=None): """Seed playwright MCP files + config into a tenant codex-home. - MCP scripts are SYSTEM files, not tenant data: a tenant never - edits them (user files live in the worktree, not codex-home), - so unlike the old never-overwrite rule, a differing script is - REFRESHED to the deployed version and reported in - ``script_refreshed``. Without this, every MCP feature ships - dead for existing projects: they keep the first-seeded script - forever and only brand-new projects get the update. - Appends ``[mcp_servers.playwright]`` only when absent. No - secrets live here -- proxy vars arrive via - ``ensure_playwright_mcp_env`` from worker proxy_env, and the - browsers path is a fixed image path. + SAFE UPGRADE RULE (changed from the old refresh-anything + rule): a tenant script is auto-merged to the deployed + version ONLY when it is proven unmodified -- its bytes + match the seeding manifest, or match a + KNOWN_SHIPPED_SCRIPT_HASHES entry (legacy tenants), or + the file is missing (fresh seed) or already current. A + script that differs AND matches neither proof is + tenant-modified: it is NEVER overwritten, it is reported + in ``script_skipped_modified`` (alias ``script_differs``), + and one deduped audit record is appended to + ``playwright_scripts_audit.jsonl`` for manual review. + The upgrade never deletes a tenant file and never rewrites + existing config semantics: ``[mcp_servers.playwright]`` + is appended only when absent, and the manifest/audit files + are this helper's own additive metadata. No secrets live + here -- proxy vars arrive via ``ensure_playwright_mcp_env`` + from worker proxy_env, and the browsers path is a fixed + image path. """ import shutil home = os.path.abspath(codex_home_dir) os.makedirs(home, exist_ok=True) here = os.path.abspath(script_dir or os.path.dirname(__file__)) + manifest = _read_script_manifest(home) script_changed = False script_differs = [] script_refreshed = [] - for name in ("playwright_browser.py", "playwright_mcp.py", - "playwright_remote.py"): + script_skipped = [] + audit_path = os.path.join(home, _AUDIT_NAME) + new_hashes = {} + for name in PLAYWRIGHT_SCRIPT_NAMES: src = os.path.join(here, name) dst = os.path.join(home, name) if not os.path.exists(src): continue + with open(src, "rb") as fh: + src_bytes = fh.read() + src_hash = _sha256_bytes(src_bytes) if not os.path.exists(dst): shutil.copyfile(src, dst) script_changed = True + new_hashes[name] = src_hash + continue + with open(dst, "rb") as fh: + dst_bytes = fh.read() + dst_hash = _sha256_bytes(dst_bytes) + if dst_bytes == src_bytes: + new_hashes[name] = src_hash + continue + proven_unmodified = ( + manifest.get(name) == dst_hash + or KNOWN_SHIPPED_SCRIPT_HASHES.get(name) == dst_hash) + if proven_unmodified: + shutil.copyfile(src, dst) + script_refreshed.append(name) + script_changed = True + new_hashes[name] = src_hash else: - with open(src, "rb") as fh: - src_bytes = fh.read() - with open(dst, "rb") as fh: - dst_bytes = fh.read() - if src_bytes != dst_bytes: - shutil.copyfile(src, dst) - script_refreshed.append(name) - script_changed = True - continue + script_differs.append(name) + script_skipped.append(name) + _audit_script_skipped(home, name, dst_hash, src_hash) + # NEVER record the tenant-modified hash as proven: + # writing dst_hash into the manifest would make the + # NEXT upgrade mistake this modified file for an + # untouched one and overwrite it. Keep any older + # proof entry (so a revert to that version can still + # auto-merge later); otherwise drop the entry, which + # fails closed to skip again. + if name in manifest: + new_hashes[name] = manifest[name] + if new_hashes: + _write_script_manifest(home, new_hashes) config_path = os.path.join(home, "config.toml") if command_args is None: command_args = ["python3", @@ -793,7 +1028,9 @@ def ensure_codex_home_playwright(codex_home_dir, command_args=None, "config_changed": config_changed, "script_changed": script_changed, "script_differs": script_differs, - "script_refreshed": script_refreshed} + "script_refreshed": script_refreshed, + "script_skipped_modified": script_skipped, + "audit_path": audit_path} def ensure_playwright_mcp_env(codex_home_dir, env): diff --git a/cloud/playwright_mcp.py b/cloud/playwright_mcp.py index c476cad..bf9d41e 100644 --- a/cloud/playwright_mcp.py +++ b/cloud/playwright_mcp.py @@ -72,7 +72,12 @@ _TOOLS = [ "description": ( "Click an element by CSS selector or Playwright " "selector (e.g. 'button.submit', 'text=Next'). " - "Then snapshot to see the result." + "Then snapshot to see the result. When " + "BROWSER_ALLOWED_HOSTS is set, an anchor click to " + "a non-listed host is refused BEFORE the click; a " + "click whose target cannot be resolved beforehand " + "(button, JS handler) is reported after it lands " + "on a blocked host instead." ), "inputSchema": { "type": "object", @@ -120,7 +125,11 @@ _TOOLS = [ "top-left of the 1280x800 viewport). This is the tool " "a live viewer drives: the user clicks on a screenshot " "and the coordinates arrive here. Then snapshot to see " - "the result." + "the result. Coordinates identify no element, so " + "there is no pre-click allowlist gate here: when " + "BROWSER_ALLOWED_HOSTS is set, a landing on a " + "blocked host is reported after the click, and the " + "egress proxy 403 is the backstop that stops the load." ), "inputSchema": { "type": "object", @@ -165,7 +174,10 @@ _TOOLS = [ "name": "browser_click_ref", "description": ( "Click an element by its ref from browser_snapshot_refs. " - "More robust than a CSS selector across re-renders." + "More robust than a CSS selector across re-renders. " + "Same allowlist rule as browser_click: an anchor ref " + "to a non-listed host is refused before the click " + "when BROWSER_ALLOWED_HOSTS is set." ), "inputSchema": { "type": "object", diff --git a/cloud/skills/browser-automation/SKILL.md b/cloud/skills/browser-automation/SKILL.md index 418780d..8fbc339 100644 --- a/cloud/skills/browser-automation/SKILL.md +++ b/cloud/skills/browser-automation/SKILL.md @@ -82,9 +82,36 @@ navigate / snapshot / click / fill / screenshot 都针对该页； 就检查：`file://` 一律放行，`http(s)` 的 host 必须命中列表 （前导点条目做后缀匹配：`.python.org` 同时放行 `python.org` 与 `docs.python.org`，不放行 -`notpython.org`），大小写与端口不影响匹配，未命中直接 -报错且不启动 Chromium。`localhost` / `127.0.0.1` 没有 -特殊待遇：本地开发服务器要自己把它列进变量，否则会被挡。 +`notpython.org`；注意：不带前导点的 `python.org` 只精确 +匹配自身，不放行 `docs.python.org`），大小写与端口不影响 +匹配，未命中直接报错且不启动 Chromium。`localhost` / +`127.0.0.1` 没有特殊待遇：本地开发服务器要自己把它列进 +变量，否则会被挡。 + +页内点击白名单（本轮补齐，与 navigate 同款检查）： +`browser_click` / `browser_click_ref` 点到带 href 的链接时， +先解析 href（相对链接按当前页 URL 展开）再检查：未命中 +直接报错、点击根本不执行；点到没有 href 的目标（按钮、 +JS 处理函数）时事前无从得知目的地，只能在点击后检查落点 +URL、命中 blocked host 时报错（此时点击已发生，只是事后 +发现，不是事前拦截）。`browser_click_at` 是坐标点击，坐标 +本身对应不了任何元素，所以没有事前检查，只有事后落点检查。 +真正拦不住的路径（不要假装已拦）：页内 JS 自行跳转 +（`location.href`、表单提交、SPA 路由、回车提交）、 +`browser_evaluate` 里跑的任意 JS、meta refresh、弹窗新页。 +这些路径没有“点击前已知目的地”可供检查，只能靠 egress +代理 403 兜底——白名单与代理白名单取同一域名集合时，被挡 +域名在代理层就加载失败，不会真的打开目标页；但 evaluate +与 shell 同权，白名单从来不是它的沙箱边界，不要把它当 +成防绕过手段。 + +老租户脚本升级（本轮改动）：播种器只对已证明未改动的 +租户脚本自动合并新版（播种 manifest 命中，或命中 +2a569ac 已发布版本的哈希，或文件缺失/已是最新）；租户 +改过的脚本绝不覆盖，原样不动、记一条审计 +（租户 codex-home 里的 `playwright_scripts_audit.jsonl`）， +留待人工处理——下次任务播种时它仍会被跳过，不会悄悄 +被新版盖掉。 清理：`browser_close` 会关掉全部页，并尽力删除工作区里的 `browser-live.png` 与 `browser-live.json`（删不掉也不报错）， diff --git a/cloud/tests/test_playwright_activation.py b/cloud/tests/test_playwright_activation.py new file mode 100644 index 0000000..4cee623 --- /dev/null +++ b/cloud/tests/test_playwright_activation.py @@ -0,0 +1,209 @@ +"""BROWSER_ALLOWED_HOSTS activation-candidate validation. + +The candidate value is the ALREADY-DEPLOYED Tinyproxy +programming-domain set (deploy/egress/tinyproxy-filter) plus the +already-present github.com / pypi.org / npmjs.org -- no invented +domains. NOTE: the task text said "npmjs.com", but the deployed +filter (tinyproxy-filter line 23, new-host-checklist.md, and +docs/platform-browser.md) all say npmjs.org, so the candidate +follows the DEPLOYED set (npmjs.org); npmjs.com is deliberately +NOT in the candidate -- adding it would widen the allowlist +beyond the proxy, breaking allowlist==proxy parity. If the npm +website itself must be browsable, that is a separate filter +change for the parent to decide, not something this round +invents. Because the Tinyproxy lines are subdomain-inclusive +(``^\\(.*\\.\\)*domain$``), the env value uses leading-dot +entries, which are the allowlist encoding of the same +subdomain-inclusive set; plain entries would NOT match +subdomains (proven by test_plain_entries_would_break_subdomains) +and would therefore regress docs.python.org etc. + +All checks run against fake browsers / the filter file itself: +no production tenant, no network, no real credentials. +""" + +import os +import re +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from playwright_browser import Browser, BrowserError # noqa: E402 +from playwright_browser import _check_allowed_url # noqa: E402 + +# The activation candidate: the deployed Tinyproxy programming +# set, encoded subdomain-inclusively. Domain SET is fixed by the +# task (no new domains); only the encoding (dot prefix) follows +# from the filter's own subdomain rule. +CANDIDATE_DOMAINS = [ + "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", + "github.com", "pypi.org", "npmjs.org", +] +CANDIDATE = ",".join("." + d for d in CANDIDATE_DOMAINS) + +REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), + "..", "..")) +FILTER_PATH = os.path.join(REPO_ROOT, "deploy", "egress", + "tinyproxy-filter") + + +class _FakePage: + def __init__(self): + self.url = "" + self.calls = [] + self.mouse = self + + def title(self): + return "T" + + def goto(self, url, timeout=None, wait_until=None): + self.calls.append(("goto", url)) + self.url = url + + def locator(self, selector): + raise AssertionError("no locator use in activation tests") + + def click(self, selector, timeout=None): + self.calls.append(("click", selector)) + + def close(self): + pass + + +class _FakeHandle: + def __init__(self, page): + self._page = page + + def new_page(self, viewport=None): + return self._page + + def close(self): + pass + + +class _FakePW: + def __init__(self, page): + self._page = page + self.launch_count = 0 + + def start(self): + return self + + @property + def chromium(self): + outer = self + + class _Chromium: + def launch(self, **kwargs): + outer.launch_count += 1 + return _FakeHandle(outer._page) + + return _Chromium() + + def stop(self): + pass + + +class _EnvGuard: + def setUp(self): + self._old = os.environ.get("BROWSER_ALLOWED_HOSTS") + + def tearDown(self): + if self._old is None: + os.environ.pop("BROWSER_ALLOWED_HOSTS", None) + else: + os.environ["BROWSER_ALLOWED_HOSTS"] = self._old + + +def _filter_domains(): + """Domains covered by subdomain-inclusive filter lines.""" + with open(FILTER_PATH, encoding="utf-8") as fh: + lines = fh.read().splitlines() + out = set() + for line in lines: + m = re.match(r"^\^\\\(\.\*\\.\\\)\*(.+)\$$", line) + if m: + out.add(m.group(1).replace("\\.", ".")) + return out + + +class ActivationCandidateTest(_EnvGuard, unittest.TestCase): + def setUp(self): + _EnvGuard.setUp(self) + os.environ["BROWSER_ALLOWED_HOSTS"] = CANDIDATE + self._tmp = tempfile.TemporaryDirectory() + self.page = _FakePage() + self.factory = _FakePW(self.page) + 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_candidate_matches_deployed_filter_set(self): + # Every candidate domain has a subdomain-inclusive line + # in the deployed filter; the candidate invents nothing. + filt = _filter_domains() + for domain in CANDIDATE_DOMAINS: + self.assertIn(domain, filt, domain) + + def test_candidate_allows_bare_and_subdomains(self): + for domain in CANDIDATE_DOMAINS: + _check_allowed_url("https://%s/" % domain) + _check_allowed_url("https://www.%s/docs" % domain) + # The cases the filter explicitly exists for. + _check_allowed_url("https://docs.python.org/3/") + _check_allowed_url("https://docs.rust-lang.org/") + _check_allowed_url("https://api.github.com/") + _check_allowed_url("https://pypi.org/project/x/") + + def test_candidate_blocks_suffix_tricks_and_outsiders(self): + for bad in ("https://notpython.org/", + "https://python.org.evil.com/", + "https://evilgithub.com/", + "https://api.diante.me/", + "https://www.amazon.com/", + "https://mail.google.com/"): + with self.assertRaises(BrowserError, msg=bad): + _check_allowed_url(bad) + + def test_plain_entries_would_break_subdomains(self): + # Why the candidate uses dot forms: plain "python.org" + # does not match docs.python.org under the established + # navigate semantics, so a plain-entry activation would + # regress exactly the docs the filter was added for. + os.environ["BROWSER_ALLOWED_HOSTS"] = ( + "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,github.com,pypi.org,npmjs.org") + with self.assertRaises(BrowserError): + _check_allowed_url("https://docs.python.org/3/") + + def test_file_allowed_localhost_blocked_under_candidate(self): + out = self.browser.navigate("file:///tmp/page.html") + self.assertTrue(out["ok"]) + with self.assertRaises(BrowserError): + self.browser.navigate("http://localhost:8000/") + # localhost blocking is a behaviour CHANGE vs unset: + # local dev servers need an explicit localhost entry. + self.assertEqual(self.factory.launch_count, 1) + + def test_task_level_navigate_allow_and_block(self): + out = self.browser.navigate("https://docs.python.org/3/") + self.assertTrue(out["ok"]) + launches = self.factory.launch_count + with self.assertRaises(BrowserError): + self.browser.navigate("https://evil.com/") + self.assertEqual(self.factory.launch_count, launches) + + def test_unset_means_no_change(self): + os.environ.pop("BROWSER_ALLOWED_HOSTS", None) + out = self.browser.navigate("https://anything.example/") + self.assertTrue(out["ok"]) + _check_allowed_url("https://anything.example/") diff --git a/cloud/tests/test_playwright_allowlist_click.py b/cloud/tests/test_playwright_allowlist_click.py new file mode 100644 index 0000000..9350eb0 --- /dev/null +++ b/cloud/tests/test_playwright_allowlist_click.py @@ -0,0 +1,309 @@ +"""Click-path allowlist tests for the Playwright browser (fake only). + +Covers the continuation of the navigate-only allowlist: click / +click_ref get the SAME check before the click when the target +anchor href is resolvable, and all three click paths get a +post-click URL check. click_at has no pre-click gate by design +(coordinates identify no element). No real Chromium, no network, +no real credentials -- values here are test placeholders. +""" + +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from playwright_browser import Browser, BrowserError # noqa: E402 +from playwright_browser import _check_navigated_url # noqa: E402 + + +class _FakeLocator: + def __init__(self, page, href): + self._page = page + self._href = href + + @property + def first(self): + return self + + def click(self, timeout=None): + self._page.calls.append(("locator_click",)) + if self._page.click_landing is not None: + self._page.url = self._page.click_landing + + def get_attribute(self, name): + if name == "href": + return self._href + return None + + def inner_text(self, timeout=None): + return "text" + + +class _FakeMouse: + def __init__(self, page): + self._page = page + + def click(self, x, y): + self._page.calls.append(("mouse_click", x, y)) + if self._page.click_landing is not None: + self._page.url = self._page.click_landing + + def move(self, x, y): + self._page.calls.append(("mouse_move", x, y)) + + +class _FakePage: + def __init__(self): + self.url = "https://example.com/" + self.calls = [] + self.href_map = {} + self.click_landing = None + self.mouse = _FakeMouse(self) + + def title(self): + return "Example" + + def goto(self, url, timeout=None, wait_until=None): + self.calls.append(("goto", url)) + self.url = url + + def locator(self, selector): + self.calls.append(("locator", selector)) + return _FakeLocator(self, self.href_map.get(selector)) + + def click(self, selector, timeout=None): + self.calls.append(("click", selector)) + if self.click_landing is not None: + self.url = self.click_landing + + def fill(self, selector, text, timeout=None): + self.calls.append(("fill", selector, text)) + + def evaluate(self, expression): + self.calls.append(("evaluate", expression)) + return None + + def screenshot(self, path=None): + with open(path, "wb") as fh: + fh.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 10) + + def close(self): + pass + + +class _FakeHandle: + def __init__(self, page): + self._page = page + + def new_page(self, viewport=None): + return self._page + + def close(self): + pass + + +class _FakePW: + def __init__(self, page): + self._page = page + self.launch_count = 0 + + def start(self): + return self + + @property + def chromium(self): + outer = self + + class _Chromium: + def launch(self, **kwargs): + outer.launch_count += 1 + return _FakeHandle(outer._page) + + return _Chromium() + + def stop(self): + pass + + +class _EnvGuard: + def setUp(self): + self._old = os.environ.get("BROWSER_ALLOWED_HOSTS") + + def tearDown(self): + if self._old is None: + os.environ.pop("BROWSER_ALLOWED_HOSTS", None) + else: + os.environ["BROWSER_ALLOWED_HOSTS"] = self._old + + +class ClickAllowlistTest(_EnvGuard, unittest.TestCase): + def setUp(self): + _EnvGuard.setUp(self) + self._tmp = tempfile.TemporaryDirectory() + self.page = _FakePage() + self.factory = _FakePW(self.page) + 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 _start(self, url="https://example.com/"): + return self.browser.navigate(url) + + def test_click_allowed_href_executes(self): + os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com, .python.org" + self._start() + self.page.href_map["a.docs"] = "https://docs.python.org/3/" + self.page.click_landing = "https://docs.python.org/3/" + out = self.browser.click("a.docs") + self.assertTrue(out["ok"]) + self.assertIn(("click", "a.docs"), self.page.calls) + self.assertEqual(out["url"], "https://docs.python.org/3/") + + def test_click_blocked_href_refused_before_click(self): + os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com" + self._start() + self.page.href_map["a.evil"] = "https://evil.com/x" + self.page.click_landing = "https://evil.com/x" + with self.assertRaises(BrowserError) as ctx: + self.browser.click("a.evil") + self.assertIn("evil.com", str(ctx.exception)) + self.assertNotIn(("click", "a.evil"), self.page.calls) + self.assertEqual(self.browser._current_url, + "https://example.com/") + + def test_click_suffix_trick_blocked(self): + os.environ["BROWSER_ALLOWED_HOSTS"] = ".python.org" + self._start("https://python.org/") + for bad in ("https://notpython.org/", + "https://python.org.evil.com/"): + self.page.href_map["a.bad"] = bad + with self.assertRaises(BrowserError): + self.browser.click("a.bad") + self.assertNotIn(("click", "a.bad"), self.page.calls) + self.page.calls.clear() + + def test_click_plain_entry_is_exact_only(self): + # Same semantics as navigate: a plain entry does NOT + # match subdomains (that is what leading-dot entries + # are for). Documented, not a bug: the activation + # candidate therefore uses dot forms. + os.environ["BROWSER_ALLOWED_HOSTS"] = "python.org" + self._start("https://python.org/") + self.page.href_map["a.docs"] = "https://docs.python.org/" + with self.assertRaises(BrowserError): + self.browser.click("a.docs") + self.assertNotIn(("click", "a.docs"), self.page.calls) + + def test_click_port_and_case_insensitive(self): + os.environ["BROWSER_ALLOWED_HOSTS"] = "Example.COM" + self._start() + self.page.href_map["a.p"] = "https://EXAMPLE.com:8443/p" + self.page.click_landing = "https://EXAMPLE.com:8443/p" + out = self.browser.click("a.p") + self.assertTrue(out["ok"]) + + def test_click_file_href_allowed(self): + os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com" + self._start() + self.page.href_map["a.file"] = "file:///tmp/page.html" + self.page.click_landing = "file:///tmp/page.html" + out = self.browser.click("a.file") + self.assertTrue(out["ok"]) + + def test_click_localhost_needs_explicit_listing(self): + os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com" + self._start() + self.page.href_map["a.local"] = "http://localhost:8000/" + with self.assertRaises(BrowserError): + self.browser.click("a.local") + self.assertNotIn(("click", "a.local"), self.page.calls) + os.environ["BROWSER_ALLOWED_HOSTS"] = "localhost" + self.page.click_landing = "http://localhost:8000/" + out = self.browser.click("a.local") + self.assertTrue(out["ok"]) + + def test_click_relative_href_resolves_against_current(self): + os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com" + self._start("https://example.com/dir/") + self.page.href_map["a.rel"] = "/other" + self.page.click_landing = "https://example.com/other" + out = self.browser.click("a.rel") + self.assertTrue(out["ok"]) + # Fragment-only hrefs never navigate to another host. + self.page.href_map["a.frag"] = "#section" + self.page.click_landing = None + out = self.browser.click("a.frag") + self.assertTrue(out["ok"]) + + def test_click_unknown_target_post_click_detection(self): + # Button / JS handler: no href, so no pre-click gate. + # The click DOES execute; a blocked landing is then + # reported (detection, not prevention -- the proxy 403 + # is the layer that stops the load in production). + os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com" + self._start() + self.page.href_map["button.go"] = None + self.page.click_landing = "https://evil.com/" + with self.assertRaises(BrowserError) as ctx: + self.browser.click("button.go") + self.assertIn("evil.com", str(ctx.exception)) + self.assertIn(("click", "button.go"), self.page.calls) + # Same target landing on an allowed host is fine. + self.page.click_landing = "https://example.com/ok" + out = self.browser.click("button.go") + self.assertTrue(out["ok"]) + + def test_click_unset_allowlist_unchanged(self): + os.environ.pop("BROWSER_ALLOWED_HOSTS", None) + self._start("https://anything.example/") + self.page.href_map["a.evil"] = "https://evil.com/" + self.page.click_landing = "https://evil.com/" + out = self.browser.click("a.evil") + self.assertTrue(out["ok"]) + self.assertIn(("click", "a.evil"), self.page.calls) + + def test_click_ref_blocked_href_refused_before_click(self): + os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com" + self._start() + self.browser._ref_paths = {"e0": "a.evil", "e1": "a.docs"} + self.page.href_map["a.evil"] = "https://evil.com/" + self.page.href_map["a.docs"] = "https://example.com/docs" + self.page.click_landing = "https://evil.com/" + with self.assertRaises(BrowserError): + self.browser.click_ref("e0") + self.assertNotIn(("locator_click",), self.page.calls) + self.page.click_landing = "https://example.com/docs" + out = self.browser.click_ref("e1") + self.assertTrue(out["ok"]) + self.assertIn(("locator_click",), self.page.calls) + + def test_click_at_post_click_only(self): + # click_at has no pre-click gate by design; a blocked + # landing raises after the click, an allowed one passes. + os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com" + self._start() + self.page.click_landing = "https://evil.com/" + with self.assertRaises(BrowserError): + self.browser.click_at(100, 200) + self.assertIn(("mouse_click", 100.0, 200.0), self.page.calls) + self.page.click_landing = "https://example.com/" + out = self.browser.click_at(100, 200) + self.assertTrue(out["ok"]) + + def test_check_navigated_url_skips_non_page_schemes(self): + os.environ["BROWSER_ALLOWED_HOSTS"] = "example.com" + # about:blank / mailto: / javascript: never navigate + # the page to a host, so they must not raise here. + _check_navigated_url("about:blank") + _check_navigated_url("mailto:someone@example.com") + _check_navigated_url("javascript:void(0)") + _check_navigated_url("file:///tmp/page.html") + with self.assertRaises(BrowserError): + _check_navigated_url("https://evil.com/") diff --git a/cloud/tests/test_playwright_upgrade.py b/cloud/tests/test_playwright_upgrade.py new file mode 100644 index 0000000..f5ad648 --- /dev/null +++ b/cloud/tests/test_playwright_upgrade.py @@ -0,0 +1,216 @@ +"""Safe tenant-script upgrade tests (no Docker, no network). + +The rule under test: only a PROVEN-unmodified tenant script +(manifest match, known-shipped-hash match, missing, or already +current) is auto-merged. A tenant-modified script is NEVER +overwritten: it stays byte-identical, is reported as skipped, +and gets exactly one deduped audit record. Reruns are idempotent. +No real credentials appear here. +""" + +import hashlib +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +import playwright_browser as pb # noqa: E402 +from playwright_browser import ensure_codex_home_playwright # noqa: E402 + +NAMES = pb.PLAYWRIGHT_SCRIPT_NAMES + + +def _write_scripts(d, contents): + for name in NAMES: + with open(os.path.join(d, name), "w", encoding="utf-8") as fh: + fh.write(contents[name]) + + +def _read_scripts(d): + out = {} + for name in NAMES: + with open(os.path.join(d, name), encoding="utf-8") as fh: + out[name] = fh.read() + return out + + +def _v1(): + return {name: "# v1 %s\nprint('v1')\n" % name for name in NAMES} + + +def _v2(): + return {name: "# v2 %s\nprint('v2')\n" % name for name in NAMES} + + +class UpgradeTest(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.src_v1 = os.path.join(self._tmp.name, "src-v1") + self.src_v2 = os.path.join(self._tmp.name, "src-v2") + self.home = os.path.join(self._tmp.name, "home") + os.makedirs(self.src_v1) + os.makedirs(self.src_v2) + os.makedirs(self.home) + _write_scripts(self.src_v1, _v1()) + _write_scripts(self.src_v2, _v2()) + + def tearDown(self): + self._tmp.cleanup() + + def _audit_records(self): + path = os.path.join(self.home, "playwright_scripts_audit.jsonl") + if not os.path.exists(path): + return [] + with open(path, encoding="utf-8") as fh: + return [json.loads(line) for line in fh if line.strip()] + + def test_fresh_seed_and_idempotent_rerun(self): + out = ensure_codex_home_playwright(self.home, + script_dir=self.src_v2) + self.assertEqual(_read_scripts(self.home), _v2()) + self.assertFalse(out["script_skipped_modified"]) + self.assertEqual(self._audit_records(), []) + # Config section appended exactly once. + with open(os.path.join(self.home, "config.toml"), + encoding="utf-8") as fh: + config = fh.read() + self.assertEqual(config.count("[mcp_servers.playwright]"), 1) + out2 = ensure_codex_home_playwright(self.home, + script_dir=self.src_v2) + self.assertFalse(out2["script_changed"]) + self.assertFalse(out2["config_changed"]) + self.assertFalse(out2["script_refreshed"]) + self.assertEqual(_read_scripts(self.home), _v2()) + with open(os.path.join(self.home, "config.toml"), + encoding="utf-8") as fh: + self.assertEqual( + fh.read().count("[mcp_servers.playwright]"), 1) + + def test_unmodified_script_auto_merges(self): + ensure_codex_home_playwright(self.home, script_dir=self.src_v1) + out = ensure_codex_home_playwright(self.home, + script_dir=self.src_v2) + self.assertEqual(set(out["script_refreshed"]), set(NAMES)) + self.assertFalse(out["script_skipped_modified"]) + self.assertEqual(_read_scripts(self.home), _v2()) + manifest_path = os.path.join( + self.home, "playwright_scripts_manifest.json") + with open(manifest_path, encoding="utf-8") as fh: + manifest = json.load(fh) + for name in NAMES: + self.assertEqual( + manifest[name], + hashlib.sha256(_v2()[name].encode()).hexdigest()) + + def test_modified_script_untouched_with_audit(self): + ensure_codex_home_playwright(self.home, script_dir=self.src_v1) + tenant_path = os.path.join(self.home, NAMES[0]) + with open(tenant_path, "a", encoding="utf-8") as fh: + fh.write("# tenant tweak: custom timeout\n") + with open(tenant_path, encoding="utf-8") as fh: + tenant_bytes = fh.read() + out = ensure_codex_home_playwright(self.home, + script_dir=self.src_v2) + self.assertEqual(out["script_skipped_modified"], [NAMES[0]]) + self.assertEqual(out["script_differs"], [NAMES[0]]) + # Modified file byte-identical; the other two auto-merged. + with open(tenant_path, encoding="utf-8") as fh: + self.assertEqual(fh.read(), tenant_bytes) + rest = _read_scripts(self.home) + for name in NAMES[1:]: + self.assertEqual(rest[name], _v2()[name]) + records = self._audit_records() + self.assertEqual(len(records), 1) + self.assertEqual(records[0]["action"], "skipped_modified") + self.assertEqual(records[0]["script"], NAMES[0]) + self.assertEqual( + records[0]["dst_sha256"], + hashlib.sha256(tenant_bytes.encode()).hexdigest()) + + def test_modified_rerun_does_not_rewrite_or_duplicate_audit(self): + ensure_codex_home_playwright(self.home, script_dir=self.src_v1) + tenant_path = os.path.join(self.home, NAMES[0]) + with open(tenant_path, "a", encoding="utf-8") as fh: + fh.write("# tenant tweak\n") + with open(tenant_path, encoding="utf-8") as fh: + tenant_bytes = fh.read() + ensure_codex_home_playwright(self.home, script_dir=self.src_v2) + out = ensure_codex_home_playwright(self.home, + script_dir=self.src_v2) + self.assertFalse(out["script_changed"]) + self.assertEqual(out["script_skipped_modified"], [NAMES[0]]) + self.assertEqual(len(self._audit_records()), 1) + with open(tenant_path, encoding="utf-8") as fh: + self.assertEqual(fh.read(), tenant_bytes) + # The manifest must NOT have learned the tenant hash: + # otherwise the next upgrade would mistake the modified + # file for untouched and overwrite it. + with open(os.path.join( + self.home, "playwright_scripts_manifest.json"), + encoding="utf-8") as fh: + manifest = json.load(fh) + self.assertNotEqual( + manifest.get(NAMES[0]), + hashlib.sha256(tenant_bytes.encode()).hexdigest()) + + def test_legacy_no_manifest_known_hash_merges(self): + # Legacy tenant: files seeded before manifests existed, + # content exactly a known shipped version -> proven + # unmodified, safe to merge even with no manifest. + old = {name: "# legacy %s\n" % name for name in NAMES} + _write_scripts(self.home, old) + patched = {name: hashlib.sha256(old[name].encode()).hexdigest() + for name in NAMES} + saved = pb.KNOWN_SHIPPED_SCRIPT_HASHES + pb.KNOWN_SHIPPED_SCRIPT_HASHES = patched + try: + out = ensure_codex_home_playwright(self.home, + script_dir=self.src_v2) + finally: + pb.KNOWN_SHIPPED_SCRIPT_HASHES = saved + self.assertEqual(set(out["script_refreshed"]), set(NAMES)) + self.assertEqual(_read_scripts(self.home), _v2()) + + def test_legacy_no_manifest_unknown_hash_skips(self): + old = {name: "# tenant-custom %s\n" % name for name in NAMES} + _write_scripts(self.home, old) + saved = pb.KNOWN_SHIPPED_SCRIPT_HASHES + pb.KNOWN_SHIPPED_SCRIPT_HASHES = {} + try: + out = ensure_codex_home_playwright(self.home, + script_dir=self.src_v2) + finally: + pb.KNOWN_SHIPPED_SCRIPT_HASHES = saved + self.assertEqual(set(out["script_skipped_modified"]), set(NAMES)) + self.assertEqual(_read_scripts(self.home), old) + self.assertEqual(len(self._audit_records()), 3) + + def test_upgrade_never_touches_tenant_data_or_config(self): + notes = os.path.join(self.home, "notes.txt") + with open(notes, "w", encoding="utf-8") as fh: + fh.write("tenant notes, must survive\n") + config = os.path.join(self.home, "config.toml") + with open(config, "w", encoding="utf-8") as fh: + fh.write('[user]\nname = "tenant"\n\n' + '[mcp_servers.playwright.env]\n' + 'FOO = "bar"\n') + ensure_codex_home_playwright(self.home, script_dir=self.src_v2) + ensure_codex_home_playwright(self.home, script_dir=self.src_v2) + with open(notes, encoding="utf-8") as fh: + self.assertEqual(fh.read(), "tenant notes, must survive\n") + with open(config, encoding="utf-8") as fh: + text = fh.read() + # Existing sections intact, playwright section not + # duplicated (it was absent, so appended exactly once). + self.assertIn('[user]\nname = "tenant"', text) + self.assertIn('FOO = "bar"', text) + self.assertEqual(text.count("[mcp_servers.playwright]"), 1) + + def test_known_shipped_hashes_are_well_formed(self): + self.assertEqual(set(pb.KNOWN_SHIPPED_SCRIPT_HASHES), set(NAMES)) + for value in pb.KNOWN_SHIPPED_SCRIPT_HASHES.values(): + self.assertEqual(len(value), 64) + int(value, 16) # must be hex diff --git a/cloud/worker.py b/cloud/worker.py index bd4fcae..805f865 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -2765,9 +2765,10 @@ class Worker: # 与 imagegen 同款：播种失败直接抛错、任务不派发 # ——静默无浏览器比 loud 失败更糟，尤其对已宣传 # “能开浏览器”的任务。代理与浏览器路径由 - # _prepare_tenant 的 env 块给。MCP 脚本是系统文件： - # 播种器在版本变化时刷新老租户的旧脚本（见 - # ensure_codex_home_playwright），刷新记录打日志。 + # _prepare_tenant 的 env 块给。播种器只对已证明 + # 未改动的租户脚本自动合并新版（见 + # ensure_codex_home_playwright）；租户改过的脚本 + # 绝不覆盖，只记审计并打 warning 留待人工处理。 _seeded = _ensure_playwright_mcp( dirs["codex_home"], command_args=[ @@ -2779,6 +2780,12 @@ class Worker: "playwright MCP scripts refreshed for project %s: %s", task.get("project_id"), ",".join(_seeded["script_refreshed"])) + if _seeded.get("script_skipped_modified"): + self.logger.warning( + "playwright MCP scripts skipped (tenant-modified, " + "left untouched, see audit) for project %s: %s", + task.get("project_id"), + ",".join(_seeded["script_skipped_modified"])) if self.seed_skills: # 技能播种失败绝不阻塞任务：技能是可选的流程 # 说明，不是 websearch 那种“已宣传就必须有”的 diff --git a/docs/platform-browser.md b/docs/platform-browser.md index 783fd23..458710a 100644 --- a/docs/platform-browser.md +++ b/docs/platform-browser.md @@ -82,9 +82,109 @@ ## 本轨补充：多标签页、导航白名单、关闭清理 - 多标签页：`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_ALLOWED_HOSTS`（逗号列表，前导点后缀匹配，大小写与端口不影响匹配）。未设置时行为不变；设置后在启动 Chromium 之前检查，未命中直接报错且不启动；`file://` 一律放行；`localhost/127.0.0.1` 无特殊待遇、须显式列出（本地开发服务器要自己加 localhost）。接线：worker 在容器启动前把 worker 环境里的 `BROWSER_ALLOWED_HOSTS`（已设置且非空时）写进租户 config 的 `[mcp_servers.playwright.env]`，MCP 进程由此拿到它；未设置时不写该键、白名单不生效（见未验证一节）。 +- 页内跳转白名单（本轮补齐，上一轮的缺口已收窄、未归零）：`browser_click` / `browser_click_ref` 在目标是带 href 的链接时先解析 href（相对链接按当前页展开）再做同款检查，未命中报错且点击不执行；`click` / `click_ref` / `click_at` 在点击后都会检查落点 URL，落到 blocked host 时报错。已拦住的：navigate、带 href 的 click / click_ref（事前）。确实拦不住的（经代码确认，不硬编检查）：`click_at` 的事前检查（坐标对应不了元素，elementFromPoint 探针受视口/滚动影响、且漏掉 JS 驱动的跳转，故只做事后检查）；无 href 目标的点击（按钮、JS 处理函数，事前无目的地）；页内 JS 自行跳转（`location.href`、表单提交、SPA 路由、回车提交）、`browser_evaluate` 内的任意 JS、meta refresh、弹窗新页（`page.click` 不跟随新页）。这些路径的兜底只能是 Tinyproxy 代理 403：当白名单与代理白名单取同一域名集合时，被挡域名在代理层就加载失败；但 evaluate 与 shell 同权，白名单不是它的沙箱边界。测试：`cloud/tests/test_playwright_allowlist_click.py`（13 项，fake browser，不联网）：命中/未命中/后缀 trick（`notpython.org`、`python.org.evil.com`）/端口/大小写/file/localhost 正反例、事前拒点击不执行、事后落点报错、未设置时行为不变。 - 关闭清理：`browser_close()` 关掉全部页并尽力删除 cwd 下的 `browser-live.png` 与 `browser-live.json`（删失败不报错）；自定义文件名的截图文件不在清理范围内。 +## 本轮补齐：老租户脚本安全升级 + +上一轮的播种器是“版本变化就刷新一切不同脚本”；本轮改为 +安全升级（`cloud/playwright_browser.py` 的 +`ensure_codex_home_playwright`，worker 播种时调用）： + +- 自动合并（只对已证明未改动者）：播种 manifest + （租户 codex-home 里的 `playwright_scripts_manifest.json`， + 播种器自己的元数据，不是租户数据）命中、或命中 2a569ac + 已发布三脚本的 SHA-256（`KNOWN_SHIPPED_SCRIPT_HASHES`， + 已对 origin/main blob 验过哈希）、或文件缺失（全新播种）、 + 或已是最新（不写）。 +- 绝不覆盖：内容已改动、且两项证明都不命中的脚本，原样 + 不动、列入返回值的 `script_skipped_modified`（兼容旧字段 + `script_differs`）、在租户 codex-home 的 + `playwright_scripts_audit.jsonl` 里记一条 + `skipped_modified` 审计（脚本名、租户哈希、新版哈希）， + worker 同时打 warning 日志；留待人工处理。 +- 幂等与防误判：重跑不重复写（已是最新不写、审计按 + 脚本+租户哈希去重）；跳过时 manifest 绝不记录租户哈希 + ——否则下次升级会把改动过的脚本误判为“未改动”而覆盖 + （本轮修掉的正是这一处隐患，见测试 + `test_modified_rerun_does_not_rewrite_or_duplicate_audit`）。 +- 升级不删除任何租户文件、不改租户已有配置语义： + `[mcp_servers.playwright]` 只在缺失时追加，现有配置段 + 原样保留；manifest 与审计文件只是播种器自己的附加元数据。 +- 测试：`cloud/tests/test_playwright_upgrade.py`（8 项）： + 全新播种、未变脚本自动合并、已变脚本原样不动且有审计、 + 幂等重跑不重复写、无 manifest 老租户（已知哈希合并 / + 未知哈希跳过）、租户数据与配置不受影响。 + +## 本轮补齐：BROWSER_ALLOWED_HOSTS 激活方案（本轮不执行主机激活） + +候选值（域名集合 = 已部署的 Tinyproxy 编程域名集合，不编 +新清单；编码用前导点形式，因为 filter 行本身就是子域名 +包含式，见下文编码说明）： + +``` +BROWSER_ALLOWED_HOSTS=.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,.github.com,.pypi.org,.npmjs.org +``` + +编码说明（关键，不可照抄纯域名版）：filter 行 +`^\(.*\.\)*python\.org$` 放行 `python.org` 及其全部子域名， +而白名单的不带点条目 `python.org` 只精确匹配自身、不放行 +`docs.python.org`；所以候选值必须用前导点条目，否则激活后 +`docs.python.org` / `docs.rust-lang.org` 会被白名单挡掉、 +相对代理白名单是回归（测试 +`test_plain_entries_would_break_subdomains` 已证明这一点）。 +出入说明：任务文本写的是 `npmjs.com`，但已部署 filter +（`deploy/egress/tinyproxy-filter:23`）、`deploy/new-host-checklist.md` +与本文档旧文均为 `npmjs.org`，候选值按已部署集合取 +`npmjs.org`，`npmjs.com` 不在候选值内；如需放行 npm 官网 +本身，那是另起的 filter 变更，不由本轮擅自决定。 + +测试租户/测试任务层面验证（本轮已做，全部 mock/fake， +未碰生产租户）：`cloud/tests/test_playwright_activation.py` +（7 项）：候选值域名集合与已部署 filter 一致；裸域名与子 +域名放行、后缀 trick 与圈外域名拦截；file 放行、localhost +在候选值下被挡（见行为变更）；fake 任务级 navigate 放行与 +拦截（拦截时不启动 Chromium）；未设置时行为不变。 + +对已有浏览器任务的行为变更： + +- 未设置 `BROWSER_ALLOWED_HOSTS`：行为完全不变（只靠代理 + 403），已有任务不受影响。 +- 设置候选值后：navigate 与带 href 的 click / click_ref 在 + 命中时不变、未命中时事前报错（navigate 不启动浏览器、 + click 不执行）；无 href 点击与 click_at 只在事后落点检查； + file:// 一律放行；localhost / 127.0.0.1 被挡——本地开发 + 服务器任务须在值里另加 `localhost`，否则会新报错，这是 + 已知行为变更，不是 bug。 +- JS 自行跳转等拦不住的路径：行为上只剩代理 403 兜底， + 与激活前一致（激活前也是代理 403）。 + +主机激活步骤（只交付步骤，本轮不执行）： + +1. 在主机 worker/service 的环境里设置上述 + `BROWSER_ALLOWED_HOSTS`（与 worker 透传链路一致：worker + 环境变量 → 租户 config 的 `[mcp_servers.playwright.env]`， + 未设置时不写该键）。 +2. 重启 worker 相关 unit（只重启、不改 unit 文件与 filter）。 +3. 用一个测试任务验证：navigate 到 `docs.python.org` 成功、 + navigate 到圈外域名事前报错且不启动浏览器、click 圈外 + 链接事前拒点、localhost 任务按预期被挡（或按需加 + `localhost` 后放行）。 + +回滚：清空该变量（从 worker 环境里删除 +`BROWSER_ALLOWED_HOSTS` 并重启相关 unit），即回到未设置 +行为——已有租户 config 里的旧 env 块由下次任务准备时的 +`ensure_playwright_mcp_env` 覆盖（未设置时不写该键；注意： +已写入旧值的租户 config 需一次任务准备刷新后才彻底回到 +未设置行为，见 worker 透传说明）。回滚不改 filter、不删 +租户数据。 + +是否执行主机激活：本轮不执行。执行条件是本轮测试通过 +（本轮测试已通过，见验证一节）且由 parent/用户拍板—— +测试不通过时只交付方案与回滚、不强行激活；本轮测试通过 +不等于主机已激活，主机激活仍待明确指令。 + ## 安全硬线 - 任何真实密码、cookie、token 绝不写进代码、测试、文档、回执。