diff --git a/chv-guest/browser-blocklist.txt b/chv-guest/browser-blocklist.txt new file mode 100644 index 0000000..7a4690a --- /dev/null +++ b/chv-guest/browser-blocklist.txt @@ -0,0 +1,18 @@ +# CHV 浏览器静态拦截名单（对齐 Muse P4） +# +# 格式：一行一个条目，# 开头是注释，空行跳过。 +# example.com 拦截该域名及其全部子域 +# .example.com 同上（前导点可写可不写） +# 匹配规则见 chv_browser_guard.py 的 check_blocklist。 +# +# 来源与更新机制（诚实口径）： +# 本文件随镜像发布，是静态名单。首版只含 RFC 2606 保留 .test 域 +# 的格式示例与自测条目（.test 域名保证不可解析，永不误拦真实站）， +# 不编造真实恶意域名。生产名单由运维在构建时用公开情报源 +# （如 URLhaus / Google Safe Browsing 导出）的域名替换本文件后 +# 重建镜像；更新 = 换文件 + 重建镜像，无在线更新（在线更新列为 +# 后续，不冒充已做）。名单缺失时视为空名单，不静默失败。 +# +# 自测条目（.test 保留域，仅供测试与格式示例）： +malware.test +.phishing.test diff --git a/chv-guest/build.sh b/chv-guest/build.sh index 09a9fc6..ea73e23 100755 --- a/chv-guest/build.sh +++ b/chv-guest/build.sh @@ -200,6 +200,13 @@ if [ -f "$REPO_DIR/cloud/chv/authd.py" ] && [ -f "$REPO_DIR/cloud/chv/sentinel.p else echo "警告：Muse P3 云端叶子模块不在仓库里（cloud/chv/authd.py 与 sentinel.py），镜像暂不含 authd 叶子模块" >&2 fi +# 对齐 Muse P4：CDP broker / guard 共享模块 / 拦截名单三件缺一即失败。 +for p4_file in chv-cdp-broker.py chv_browser_guard.py browser-blocklist.txt; do + if [ ! -f "$SCRIPT_DIR/$p4_file" ]; then + echo "缺少 $p4_file（Muse P4 客机文件）：$SCRIPT_DIR" >&2 + exit 1 + fi +done if [ "$DRY_RUN" = "0" ]; then if [ ! -e /dev/kvm ]; then echo "CHV_GUEST_BUILD needs a KVM host: /dev/kvm 不存在" >&2 @@ -305,6 +312,10 @@ VC_ARGS=(-a "$OUT/disk.raw" --run-command 'mv /usr/local/bin/authd.py /usr/local/bin/chv-authd && chmod +x /usr/local/bin/chv-authd /usr/local/bin/git-credential-chv-authd' --run-command 'systemctl enable chv-authd || true' --run-command 'printf "runcmd:\n - systemctl restart ssh || service ssh restart || true\n - systemctl start chv-authd || true\n - /usr/local/bin/chv-sync boot >>/var/log/chv-sync.log 2>&1 || true\n" > /etc/cloud/cloud.cfg.d/99-chv-sync.cfg' + --copy-in "$SCRIPT_DIR/chv-cdp-broker.py:/usr/local/bin/" + --copy-in "$SCRIPT_DIR/chv_browser_guard.py:/usr/local/bin/" + --copy-in "$SCRIPT_DIR/browser-blocklist.txt:/usr/local/share/chv/" + --run-command 'mv /usr/local/bin/chv-cdp-broker.py /usr/local/bin/chv-cdp-broker && chmod +x /usr/local/bin/chv-cdp-broker && chmod 0644 /usr/local/bin/chv_browser_guard.py /usr/local/share/chv/browser-blocklist.txt' --run-command 'git config --system user.name chv && git config --system user.email chv@localhost && git config --system credential.helper store && git config --system init.defaultBranch main' --run-command "$GT_INSTALL_CMD" --run-command '/usr/local/bin/gt version > /app/guest-versions-gt.txt' @@ -314,7 +325,8 @@ VC_ARGS=(-a "$OUT/disk.raw" --run-command 'useradd --system --home-dir /var/lib/chv-browser --shell /usr/sbin/nologin chv-browser && mkdir -p /var/lib/chv-browser /run/chv/shots && chown chv-browser:chv-browser /var/lib/chv-browser /run/chv/shots && chmod 0755 /run/chv /run/chv/shots' --run-command 'printf "[Unit]\nDescription=CHV cell broker (Muse P2)\nAfter=network.target\n\n[Service]\nType=simple\nExecStart=/usr/bin/python3 /usr/local/bin/chv-broker\nRestart=on-failure\nRestartSec=2\n\n[Install]\nWantedBy=multi-user.target\n" > /etc/systemd/system/chv-broker.service' --run-command 'printf "[Unit]\nDescription=CHV browser sub-agent (Muse P2)\nAfter=network.target\n\n[Service]\nType=simple\nUser=chv-browser\nGroup=chv-browser\nExecStart=/usr/bin/python3 /usr/local/bin/chv-browser\nRestart=on-failure\nRestartSec=2\n\n[Install]\nWantedBy=multi-user.target\n" > /etc/systemd/system/chv-browser.service' - --run-command 'printf "d /run/chv 0755 root root -\nd /run/chv/shots 0755 chv-browser chv-browser -\n" > /etc/tmpfiles.d/chv.conf && systemctl enable chv-broker chv-browser || true') + --run-command 'printf "[Unit]\nDescription=CHV CDP broker (Muse P4)\nAfter=network.target\n\n[Service]\nType=simple\nUser=chv-browser\nGroup=chv-browser\nExecStart=/usr/bin/python3 /usr/local/bin/chv-cdp-broker\nRestart=on-failure\nRestartSec=2\n\n[Install]\nWantedBy=multi-user.target\n" > /etc/systemd/system/chv-cdp-broker.service' + --run-command 'printf "d /run/chv 0755 root root -\nd /run/chv/shots 0755 chv-browser chv-browser -\nd /run/chv/ask-queue 0750 chv-browser chv-browser -\n" > /etc/tmpfiles.d/chv.conf && systemctl enable chv-broker chv-browser chv-cdp-broker || true') # P3 交接（Muse P3 的 authd/sentinel 叶子模块）：文件在仓库里存在时 # 拷入镜像（本分支已有，合入后自动生效）；不存在时只警告、不失败—— # M4 既定 build.sh 不硬要求 cloud/，见上。缺叶子模块时 chv-authd diff --git a/chv-guest/buildinfo.py b/chv-guest/buildinfo.py index e18c821..1b644e2 100644 --- a/chv-guest/buildinfo.py +++ b/chv-guest/buildinfo.py @@ -45,6 +45,9 @@ BASE_CONTENTS = [ # 它们是条件拷入（缺失只警告、不失败），存在时才经 # extra_contents 登记，绝不在缺失的构建里谎称已进镜像。 "chv-authd", "git-credential-chv-authd", "chv-authd.service", + # 对齐 Muse P4：单元外 CDP broker + 浏览器安全共享模块 + + # 静态拦截名单。 + "chv-cdp-broker", "chv-browser-guard", "browser-blocklist", ] # 执行器默认内核 cmdline（config.py），构建产物须与它一起记录（B-12）， diff --git a/chv-guest/chv-browser.py b/chv-guest/chv-browser.py index 743feb9..5526d28 100644 --- a/chv-guest/chv-browser.py +++ b/chv-guest/chv-browser.py @@ -1,217 +1,200 @@ #!/usr/bin/env python3 """chv-browser：浏览器子代理（客机侧、单元外独立进程）。 -对齐 Muse P2（见 docs/chv-muse-p2-gap.md）：浏览器任务由专门子代理 -完成，与 codex 单元分离。子代理只向单元返回页面无障碍树快照 -（Playwright accessibility.snapshot），不见原始 DOM，不能在页面 -上下文跑 JS，不暴露 evaluate/content，不开 DevTools。 +对齐 Muse P4（见 docs/chv-muse-p4-gap.md）：本文件源自 P2 的 +chv-browser.py（c61484a），P4 在此基础上重构——子代理不再直连 +Playwright/Chromium，Chromium 的所有权在单元外 CDP broker +（chv-cdp-broker.py）手里；本进程只是 broker 的客户端。 + +子代理只向单元返回页面无障碍树快照（Playwright +accessibility.snapshot，经 broker 取回），不见原始 DOM，不能 +在页面上下文跑 JS，不暴露交互操作，不开 DevTools。 操作（Unix socket /run/chv/browser.sock，JSON 行）： - open {url} -> 导航并返回 {title, url, a11y, truncated} - snapshot {} -> 当前页的 {title, url, a11y, truncated} - screenshot {} -> 截图落盘 /run/chv/shots/.png，返回 {path} - close {} -> 关当前页（浏览器进程继续跑，等下一次 open） + open {url} -> 经 broker 导航并返回 {title, url, a11y, + truncated, untrusted_input, injection_flags, + form_risk} + snapshot {} -> 当前页同上 + screenshot {} -> 截图落盘，返回 {path} + close {} -> 关当前页 + pause {} -> 暂停（用户接管/填表/自动暂停的统一状态） + resume {} -> 解除暂停 + status {} -> {paused, url, form_risk} + list_asks {} -> 待审批 ask 列表 + resolve_ask {ask_id, decision} -> 写回 allow/denied ping {} -> {"ok": true} -URL 只允许 http/https；localhost 与私有地址字面量直接拒绝（主机名 -判定，不做 DNS 解析——DNS 重绑定防护是 P3 Sentinel 的事，这里不 -冒充已做）。无障碍树截断到 200 节点 / 64KB，超了标 truncated=true， -绝不把整棵树无上限灌进 socket。 +暂停时除 ping/resume/status/list_asks/resolve_ask 外一切操作 +拒绝——用户接管或凭证注入填表时，代理完全不能行动。 -playwright 是懒加载：import 本模块不需要 playwright 已装（单测 -在无 playwright 环境跑）；真正 open 时才加载，缺失即报错。 +URL 校验、快照截断、错误消毒、禁用操作表都在 +chv_browser_guard.py（纯函数模块，两脚本共用）。本层再做一次 +validate_url 与 truncate_a11y（纵深防御），回包先过泄露检查。 -只用标准库 + playwright（运行时）。运行身份：系统用户 chv-browser， -socket 0600（broker 以 root 连，不受模式限制）。 +只用标准库。运行身份：系统用户 chv-browser，socket 0600 +（broker 以 root 连，不受模式限制）。 """ +import importlib.util import json import os -import re import socket import sys -import uuid BROWSER_SOCK = "/run/chv/browser.sock" -SHOTS_DIR = "/run/chv/shots" -PLAYWRIGHT_BROWSERS_PATH = "/usr/local/share/ms-playwright" +CDP_SOCK = "/run/chv/cdp-broker.sock" MAX_REQUEST_BYTES = 65536 -MAX_A11Y_NODES = 200 -MAX_A11Y_CHARS = 65536 -NAV_TIMEOUT_MS = 30000 -# 拒绝的 URL 主机：localhost 与私有地址字面量（主机名判定）。 -_PRIVATE_HOST_RE = re.compile( - r"^(localhost|127\.\d+\.\d+\.\d+|10\.\d+\.\d+\.\d+|" - r"172\.(1[6-9]|2\d|3[01])\.\d+\.\d+|192\.168\.\d+\.\d+|" - r"\[::1\]|\[fe8[0-9a-f]*:.*\])$", - re.IGNORECASE) +_GUARD = None -def validate_url(url): - """URL 校验：只允许 http/https，且主机不是 localhost/私有地址 - 字面量。返回 (ok, error)。DNS 重绑定不在本层防（见模块 docstring）。""" - if not isinstance(url, str) or not url: - return False, "url 不能为空" - m = re.match(r"^(https?)://([^/]+)", url, re.IGNORECASE) - if not m: - return False, "只允许 http/https URL：%s" % url[:80] - host = m.group(2).split("@")[-1].split(":")[0].lower() - if _PRIVATE_HOST_RE.match(host): - return False, "拒绝 localhost/私有地址：%s" % host - return True, "" - - -def truncate_a11y(node, max_nodes=MAX_A11Y_NODES, - max_chars=MAX_A11Y_CHARS): - """截断无障碍树：节点数超上限时剪掉多余子树。返回 (tree, truncated)。 - 纯函数，可单测。节点计数按 dict 节点数（含 children 展开）。""" - state = {"count": 0, "truncated": False} - - def walk(n): - if not isinstance(n, dict): - return n - state["count"] += 1 - if state["count"] > max_nodes: - state["truncated"] = True - return None - out = {} - for key, value in n.items(): - if key == "children" and isinstance(value, list): - children = [] - for child in value: - walked = walk(child) - if walked is None: - state["truncated"] = True - break - children.append(walked) - out[key] = children - else: - out[key] = value - return out - - tree = walk(node) - text = json.dumps(tree, ensure_ascii=False) - if len(text) > max_chars: - state["truncated"] = True - # 字符超限时只留顶层角色/名字/URL 信息，不灌整棵树。 - tree = {"role": (node or {}).get("role"), - "name": (node or {}).get("name"), - "truncated_chars": True} - return tree, state["truncated"] - - -def _load_playwright(): - """懒加载 playwright；缺失时报明错（不静默假装有浏览器）。""" - os.environ.setdefault("PLAYWRIGHT_BROWSERS_PATH", - PLAYWRIGHT_BROWSERS_PATH) - try: - from playwright.sync_api import sync_playwright - except ImportError as e: - raise RuntimeError( - "chv-browser: playwright 未装（%s），浏览器子代理不可用" - % e) - return sync_playwright +def _load_guard(): + """加载同目录的 chv_browser_guard（纯函数模块）。""" + global _GUARD + if _GUARD is not None: + return _GUARD + here = os.path.dirname(os.path.abspath(__file__)) + path = os.path.join(here, "chv_browser_guard.py") + spec = importlib.util.spec_from_file_location( + "chv_browser_guard", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + _GUARD = module + return module class BrowserAgent: - """单页浏览器子代理。playwright_factory 供单测注入假实现； - 正常运行时用 _load_playwright()。""" + """浏览器子代理：CDP broker 的客户端。broker_call 供单测注入 + 假 broker（收到 req dict 返回 resp dict）；正常运行时走 + Unix socket 直连 broker。""" - def __init__(self, playwright_factory=None, shots_dir=SHOTS_DIR): - self._factory = playwright_factory or _load_playwright - self._shots_dir = shots_dir - self._pw = None - self._browser = None - self._page = None + def __init__(self, cdp_sock_path=CDP_SOCK, broker_call=None): + self._guard = _load_guard() + self._cdp_sock_path = cdp_sock_path + self._broker_call = broker_call + self._paused = False - def _ensure_browser(self): - if self._browser is not None: - return - self._pw = self._factory() - # sync_playwright() 返回上下文管理器；__enter__ 后拿 chromium。 - entered = self._pw.__enter__() - self._browser = entered.chromium.launch(headless=True) + def _call(self, req): + if self._broker_call is not None: + return self._broker_call(req) + try: + conn = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + conn.settimeout(120) + conn.connect(self._cdp_sock_path) + except OSError as e: + return {"ok": False, + "error": "CDP broker 未就绪：%s" % e} + try: + conn.sendall((json.dumps(req, ensure_ascii=False) + + "\n").encode()) + data = b"" + while len(data) < MAX_REQUEST_BYTES: + chunk = conn.recv(4096) + if not chunk: + break + data += chunk + if data.endswith(b"\n"): + break + except OSError as e: + return {"ok": False, + "error": "CDP broker 通信失败：%s" % e} + finally: + conn.close() + try: + resp = json.loads(data.decode()) + except (ValueError, UnicodeDecodeError): + return {"ok": False, "error": "CDP broker 回包不是 JSON"} + if not isinstance(resp, dict) or "ok" not in resp: + return {"ok": False, "error": "CDP broker 回包格式不对"} + return resp + + def _sync_paused(self, resp): + """从回包同步暂停镜像状态（broker 暂停时本层同样拒绝）。""" + if isinstance(resp, dict) and "已暂停" in str( + resp.get("error", "")): + self._paused = True + return resp + + def _blocked_while_paused(self): + return {"ok": False, + "error": "浏览器已暂停（用户接管/填表/自动暂停）"} def open(self, url): - ok, err = validate_url(url) + if self._paused: + return self._blocked_while_paused() + ok, err = self._guard.validate_url(url) if not ok: return {"ok": False, "error": err} - try: - self._ensure_browser() - if self._page is not None: - try: - self._page.close() - except Exception: - pass - self._page = self._browser.new_page() - self._page.goto(url, timeout=NAV_TIMEOUT_MS) - self._page.wait_for_load_state("domcontentloaded", - timeout=NAV_TIMEOUT_MS) - except Exception as e: - return {"ok": False, - "error": "打开 %s 失败：%s" % (url[:80], e)} - return self.snapshot() + resp = self._call({"op": "open", "url": url}) + if resp.get("ok") and isinstance(resp.get("a11y"), dict): + tree, truncated = self._guard.truncate_a11y(resp["a11y"]) + resp = dict(resp) + resp["a11y"] = tree + resp["truncated"] = bool(resp.get("truncated") + or truncated) + return self._sync_paused(resp) def snapshot(self): - if self._page is None: - return {"ok": False, "error": "还没有打开的页面（先 open）"} - try: - title = self._page.title() - url = self._page.url - tree = self._page.accessibility.snapshot() - except Exception as e: - return {"ok": False, "error": "取快照失败：%s" % e} - tree, truncated = truncate_a11y(tree or {}) - return {"ok": True, "title": title, "url": url, - "a11y": tree, "truncated": truncated} + if self._paused: + return self._blocked_while_paused() + resp = self._call({"op": "snapshot"}) + if resp.get("ok") and isinstance(resp.get("a11y"), dict): + tree, truncated = self._guard.truncate_a11y(resp["a11y"]) + resp = dict(resp) + resp["a11y"] = tree + resp["truncated"] = bool(resp.get("truncated") + or truncated) + return self._sync_paused(resp) def screenshot(self): - if self._page is None: - return {"ok": False, "error": "还没有打开的页面（先 open）"} - try: - os.makedirs(self._shots_dir, exist_ok=True) - path = os.path.join(self._shots_dir, "%s.png" - % uuid.uuid4().hex) - self._page.screenshot(path=path) - except Exception as e: - return {"ok": False, "error": "截图失败：%s" % e} - return {"ok": True, "path": path} + if self._paused: + return self._blocked_while_paused() + return self._sync_paused(self._call({"op": "screenshot"})) def close(self): - if self._page is not None: - try: - self._page.close() - except Exception: - pass - self._page = None - return {"ok": True, "closed": True} + if self._paused: + return self._blocked_while_paused() + resp = self._call({"op": "close"}) + return self._sync_paused(resp) - def shutdown(self): - try: - if self._browser is not None: - self._browser.close() - except Exception: - pass - try: - if self._pw is not None: - self._pw.__exit__(None, None, None) - except Exception: - pass + def pause(self): + resp = self._call({"op": "pause"}) + if resp.get("ok"): + self._paused = True + return resp + def resume(self): + resp = self._call({"op": "resume"}) + if resp.get("ok"): + self._paused = False + return resp -# 本层不暴露的操作：evaluate / content（原始 DOM）/ DevTools。 -# 收到即拒绝，不转发、不执行。 -FORBIDDEN_OPS = ("evaluate", "content", "devtools", "js") + def status(self): + resp = self._call({"op": "status"}) + if isinstance(resp, dict) and "paused" in resp: + self._paused = bool(resp["paused"]) + return resp + + def list_asks(self): + return self._call({"op": "list_asks"}) + + def resolve_ask(self, ask_id, decision): + resp = self._call({"op": "resolve_ask", "ask_id": ask_id, + "decision": decision}) + if resp.get("ok") and resp.get("decision") == "allowed": + self._paused = False + return resp def handle_request(agent, req): """处理一个请求 dict，返回响应 dict（纯逻辑，可单测）。""" + guard = _load_guard() if not isinstance(req, dict): return {"ok": False, "error": "请求须是 JSON 对象"} op = req.get("op") - if op in FORBIDDEN_OPS: + if op in guard.FORBIDDEN_OPS: return {"ok": False, - "error": "操作 %r 不暴露（无 JS/无原始 DOM）" % op} + "error": "操作 %r 不暴露（无 JS/无原始 DOM/无交互）" + % op} if op == "ping": return {"ok": True, "browser": True} if op == "open": @@ -222,14 +205,54 @@ def handle_request(agent, req): return agent.screenshot() if op == "close": return agent.close() + if op == "pause": + return agent.pause() + if op == "resume": + return agent.resume() + if op == "status": + return agent.status() + if op == "list_asks": + return agent.list_asks() + if op == "resolve_ask": + return agent.resolve_ask(req.get("ask_id"), + req.get("decision")) return {"ok": False, "error": "未知操作：%r（白名单：open/snapshot/screenshot/" - "close/ping）" % (op,)} + "close/pause/resume/status/list_asks/resolve_ask/" + "ping）" % (op,)} -def serve(sock_path=BROWSER_SOCK, shots_dir=SHOTS_DIR): - """监听 Unix socket，逐连接处理。socket 0600（只给属主 + - broker 的 root 连）。建不起来即抛错，不要静默带病跑。""" +def _handle_conn(agent, conn): + """一个连接一请求一响应；回包先过泄露检查，不过即换通用错； + 错误信息过消毒，绝不把页面原文原样放进回包。""" + guard = _load_guard() + conn.settimeout(120) + data = b"" + while len(data) < MAX_REQUEST_BYTES: + chunk = conn.recv(4096) + if not chunk: + break + data += chunk + if data.endswith(b"\n"): + break + try: + req = json.loads(data.decode()) + except (ValueError, UnicodeDecodeError): + resp = {"ok": False, "error": "请求不是合法 JSON"} + else: + try: + resp = handle_request(agent, req) + except Exception as e: + resp = {"ok": False, + "error": guard.sanitize_error(e)} + if guard.response_leaks(resp): + resp = {"ok": False, "error": "响应泄露检查未通过（已拦截）"} + conn.sendall((json.dumps(resp, ensure_ascii=False) + "\n").encode()) + + +def serve(sock_path=BROWSER_SOCK, cdp_sock_path=CDP_SOCK): + """监听 Unix socket，逐连接处理。socket 0600。建不起来即抛错， + 不要静默带病跑。""" if os.path.exists(sock_path): os.remove(sock_path) os.makedirs(os.path.dirname(sock_path), exist_ok=True) @@ -238,7 +261,7 @@ def serve(sock_path=BROWSER_SOCK, shots_dir=SHOTS_DIR): srv.listen(16) os.chmod(sock_path, 0o600) print("chv-browser: 监听 %s" % sock_path, file=sys.stderr, flush=True) - agent = BrowserAgent(shots_dir=shots_dir) + agent = BrowserAgent(cdp_sock_path=cdp_sock_path) try: while True: try: @@ -246,54 +269,35 @@ def serve(sock_path=BROWSER_SOCK, shots_dir=SHOTS_DIR): except OSError: continue try: - conn.settimeout(120) - data = b"" - while len(data) < MAX_REQUEST_BYTES: - chunk = conn.recv(4096) - if not chunk: - break - data += chunk - if data.endswith(b"\n"): - break - try: - req = json.loads(data.decode()) - except (ValueError, UnicodeDecodeError): - resp = {"ok": False, "error": "请求不是合法 JSON"} - else: - try: - resp = handle_request(agent, req) - except Exception as e: - resp = {"ok": False, - "error": "浏览器子代理内部错误：%s" % e} - conn.sendall((json.dumps(resp, ensure_ascii=False) - + "\n").encode()) + _handle_conn(agent, conn) except OSError as e: - print("chv-browser: 连接处理异常：%s" % e, file=sys.stderr, - flush=True) + print("chv-browser: 连接处理异常：%s" + % _load_guard().sanitize_error(e), + file=sys.stderr, flush=True) finally: try: conn.close() except OSError: pass finally: - agent.shutdown() + pass def main(argv): sock_path = BROWSER_SOCK - shots_dir = SHOTS_DIR + cdp_sock_path = CDP_SOCK i = 0 while i < len(argv): if argv[i] == "--socket" and i + 1 < len(argv): sock_path = argv[i + 1] i += 2 - elif argv[i] == "--shots-dir" and i + 1 < len(argv): - shots_dir = argv[i + 1] + elif argv[i] == "--cdp-socket" and i + 1 < len(argv): + cdp_sock_path = argv[i + 1] i += 2 else: print("chv-browser: 未知参数：%s" % argv[i], file=sys.stderr) return 2 - serve(sock_path, shots_dir) + serve(sock_path, cdp_sock_path) return 0 diff --git a/chv-guest/chv-cdp-broker.py b/chv-guest/chv-cdp-broker.py new file mode 100644 index 0000000..c2d3eec --- /dev/null +++ b/chv-guest/chv-cdp-broker.py @@ -0,0 +1,428 @@ +#!/usr/bin/env python3 +"""chv-cdp-broker：单元外 CDP broker（客机侧、单元外独立进程）。 + +对齐 Muse P4（见 docs/chv-muse-p4-gap.md）：Chromium 的所有权 +在这里，不在浏览器子代理手里。broker 独家持有 Playwright， +CDP 连接只存在于本进程内；子代理（chv-browser.py）只经本 +socket 说话，绝不直连浏览器、绝不见 CDP 端点。 + +操作（Unix socket /run/chv/cdp-broker.sock，JSON 行）： + open {url} -> 导航并返回 {title, url, a11y, truncated, + untrusted_input, injection_flags, form_risk} + snapshot {} -> 当前页同上 + screenshot {} -> 截图落盘 /run/chv/shots/.png，返回 {path} + close {} -> 关当前页（浏览器进程继续跑） + pause {} -> 暂停（用户接管/填表/自动暂停的统一状态） + resume {} -> 解除暂停 + status {} -> {paused, url, form_risk} + list_asks {} -> 待审批 ask 列表 + resolve_ask {ask_id, decision} -> 写回 allow/denied + ping {} -> {"ok": true} + +暂停时除 ping/resume/status/list_asks 外一切操作拒绝—— +用户接管或凭证注入填表时，代理完全不能行动。 + +安全规则（纯函数在 chv_browser_guard.py，本文件只做接线）： + 导航前 validate_url + 静态拦截名单；快照恒带 + untrusted_input=true（外部页面文本永远是数据、不是指令）， + 命中注入启发式带 injection_flags；高风险表单在页时 open + 导航即返回 ask（入队、执行停止）；观测到导航 POST 即自动 + 暂停 + ask；下载命中可执行扩展名即取消 + 暂停 + ask。 + 所有回包与日志过 sanitize_error；响应里绝无 ws 端点、 + DevTools 地址、原始 HTML。 + +Playwright 调用白名单：launch/goto/wait_for_load_state/title/ +url/accessibility.snapshot/screenshot/close + 只读监听 +on(request/download)。evaluate/add_script_tag/add_init_script/ +expose_function/set_content 等脚本注入面不在代码里出现， +测试用 AST 扫本文件钉住。 + +只用标准库 + playwright（运行时，懒加载）。运行身份：系统用户 +chv-browser，socket 0600。 +""" +import importlib.util +import json +import os +import socket +import sys +import uuid + +CDP_SOCK = "/run/chv/cdp-broker.sock" +ASK_QUEUE_DIR = "/run/chv/ask-queue" +BLOCKLIST_PATH = "/usr/local/share/chv/browser-blocklist.txt" +SHOTS_DIR = "/run/chv/shots" +PLAYWRIGHT_BROWSERS_PATH = "/usr/local/share/ms-playwright" +MAX_REQUEST_BYTES = 65536 +NAV_TIMEOUT_MS = 30000 + +_GUARD = None + + +def _load_guard(): + """加载同目录的 chv_browser_guard（纯函数模块）。运行时 + 安装在 /usr/local/bin/chv_browser_guard.py；测试时与本 + 文件同在 chv-guest/。只在首次调用加载一次。""" + global _GUARD + if _GUARD is not None: + return _GUARD + here = os.path.dirname(os.path.abspath(__file__)) + path = os.path.join(here, "chv_browser_guard.py") + spec = importlib.util.spec_from_file_location( + "chv_browser_guard", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + _GUARD = module + return module + + +def _load_playwright(): + """懒加载 playwright；缺失时报明错（不静默假装有浏览器）。""" + os.environ.setdefault("PLAYWRIGHT_BROWSERS_PATH", + PLAYWRIGHT_BROWSERS_PATH) + try: + from playwright.sync_api import sync_playwright + except ImportError as e: + raise RuntimeError( + "chv-cdp-broker: playwright 未装（%s）" % e) + return sync_playwright + + +class CdpBroker: + """CDP broker：独家持有 Chromium。playwright_factory 供单测 + 注入假实现；正常运行时用 _load_playwright()。""" + + def __init__(self, playwright_factory=None, + blocklist_path=BLOCKLIST_PATH, + ask_queue_dir=ASK_QUEUE_DIR, shots_dir=SHOTS_DIR): + self._guard = _load_guard() + self._factory = playwright_factory or _load_playwright + self._blocklist_path = blocklist_path + self._ask_queue_dir = ask_queue_dir + self._shots_dir = shots_dir + self._pw = None + self._browser = None + self._page = None + self._paused = False + self._form_risk = {"level": "none", "fields": []} + self._current_url = "" + + def _ensure_browser(self): + if self._browser is not None: + return + ok, err = self._guard.check_launch_args( + self._guard.LAUNCH_ARGS) + if not ok: + raise RuntimeError("启动参数不合规：%s" % err) + self._pw = self._factory() + entered = self._pw.__enter__() + self._browser = entered.chromium.launch( + headless=True, args=list(self._guard.LAUNCH_ARGS)) + + def _new_page(self): + if self._page is not None: + try: + self._page.close() + except Exception: + pass + self._page = self._browser.new_page() + self._page.on("request", self._on_request) + self._page.on("download", self._on_download) + + def _on_request(self, request): + """只读监听：导航 POST（表单提交）即自动暂停 + 入队。 + 非导航 POST（beacon/fetch）不在此处理——它不代表提交。""" + try: + method = request.method + is_nav = request.is_navigation_request() + except Exception: + return + if method != "POST" or not is_nav: + return + self._paused = True + self._guard.enqueue_ask( + self._ask_queue_dir, "form_submit_observed", + self._current_url, "观测到导航 POST（疑似表单提交）") + + def _on_download(self, download): + """只读监听：命中可执行扩展名即取消下载 + 暂停 + 入队。""" + try: + name = download.suggested_filename + except Exception: + return + blocked, reason = self._guard.scan_download(name) + if not blocked: + return + try: + download.cancel() + except Exception: + pass + self._paused = True + self._guard.enqueue_ask( + self._ask_queue_dir, "download_blocked", + self._current_url, "下载命中拦截规则：%s" % reason) + + def _paused_resp(self): + return {"ok": False, + "error": "浏览器已暂停（用户接管/填表/自动暂停）"} + + def _ask_resp(self, ask_id, detail): + return {"ok": False, "ask": True, "ask_id": ask_id, + "error": "需审批：%s" % detail} + + def open(self, url): + if self._paused: + return self._paused_resp() + ok, err = self._guard.validate_url(url) + if not ok: + return {"ok": False, "error": err} + host = self._guard.url_host(url) + entries = self._guard.load_blocklist(self._blocklist_path) + if self._guard.check_blocklist(host, entries): + return {"ok": False, "error": "命中恶意站拦截名单：%s" + % host} + if self._form_risk.get("level") != "none": + ask_id = self._guard.enqueue_ask( + self._ask_queue_dir, "form_submit", url, + "高风险表单在页（%s），导航须审批" + % self._form_risk["level"]) + return self._ask_resp(ask_id, "高风险表单在页，导航须审批") + try: + self._ensure_browser() + self._new_page() + self._page.goto(url, timeout=NAV_TIMEOUT_MS) + self._page.wait_for_load_state("domcontentloaded", + timeout=NAV_TIMEOUT_MS) + except Exception as e: + return {"ok": False, + "error": self._guard.sanitize_error(e)} + return self.snapshot() + + def snapshot(self): + if self._paused: + return self._paused_resp() + if self._page is None: + return {"ok": False, "error": "还没有打开的页面（先 open）"} + try: + title = self._page.title() + url = self._page.url + tree = self._page.accessibility.snapshot() + except Exception as e: + return {"ok": False, + "error": self._guard.sanitize_error(e)} + tree, truncated = self._guard.truncate_a11y(tree or {}) + flags = self._guard.detect_injection(tree) + self._form_risk = self._guard.detect_form_risk(tree) + self._current_url = url + return {"ok": True, "title": title, "url": url, + "a11y": tree, "truncated": truncated, + "untrusted_input": True, "injection_flags": flags, + "form_risk": self._form_risk} + + def screenshot(self): + if self._paused: + return self._paused_resp() + if self._page is None: + return {"ok": False, "error": "还没有打开的页面（先 open）"} + try: + os.makedirs(self._shots_dir, exist_ok=True) + path = os.path.join(self._shots_dir, "%s.png" + % uuid.uuid4().hex) + self._page.screenshot(path=path) + except Exception as e: + return {"ok": False, + "error": self._guard.sanitize_error(e)} + return {"ok": True, "path": path} + + def close(self): + if self._page is not None: + try: + self._page.close() + except Exception: + pass + self._page = None + self._form_risk = {"level": "none", "fields": []} + self._current_url = "" + return {"ok": True, "closed": True} + + def pause(self): + self._paused = True + return {"ok": True, "paused": True} + + def ping(self): + return {"ok": True, "broker": True} + + def resume(self): + self._paused = False + return {"ok": True, "paused": False} + + def status(self): + return {"ok": True, "paused": self._paused, + "url": self._current_url, + "form_risk": self._form_risk} + + def list_asks(self): + return {"ok": True, + "asks": self._guard.list_asks(self._ask_queue_dir)} + + def resolve_ask(self, ask_id, decision): + ok, err = self._guard.resolve_ask( + self._ask_queue_dir, ask_id, decision) + if not ok: + return {"ok": False, "error": err} + if decision == "allowed": + self._paused = False + # 批准即放行这次操作：清掉快照缓存的表单风险（下一次 + # snapshot 会按新页面重算，不会丢检测）。 + self._form_risk = {"level": "none", "fields": []} + else: + self.close() + self._paused = False + return {"ok": True, "ask_id": ask_id, + "decision": decision} + + def shutdown(self): + try: + if self._browser is not None: + self._browser.close() + except Exception: + pass + try: + if self._pw is not None: + self._pw.__exit__(None, None, None) + except Exception: + pass + + +def handle_request(broker, req): + """处理一个请求 dict，返回响应 dict（纯逻辑，可单测）。 + 未知操作与禁用操作一律拒绝，不转发、不执行。""" + guard = _load_guard() + if not isinstance(req, dict): + return {"ok": False, "error": "请求须是 JSON 对象"} + op = req.get("op") + if op in guard.FORBIDDEN_OPS: + return {"ok": False, + "error": "操作 %r 不暴露（无 JS/无原始 DOM/无交互）" + % op} + if op == "ping": + return {"ok": True, "broker": True} + if op == "open": + return broker.open(req.get("url")) + if op == "snapshot": + return broker.snapshot() + if op == "screenshot": + return broker.screenshot() + if op == "close": + return broker.close() + if op == "pause": + return broker.pause() + if op == "resume": + return broker.resume() + if op == "status": + return broker.status() + if op == "list_asks": + return broker.list_asks() + if op == "resolve_ask": + return broker.resolve_ask(req.get("ask_id"), + req.get("decision")) + return {"ok": False, + "error": "未知操作：%r（白名单：open/snapshot/screenshot/" + "close/pause/resume/status/list_asks/resolve_ask/" + "ping）" % (op,)} + + +def _handle_conn(broker, conn): + """一个连接一请求一响应；回包先过泄露检查，不过即换通用错。""" + guard = _load_guard() + conn.settimeout(120) + data = b"" + while len(data) < MAX_REQUEST_BYTES: + chunk = conn.recv(4096) + if not chunk: + break + data += chunk + if data.endswith(b"\n"): + break + try: + req = json.loads(data.decode()) + except (ValueError, UnicodeDecodeError): + resp = {"ok": False, "error": "请求不是合法 JSON"} + else: + try: + resp = handle_request(broker, req) + except Exception as e: + resp = {"ok": False, + "error": guard.sanitize_error(e)} + if guard.response_leaks(resp): + resp = {"ok": False, "error": "响应泄露检查未通过（已拦截）"} + conn.sendall((json.dumps(resp, ensure_ascii=False) + "\n").encode()) + + +def serve(sock_path=CDP_SOCK, shots_dir=SHOTS_DIR, + blocklist_path=BLOCKLIST_PATH, ask_queue_dir=ASK_QUEUE_DIR, + playwright_factory=None): + """监听 Unix socket，逐连接处理。socket 0600。建不起来即抛错， + 不要静默带病跑。playwright_factory 只供测试注入假实现。""" + if os.path.exists(sock_path): + os.remove(sock_path) + os.makedirs(os.path.dirname(sock_path), exist_ok=True) + srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + srv.bind(sock_path) + srv.listen(16) + os.chmod(sock_path, 0o600) + print("chv-cdp-broker: 监听 %s" % sock_path, + file=sys.stderr, flush=True) + broker = CdpBroker(playwright_factory=playwright_factory, + shots_dir=shots_dir, + blocklist_path=blocklist_path, + ask_queue_dir=ask_queue_dir) + try: + while True: + try: + conn, _ = srv.accept() + except OSError: + continue + try: + _handle_conn(broker, conn) + except OSError as e: + print("chv-cdp-broker: 连接处理异常：%s" + % _load_guard().sanitize_error(e), + file=sys.stderr, flush=True) + finally: + try: + conn.close() + except OSError: + pass + finally: + broker.shutdown() + + +def main(argv): + sock_path = CDP_SOCK + shots_dir = SHOTS_DIR + blocklist_path = BLOCKLIST_PATH + ask_queue_dir = ASK_QUEUE_DIR + i = 0 + while i < len(argv): + if argv[i] == "--socket" and i + 1 < len(argv): + sock_path = argv[i + 1] + i += 2 + elif argv[i] == "--shots-dir" and i + 1 < len(argv): + shots_dir = argv[i + 1] + i += 2 + elif argv[i] == "--blocklist" and i + 1 < len(argv): + blocklist_path = argv[i + 1] + i += 2 + elif argv[i] == "--ask-queue" and i + 1 < len(argv): + ask_queue_dir = argv[i + 1] + i += 2 + else: + print("chv-cdp-broker: 未知参数：%s" % argv[i], + file=sys.stderr) + return 2 + serve(sock_path, shots_dir, blocklist_path, ask_queue_dir) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/chv-guest/chv_browser_guard.py b/chv-guest/chv_browser_guard.py new file mode 100644 index 0000000..195fc16 --- /dev/null +++ b/chv-guest/chv_browser_guard.py @@ -0,0 +1,386 @@ +#!/usr/bin/env python3 +"""chv_browser_guard：浏览器安全共享纯函数（客机侧）。 + +对齐 Muse P4（见 docs/chv-muse-p4-gap.md）：本模块只放纯函数， +不 import playwright、不起进程、不连网络，供 chv-cdp-broker.py +与 chv-browser.py 两脚本共用，避免三份拷贝漂移。 + +内容： + validate_url P2 的 URL 规则：只 http/https，拒 localhost/ + 私有地址字面量（DNS 重绑定是 Sentinel 的事）。 + check_blocklist 静态恶意站名单（browser-blocklist.txt）后缀匹配。 + truncate_a11y P2 的无障碍树截断（200 节点/64KB）。 + sanitize_error 错误信息消毒：去 HTML 标签、截断 200 字符、 + 仍含尖括号即只留异常类型名——快照路径上不许有 + 原始 DOM/HTML 外泄口子。 + detect_injection DOM 内提示注入启发式（中英文模式表，命中返标签； + 是规则，不是模型分类器）。 + detect_form_risk 无障碍树扫密码/支付字段。 + scan_download 下载扫描钩子：一条可执行扩展名规则。 + enqueue/list/resolve_ask 文件型待审批队列（与 P3 ApprovalQueue + 同语义：pending/allowed/denied）。 + LAUNCH_ARGS / FORBIDDEN_PAGE_ATTRS Chromium 启动参数白名单与 + 禁用的 Playwright 页面调用名表。 +""" +import json +import os +import re +import time +import uuid + +MAX_A11Y_NODES = 200 +MAX_A11Y_CHARS = 65536 +MAX_ERROR_CHARS = 200 + +# 拒绝的 URL 主机：localhost 与私有地址字面量（P2 原样）。 +_PRIVATE_HOST_RE = re.compile( + r"^(localhost|127\.\d+\.\d+\.\d+|10\.\d+\.\d+\.\d+|" + r"172\.(1[6-9]|2\d|3[01])\.\d+\.\d+|192\.168\.\d+\.\d+|" + r"\[::1\]|\[fe8[0-9a-f]*:.*\])$", + re.IGNORECASE) + +# Chromium 启动参数白名单：均为真实开关；刻意没有 +# --remote-debugging-port（DevTools 不暴露）与 --no-sandbox +#（浏览器以非 root 的 chv-browser 身份跑，不降级沙箱）。 +LAUNCH_ARGS = [ + "--disable-gpu", + "--no-first-run", + "--disable-dev-shm-usage", + "--disable-extensions", + "--disable-background-networking", + "--disable-sync", + "--disable-translate", + "--mute-audio", +] + +# 禁用的 Playwright 页面调用名：页面上下文 JS 与进程内 exec +# 的全部入口。broker 与子代理的代码里不得出现这些属性名， +# 测试用 AST 扫两个脚本钉住（字符串常量里的出现不算）。 +FORBIDDEN_PAGE_ATTRS = ( + "evaluate", "evaluate_handle", "add_script_tag", + "add_init_script", "expose_function", "set_content", + "keyboard", "mouse", "touchscreen", +) + +# 浏览器 socket 白名单之外的操作名（子代理层拒绝）。 +FORBIDDEN_OPS = ( + "evaluate", "content", "devtools", "js", "click", "fill", + "type", "press", "hover", "select", "check", "route", + "expose", "add_init_script", "add_script_tag", "set_content", + "eval", +) + +# 下载拦截：可执行扩展名（一条规则；后续规则只加条目）。 +BLOCKED_DOWNLOAD_EXTS = frozenset({ + ".exe", ".msi", ".bat", ".cmd", ".ps1", ".sh", ".dmg", + ".pkg", ".apk", ".scr", ".vbs", ".com", ".gadget", +}) + +# 提示注入启发式模式表（中英文各 6 条）：字面匹配，不是语义判断。 +# 命中只做标注（untrusted + injection_flags），不做拦截决定。 +INJECTION_PATTERNS = ( + ("en_ignore_instructions", r"ignore (all )?(previous|prior|above) instructions"), + ("en_you_are_now", r"you are now (a|an|the)"), + ("en_system_prompt", r"(reveal|show|print|disregard).{0,40}system prompt"), + ("en_disregard", r"disregard (all )?(previous|prior|above)"), + ("en_new_instructions", r"new instructions:? (you|to|do)"), + ("en_jailbreak", r"(jailbreak|DAN mode|do anything now)"), + ("zh_ignore", r"忽略(以上|之前|所有|先前)的?(指令|指示|提示)"), + ("zh_pretend", r"假装你是"), + ("zh_now_you_are", r"你现在是"), + ("zh_leak_prompt", r"(泄露|泄漏|输出|打印).{0,20}系统提示"), + ("zh_admin", r"以管理员(身份|权限)"), + ("zh_no_limit", r"(无视|绕过).{0,20}(限制|规则|审查)"), +) + +# 表单风险：无障碍树节点名/autocomplete 里的敏感词（小写匹配）。 +PASSWORD_HINTS = ("password", "密码", "current-password", + "new-password", "passwd", "pass word") +PAYMENT_HINTS = ("payment", "支付", "信用卡", "银行卡", "卡号", + "cc-number", "cc-exp", "cc-csc", "cc-name", + "card number", "cvv", "expiry", "有效期") + + +def validate_url(url): + """URL 校验：只允许 http/https，且主机不是 localhost/私有地址 + 字面量。返回 (ok, error)。""" + if not isinstance(url, str) or not url: + return False, "url 不能为空" + m = re.match(r"^(https?)://([^/]+)", url, re.IGNORECASE) + if not m: + return False, "只允许 http/https URL：%s" % url[:80] + host = m.group(2).split("@")[-1].split(":")[0].lower() + if _PRIVATE_HOST_RE.match(host): + return False, "拒绝 localhost/私有地址：%s" % host + return True, "" + + +def url_host(url): + """取 URL 的主机小写形式；取不到返回空串。""" + m = re.match(r"^(https?)://([^/]+)", url or "", re.IGNORECASE) + if not m: + return "" + return m.group(2).split("@")[-1].split(":")[0].lower() + + +def load_blocklist(path): + """读静态拦截名单：每行一个条目（域名或 .后缀），# 开头是注释， + 空行跳过。文件缺失即空名单（不静默编造条目）。""" + entries = set() + try: + with open(path) as fh: + for line in fh: + line = line.strip() + if not line or line.startswith("#"): + continue + entries.add(line.lower()) + except OSError: + return set() + return entries + + +def check_blocklist(host, entries): + """主机是否命中拦截名单：精确匹配或后缀匹配（条目 .example.com + 命中 a.example.com；条目 example.com 同样命中子域）。""" + host = (host or "").lower() + if not host or not entries: + return False + for entry in entries: + entry = entry.lstrip(".") + if host == entry or host.endswith("." + entry): + return True + return False + + +def truncate_a11y(node, max_nodes=MAX_A11Y_NODES, + max_chars=MAX_A11Y_CHARS): + """截断无障碍树：节点数超上限时剪掉多余子树。返回 (tree, truncated)。 + P2 原样搬入（两脚本共用，不再各存一份）。""" + state = {"count": 0, "truncated": False} + + def walk(n): + if not isinstance(n, dict): + return n + state["count"] += 1 + if state["count"] > max_nodes: + state["truncated"] = True + return None + out = {} + for key, value in n.items(): + if key == "children" and isinstance(value, list): + children = [] + for child in value: + walked = walk(child) + if walked is None: + state["truncated"] = True + break + children.append(walked) + out[key] = children + else: + out[key] = value + return out + + tree = walk(node) + text = json.dumps(tree, ensure_ascii=False) + if len(text) > max_chars: + state["truncated"] = True + tree = {"role": (node or {}).get("role"), + "name": (node or {}).get("name"), + "truncated_chars": True} + return tree, state["truncated"] + + +def sanitize_error(exc): + """错误信息消毒：只留异常类型名 + 去 HTML 标签、截断后的首行； + 凭据形内容（token=/password= 等 key=value 与 sk- 形 token）打码； + 清洗后仍含尖括号或 javascript: 即丢弃原文。绝不把页面原文 + 或凭据原样放进回包或日志。""" + name = type(exc).__name__ if not isinstance(exc, str) else "Error" + text = exc if isinstance(exc, str) else str(exc) + text = re.sub(r"<[^>]{0,500}>", "", text) + text = re.sub(r"(?i)(token|password|passwd|secret|api[-_]?key)" + r"\s*=\s*[^\s&;\"']+", r"\1=", text) + text = re.sub(r"sk-[A-Za-z0-9_\-]{8,}", "", text) + text = " ".join(text.split())[:MAX_ERROR_CHARS] + if "<" in text or ">" in text or "javascript:" in text.lower(): + return "浏览器操作失败（%s）" % name + if not text: + return "浏览器操作失败（%s）" % name + return "浏览器操作失败（%s）：%s" % (name, text) + + +def _collect_text(node, out): + """收集无障碍树里全部字符串值（name/value/description 等）， + 供注入启发式扫描。只收字符串，不收 role 名以外的结构。""" + if isinstance(node, dict): + for value in node.values(): + _collect_text(value, out) + elif isinstance(node, list): + for item in node: + _collect_text(item, out) + elif isinstance(node, str): + out.append(node) + + +def detect_injection(tree): + """DOM 内提示注入启发式：扫描快照全部文本，返回命中的模式标签 + 列表（去重、按模式表顺序）。无命中即空列表。规则而已，不是 + 模型分类器——命中只标注，不拦截。""" + texts = [] + _collect_text(tree or {}, texts) + blob = "\n".join(texts) + hits = [] + for label, pattern in INJECTION_PATTERNS: + if re.search(pattern, blob, re.IGNORECASE): + hits.append(label) + return hits + + +def _node_hints(node, hints): + """一个无障碍树节点的 name/autocomplete/type 里是否含敏感词。""" + blob = " ".join(str(node.get(k, "")) for k in + ("name", "autocomplete", "type", "description")) + blob = blob.lower() + return any(h in blob for h in hints) + + +def _walk_form_risk(node, found): + if not isinstance(node, dict): + return + role = str(node.get("role", "")).lower() + if role in ("textbox", "textfield", "combobox"): + if _node_hints(node, PASSWORD_HINTS): + found.setdefault("password", []).append( + str(node.get("name", ""))[:40]) + elif _node_hints(node, PAYMENT_HINTS): + found.setdefault("payment", []).append( + str(node.get("name", ""))[:40]) + for child in node.get("children") or []: + _walk_form_risk(child, found) + + +def detect_form_risk(tree): + """扫无障碍树里的密码/支付字段。返回 {level, fields}：level 是 + none/password/payment（支付优先于密码），fields 是命中的字段名 + （最多 5 个，供审批记录看，不进模型上下文原样只作标注）。""" + found = {} + _walk_form_risk(tree or {}, found) + if found.get("payment"): + return {"level": "payment", + "fields": found["payment"][:5]} + if found.get("password"): + return {"level": "password", + "fields": found["password"][:5]} + return {"level": "none", "fields": []} + + +def scan_download(filename): + """下载扫描钩子（一条规则）：可执行扩展名、或中间段含可执行 + 扩展名（双扩展名把戏，如 report.pdf.exe / archive.exe.pdf）， + 即拦。返回 (blocked, reason)。""" + name = (filename or "").strip().lower() + if not name: + return False, "" + parts = [p for p in name.split(".") if p] + if len(parts) < 2: + return False, "" + final_ext = "." + parts[-1] + if final_ext in BLOCKED_DOWNLOAD_EXTS: + return True, "可执行扩展名：%s" % final_ext + for middle in parts[1:-1]: + if ("." + middle) in BLOCKED_DOWNLOAD_EXTS: + return True, "双扩展名可疑：%s" % filename[:60] + return False, "" + + +def check_launch_args(args): + """Chromium 启动参数校验：不得含 --remote-debugging-*（DevTools + 不暴露），不得含 devtools 字串。返回 (ok, error)。""" + for arg in args or []: + if arg.startswith("--remote-debugging"): + return False, "启动参数不得含 %s" % arg + if "devtools" in arg.lower(): + return False, "启动参数不得含 DevTools：%s" % arg + return True, "" + + +def _ask_path(queue_dir, ask_id): + return os.path.join(queue_dir, "ask-%s.json" % ask_id) + + +def enqueue_ask(queue_dir, kind, url, detail): + """待审批入队：一条 ask 一个 JSON 文件（原子写：先写 tmp 再 + rename）。返回 ask_id。队列目录不存在即建。""" + os.makedirs(queue_dir, exist_ok=True) + ask_id = uuid.uuid4().hex[:12] + record = {"ask_id": ask_id, "kind": kind, "url": url or "", + "detail": detail or "", "status": "pending", + "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", + time.gmtime())} + tmp_path = _ask_path(queue_dir, ask_id) + ".tmp" + with open(tmp_path, "w") as fh: + json.dump(record, fh, ensure_ascii=False) + fh.write("\n") + os.replace(tmp_path, _ask_path(queue_dir, ask_id)) + return ask_id + + +def list_asks(queue_dir): + """列出待审批（status pending）的 ask，按创建时间排序。""" + out = [] + try: + names = sorted(os.listdir(queue_dir)) + except OSError: + return [] + for name in names: + if not (name.startswith("ask-") and name.endswith(".json")): + continue + try: + with open(os.path.join(queue_dir, name)) as fh: + record = json.load(fh) + except (OSError, ValueError): + continue + if record.get("status") == "pending": + out.append(record) + out.sort(key=lambda r: r.get("created_at", "")) + return out + + +def resolve_ask(queue_dir, ask_id, decision): + """写回审批决定：decision 只认 allowed/denied。返回 (ok, error)。 + 不存在的 ask_id 即报错，不静默成功。""" + if decision not in ("allowed", "denied"): + return False, "decision 只认 allowed/denied：%s" % decision + path = _ask_path(queue_dir, ask_id) + try: + with open(path) as fh: + record = json.load(fh) + except (OSError, ValueError): + return False, "ask 不存在：%s" % ask_id + if record.get("status") != "pending": + return False, "ask 已处理：%s" % ask_id + record["status"] = decision + tmp_path = path + ".tmp" + with open(tmp_path, "w") as fh: + json.dump(record, fh, ensure_ascii=False) + fh.write("\n") + os.replace(tmp_path, path) + return True, "" + + +def response_leaks(resp): + """回包泄露检查（测试与 broker 自检共用）：递归扫响应里的字符串， + 命中 ws://、DevTools 字串或 HTML 标签即返回问题列表。空列表即干净。""" + problems = [] + text = json.dumps(resp, ensure_ascii=False) + if "ws://" in text or "wss://" in text: + problems.append("响应含 ws 端点") + # 只查 CDP 形态（devtools/ 路径、remote-debugging），不查裸词—— + # 正常 URL 里也可能含 devtools（如 devtools.example.com）。 + if re.search(r"devtools/", text, re.IGNORECASE): + problems.append("响应含 DevTools 路径") + if "remote-debugging" in text.lower(): + problems.append("响应含 remote-debugging") + if re.search(r"<[a-zA-Z][^>]{0,200}>", text): + problems.append("响应含 HTML 标签") + return problems diff --git a/cloud/tests/test_chv_muse_p2.py b/cloud/tests/test_chv_muse_p2.py index 0beeaa3..e67f051 100644 --- a/cloud/tests/test_chv_muse_p2.py +++ b/cloud/tests/test_chv_muse_p2.py @@ -2,7 +2,7 @@ 只测本轨交付物（chv-guest/chv-cell.py、chv-broker.py、chv-browser.py、 build.sh 干跑、buildinfo.py 内容项）。全部用假 socket/假进程/假 -playwright，绝不连真实网络、不起真实 VM、不改生产行为。 +CDP broker，绝不连真实网络、不起真实 VM、不改生产行为。 """ import importlib.util import json @@ -33,6 +33,7 @@ def _load_script(name): CELL = _load_script("chv-cell.py") BROKER = _load_script("chv-broker.py") BROWSER = _load_script("chv-browser.py") +GUARD = _load_script("chv_browser_guard.py") BUILDINFO = _load_script("buildinfo.py") @@ -329,21 +330,21 @@ class BrokerTest(unittest.TestCase): class BrowserTest(unittest.TestCase): def test_validate_url(self): - self.assertTrue(BROWSER.validate_url("https://example.com")[0]) - self.assertTrue(BROWSER.validate_url("http://example.com/x")[0]) + self.assertTrue(GUARD.validate_url("https://example.com")[0]) + self.assertTrue(GUARD.validate_url("http://example.com/x")[0]) self.assertTrue( - BROWSER.validate_url("https://user@example.com/")[0]) + GUARD.validate_url("https://user@example.com/")[0]) for bad in ("ftp://example.com", "file:///etc/passwd", "https://localhost:8080/", "http://127.0.0.1/", "https://192.168.1.1/", "https://10.0.0.5/", "https://172.17.0.2/", "", None): - ok, _ = BROWSER.validate_url(bad) + ok, _ = GUARD.validate_url(bad) self.assertFalse(ok, bad) def test_truncate_a11y_small_tree_untouched(self): tree = {"role": "WebArea", "name": "Example", "children": [{"role": "heading", "name": "Hi"}]} - out, truncated = BROWSER.truncate_a11y(tree) + out, truncated = GUARD.truncate_a11y(tree) self.assertFalse(truncated) self.assertEqual(out, tree) @@ -351,13 +352,13 @@ class BrowserTest(unittest.TestCase): tree = {"role": "WebArea", "children": [{"role": "link", "name": "l%d" % i} for i in range(300)]} - out, truncated = BROWSER.truncate_a11y(tree) + out, truncated = GUARD.truncate_a11y(tree) self.assertTrue(truncated) self.assertLessEqual(len(out["children"]), 200) def test_truncate_a11y_char_cap(self): tree = {"role": "WebArea", "name": "x" * 70000, "children": []} - out, truncated = BROWSER.truncate_a11y(tree) + out, truncated = GUARD.truncate_a11y(tree) self.assertTrue(truncated) self.assertTrue(out.get("truncated_chars")) @@ -371,28 +372,26 @@ class BrowserTest(unittest.TestCase): resp = BROWSER.handle_request(agent, {"op": "bogus"}) self.assertFalse(resp["ok"]) - def test_open_rejects_bad_url_without_loading_playwright(self): - def exploding_factory(): - raise AssertionError("坏 URL 不应加载 playwright") + def test_open_rejects_bad_url_without_calling_broker(self): + def exploding_broker(req): + raise AssertionError("坏 URL 不应调 broker") - agent = BROWSER.BrowserAgent(playwright_factory=exploding_factory) + agent = BROWSER.BrowserAgent(broker_call=exploding_broker) resp = agent.open("https://127.0.0.1/secret") self.assertFalse(resp["ok"]) def test_snapshot_and_screenshot_need_open_first(self): - agent = BROWSER.BrowserAgent( - playwright_factory=lambda: (_ for _ in ()).throw( - AssertionError("不应加载 playwright"))) - self.assertFalse(agent.snapshot()["ok"]) - self.assertFalse(agent.screenshot()["ok"]) - self.assertTrue(agent.close()["ok"]) - - def test_open_returns_a11y_snapshot_with_fake_playwright(self): - page = FakePage() with tempfile.TemporaryDirectory() as tmp: agent = BROWSER.BrowserAgent( - playwright_factory=lambda: FakePlaywright(page), - shots_dir=tmp) + broker_call=FakeCdpBroker(tmp)) + self.assertFalse(agent.snapshot()["ok"]) + self.assertFalse(agent.screenshot()["ok"]) + self.assertTrue(agent.close()["ok"]) + + def test_open_returns_a11y_snapshot_with_fake_cdp_broker(self): + with tempfile.TemporaryDirectory() as tmp: + agent = BROWSER.BrowserAgent( + broker_call=FakeCdpBroker(tmp)) resp = agent.open("https://example.com") self.assertTrue(resp["ok"]) self.assertEqual(resp["title"], "Example Domain") @@ -407,62 +406,43 @@ class BrowserTest(unittest.TestCase): self.assertTrue(agent.close()["ok"]) -class FakePage: - def __init__(self): - self.url = "https://example.com/" - self._closed = False +class FakeCdpBroker: + """假 CDP broker：收到 req dict 返回 resp dict；open 后才有页。 - def goto(self, url, timeout=None): - self.url = url.rstrip("/") + "/" + round-12：P4 起 BrowserAgent 不再直连 Playwright，Chromium + 的所有权在单元外 CDP broker 手里；本测试用假 broker 走同一 + 协议，断言与 P2 原版一致。""" - def wait_for_load_state(self, state, timeout=None): - pass + def __init__(self, shots_dir): + self._shots_dir = shots_dir + self._open = False + self.calls = [] - def title(self): - return "Example Domain" - - @property - def accessibility(self): - return self - - def snapshot(self): - return {"role": "WebArea", "name": "Example Domain", + def __call__(self, req): + self.calls.append(req.get("op")) + op = req.get("op") + tree = {"role": "WebArea", "name": "Example Domain", "children": [{"role": "heading", "name": "Example"}]} - - def screenshot(self, path=None): - with open(path, "wb") as fh: - fh.write(b"\x89PNG") - - def close(self): - self._closed = True - - -class FakeBrowser: - def __init__(self, page): - self._page = page - - def new_page(self): - return self._page - - def close(self): - pass - - -class FakeEntered: - def __init__(self, page): - self.chromium = mock.Mock() - self.chromium.launch.return_value = FakeBrowser(page) - - -class FakePlaywright: - def __init__(self, page): - self._page = page - - def __enter__(self): - return FakeEntered(self._page) - - def __exit__(self, *exc): - return False + if op == "open": + self._open = True + return {"ok": True, "title": "Example Domain", + "url": "https://example.com/", "a11y": tree} + if op == "snapshot": + if not self._open: + return {"ok": False, "error": "尚未 open"} + return {"ok": True, "title": "Example Domain", + "url": "https://example.com/", "a11y": tree} + if op == "screenshot": + if not self._open: + return {"ok": False, "error": "尚未 open"} + path = os.path.join(self._shots_dir, "shot.png") + with open(path, "wb") as fh: + fh.write(b"\x89PNG") + return {"ok": True, "path": path} + if op == "close": + self._open = False + return {"ok": True} + return {"ok": False, "error": "未知操作"} class CellCliTest(unittest.TestCase): @@ -570,7 +550,9 @@ class BuildDryRunTest(unittest.TestCase): "generate_lockfile.py", "chv-sync.sh", "chv-cell.py", "chv-broker.py", "chv-browser.py", "authd.py", "git-credential-chv-authd", - "chv-authd.service"): + "chv-authd.service", + "chv-cdp-broker.py", "chv_browser_guard.py", + "browser-blocklist.txt"): shutil.copy2(ROOT / "chv-guest" / name, os.path.join(guest, name)) shutil.copy2(ROOT / "host.py", os.path.join(repo, "host.py")) diff --git a/cloud/tests/test_chv_muse_p4.py b/cloud/tests/test_chv_muse_p4.py new file mode 100644 index 0000000..1cfe873 --- /dev/null +++ b/cloud/tests/test_chv_muse_p4.py @@ -0,0 +1,777 @@ +"""对齐 Muse P4 测试：浏览器完全对齐。 + +只测本轨交付物（chv-guest/chv_browser_guard.py、 +chv-cdp-broker.py、chv-browser.py、browser-blocklist.txt、 +build.sh 干跑、buildinfo.py 内容项）。全部用假 socket/假进程/ +假 playwright，绝不连真实网络、不起真实 VM、不改生产行为。 +""" +import ast +import importlib.util +import json +import os +import shutil +import socket +import subprocess +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from unittest import mock + +ROOT = Path(__file__).resolve().parents[2] + + +def _load_script(name): + """把 chv-guest 下的独立脚本当模块加载（无包结构）。""" + path = ROOT / "chv-guest" / name + spec = importlib.util.spec_from_file_location( + "chv_guest_p4_" + name[:-3].replace("-", "_"), str(path)) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +GUARD = _load_script("chv_browser_guard.py") +CDP = _load_script("chv-cdp-broker.py") +BROWSER = _load_script("chv-browser.py") +BUILDINFO = _load_script("buildinfo.py") + + +class GuardTest(unittest.TestCase): + def test_validate_url(self): + self.assertTrue(GUARD.validate_url("https://example.com")[0]) + self.assertTrue(GUARD.validate_url("http://example.com/x")[0]) + for bad in ("ftp://example.com", "file:///etc/passwd", + "https://localhost:8080/", "http://127.0.0.1/", + "https://192.168.1.1/", "https://10.0.0.5/", + "https://172.17.0.2/", "", None): + ok, _ = GUARD.validate_url(bad) + self.assertFalse(ok, bad) + + def test_blocklist_load_and_match(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, "blocklist.txt") + with open(path, "w") as fh: + fh.write("# comment\n\nmalware.test\n.phishing.test\n") + entries = GUARD.load_blocklist(path) + self.assertEqual(entries, {"malware.test", + ".phishing.test"}) + self.assertTrue(GUARD.check_blocklist("malware.test", + entries)) + self.assertTrue(GUARD.check_blocklist("a.malware.test", + entries)) + self.assertTrue(GUARD.check_blocklist("x.phishing.test", + entries)) + self.assertFalse(GUARD.check_blocklist("example.com", + entries)) + self.assertFalse(GUARD.check_blocklist("notmalware.test", + entries)) + self.assertEqual(GUARD.load_blocklist("/nonexistent/x"), set()) + + def test_truncate_a11y_small_tree_untouched(self): + tree = {"role": "WebArea", "name": "Example", + "children": [{"role": "heading", "name": "Hi"}]} + out, truncated = GUARD.truncate_a11y(tree) + self.assertFalse(truncated) + self.assertEqual(out, tree) + + def test_truncate_a11y_char_cap(self): + tree = {"role": "WebArea", "name": "x" * 70000, "children": []} + out, truncated = GUARD.truncate_a11y(tree) + self.assertTrue(truncated) + self.assertTrue(out.get("truncated_chars")) + + def test_sanitize_error_strips_html_and_secrets(self): + err = RuntimeError( + "timeout token=sk-secret-abc") + clean = GUARD.sanitize_error(err) + self.assertIn("RuntimeError", clean) + self.assertNotIn(" sk-secret-abc") + resp = broker.open("https://example.com") + self.assertFalse(resp["ok"]) + self.assertNotIn("