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)
This commit is contained in:
host committed 2026-11-01 09:06:28 +00:00
1 parent 1141d0d0f8
commit 125b436ab3
8 files changed
+1150 -33

No files matched your search

+260 -23
View File
@@ -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):
+15 -3
View File
@@ -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",
+30 -3
View File
@@ -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`(删不掉也不报错),
+209
View File
@@ -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/")
@@ -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/")
+216
View File
@@ -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
+10 -3
View File
@@ -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 那种“已宣传就必须有”的
+101 -1
View File
@@ -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 绝不写进代码、测试、文档、回执。