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_code_search_ext.py b/cloud/tests/test_code_search_ext.py new file mode 100644 index 0000000..0dd6662 --- /dev/null +++ b/cloud/tests/test_code_search_ext.py @@ -0,0 +1,220 @@ +"""Task-level code-scope extension (search(..., allowed_hosts=[...])). + +All HTTP is mocked; no network, no real keys. Fake key values are +distinctive so a leak into an error string or result is caught by +assertion. Extension values are the caller's own input echoed in +its own result only -- no test here writes them to a log, and the +code under test never formats them into error strings. +""" + +import json +import sys +import unittest +from pathlib import Path + +try: + from cloud import websearch as _ws + from cloud.websearch import ( + CODE_SEARCH_DOMAINS, WebSearchError, _is_code_domain, + _normalize_allowed_hosts, search) + from cloud.websearch_mcp import handle_request +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud import websearch as _ws + from cloud.websearch import ( + CODE_SEARCH_DOMAINS, WebSearchError, _is_code_domain, + _normalize_allowed_hosts, search) + from cloud.websearch_mcp import handle_request + +_FAKE_KEYS = { + "PERPLEXITY_API_KEY": "pplx-SECRET-111", + "EXA_API_KEY": "exa-SECRET-222", + "TAVILY_API_KEY": "tvly-SECRET-333", + "BRAVE_API_KEY": "brave-SECRET-444", +} + +_EXTRA = "docs.example-project.dev" + + +class FakeHttp: + """post_json/get keyed by URL substring, in route order.""" + + def __init__(self, routes): + self.routes = routes + self.seen = [] + + def post_json(self, url, headers, payload): + self.seen.append(("POST", url, payload)) + return self._route(url) + + def get(self, url, headers=None): + self.seen.append(("GET", url, headers)) + return self._route(url) + + def _route(self, url): + for needle, body, status in self.routes: + if needle in url: + if status != 200: + raise WebSearchError("HTTP %s" % status) + if isinstance(body, dict): + return 200, json.dumps(body).encode(), \ + "application/json" + return 200, body.encode(), "text/html" + raise WebSearchError("HTTP 404") + + +def _tavily_body(urls): + return {"results": [{"title": "T %s" % u, "url": u, + "content": "snippet %s" % u} for u in urls]} + + +class ExtensionTestCase(unittest.TestCase): + def test_specified_extra_domain_returned_and_native_param_sent(self): + http = FakeHttp([ + ("api.tavily.com", _tavily_body([ + "https://docs.example-project.dev/guide", + "https://github.com/example/project", + "https://random-blog.example.com/post", + ]), 200), + ]) + out = search("project guide", provider="tavily", scope="code", + allowed_hosts=[_EXTRA], + env=dict(_FAKE_KEYS), http=http) + payload = http.seen[0][2] + self.assertIn(_EXTRA, payload["include_domains"]) + self.assertEqual(out["code_domains"], + list(CODE_SEARCH_DOMAINS) + [_EXTRA]) + self.assertEqual( + [r["url"] for r in out["results"]], + ["https://docs.example-project.dev/guide", + "https://github.com/example/project"]) + self.assertEqual(out["attempted"][0]["domain_filtered_out"], 1) + for secret in _FAKE_KEYS.values(): + self.assertNotIn(secret, json.dumps(out)) + + def test_unspecified_extra_domain_still_filtered_out(self): + http = FakeHttp([ + ("api.tavily.com", _tavily_body([ + "https://docs.example-project.dev/guide", + "https://github.com/example/project", + ]), 200), + ]) + out = search("project guide", provider="tavily", scope="code", + env=dict(_FAKE_KEYS), http=http) + self.assertEqual(out["code_domains"], list(CODE_SEARCH_DOMAINS)) + self.assertEqual([r["url"] for r in out["results"]], + ["https://github.com/example/project"]) + + def test_dot_boundary_positive_and_negative(self): + domains = list(CODE_SEARCH_DOMAINS) + [_EXTRA] + self.assertTrue(_is_code_domain( + "https://api.docs.example-project.dev/v1", domains)) + self.assertTrue(_is_code_domain( + "https://docs.example-project.dev/", domains)) + self.assertFalse(_is_code_domain( + "https://evildocs.example-project.dev/", domains)) + self.assertFalse(_is_code_domain( + "https://docs.example-project.dev.evil.com/", domains)) + # Same boundary rule for a builtin domain, unchanged. + self.assertFalse(_is_code_domain( + "https://evilgithub.com/x", domains)) + self.assertTrue(_is_code_domain( + "https://gist.github.com/x", domains)) + + def test_builtin_entries_are_deduped_not_errors(self): + self.assertEqual( + _normalize_allowed_hosts(["github.com", _EXTRA]), [_EXTRA]) + self.assertEqual( + _normalize_allowed_hosts(["Docs.Example-Project.Dev."]), + ["docs.example-project.dev"]) + + def test_empty_and_illegal_values_raise_before_http(self): + bad_values = [ + [], [""], [" "], ["*.example.com"], + ["http://example.com"], ["https://example.com/x"], + ["example.com:8080"], ["example.com/path"], + [".example.com"], ["example.com..dev"], ["exa mple.com"], + ["user@example.com"], [123], [None], ["github.com"], + ] + for bad in bad_values: + http = FakeHttp([("api.tavily.com", _tavily_body([]), 200)]) + with self.assertRaises(ValueError, msg=repr(bad)): + search("q", provider="tavily", scope="code", + allowed_hosts=bad, env=dict(_FAKE_KEYS), + http=http) + self.assertEqual(http.seen, [], + "invalid %r must fail before any HTTP" + % (bad,)) + # A bare string is iterable but is not a list of hostnames. + with self.assertRaises(ValueError): + search("q", provider="tavily", scope="code", + allowed_hosts="example.com", + env=dict(_FAKE_KEYS), http=FakeHttp([])) + + def test_allowed_hosts_with_general_scope_raises_not_ignored(self): + http = FakeHttp([("api.tavily.com", _tavily_body( + ["https://docs.example-project.dev/guide"]), 200)]) + with self.assertRaises(ValueError): + search("q", provider="tavily", scope="general", + allowed_hosts=[_EXTRA], + env=dict(_FAKE_KEYS), http=http) + self.assertEqual(http.seen, []) + + def test_perplexity_over_cap_skips_native_filter_keeps_postfilter(self): + extras = ["extra%d.example.com" % i for i in range(6)] + self.assertGreater(len(CODE_SEARCH_DOMAINS) + len(extras), 20) + http = FakeHttp([ + ("api.perplexity.ai", + {"choices": [{"message": {"content": "answer"}}], + "search_results": [ + {"title": "P", + "url": "https://extra0.example.com/doc", + "snippet": "extra"}, + {"title": "Ad", "url": "https://shop.example.com/x", + "snippet": "ad"}]}, 200), + ]) + out = search("q", provider="perplexity", scope="code", + allowed_hosts=extras, + env=dict(_FAKE_KEYS), http=http) + payload = http.seen[0][2] + self.assertNotIn("search_domain_filter", payload) + self.assertEqual([r["url"] for r in out["results"]], + ["https://extra0.example.com/doc"]) + self.assertEqual(len(out["code_domains"]), 21) + + +class McpPassthroughTestCase(unittest.TestCase): + def test_mcp_forwards_allowed_hosts_when_supplied(self): + captured = {} + + def fake_search(query, limit=5, provider="auto", scope="general", + allowed_hosts=None): + captured["allowed_hosts"] = allowed_hosts + return {"provider": "fake", "results": [], "answer": None, + "attempted": [], "scope": scope} + + resp = handle_request( + {"jsonrpc": "2.0", "id": 1, "method": "tools/call", + "params": {"name": "web_search", + "arguments": {"query": "hi", "scope": "code", + "allowed_hosts": [_EXTRA]}}}, + search_fn=fake_search) + self.assertFalse(resp["result"]["isError"]) + self.assertEqual(captured["allowed_hosts"], [_EXTRA]) + + def test_mcp_without_allowed_hosts_keeps_old_fake_working(self): + def fake_search(query, limit=5, provider="auto", + scope="general"): + return {"provider": "fake", "results": [], "answer": None, + "attempted": [], "scope": scope} + + resp = handle_request( + {"jsonrpc": "2.0", "id": 2, "method": "tools/call", + "params": {"name": "web_search", + "arguments": {"query": "hi"}}}, + search_fn=fake_search) + self.assertFalse(resp["result"]["isError"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/cloud/tests/test_js_heavy_extract.py b/cloud/tests/test_js_heavy_extract.py new file mode 100644 index 0000000..526b980 --- /dev/null +++ b/cloud/tests/test_js_heavy_extract.py @@ -0,0 +1,155 @@ +"""JS-heavy page extraction evaluation, mock-first. + +All HTTP is mocked; no network, no real keys. The three page shapes +below are the mock stand-ins for real JS-heavy pages (a static shell +whose body is rendered by JS, a shell whose pagination is JS-loaded, +and a shell whose content lives only in a + +Next (static link) +""" + +# Shape 2: paginated list; page 2 items arrive only via JS fetch. +_PAGED_SHELL_HTML = """List + + + +""" + + +class FakeHttp: + """get/post_json keyed by URL substring, in route order.""" + + def __init__(self, routes): + self.routes = routes + self.seen = [] + + def post_json(self, url, headers, payload): + self.seen.append(("POST", url, payload)) + return self._route(url) + + def get(self, url, headers=None): + self.seen.append(("GET", url, headers)) + return self._route(url) + + def _route(self, url): + for needle, body, status in self.routes: + if needle in url: + if status != 200: + raise WebSearchError("HTTP %s" % status) + if isinstance(body, dict): + return 200, json.dumps(body).encode(), \ + "application/json" + return 200, body.encode(), "text/html" + raise WebSearchError("HTTP 404") + + +class JsHeavyDirectFetchTestCase(unittest.TestCase): + def test_direct_fetch_gets_shell_only_not_js_rendered_body(self): + http = FakeHttp([("example.com/guide", _SHELL_HTML, 200)]) + out = fetch_url("https://example.com/guide", http=http, env={}) + # Static shell survives: title + placeholder + static link. + self.assertEqual(out["title"], "Guide") + self.assertIn("Loading", out["text"]) + self.assertIn("https://example.com/guide/page2", out["links"]) + # JS-rendered body does not: