From ad77092ce9292b37e6c4abbed74cb33bc590e0bb Mon Sep 17 00:00:00 2001 From: tizerluo <192086140+tizerluo@users.noreply.github.com> Date: Sun, 20 Sep 2026 02:00:15 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20=E5=AE=A1=E6=9F=A5=E5=8F=AA=E8=AF=BB?= =?UTF-8?q?=E6=8A=A4=E6=A0=8F=20P0=20=E4=B8=89=E8=BF=9E=E4=BF=AE=20(issue?= =?UTF-8?q?=20#49)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 洞一: --disallowed-tools 黑名单补动态工作流族 (0.16.9 新增的 EvalWorkflowSnippet/CreateWorkflow/AmendWorkflow/SaveWorkflow/ ListWorkflowRuns/GetWorkflowRun/ResumeWorkflowRun/ListSavedWorkflows/ ResolveWorkflowQuestion) — 漏禁任一即等于放行任意命令执行与文件写入 (world.run / artifact 写)。 洞三: 新增项目配置护栏 fail-closed。被审目录链 (cwd 到 git 根) 存在 zcode.json / .zcode/config.json 时直接拒审 (zcode 启动时会 spawn 其中 mcp.servers 定义的命令, 不经模型不受黑名单约束)。拒审文案首行带 ZOB-REFUSED: project_config 机器可读标记; ZCODE_BRIDGE_TRUST_PROJECT_CONFIG=1 可显式放行可信仓库。 配套: 审查子进程 --cwd 改为隔离临时目录 (内建空 .git 阻断向上发现), 被审仓库根目录改由 prompt 以绝对路径提供; review-gate 识别 ZOB-REFUSED 标记, 拒审按终态处理 (gave_up + 🚫 告警评论), 不再当临时失败退避重试 (重试无意义且 5 次后静默 gave_up 会吞掉投毒信号)。 测试: PG1-PG10 (护栏检测/链上溯/拒审/放行/标记在 500 字符截断窗内/ 拒审不起子进程) + RF1-RF4 + 拒审终态/评论缓存 e2e + 工作流族黑名单。 537 tests pass; 三场景实测 (陷阱目录拒审且陷阱未执行 / TRUST=1 陷阱 仍不执行 / 干净目录正常出报告)。 --- packages/mcp-server/zcode-mcp-server | 143 ++++++++++++++++-- packages/review-gate/zcode-review-gate | 61 ++++++++ tests/test_review_gate.py | 71 +++++++++ tests/test_security_review.py | 192 ++++++++++++++++++++++++- 4 files changed, 457 insertions(+), 10 deletions(-) diff --git a/packages/mcp-server/zcode-mcp-server b/packages/mcp-server/zcode-mcp-server index fc34355..ca654ac 100755 --- a/packages/mcp-server/zcode-mcp-server +++ b/packages/mcp-server/zcode-mcp-server @@ -27,6 +27,7 @@ import select import shutil import subprocess import sys +import tempfile import time from datetime import datetime, timedelta, timezone from pathlib import Path @@ -321,12 +322,23 @@ class ReviewFileLock: # 而是直接报 "No permission client configured" 失败; # - --allowed-tools / --max-turns 帮助文案有, 但未注册 parseArgs (实测 # Unknown option), 不可用; auto 模式是保留值未实现。 +# App 3.14.0 / CLI 0.16.9 补充 (2026-09-19 实测, 与 Node REPL 同族问题): +# - 动态工作流工具族 (EvalWorkflowSnippet/CreateWorkflow/...) 是新的执行原语: +# snippet 里的 world.run 可跑任意命令, 不在此黑名单时 Bash 禁令形同虚设 +# (实测: 黑名单原样下 world.run("node", [...fs.writeFileSync...]) 成功写入 +# 文件, 平台自身遥测把 EvalWorkflowSnippet 标为 read_only=0); +# - 黑名单按工具名精确匹配 (Write/Bash 被挡住即证), 故按名逐个列; 工具面随 +# 版本增长, 新版本上线后需重新核对下发工具清单 (见 docs/recheck-3.14.0.md)。 # 组合: --mode yolo (免授权) + 全量写/执行黑名单 (物理只读, 读工具全开)。 # ============================================================ REVIEW_DISALLOWED_TOOLS = ( "Write Edit MultiEdit ApplyPatch Bash " "js js_reset js_add_node_module_dir " - "mcp__node_repl__js mcp__node_repl__js_reset mcp__node_repl__js_add_node_module_dir" + "mcp__node_repl__js mcp__node_repl__js_reset mcp__node_repl__js_add_node_module_dir " + # 动态工作流一族 (0.16.9): 漏禁即等于放行任意命令执行与文件写入 + "EvalWorkflowSnippet CreateWorkflow AmendWorkflow SaveWorkflow " + "ListWorkflowRuns GetWorkflowRun ResumeWorkflowRun " + "ListSavedWorkflows ResolveWorkflowQuestion" ) # deep 轮询: status 单次响应超时的最大连续容忍次数 (复审 P1-3: @@ -936,7 +948,9 @@ TOOLS = [ }, "cwd": { "type": "string", - "description": "工作目录 (影响 zcode 的项目上下文),默认当前目录", + "description": "被审仓库根目录 (经 prompt 告知审查模型, 供其用绝对路径" + "读上下文),默认当前目录。注: zcode 子进程的 --cwd 恒为" + "隔离沙箱 (防被审目录的项目配置触发命令执行, issue #49)", }, }, "additionalProperties": False, @@ -985,7 +999,8 @@ TOOLS = [ }, "cwd": { "type": "string", - "description": "zcode 工作目录,默认与 path 相同", + "description": "被审仓库根目录 (经 prompt 告知审查模型),默认与 path 相同。" + "注: zcode 子进程的 --cwd 恒为隔离沙箱 (issue #49)", }, }, "additionalProperties": False, @@ -1039,7 +1054,8 @@ TOOLS = [ }, "cwd": { "type": "string", - "description": "zcode 工作目录,默认与 path 相同", + "description": "被审仓库根目录 (经 prompt 告知审查模型),默认与 path 相同。" + "注: zcode 子进程的 --cwd 恒为隔离沙箱 (issue #49)", }, }, "additionalProperties": False, @@ -1103,14 +1119,103 @@ def _extract_response(output): return output, None -def _build_review_cmd(prompt, cwd): - """构造只读 review 的 zcode 命令 (yolo 免授权 + 写/执行工具物理禁用 + JSON 输出)。""" +def _env_truthy(name): + """读布尔型环境开关: 非空且非 0/false/no 视为开。""" + v = (os.environ.get(name) or "").strip().lower() + return v not in ("", "0", "false", "no") + + +# ============================================================ +# 项目配置防护 (2026-09-19, issue #49 洞三) +# ------------------------------------------------------------ +# 被审目录里的 zcode.json / .zcode/config.json 会被 zcode 启动时装载, 其中 +# mcp.servers 定义的工具**进程会被直接 spawn** — 不经过模型、不经过工具黑名单 +# (实测: 黑名单全量在位的 review 调用下, 该文件里的命令照常执行)。发现范围是 +# 从工作目录向上直到 git 根 (实测逆向 zcode bundle 的发现顺序), 所以只清仓库 +# 根不够, 要沿链看。 +# +# 对策两层: +# ① 主门锁 — 审查子进程的 --cwd 改指一个只含空 .git 的临时目录 (发现链就 +# 此终止, 看不到被审仓库); 被审仓库路径改由 prompt 提供。4 样本实测 +# (2 注入真 bug + 2 真实 diff) 检出能力与读上下文量均不降 (详见 +# docs/recheck-3.14.0.md)。 +# ② 报警器 — 本函数: 被审目录链上存在这两个文件时 fail-closed 拒审, 既是 +# 拦截也是"有人在试探"的信号。确信任用时可用环境变量放行。 +# ============================================================ +_PROJECT_CONFIG_RELS = ("zcode.json", os.path.join(".zcode", "config.json")) +_TRUST_PROJECT_CONFIG_ENV = "ZCODE_BRIDGE_TRUST_PROJECT_CONFIG" + +_scratch_cwd_cache = None + + +def _review_scratch_cwd(): + """审查用的干净工作目录 (只含一个空 .git, 令配置发现链止步于此)。""" + global _scratch_cwd_cache + if _scratch_cwd_cache and os.path.isdir(_scratch_cwd_cache): + return _scratch_cwd_cache + d = tempfile.mkdtemp(prefix="zcode-review-cwd-") + os.mkdir(os.path.join(d, ".git")) # 空目录即可: zcode 见到 .git 就停止上溯 + _scratch_cwd_cache = d + return d + + +def _project_config_files(root): + """沿 root → git 根 (含) 列出存在的 zcode 项目配置文件 (与 zcode 发现范围一致)。""" + found = [] + cur = os.path.abspath(root) + while True: + for rel in _PROJECT_CONFIG_RELS: + p = os.path.join(cur, rel) + if os.path.exists(p): + found.append(p) + if os.path.exists(os.path.join(cur, ".git")): + break + parent = os.path.dirname(cur) + if parent == cur: + break + cur = parent + return found + + +def _project_config_guard(root): + """被审目录链上有 zcode 项目配置文件则返回拒绝 result, 否则 None。""" + if _env_truthy(_TRUST_PROJECT_CONFIG_ENV): + return None + bad = _project_config_files(root) + if not bad: + return None + log(f"⛔ 拒审: 被审目录链上存在 zcode 项目配置 {bad}") + return {"content": [{"type": "text", "text": ( + # 机器可读标记必须在首行: review-gate 只取失败文本前 500 字符 + # (run_review 的 detail 截断), 放末尾会被长路径/多文件顶出窗口, + # 导致拒审被误判为临时失败而退避重试 (告警随之被吞)。 + "ZOB-REFUSED: project_config\n" + "拒绝审查: 被审目录 (或其上层路径) 存在 ZCode 项目配置文件:\n" + + "\n".join(f" - {p}" for p in bad) + + "\n\n原因: zcode 启动时会装载该文件的 mcp.servers 并**直接执行**其中定义的命令" + " (进程 spawn, 不经过模型也不受工具黑名单约束) — 审查不受信内容时等于把" + "执行权交给被审内容 (issue #49 洞三)。\n" + "处理: ① 审查他人代码/PR: 请在副本中删除这两个文件后再审 (review-gate 应在" + "克隆副本里做); ② 确认该目录可信 (自己的仓库且确有项目配置): 设 " + f"{_TRUST_PROJECT_CONFIG_ENV}=1 放行。\n" + "注意: 沿工作目录到 git 根的整条链都会被 zcode 检查, 不只仓库根。" + )}], "isError": True} + + +def _build_review_cmd(prompt, repo): + """构造只读 review 的 zcode 命令 (yolo 免授权 + 写/执行工具物理禁用 + JSON 输出)。 + + --cwd 固定指向干净临时目录 (项目配置防护①): 被审仓库的真实路径由调用方写进 + prompt ("被审仓库根目录: "), 审查模型用绝对路径读上下文 — 4 样本实测 + 与 --cwd 指向仓库等价 (读取类工具调用量不降, 注入 bug 检出率相同)。 + """ cmd = [ZCODE_BIN, "--prompt", prompt, "--mode", "yolo", "--disallowed-tools", REVIEW_DISALLOWED_TOOLS, - "--no-color", "--json"] - if cwd: - cmd += ["--cwd", cwd] + "--no-color", "--json", + "--cwd", _review_scratch_cwd()] + if repo: + log(f"审查工作目录: {_review_scratch_cwd()} (被审仓库 {repo} 经 prompt 提供)") return cmd @@ -1280,10 +1385,17 @@ def tool_zcode_review(args): f"\n\n[... code 超 {code_max} 字节已截断; " "超大代码建议改用 files 传文件路径 ...]") + # 项目配置防护②: 被审目录链上有 zcode 项目配置 → fail-closed (issue #49 洞三) + guard = _project_config_guard(cwd or os.getcwd()) + if guard: + return guard + # 构造 prompt — "只审不修"职责钉死在文本层, 写工具在工具集层物理禁用, # 双保险防 "审着审着顺手改了" (plan→build 惯性, 2026-08-08 重构动因)。 prompt = ( f"你是资深代码审查专家。请审查附件中的代码,重点: {focus}。\n" + f"被审仓库根目录: {os.path.abspath(cwd or os.getcwd())}" + " (用绝对路径 + 只读工具读上下文; 当前工作目录是隔离沙箱, 不是你审查的仓库)\n" "规则:\n" "1. 你的任务只是审查并输出报告 — 绝对不要修改、创建或删除任何文件" " (写/执行工具已被物理禁用,不要尝试)。\n" @@ -1357,6 +1469,10 @@ def tool_zcode_security_review(args): return {"content": [{"type": "text", "text": f"扫描目录不存在或不是目录: {scan_path}"}], "isError": True} + # 项目配置防护②: 被审目录链上有 zcode 项目配置 → fail-closed (issue #49 洞三) + guard = _project_config_guard(scan_path) + if guard: + return guard root = _find_mimosa_root() if not root: @@ -1395,6 +1511,8 @@ def tool_zcode_security_review(args): prompt = ( f"你是资深安全审查专家。附件是安全扫描器 (mimosa,确定性规则引擎,depth={depth})" "对本项目的扫描结果。\n" + f"被审仓库根目录: {os.path.abspath(scan_path)}" + " (用绝对路径 + 只读工具读上下文; 当前工作目录是隔离沙箱, 不是你审查的仓库)\n" "任务:\n" "1. 逐条核实每个 finding: 用只读工具 (Read/Grep/Glob) 阅读相关代码,判定为" "【确认漏洞】/【误报】/【存疑】,给出依据 (可利用的攻击路径,或为什么不可利用)。\n" @@ -1446,6 +1564,11 @@ def tool_zcode_pr_review(args): return {"content": [{"type": "text", "text": f"仓库目录不存在或不是目录: {scan_path}"}], "isError": True} + # 项目配置防护②: 被审目录链上有 zcode 项目配置 → fail-closed (issue #49 洞三)。 + # PR 审查场景最危险: path 通常是刚 check out 的 PR head 副本, 内容不受信。 + guard = _project_config_guard(scan_path) + if guard: + return guard # ① git diff (只读) rc, _, err = _git(scan_path, "rev-parse", "--git-dir") @@ -1611,6 +1734,8 @@ def tool_zcode_pr_review(args): "你是资深安全审查专家。这是一次 PR 审查: 附件含 ① 本 PR 的完整 diff " f"(base={base} → {head}, 改动 {len(changed)} 个文件) ② mimosa 安全扫描结果" f"(depth={depth}, 业务逻辑复核优先聚焦本次改动文件{findings_scope_note})。\n" + f"被审仓库根目录: {os.path.abspath(scan_path)}" + " (用绝对路径 + 只读工具读上下文; 当前工作目录是隔离沙箱, 不是你审查的仓库)\n" "任务:\n" "1. 审查 diff 中新引入的问题, 按 P0(阻断合并)/P1(应修)/P2(建议)/P3(提示) 分级," "每个问题含文件位置/判定依据/修复建议。重点是'新引入';" diff --git a/packages/review-gate/zcode-review-gate b/packages/review-gate/zcode-review-gate index 8d7df65..564ec4e 100755 --- a/packages/review-gate/zcode-review-gate +++ b/packages/review-gate/zcode-review-gate @@ -734,6 +734,39 @@ def parse_verdict_marker(report): m.group(5) == "true") +# 拒审标记 (issue #49 洞三): mcp-server 的项目配置护栏在拒审文案末尾带此行。 +# review-gate 据此把"安全策略拒绝"与"临时失败"分开 — 前者终态 + 告警评论 +# (重试不会变, 且 5 次后静默 gave_up 会吞掉"有人在投毒"的信号), 后者照旧退避。 +# 只在审查失败 (ok=False) 的载荷上解析: 该文案由 mcp-server 生成, 非模型输出, +# 无伪造面 (审查报告正文不参与判定)。 +_RE_ZOB_REFUSED = re.compile(r"^ZOB-REFUSED:\s*([A-Za-z0-9_]+)\s*$", re.M) + + +def parse_refusal_marker(text): + """读 ZOB-REFUSED 标记, 返回原因串 (如 "project_config") 或 None。""" + matches = list(_RE_ZOB_REFUSED.finditer(text or "")) + return matches[-1].group(1) if matches else None + + +def build_refusal_comment(head_sha, detail): + """拒审告警评论体 (供人工立即看见; 不参与 verdict 解析)。""" + return ( + "## 🚫 审查被拒绝（安全策略）\n\n" + "被审分支的仓库里存在 ZCode 项目配置文件（`zcode.json` 或 " + "`.zcode/config.json`）。该文件会在审查进程启动时被装载，其中 " + "`mcp.servers` 定义的命令会被**直接执行**（进程 spawn，不经过模型、" + "不受工具黑名单约束）——等于把执行权交给被审内容；因此本次审查按 " + "fail-closed 策略拒绝（zcode-open-bridge issue #49 洞三）。\n\n" + f"- head: `{(head_sha or '')[:12]}`\n" + "- 同 head 不再重试；推送新 commit 会自动重新触发审查。\n" + "- **若这是本 PR 新增的文件**：请从 PR 中移除后重推。\n" + "- **若该文件本就在目标分支上**：属仓库自身配置，需人工确认后处理" + "（可设 `ZCODE_BRIDGE_TRUST_PROJECT_CONFIG=1` 放行，或在审查副本中移除）。\n\n" + "
护栏原始输出\n\n" + f"```\n{(detail or '')[:1500]}\n```\n\n
" + ) + + def parse_severity_counts(report): """解析 (P0, P1, P2, P3) 条数; P0/P1/P2 任一缺失 → None, P3 缺省计 0。 @@ -1368,6 +1401,34 @@ def process_pr(cfg, state, token, owner, repo, clone, info): f"base=origin/{info['base_ref']} depth={cfg.review_depth}") ok, payload, rinfo = run_review(cfg, clone, info["base_ref"], head_sha) if not ok: + # 拒审 (issue #49 洞三): 安全策略拒绝 → 终态 + 告警评论。 + # 不能当临时失败退避重试: 重试结果不会变, 且 5 次后静默 gave_up + # 会把"有人在投毒"的信号吞掉 (mcp-server 侧带 ZOB-REFUSED 标记)。 + # 复用 gave_up 语义: 同 head 不再动, 新 head 自动复活重审。 + refused = parse_refusal_marker(payload) + if refused: + entry["error"] = payload[:500] + entry["verdict"] = None + entry["status"] = "gave_up" + entry["next_retry_at"] = 0.0 + log(f"{key}: ⛔ 拒审 (ZOB-REFUSED: {refused}), 终态 " + f"(head={head_sha[:12]}) — 发告警评论") + if cfg.comment_enabled and token: + try: + resp = post_comment( + token, cfg.github_api, owner, repo, info["number"], + build_refusal_comment(head_sha, payload)) + log(f"{key}: 拒审告警已发布 " + f"{(resp or {}).get('html_url') or ''}") + except (RetryableError, RepoHardError) as e: + # 告警发不出去也不能丢: 缓存拒审原文, 下轮只补评论 + # (与审查成功但评论失败同一条缓存补发路径) + entry["report"] = payload[:cfg.comment_max_body] + mark_failure(entry, f"拒审告警发布失败: {e}", now, cfg, + status="comment_failed") + else: + log(f"{key}: 拒审告警未发 (评论禁用或无 token), 详见 state/日志") + return # 判据优先级 (#35①): 结构化 quota_limit 键优先 (mcp-server 侧已转 # unix 秒), 键缺失/形状不符退正则兜底 (旧 mcp-server 兼容)。两条 # 路径汇成同一个 reset_ts 后过同一道 is_valid_quota_window — diff --git a/tests/test_review_gate.py b/tests/test_review_gate.py index c808a7c..78183e8 100644 --- a/tests/test_review_gate.py +++ b/tests/test_review_gate.py @@ -695,6 +695,34 @@ def test_baseline_header_line_does_not_disturb_parsing(self): # ============================================================ # 评论 body # ============================================================ +class TestRefusalMarker(_GateCase): + """ZOB-REFUSED 拒审标记解析 (issue #49 洞三: 拒审 ≠ 临时失败)""" + + def test_no_marker_returns_none(self): + """RF1: 普通失败载荷无标记 → None (照走退避重试)""" + self.assertIsNone(self.mod.parse_refusal_marker("限流失败: 1308 额度墙")) + self.assertIsNone(self.mod.parse_refusal_marker("")) + self.assertIsNone(self.mod.parse_refusal_marker(None)) + + def test_parses_reason(self): + """RF2: 标记在文末 → 取出原因串""" + text = ("拒绝审查: 被审目录存在 ZCode 项目配置文件:\n" + " - /clone/.zcode/config.json\n" + "ZOB-REFUSED: project_config") + self.assertEqual(self.mod.parse_refusal_marker(text), "project_config") + + def test_requires_full_line_form(self): + """RF3: 只认整行形态 — 嵌在句中/缺原因值不采信 (与 verdict 标记同策略)""" + self.assertIsNone(self.mod.parse_refusal_marker( + "报告提到 ZOB-REFUSED: project_config 这个词但非独立行")) + self.assertIsNone(self.mod.parse_refusal_marker("ZOB-REFUSED:")) + + def test_last_match_wins(self): + """RF4: 多个标记取最后一个""" + text = "ZOB-REFUSED: other\n中间\nZOB-REFUSED: project_config\n" + self.assertEqual(self.mod.parse_refusal_marker(text), "project_config") + + class TestCommentBody(_GateCase): SHA = "abcdef1234567890" + "0" * 24 @@ -1385,6 +1413,49 @@ def test_review_failure_goes_backoff_then_skip(self): entry = self._state_entry() self.assertEqual(entry["attempts"], 1) # 没增加 + def test_refusal_terminal_with_alert_comment(self): + """拒审 (ZOB-REFUSED, issue #49 洞三): 终态 gave_up + 告警评论。 + + 与"临时失败"语义相反: 重试不会变, 且不能让告警被退避吞掉 — 同 head + 立即终态, 但必须发一条评论让人看见; 新 head 仍会自动复活重审。 + """ + cfg_path = self._write_config() + self.mcp_rc = 2 + self.mcp_stdout = self._err_stdout( + "拒绝审查: 被审目录 (或其上层路径) 存在 ZCode 项目配置文件:\n" + " - /clone/octo__hello/.zcode/config.json\n" + "ZOB-REFUSED: project_config") + + self.assertEqual(self._run_once(cfg_path), 0) + entry = self._state_entry() + self.assertEqual(entry["status"], "gave_up", "拒审应终态, 不进失败退避") + self.assertIsNone(entry["verdict"]) + posts = [c for c in self.api_calls if c[0] == "POST"] + self.assertEqual(len(posts), 1, "拒审必须发告警评论") + body = posts[0][2]["body"] + self.assertIn("审查被拒绝", body) + self.assertIn(".zcode/config.json", body) + self.assertIn("/repos/octo/hello/issues/5/comments", posts[0][1]) + + # 同 head 第二轮: 终态不处理 (无第二次审查调用, 也不重复评论) + self.assertEqual(self._run_once(cfg_path), 0) + self.assertEqual(len(self.mcp_calls), 1, "同 head 拒审不应重试") + self.assertEqual(len([c for c in self.api_calls if c[0] == "POST"]), 1) + + def test_refusal_comment_failure_cached_for_retry(self): + """拒审告警发不出去: 转 comment_failed 缓存拒审原文, 下轮只补评论""" + cfg_path = self._write_config() + self.mcp_rc = 2 + self.mcp_stdout = self._err_stdout( + "拒绝审查: 命中项目配置护栏\nZOB-REFUSED: project_config") + self.comment_behavior = "retryable" + + self.assertEqual(self._run_once(cfg_path), 0) + entry = self._state_entry() + self.assertEqual(entry["status"], "comment_failed") + self.assertIn("ZOB-REFUSED", entry.get("report", ""), + "拒审原文须缓存, 供补评论使用") + def test_comment_retryable_failure_caches_and_comment_only_retry(self): """审查成功+评论 RetryableError → comment_failed+退避+缓存; 退避到点后的下一轮: 只补评论, 不重跑 mcp 审查""" diff --git a/tests/test_security_review.py b/tests/test_security_review.py index edacf9a..c38fd61 100644 --- a/tests/test_security_review.py +++ b/tests/test_security_review.py @@ -24,8 +24,10 @@ import json import os +import shutil import subprocess import sys +import tempfile import types import unittest @@ -85,7 +87,9 @@ class _EnvGuard(unittest.TestCase): "ZCODE_BRIDGE_PR_DIFF_MAX", # issue #26/#27: PR findings 过滤/基线开关与基线目录 "ZCODE_BRIDGE_PR_FINDINGS_SCOPE", "ZCODE_BRIDGE_PR_BASELINE", - "ZCODE_BRIDGE_BASELINE_DIR") + "ZCODE_BRIDGE_BASELINE_DIR", + # issue #49 洞三: 项目配置拒审的显式放行开关 + "ZCODE_BRIDGE_TRUST_PROJECT_CONFIG") def setUp(self): self._saved = {k: os.environ.get(k) for k in self.ENV_KEYS} @@ -141,6 +145,21 @@ def test_rc1_denylist_blocks_write_and_repl(self): "mcp__node_repl__js_add_node_module_dir"): self.assertIn(tool, deny.split(), f"黑名单缺 {tool}") + def test_rc1b_denylist_blocks_dynamic_workflow_family(self): + """RC1b: 黑名单含动态工作流一族 (0.16.9 新增的执行原语)。 + + EvalWorkflowSnippet 的 world.run 等价任意命令执行 + 文件写入, 与 Node REPL + 同族 (2026-09-19 实测复现: 黑名单原样下仍写入成功, 平台遥测标该工具 + read_only=0; 详见 docs/recheck-3.14.0.md)。漏禁则 Bash/Write 禁令被绕开。 + """ + cmd, _ = self._capture_cmd() + deny = cmd[cmd.index("--disallowed-tools") + 1].split() + for tool in ("EvalWorkflowSnippet", "CreateWorkflow", "AmendWorkflow", + "SaveWorkflow", "ListWorkflowRuns", "GetWorkflowRun", + "ResumeWorkflowRun", "ListSavedWorkflows", + "ResolveWorkflowQuestion"): + self.assertIn(tool, deny, f"黑名单缺动态工作流工具 {tool}") + def test_rc2_json_output(self): """RC2: 带 --json (结构化输出)""" cmd, _ = self._capture_cmd() @@ -220,6 +239,177 @@ def fake_run(cmd, *a, **kw): self.assertLess(len(captured["content"]), 11000, "截断后应在上限附近") +class TestProjectConfigGuard(_EnvGuard): + """issue #49 洞三防护: 被审目录链上的 zcode 项目配置 → 拒审 + 干净工作目录。 + + 背景: zcode 启动时装载 cwd 链上的 zcode.json / .zcode/config.json, 其中的 + mcp.servers **进程会被直接 spawn** (不经过模型/黑名单)。两条护栏: + ① _build_review_cmd 的 --cwd 指向只含空 .git 的临时目录 (发现链止步); + ② 被审目录链上有这两个文件时 fail-closed 拒审 (可显式放行)。 + """ + + @classmethod + def setUpClass(cls): + cls.mod = _load_mcp_module() + + def setUp(self): + super().setUp() + mod = self.mod + self._tmp = tempfile.mkdtemp(prefix="zob-guard-test-") + # scratch 缓存跨测试复用, 置空避免互相影响 + self._saved_scratch = mod._scratch_cwd_cache + mod._scratch_cwd_cache = None + + def tearDown(self): + self.mod._scratch_cwd_cache = self._saved_scratch + shutil.rmtree(self._tmp, ignore_errors=True) + super().tearDown() + + def _mk_repo(self, name, rel=None, git=True): + root = os.path.join(self._tmp, name) + os.makedirs(root, exist_ok=True) + if git: + os.mkdir(os.path.join(root, ".git")) + if rel: + p = os.path.join(root, rel) + os.makedirs(os.path.dirname(p), exist_ok=True) + with open(p, "w") as f: + f.write("{}") + return root + + # ---------- _project_config_files ---------- + + def test_pg1_finds_dot_zcode_config(self): + """PG1: /.zcode/config.json 被识别""" + root = self._mk_repo("r1", rel=".zcode/config.json") + got = self.mod._project_config_files(root) + self.assertEqual(len(got), 1) + self.assertTrue(got[0].endswith(".zcode/config.json")) + + def test_pg2_finds_zcode_json(self): + """PG2: /zcode.json 被识别""" + root = self._mk_repo("r2", rel="zcode.json") + got = self.mod._project_config_files(root) + self.assertEqual(len(got), 1) + self.assertTrue(got[0].endswith("zcode.json")) + + def test_pg3_clean_repo_is_empty(self): + """PG3: 干净仓库无命中""" + root = self._mk_repo("r3") + self.assertEqual(self.mod._project_config_files(root), []) + + def test_pg4_walks_up_to_git_root(self): + """PG4: 从子目录向上到 git 根 (含) — 仓库根的配置也被发现""" + root = self._mk_repo("r4", rel="zcode.json") + sub = os.path.join(root, "pkg", "inner") + os.makedirs(sub, exist_ok=True) + got = self.mod._project_config_files(sub) + self.assertEqual(len(got), 1, "应沿子目录→git 根发现仓库根的配置") + + def test_pg5_stops_at_git_root(self): + """PG5: git 根之上不再上溯 (与 zcode 的发现顺序一致)""" + outside = os.path.join(self._tmp, "pg5-outside") + os.makedirs(outside, exist_ok=True) + with open(os.path.join(outside, "zcode.json"), "w") as f: + f.write("{}") + repo = os.path.join(outside, "repo") + os.makedirs(repo) + os.mkdir(os.path.join(repo, ".git")) + self.assertEqual(self.mod._project_config_files(repo), [], + "git 根之上的配置不属于该项目, 不应误报") + + # ---------- _project_config_guard ---------- + + def test_pg6_guard_blocks_and_allows(self): + """PG6: 命中即拒审 (isError); 显式放行开关生效""" + root = self._mk_repo("r6", rel=".zcode/config.json") + res = self.mod._project_config_guard(root) + self.assertIsNotNone(res) + self.assertTrue(res["isError"]) + self.assertIn("拒绝审查", res["content"][0]["text"]) + self.assertIn("ZCODE_BRIDGE_TRUST_PROJECT_CONFIG", res["content"][0]["text"]) + # 机器可读标记: review-gate 靠它把"拒审"与"临时失败"分开 + self.assertIn("ZOB-REFUSED: project_config", res["content"][0]["text"]) + os.environ["ZCODE_BRIDGE_TRUST_PROJECT_CONFIG"] = "1" + try: + self.assertIsNone(self.mod._project_config_guard(root)) + finally: + os.environ.pop("ZCODE_BRIDGE_TRUST_PROJECT_CONFIG", None) + + def test_pg7_guard_passes_clean(self): + """PG7: 干净目录放行""" + self.assertIsNone(self.mod._project_config_guard(self._mk_repo("r7"))) + + def test_pg7b_marker_survives_gate_truncation(self): + """PG7b: 标记必须落在失败文本前 500 字符内。 + + review-gate 的 run_review 只取 detail[:500] 再解析 ZOB-REFUSED — + 标记若被长路径/多文件顶出窗口, 拒审会被误判成临时失败 (退避重试, + 告警被吞)。故钉住: 真实长路径下标记仍在窗口内。 + """ + deep = self._mk_repo("r7b-deep", rel=".zcode/config.json", git=False) + # 造一个足够长的路径 (贴近 GC-8G 克隆目录的真实长度) + long_dir = os.path.join(deep, *(["clones", "openagentemail__openagentemail"] * 2)) + os.makedirs(long_dir, exist_ok=True) + with open(os.path.join(long_dir, "zcode.json"), "w") as f: + f.write("{}") + res = self.mod._project_config_guard(deep) + self.assertIsNotNone(res) + text = res["content"][0]["text"] + self.assertIn("ZOB-REFUSED: project_config", text[:500], + "标记被顶出 500 字符窗口 → 闸会误判为临时失败") + self.assertTrue(text.startswith("ZOB-REFUSED: project_config")) + + # ---------- 命令构造: 干净工作目录 + prompt 带仓库根 ---------- + + def test_pg8_review_cmd_uses_scratch_not_repo(self): + """PG8: --cwd 指向隔离沙箱 (含空 .git), 不是被审仓库""" + repo = self._mk_repo("r8") + cmd = self.mod._build_review_cmd("PROMPT", repo) + cwd = cmd[cmd.index("--cwd") + 1] + self.assertNotEqual(os.path.abspath(cwd), os.path.abspath(repo)) + self.assertTrue(os.path.isdir(os.path.join(cwd, ".git")), + "沙箱须含空 .git 令 zcode 的配置发现链止步") + self.assertFalse(os.path.exists(os.path.join(cwd, "zcode.json"))) + + def test_pg9_review_prompt_carries_repo_root(self): + """PG9: tool_zcode_review 的 prompt 带被审仓库绝对路径""" + repo = self._mk_repo("r9") + mod = self.mod + captured = {} + + def fake_run(cmd, *a, **kw): + captured["cmd"] = cmd + return _FakeCompletedProcess(returncode=0, stdout="OK", stderr="") + + saved = mod.subprocess.run + mod.subprocess.run = fake_run + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + try: + result = mod.tool_zcode_review({"code": "print(1)", "cwd": repo}) + finally: + mod.subprocess.run = saved + self.assertNotIn("isError", result) + prompt = captured["cmd"][captured["cmd"].index("--prompt") + 1] + self.assertIn(f"被审仓库根目录: {os.path.abspath(repo)}", prompt) + + def test_pg10_tool_refuses_project_config(self): + """PG10: 被审目录含项目配置 → tool_zcode_review 直接拒审, 不调 zcode""" + repo = self._mk_repo("r10", rel=".zcode/config.json") + mod = self.mod + calls = [] + saved = mod.subprocess.run + mod.subprocess.run = lambda *a, **kw: calls.append(a) or _FakeCompletedProcess(0, "OK", "") + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + try: + result = mod.tool_zcode_review({"code": "print(1)", "cwd": repo}) + finally: + mod.subprocess.run = saved + self.assertTrue(result["isError"]) + self.assertIn("拒绝审查", result["content"][0]["text"]) + self.assertEqual(calls, [], "拒审时不应启动 zcode 子进程") + + class TestExtractResponse(unittest.TestCase): """--json 输出的 (response, usage.totalTokens) 提取""" From 707ae56f050346298f2eaf6c07262dfaa69a82b2 Mon Sep 17 00:00:00 2001 From: tizerluo <192086140+tizerluo@users.noreply.github.com> Date: Sun, 20 Sep 2026 03:19:18 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20R1=20=E5=AE=A1=E6=9F=A5=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=20(2P1=20+=204P2=20+=204P3)=20+=20=E7=B2=BE=E7=AE=80?= =?UTF-8?q?=20pass=20=E9=87=87=E7=BA=B3=204/5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 外部审查 (code-reviewer + simplifier) 全部属实项落地: P1-1 拒审告警撞限流穿透: _post_refusal_alert 补 RateLimited 分支 (缓存拒审原文 + comment_failed 不烧 attempts + 上抛停轮), 与正常 评论路径同口径 — 原实现 gave_up 已写而缓存未落, 下一轮 needs_review 跳过 gave_up, 告警永久丢失。 P1-2 拒审告警补发被销毁: comment_failed 重试轮识别 entry.refusal, 重发拒审评论并回 gave_up 终态 — 原实现走通用补评论路径, 把安全 拒审发成 concerns 质量评论并翻成 reviewed。 P2-2 拒审判定改结构化键 (防伪造): mcp-server 护栏 result 顶层附 "refusal" 键, gate 只认键 (run_review 原样透传); 文本 ZOB-REFUSED 行降级为仅供人看 — mimosa 错误体/子进程回显可被攻击者影响, 文本 解析可把临时失败伪造成终态拒审。键与解析器同 PR 发布, 无旧版兼容 场景, 不留文本回退。RF1-4 重写为 TestRefusalKeyInRunReview (含 伪造文本不生效的行为钉子), PG7b 改钉结构化键 (500 字符窗口问题 随之消失)。 P2-3 沙箱目录泄漏: atexit 注册 _cleanup_scratch_cwd (--call 模式 一次审查一进程, 不清理则每审漏一个目录), PG8b 钉清理+重建。 P2-4 README 补 ZCODE_BRIDGE_TRUST_PROJECT_CONFIG 表项 + gate 拒审 终态行为说明。 P3: TRUST 开关假值集扩 off/disabled (内联进 guard, 精简#2); gate 标记注释与实现对齐; _build_review_cmd 日志统一用 scan_path; _project_config_files 改 realpath (PG2b)。 精简采纳: #1 Path.parents 重写上溯 (-7); #2 _env_truthy 内联; #4 _capture_cmd 上移 _EnvGuard + PG9 复用 (-13); #5 PG7b 死 setup 随结构化键重写消除。不采纳 #3 (entry.update 与全文件逐键风格冲突)。 540 tests pass (净 +3), ruff clean。 --- README.md | 3 + packages/mcp-server/zcode-mcp-server | 62 +++++----- packages/review-gate/zcode-review-gate | 100 +++++++++++----- tests/test_review_gate.py | 152 +++++++++++++++++++------ tests/test_security_review.py | 117 +++++++++++-------- 5 files changed, 295 insertions(+), 139 deletions(-) diff --git a/README.md b/README.md index a23b986..fe5e0c1 100644 --- a/README.md +++ b/README.md @@ -209,6 +209,8 @@ PR 自动审查闸门守护进程（第 4 组件，experimental）：常驻轮 安装、配置参考、systemd 部署与运维详见 [packages/review-gate/README.md](packages/review-gate/README.md)。 +> **拒审 = 终态告警**（issue #49 洞三）：被审分支含 `zcode.json` / `.zcode/config.json` 时，mcp-server 返回结构化 `refusal` 键，gate 对该 head 立即 `gave_up` 并回贴「🚫 审查被拒绝（安全策略）」评论（同 head 不重试、新 head 自动复活）；告警评论发布失败（限流/网络）会缓存原文，下一轮只补发拒审评论、不重跑审查。判定只认结构化键——报告正文/错误文本里的 `ZOB-REFUSED` 字样不参与（防伪造，R1 P2-2）。 + > 配套能力：mcp-server 新增 `--call TOOL ''` 一次性调用模式（脚本化入口，exit 0/1/2 分别对应 成功 / 用法错误或 handler 异常 / tool 执行失败），stdio 模式行为不变。 ## 会话存储 @@ -261,6 +263,7 @@ ZCODE_BASE_URL=https://api.z.ai/api/anthropic ./packages/mcp-server/zcode-mcp-se | 机制 | 行为 | 配置 | |------|------|------| | **进程级文件锁** | 多个 MCP client 并发调用时，串行化 headless review，防并发触发限流 | `ZCODE_BRIDGE_REVIEW_LOCK=0` 关闭 | +| **项目配置拒审**（issue #49 洞三） | 被审目录链（cwd 到 git 根）存在 `zcode.json` / `.zcode/config.json` 时 **fail-closed 拒审**（`isError` + 顶层结构化 `refusal` 键；review-gate 据此按终态告警处理，不当临时失败重试）——该文件的 `mcp.servers` 会被 zcode 启动时直接 spawn。审查子进程 `--cwd` 恒为隔离沙箱（含空 `.git` 阻断配置上溯），被审仓库根目录经 prompt 以绝对路径提供 | `ZCODE_BRIDGE_TRUST_PROJECT_CONFIG=1` 显式放行可信仓库（`0/false/no/off/disabled` 均视为关） | | **provider 错误解析** | 识别 429 / 1302 / `Too Many Requests` / `请求过于频繁` / `retry-after`，区分限流/配额/其他 | — | | **有限重试 + 退避** | 仅对**限流**错误重试（配额/Unauthorized 不重试），退避用 retry-after 或指数退避（`2^n+1`） | `ZCODE_BRIDGE_MAX_RETRIES`（默认 3） | | **单次调用超时** | review 单次 zcode 调用超时 | `ZCODE_BRIDGE_REVIEW_TIMEOUT`（默认 300s，下限 30s） | diff --git a/packages/mcp-server/zcode-mcp-server b/packages/mcp-server/zcode-mcp-server index ca654ac..5ff9ab0 100755 --- a/packages/mcp-server/zcode-mcp-server +++ b/packages/mcp-server/zcode-mcp-server @@ -18,6 +18,7 @@ Claude Code、Cursor 等) 能标准化地发现并调用 headless zcode。 依赖: 仅 Python3 标准库 (零第三方依赖) """ +import atexit import fcntl import hashlib import json @@ -1119,12 +1120,6 @@ def _extract_response(output): return output, None -def _env_truthy(name): - """读布尔型环境开关: 非空且非 0/false/no 视为开。""" - v = (os.environ.get(name) or "").strip().lower() - return v not in ("", "0", "false", "no") - - # ============================================================ # 项目配置防护 (2026-09-19, issue #49 洞三) # ------------------------------------------------------------ @@ -1148,47 +1143,60 @@ _TRUST_PROJECT_CONFIG_ENV = "ZCODE_BRIDGE_TRUST_PROJECT_CONFIG" _scratch_cwd_cache = None +def _cleanup_scratch_cwd(): + """清掉沙箱目录并置空缓存 (atexit 注册, 幂等可重入)。""" + global _scratch_cwd_cache + if _scratch_cwd_cache: + shutil.rmtree(_scratch_cwd_cache, ignore_errors=True) + _scratch_cwd_cache = None + + def _review_scratch_cwd(): - """审查用的干净工作目录 (只含一个空 .git, 令配置发现链止步于此)。""" + """审查用的干净工作目录 (只含一个空 .git, 令配置发现链止步于此)。 + + 进程退出时清理 (R1 P2-3): gate 的 --call 模式一次审查一个进程, + 不清理则每审漏一个目录无限缓慢累积; stdio 常驻模式进程级一份, 退出即清。 + """ global _scratch_cwd_cache if _scratch_cwd_cache and os.path.isdir(_scratch_cwd_cache): return _scratch_cwd_cache d = tempfile.mkdtemp(prefix="zcode-review-cwd-") os.mkdir(os.path.join(d, ".git")) # 空目录即可: zcode 见到 .git 就停止上溯 + atexit.register(_cleanup_scratch_cwd) _scratch_cwd_cache = d return d def _project_config_files(root): - """沿 root → git 根 (含) 列出存在的 zcode 项目配置文件 (与 zcode 发现范围一致)。""" + """沿 root → git 根 (含) 列出存在的 zcode 项目配置文件 (与 zcode 发现范围一致)。 + + realpath 先解析符号链接: 与 zcode 落地后的真实 cwd 对齐, 链接链上的 + 配置不漏报 (报警器层; 主锁是 --cwd 沙箱, 不依赖本函数的完整性)。 + """ + start = Path(os.path.realpath(root)) found = [] - cur = os.path.abspath(root) - while True: - for rel in _PROJECT_CONFIG_RELS: - p = os.path.join(cur, rel) - if os.path.exists(p): - found.append(p) - if os.path.exists(os.path.join(cur, ".git")): - break - parent = os.path.dirname(cur) - if parent == cur: + for d in (start, *start.parents): + found += [str(d / rel) for rel in _PROJECT_CONFIG_RELS if (d / rel).exists()] + if (d / ".git").exists(): break - cur = parent return found def _project_config_guard(root): """被审目录链上有 zcode 项目配置文件则返回拒绝 result, 否则 None。""" - if _env_truthy(_TRUST_PROJECT_CONFIG_ENV): + trust = (os.environ.get(_TRUST_PROJECT_CONFIG_ENV) or "").strip().lower() + # off/disabled 也算关 (R1 P3-1): 关安全护栏的开关, 反直觉值不得触发放行 + if trust not in ("", "0", "false", "no", "off", "disabled"): return None bad = _project_config_files(root) if not bad: return None log(f"⛔ 拒审: 被审目录链上存在 zcode 项目配置 {bad}") return {"content": [{"type": "text", "text": ( - # 机器可读标记必须在首行: review-gate 只取失败文本前 500 字符 - # (run_review 的 detail 截断), 放末尾会被长路径/多文件顶出窗口, - # 导致拒审被误判为临时失败而退避重试 (告警随之被吞)。 + # 文本标记放首行便于肉眼/日志 grep; 机器契约是顶层 "refusal" 键 + # (R1 P2-2): 失败 detail 会混入 mimosa 错误体/子进程回显等攻击者 + # 可影响的文本, gate 解析文本存在伪造面 (可把临时失败伪造成终态 + # 拒审), 结构化键只有本护栏会写, 无伪造面。 "ZOB-REFUSED: project_config\n" "拒绝审查: 被审目录 (或其上层路径) 存在 ZCode 项目配置文件:\n" + "\n".join(f" - {p}" for p in bad) @@ -1199,7 +1207,9 @@ def _project_config_guard(root): "克隆副本里做); ② 确认该目录可信 (自己的仓库且确有项目配置): 设 " f"{_TRUST_PROJECT_CONFIG_ENV}=1 放行。\n" "注意: 沿工作目录到 git 根的整条链都会被 zcode 检查, 不只仓库根。" - )}], "isError": True} + )}], "isError": True, + # 结构化拒审键 (R1 P2-2): review-gate 只认它判定"安全策略拒绝" + "refusal": "project_config"} def _build_review_cmd(prompt, repo): @@ -1525,7 +1535,7 @@ def tool_zcode_security_review(args): if focus: prompt += f"\n额外审查重点: {focus}。" - cmd = _build_review_cmd(prompt, args.get("cwd") or scan_path) + cmd = _build_review_cmd(prompt, scan_path) cmd += ["--attach", tmp_path] log("调用 zcode 复核 mimosa findings (yolo+只读黑名单)") @@ -1753,7 +1763,7 @@ def tool_zcode_pr_review(args): if focus: prompt += f"\n额外审查重点: {focus}。" - cmd = _build_review_cmd(prompt, args.get("cwd") or scan_path) + cmd = _build_review_cmd(prompt, scan_path) cmd += ["--attach", tmp_path] log(f"调用 zcode PR 复核 (yolo+只读黑名单, {len(changed)} 个改动文件)") diff --git a/packages/review-gate/zcode-review-gate b/packages/review-gate/zcode-review-gate index 564ec4e..6fdbfad 100755 --- a/packages/review-gate/zcode-review-gate +++ b/packages/review-gate/zcode-review-gate @@ -661,6 +661,8 @@ def run_review(cfg, clone, base_ref, head_sha): info (#35) 结构化旁路信息, 从 result 顶层附加键直读: quota_limit: {"code": 1308, "reset_at": } | None — 仅 stderr 命中 1308 才有 (mcp-server 侧已完成北京 naive → unix 转换); + refusal: str | None — 安全策略拒审 (issue #49 洞三) 时 mcp-server + 护栏写入的原因串 (如 "project_config"), 机器契约只此一处; total_tokens: int — 成功时 usage.total_tokens 透传, 无则 0。 子进程异常/非 JSON 输出时为空 info。quota_limit 原样透传 (键缺失为 None), 合法性校验全在消费端 (quota_limit_reset_ts), 这里不猜不修。 @@ -676,7 +678,7 @@ def run_review(cfg, clone, base_ref, head_sha): json.dumps(args, ensure_ascii=False)] env = dict(os.environ) env["ZCODE_BRIDGE_REVIEW_TIMEOUT"] = str(ZCODE_REVIEW_TIMEOUT) - info = {"quota_limit": None, "total_tokens": 0} + info = {"quota_limit": None, "total_tokens": 0, "refusal": None} try: p = subprocess.run(cmd, capture_output=True, text=True, timeout=REVIEW_TIMEOUT, env=env) @@ -693,6 +695,9 @@ def run_review(cfg, clone, base_ref, head_sha): # quota_limit 原样透传: 形状校验全在消费端 (quota_limit_reset_ts), # 非 dict 在那里回 None → 正则兜底, 这里不重复判 info["quota_limit"] = result.get("quota_limit") + # 拒审原因串原样透传 (#49): 只有 mcp-server 的项目配置护栏会写这个键, + # 正文文本不参与判定 (伪造面, R1 P2-2) + info["refusal"] = result.get("refusal") usage = result.get("usage") if isinstance(usage, dict): tokens = usage.get("total_tokens") @@ -734,18 +739,13 @@ def parse_verdict_marker(report): m.group(5) == "true") -# 拒审标记 (issue #49 洞三): mcp-server 的项目配置护栏在拒审文案末尾带此行。 -# review-gate 据此把"安全策略拒绝"与"临时失败"分开 — 前者终态 + 告警评论 -# (重试不会变, 且 5 次后静默 gave_up 会吞掉"有人在投毒"的信号), 后者照旧退避。 -# 只在审查失败 (ok=False) 的载荷上解析: 该文案由 mcp-server 生成, 非模型输出, -# 无伪造面 (审查报告正文不参与判定)。 -_RE_ZOB_REFUSED = re.compile(r"^ZOB-REFUSED:\s*([A-Za-z0-9_]+)\s*$", re.M) - - -def parse_refusal_marker(text): - """读 ZOB-REFUSED 标记, 返回原因串 (如 "project_config") 或 None。""" - matches = list(_RE_ZOB_REFUSED.finditer(text or "")) - return matches[-1].group(1) if matches else None +# 拒审 (issue #49 洞三): mcp-server 的项目配置护栏在 isError result 顶层附 +# 结构化键 "refusal": "project_config" (文本里那行 ZOB-REFUSED 仅供人看)。 +# gate 只认结构化键判定"安全策略拒绝" (R1 P2-2): 失败 detail 会混入 mimosa +# 错误体/子进程回显等攻击者可影响的文本, 解析文本可把任意临时失败伪造成 +# 终态拒审 (跳过 5 次重试 + 假告警); 结构化键只有 mcp-server 护栏会写。 +# 拒审按终态处理 (gave_up + 🚫 告警评论), 不当临时失败退避重试: +# 重试不会变, 且 5 次后静默 gave_up 会吞掉"有人在投毒"的信号。 def build_refusal_comment(head_sha, detail): @@ -767,6 +767,40 @@ def build_refusal_comment(head_sha, detail): ) +def _post_refusal_alert(cfg, token, entry, key, head_sha, owner, repo, + pr_number, text, now): + """发拒审告警评论; 发不出去则缓存原文转 comment_failed 等下轮补发。 + + 两个调用点共用 (首发于拒审分支 / 补发于 comment_failed 重试轮), + 失败口径与正常评论路径一致: + - RateLimited (R1 P1-1): 仓级事件不烧 attempts, 缓存 + comment_failed + 后上抛停本轮 — 早期版本只捕 Retryable/RepoHard, RateLimited 穿透时 + gave_up 已写而缓存未落, 下一轮 needs_review 跳过 gave_up, 告警 + 永久丢失; + - Retryable/RepoHard: mark_failure(comment_failed) 走退避, 缓存保留。 + 返回 True=已发布, False=未尝试 (评论禁用/无 token) 或发布失败 (已转 + comment_failed, 状态见 entry)。 + """ + if not (cfg.comment_enabled and token): + log(f"{key}: 拒审告警未发 (评论禁用或无 token), 详见 state/日志") + return False + try: + resp = post_comment(token, cfg.github_api, owner, repo, pr_number, + build_refusal_comment(head_sha, text)) + log(f"{key}: 拒审告警已发布 {(resp or {}).get('html_url') or ''}") + return True + except RateLimited: + entry["report"] = text[:cfg.comment_max_body] + entry["status"] = "comment_failed" + entry["next_retry_at"] = 0.0 + raise + except (RetryableError, RepoHardError) as e: + entry["report"] = text[:cfg.comment_max_body] + mark_failure(entry, f"拒审告警发布失败: {e}", now, cfg, + status="comment_failed") + return False + + def parse_severity_counts(report): """解析 (P0, P1, P2, P3) 条数; P0/P1/P2 任一缺失 → None, P3 缺省计 0。 @@ -1372,6 +1406,21 @@ def process_pr(cfg, state, token, owner, repo, clone, info): # 审查已成功过、只是评论没发出去: 沿用缓存结果, 只补评论, # 不为一次评论失败重跑昂贵的 zcode 审查 report = entry["report"] + if entry.get("refusal"): + # 拒审告警补发 (R1 P1-2): 缓存的是拒审文案, 补发的必须是拒审 + # 评论 — 走下方通用路径会把安全拒审销毁成 "concerns 质量告警" + # 并翻成 reviewed, 下游读 state 拿到错误结论。 + if _post_refusal_alert(cfg, token, entry, key, head_sha, owner, + repo, info["number"], report, now): + entry["status"] = "gave_up" # 补发成功, 回终态 + entry["next_retry_at"] = 0.0 + entry["report"] = None + elif entry["status"] != "comment_failed": + # 未尝试 (评论被禁用/无 token): 烧退避老化, 不无限空转; + # 缓存保留, gave_up 后仍可供人工排查 + mark_failure(entry, "拒审告警未发 (评论禁用或无 token)", + now, cfg, status="comment_failed") + return verdict = entry.get("verdict") or "concerns" counts = tuple(entry["counts"]) if entry.get("counts") else None log(f"{key}: 沿用缓存审查结果 (verdict={verdict}), 只补评论") @@ -1403,31 +1452,20 @@ def process_pr(cfg, state, token, owner, repo, clone, info): if not ok: # 拒审 (issue #49 洞三): 安全策略拒绝 → 终态 + 告警评论。 # 不能当临时失败退避重试: 重试结果不会变, 且 5 次后静默 gave_up - # 会把"有人在投毒"的信号吞掉 (mcp-server 侧带 ZOB-REFUSED 标记)。 + # 会把"有人在投毒"的信号吞掉 (判定只认 mcp-server 护栏写的 + # 结构化 refusal 键, 文本标记仅供人看)。 # 复用 gave_up 语义: 同 head 不再动, 新 head 自动复活重审。 - refused = parse_refusal_marker(payload) + refused = rinfo.get("refusal") if refused: entry["error"] = payload[:500] entry["verdict"] = None + entry["refusal"] = refused entry["status"] = "gave_up" entry["next_retry_at"] = 0.0 - log(f"{key}: ⛔ 拒审 (ZOB-REFUSED: {refused}), 终态 " + log(f"{key}: ⛔ 拒审 (refusal: {refused}), 终态 " f"(head={head_sha[:12]}) — 发告警评论") - if cfg.comment_enabled and token: - try: - resp = post_comment( - token, cfg.github_api, owner, repo, info["number"], - build_refusal_comment(head_sha, payload)) - log(f"{key}: 拒审告警已发布 " - f"{(resp or {}).get('html_url') or ''}") - except (RetryableError, RepoHardError) as e: - # 告警发不出去也不能丢: 缓存拒审原文, 下轮只补评论 - # (与审查成功但评论失败同一条缓存补发路径) - entry["report"] = payload[:cfg.comment_max_body] - mark_failure(entry, f"拒审告警发布失败: {e}", now, cfg, - status="comment_failed") - else: - log(f"{key}: 拒审告警未发 (评论禁用或无 token), 详见 state/日志") + _post_refusal_alert(cfg, token, entry, key, head_sha, owner, + repo, info["number"], payload, now) return # 判据优先级 (#35①): 结构化 quota_limit 键优先 (mcp-server 侧已转 # unix 秒), 键缺失/形状不符退正则兜底 (旧 mcp-server 兼容)。两条 diff --git a/tests/test_review_gate.py b/tests/test_review_gate.py index 78183e8..d910a84 100644 --- a/tests/test_review_gate.py +++ b/tests/test_review_gate.py @@ -269,11 +269,13 @@ def _fake_run(self, cmd, *a, **kw): return _CP(returncode=0, stdout=json.dumps(payload), stderr="") raise AssertionError(f"未预期命令: {cmd}") - def _err_stdout(self, text, quota_limit=None): + def _err_stdout(self, text, quota_limit=None, refusal=None): """伪造 mcp-server --call 失败输出 (isError result, 可选结构化键)""" result = {"content": [{"type": "text", "text": text}], "isError": True} if quota_limit is not None: result["quota_limit"] = quota_limit + if refusal is not None: + result["refusal"] = refusal return json.dumps({"ok": False, "result": result}) def _write_config(self, repos=("octo/hello",), quota=None): @@ -695,32 +697,61 @@ def test_baseline_header_line_does_not_disturb_parsing(self): # ============================================================ # 评论 body # ============================================================ -class TestRefusalMarker(_GateCase): - """ZOB-REFUSED 拒审标记解析 (issue #49 洞三: 拒审 ≠ 临时失败)""" - - def test_no_marker_returns_none(self): - """RF1: 普通失败载荷无标记 → None (照走退避重试)""" - self.assertIsNone(self.mod.parse_refusal_marker("限流失败: 1308 额度墙")) - self.assertIsNone(self.mod.parse_refusal_marker("")) - self.assertIsNone(self.mod.parse_refusal_marker(None)) - - def test_parses_reason(self): - """RF2: 标记在文末 → 取出原因串""" - text = ("拒绝审查: 被审目录存在 ZCode 项目配置文件:\n" - " - /clone/.zcode/config.json\n" - "ZOB-REFUSED: project_config") - self.assertEqual(self.mod.parse_refusal_marker(text), "project_config") - - def test_requires_full_line_form(self): - """RF3: 只认整行形态 — 嵌在句中/缺原因值不采信 (与 verdict 标记同策略)""" - self.assertIsNone(self.mod.parse_refusal_marker( - "报告提到 ZOB-REFUSED: project_config 这个词但非独立行")) - self.assertIsNone(self.mod.parse_refusal_marker("ZOB-REFUSED:")) - - def test_last_match_wins(self): - """RF4: 多个标记取最后一个""" - text = "ZOB-REFUSED: other\n中间\nZOB-REFUSED: project_config\n" - self.assertEqual(self.mod.parse_refusal_marker(text), "project_config") +class TestRefusalKeyInRunReview(_GateCase): + """拒审结构化键提取 (issue #49 洞三, R1 P2-2): 只认 result 顶层 + "refusal" 键, 文本标记不参与判定 (失败 detail 混有攻击者可影响的文本)""" + + def _cfg(self): + return self.mod.GateConfig({}) + + def _run_with_stdout(self, payload): + cp = _CP(returncode=2, stdout=json.dumps(payload), stderr="") + with mock.patch.object(self.mod.subprocess, "run", + lambda *a, **kw: cp): + return self.mod.run_review(self._cfg(), "/clone", "main", "sha") + + def test_refusal_key_extracted(self): + """isError result 顶层 refusal 键原样透传到 info""" + ok, _detail, info = self._run_with_stdout({ + "ok": False, "result": { + "content": [{"type": "text", + "text": "ZOB-REFUSED: project_config\n拒绝审查…"}], + "isError": True, "refusal": "project_config"}}) + self.assertFalse(ok) + self.assertEqual(info["refusal"], "project_config") + + def test_refusal_key_absent(self): + """无 refusal 键 → None (普通失败, 照走退避)""" + ok, _detail, info = self._run_with_stdout({ + "ok": False, "result": { + "content": [{"type": "text", "text": "炸了"}], + "isError": True}}) + self.assertFalse(ok) + self.assertIsNone(info["refusal"]) + + def test_forged_text_marker_not_refusal(self): + """正文伪造的 ZOB-REFUSED 行不算拒审 (R1 P2-2 的行为钉子): + mimosa 错误体/子进程回显可被攻击者影响, 只有结构化键算数 — + 否则任意临时失败可被伪造成终态拒审 (跳过重试 + 假告警)""" + ok, _detail, info = self._run_with_stdout({ + "ok": False, "result": { + "content": [{"type": "text", + "text": "mimosa 扫描失败: boom\n" + "ZOB-REFUSED: project_config"}], + "isError": True}}) + self.assertFalse(ok) + self.assertIsNone(info["refusal"], + "文本标记不得被解析为拒审 (伪造面)") + + def test_subprocess_failure_empty_info(self): + """子进程起不来/非 JSON → 空 info (refusal=None, 不误判拒审)""" + def boom(*a, **kw): + raise OSError("gone") + with mock.patch.object(self.mod.subprocess, "run", boom): + ok, _detail, info = self.mod.run_review( + self._cfg(), "/clone", "main", "sha") + self.assertFalse(ok) + self.assertIsNone(info["refusal"]) class TestCommentBody(_GateCase): @@ -1414,7 +1445,7 @@ def test_review_failure_goes_backoff_then_skip(self): self.assertEqual(entry["attempts"], 1) # 没增加 def test_refusal_terminal_with_alert_comment(self): - """拒审 (ZOB-REFUSED, issue #49 洞三): 终态 gave_up + 告警评论。 + """拒审 (结构化 refusal 键, issue #49 洞三): 终态 gave_up + 告警评论。 与"临时失败"语义相反: 重试不会变, 且不能让告警被退避吞掉 — 同 head 立即终态, 但必须发一条评论让人看见; 新 head 仍会自动复活重审。 @@ -1422,14 +1453,16 @@ def test_refusal_terminal_with_alert_comment(self): cfg_path = self._write_config() self.mcp_rc = 2 self.mcp_stdout = self._err_stdout( + "ZOB-REFUSED: project_config\n" "拒绝审查: 被审目录 (或其上层路径) 存在 ZCode 项目配置文件:\n" - " - /clone/octo__hello/.zcode/config.json\n" - "ZOB-REFUSED: project_config") + " - /clone/octo__hello/.zcode/config.json", + refusal="project_config") self.assertEqual(self._run_once(cfg_path), 0) entry = self._state_entry() self.assertEqual(entry["status"], "gave_up", "拒审应终态, 不进失败退避") self.assertIsNone(entry["verdict"]) + self.assertEqual(entry.get("refusal"), "project_config") posts = [c for c in self.api_calls if c[0] == "POST"] self.assertEqual(len(posts), 1, "拒审必须发告警评论") body = posts[0][2]["body"] @@ -1443,19 +1476,74 @@ def test_refusal_terminal_with_alert_comment(self): self.assertEqual(len([c for c in self.api_calls if c[0] == "POST"]), 1) def test_refusal_comment_failure_cached_for_retry(self): - """拒审告警发不出去: 转 comment_failed 缓存拒审原文, 下轮只补评论""" + """拒审告警发不出去 (Retryable): 转 comment_failed 缓存拒审原文; + 下一轮补发的仍是拒审评论 (不是 verdict 评论), 补发成功回 gave_up + 终态 — 走通用补评论路径会把安全拒审销毁成 concerns (R1 P1-2)""" cfg_path = self._write_config() self.mcp_rc = 2 self.mcp_stdout = self._err_stdout( - "拒绝审查: 命中项目配置护栏\nZOB-REFUSED: project_config") + "ZOB-REFUSED: project_config\n拒绝审查: 命中项目配置护栏", + refusal="project_config") self.comment_behavior = "retryable" self.assertEqual(self._run_once(cfg_path), 0) entry = self._state_entry() self.assertEqual(entry["status"], "comment_failed") + self.assertEqual(entry.get("refusal"), "project_config") self.assertIn("ZOB-REFUSED", entry.get("report", ""), "拒审原文须缓存, 供补评论使用") + # 下一轮 (退避到点): 只补发拒审评论, 不重跑审查, 回 gave_up 终态 + self._force_retry_due() + self.comment_behavior = "ok" + self.assertEqual(self._run_once(cfg_path), 0) + entry = self._state_entry() + self.assertEqual(entry["status"], "gave_up", "补发成功应回拒审终态") + self.assertIsNone(entry["verdict"], "不得翻成 verdict 状态") + self.assertEqual(len(self.mcp_calls), 1, "补评论不得重跑 mcp 审查") + posts = [c for c in self.api_calls if c[0] == "POST"] + # 第一轮失败尝试也会被 fake 记录: 1 (失败) + 1 (补发成功) + self.assertEqual(len(posts), 2) + self.assertIn("审查被拒绝", posts[-1][2]["body"], + "补发的必须是拒审评论") + self.assertNotIn("concerns", posts[-1][2]["body"], + "不得发成通用 verdict 评论") + + def test_refusal_alert_ratelimited_then_reposted(self): + """R1 P1-1: 拒审告警撞限流 — 早期版本 RateLimited 穿透 except, + gave_up 已写而缓存未落, 下一轮 needs_review 跳过 gave_up, 告警 + 永久丢失。现在: 缓存 + comment_failed (不烧 attempts) + 上抛停轮; + 下一轮 (无退避) 补发拒审评论, 回 gave_up 终态""" + cfg_path = self._write_config() + self.mcp_rc = 2 + self.mcp_stdout = self._err_stdout( + "ZOB-REFUSED: project_config\n拒绝审查: 命中项目配置护栏", + refusal="project_config") + self.comment_behavior = "ratelimited" + + # 第一轮: 限流停轮, --once 退出码仍 0 + self.assertEqual(self._run_once(cfg_path), 0) + entry = self._state_entry() + self.assertEqual(entry["status"], "comment_failed", + "限流不得让告警丢失 (须转可补发状态)") + self.assertEqual(entry.get("refusal"), "project_config") + self.assertIn("ZOB-REFUSED", entry.get("report", "")) + self.assertEqual(entry.get("attempts") or 0, 0, "限流不烧 attempts") + self.assertEqual(entry.get("next_retry_at"), 0) + self.assertEqual(len(self.mcp_calls), 1) + + # 下一轮: 补发拒审评论成功 → 回 gave_up 终态, 不重跑审查 + self.comment_behavior = "ok" + self.assertEqual(self._run_once(cfg_path), 0) + entry = self._state_entry() + self.assertEqual(entry["status"], "gave_up") + self.assertIsNone(entry["verdict"]) + posts = [c for c in self.api_calls if c[0] == "POST"] + # 第一轮限流尝试也被 fake 记录: 1 (限流失败) + 1 (补发成功) + self.assertEqual(len(posts), 2, "限流失败一次 + 补发成功一次") + self.assertIn("审查被拒绝", posts[-1][2]["body"]) + self.assertEqual(len(self.mcp_calls), 1) + def test_comment_retryable_failure_caches_and_comment_only_retry(self): """审查成功+评论 RetryableError → comment_failed+退避+缓存; 退避到点后的下一轮: 只补评论, 不重跑 mcp 审查""" diff --git a/tests/test_security_review.py b/tests/test_security_review.py index c38fd61..2b98d04 100644 --- a/tests/test_security_review.py +++ b/tests/test_security_review.py @@ -101,16 +101,11 @@ def tearDown(self): else: os.environ[k] = v - -class TestReviewCmd(_EnvGuard): - """只读护栏: 命令构造 + prompt 约束""" - - @classmethod - def setUpClass(cls): - cls.mod = _load_mcp_module() - def _capture_cmd(self, **kwargs): - """patch subprocess.run 捕获 cmd, 返回 (cmd, result)。""" + """patch subprocess.run 捕获 cmd, 返回 (cmd, result)。 + + (R1 精简 #4: 原为 TestReviewCmd 私有, PG9 同形状需求, 上移共用。) + """ mod = self.mod captured = {} @@ -127,6 +122,14 @@ def fake_run(cmd, *a, **kw): mod.subprocess.run = saved return captured["cmd"], result + +class TestReviewCmd(_EnvGuard): + """只读护栏: 命令构造 + prompt 约束""" + + @classmethod + def setUpClass(cls): + cls.mod = _load_mcp_module() + def test_rc0_yolo_not_plan(self): """RC0: 用 --mode yolo, 绝不再出现 plan""" cmd, _ = self._capture_cmd() @@ -293,6 +296,15 @@ def test_pg2_finds_zcode_json(self): self.assertEqual(len(got), 1) self.assertTrue(got[0].endswith("zcode.json")) + def test_pg2b_symlink_resolved(self): + """PG2b: realpath 先解析符号链接 (R1 P3-4), 经链接进入的仓库照常识别""" + real = self._mk_repo("r2b-real", rel="zcode.json") + link = os.path.join(self._tmp, "r2b-link") + os.symlink(real, link) + got = self.mod._project_config_files(link) + self.assertEqual(len(got), 1) + self.assertTrue(got[0].endswith("zcode.json")) + def test_pg3_clean_repo_is_empty(self): """PG3: 干净仓库无命中""" root = self._mk_repo("r3") @@ -321,44 +333,49 @@ def test_pg5_stops_at_git_root(self): # ---------- _project_config_guard ---------- def test_pg6_guard_blocks_and_allows(self): - """PG6: 命中即拒审 (isError); 显式放行开关生效""" + """PG6: 命中即拒审 (isError, 带结构化 refusal 键); 放行开关值矩阵""" root = self._mk_repo("r6", rel=".zcode/config.json") res = self.mod._project_config_guard(root) self.assertIsNotNone(res) self.assertTrue(res["isError"]) - self.assertIn("拒绝审查", res["content"][0]["text"]) - self.assertIn("ZCODE_BRIDGE_TRUST_PROJECT_CONFIG", res["content"][0]["text"]) - # 机器可读标记: review-gate 靠它把"拒审"与"临时失败"分开 - self.assertIn("ZOB-REFUSED: project_config", res["content"][0]["text"]) - os.environ["ZCODE_BRIDGE_TRUST_PROJECT_CONFIG"] = "1" - try: - self.assertIsNone(self.mod._project_config_guard(root)) - finally: - os.environ.pop("ZCODE_BRIDGE_TRUST_PROJECT_CONFIG", None) + # 机器契约: 结构化 refusal 键 (R1 P2-2), gate 只认它判定拒审 + self.assertEqual(res.get("refusal"), "project_config") + text = res["content"][0]["text"] + self.assertIn("拒绝审查", text) + self.assertIn("ZCODE_BRIDGE_TRUST_PROJECT_CONFIG", text) + # 文本标记保留在首行, 仅供肉眼/日志 grep + self.assertIn("ZOB-REFUSED: project_config", text) + # 放行开关值矩阵 (R1 P3-1): off/disabled 是"关"不是"开" — + # 关安全护栏的开关, 反直觉值不得触发放行 + for v in ("1", "yes", "true"): + os.environ["ZCODE_BRIDGE_TRUST_PROJECT_CONFIG"] = v + self.assertIsNone( + self.mod._project_config_guard(root), f"TRUST={v!r} 应放行") + for v in ("0", "false", "no", "off", "disabled"): + os.environ["ZCODE_BRIDGE_TRUST_PROJECT_CONFIG"] = v + self.assertIsNotNone( + self.mod._project_config_guard(root), f"TRUST={v!r} 不应放行") + os.environ.pop("ZCODE_BRIDGE_TRUST_PROJECT_CONFIG", None) def test_pg7_guard_passes_clean(self): """PG7: 干净目录放行""" self.assertIsNone(self.mod._project_config_guard(self._mk_repo("r7"))) - def test_pg7b_marker_survives_gate_truncation(self): - """PG7b: 标记必须落在失败文本前 500 字符内。 + def test_pg7b_structured_refusal_key(self): + """PG7b: 拒审 result 带结构化 refusal 键 (机器契约)。 - review-gate 的 run_review 只取 detail[:500] 再解析 ZOB-REFUSED — - 标记若被长路径/多文件顶出窗口, 拒审会被误判成临时失败 (退避重试, - 告警被吞)。故钉住: 真实长路径下标记仍在窗口内。 + R1 P2-2: 失败 detail 会混入 mimosa 错误体/子进程回显等攻击者可影响 + 的文本, gate 解析文本标记存在伪造面 (可把临时失败伪造成终态拒审), + 故机器契约改为 result 顶层 "refusal" 键 — 只有本护栏会写它; + 文本首行的 ZOB-REFUSED 保留, 仅供肉眼/日志 grep。 """ - deep = self._mk_repo("r7b-deep", rel=".zcode/config.json", git=False) - # 造一个足够长的路径 (贴近 GC-8G 克隆目录的真实长度) - long_dir = os.path.join(deep, *(["clones", "openagentemail__openagentemail"] * 2)) - os.makedirs(long_dir, exist_ok=True) - with open(os.path.join(long_dir, "zcode.json"), "w") as f: - f.write("{}") - res = self.mod._project_config_guard(deep) + root = self._mk_repo("r7b", rel=".zcode/config.json", git=False) + res = self.mod._project_config_guard(root) self.assertIsNotNone(res) - text = res["content"][0]["text"] - self.assertIn("ZOB-REFUSED: project_config", text[:500], - "标记被顶出 500 字符窗口 → 闸会误判为临时失败") - self.assertTrue(text.startswith("ZOB-REFUSED: project_config")) + self.assertTrue(res["isError"]) + self.assertEqual(res.get("refusal"), "project_config") + self.assertTrue(res["content"][0]["text"].startswith( + "ZOB-REFUSED: project_config"), "文本标记保持首行, 便于日志 grep") # ---------- 命令构造: 干净工作目录 + prompt 带仓库根 ---------- @@ -372,25 +389,25 @@ def test_pg8_review_cmd_uses_scratch_not_repo(self): "沙箱须含空 .git 令 zcode 的配置发现链止步") self.assertFalse(os.path.exists(os.path.join(cwd, "zcode.json"))) + def test_pg8b_scratch_cwd_cleanup(self): + """PG8b: 沙箱目录退出时清理 (atexit), 清理后可重建 (R1 P2-3)""" + mod = self.mod + d1 = mod._review_scratch_cwd() + self.assertTrue(os.path.isdir(d1)) + mod._cleanup_scratch_cwd() + self.assertFalse(os.path.exists(d1), "清理后目录应删除") + self.assertIsNone(mod._scratch_cwd_cache) + d2 = mod._review_scratch_cwd() + self.assertNotEqual(d1, d2) + self.assertTrue(os.path.isdir(os.path.join(d2, ".git")), + "重建的沙箱同样含空 .git") + def test_pg9_review_prompt_carries_repo_root(self): """PG9: tool_zcode_review 的 prompt 带被审仓库绝对路径""" repo = self._mk_repo("r9") - mod = self.mod - captured = {} - - def fake_run(cmd, *a, **kw): - captured["cmd"] = cmd - return _FakeCompletedProcess(returncode=0, stdout="OK", stderr="") - - saved = mod.subprocess.run - mod.subprocess.run = fake_run - os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" - try: - result = mod.tool_zcode_review({"code": "print(1)", "cwd": repo}) - finally: - mod.subprocess.run = saved + cmd, result = self._capture_cmd(cwd=repo) self.assertNotIn("isError", result) - prompt = captured["cmd"][captured["cmd"].index("--prompt") + 1] + prompt = cmd[cmd.index("--prompt") + 1] self.assertIn(f"被审仓库根目录: {os.path.abspath(repo)}", prompt) def test_pg10_tool_refuses_project_config(self):