diff --git a/README.md b/README.md index 9103896..17f2c2f 100644 --- a/README.md +++ b/README.md @@ -111,7 +111,7 @@ zcode --prompt "继续" --resume sess_xxxx |------|------| | `get_zcode_capabilities` | 返回 ZCode 能力清单（调 agent-help） | | `zcode_review` | 调 ZCode 审查代码（yolo + 写/执行工具物理禁用，全程免授权但改不了文件，安全） | -| `zcode_security_review` | 安全专项审查：mimosa 确定性规则引擎全仓预扫 → ZCode 拿 findings 逐条核实（确认/误报/存疑 + 攻击路径 + 修复建议） | +| `zcode_security_review` | 安全专项审查：mimosa 确定性规则引擎预扫 → ZCode 拿 findings 逐条核实（确认/误报/存疑 + 攻击路径 + 修复建议）。`depth=normal` 秒级快扫（默认），`depth=deep` 含业务逻辑投研（异步任务管线） | > **只读原理（2026-08-08 重构，告别 `--mode plan`）**：review 体系不再用 plan 模式——plan 只禁「改文件」，读探索/子代理照样放行（限流超时主因），且 plan→build 的规划惯性容易让 review 变成「边审边修」。新方案用 `--mode yolo`（全程免授权）+ `--disallowed-tools` 把 `Write/Edit/MultiEdit/ApplyPatch/Bash` 连同 Node REPL 一族（`js` / `mcp__node_repl__js*`）一起禁掉：`--disallowed-tools` 是工具集级物理移除、先于权限层，yolo 也绕不过；Node REPL 一族必须同禁，否则可被 `execSync` 打穿 Bash 黑名单（0.16.1 实测复现）。读工具（Read/Grep/Glob）全开，不影响审查能力。prompt 层另有「只审不修」职责约束（不修改文件、不提议帮忙修复）作双保险。 @@ -247,19 +247,29 @@ ZCODE_BASE_URL=https://api.z.ai/api/anthropic ./packages/mcp-server/zcode-mcp-se 注意：zcode 内部已有自己的指数退避重试（`_retryWithExponentialBackoff`），MCP 层的重试是补充，默认保守（max 3）。 -`zcode_security_review` 额外有 mimosa 相关的两个 env： +`zcode_security_review` 额外有 mimosa 相关的 env： | 配置 | 作用 | |------|------| | `ZCODE_BRIDGE_MIMOSA_ROOT` | 指向 mimosa 插件根目录（含 `payload/dist/mcp/server.js` 的那层）；不设则自动探测 `~/.local/share/mimosa/*` 与 `~/.zcode/cli/plugins/cache/*/mimosa/*`，找不到会明确报错并建议改用 `zcode_review` | -| `ZCODE_BRIDGE_MIMOSA_TIMEOUT` | mimosa `security_scan` 快扫超时（默认 180s） | +| `ZCODE_BRIDGE_MIMOSA_TIMEOUT` | mimosa `security_scan` 快扫（depth=normal）超时（默认 180s） | | `ZCODE_BRIDGE_MIMOSA_SCAN_ROOT` | findings 回读的信任根（默认 `~/.mimosa/security-scans`）：从 mimosa 摘要解析出的 scanDir 必须落在其下才回读 `findings.json`，越界降级为仅用摘要（防路径注入导致任意文件回读） | +| `ZCODE_BRIDGE_MIMOSA_DEEP_TIMEOUT` | depth=deep 异步扫描的总预算（默认 900s），超时会 best-effort cancel 后台 job | +| `ZCODE_BRIDGE_MIMOSA_POLL_INTERVAL` | depth=deep 的 status 轮询间隔（默认 2s） | + +**depth 两档**（2026-08-08 接入，mimosa 1.0.3 实测）： + +- `normal`（默认）：同步 `security_scan`，秒级（400 文件项目 ~2s），纯规则匹配 +- `deep`：异步 `security_scan_start` → `security_scan_status` 轮询 → 完成后回读 findings，含业务逻辑投研（threatModel/validation/pathAnalysis 等阶段），400 文件项目 ~13s。与 normal 共用同一条 findings 回读管线。纯 native 引擎、零 LLM、零网络（`evidenceBoundary: static_only_no_runtime_execution`） +- `focus_files` 参数（仅 deep）：业务逻辑复核的**优先级提示**（典型用法：调用方自己算出本次改动的文件清单传入——bridge 不做 git diff 集成），不是过滤器，静态引擎永远全量扫（实测） + +异步响应解析的两个坑（已在代码里处理）：start/status/cancel/resume 的 `content[0].text` 是**嵌套 JSON 字符串**（`mimosa-mcp-security-scan-job/v1`）而非 Markdown 摘要；完成判定必须 parse JSON 看 `job.status`——running 态也含 `"completedAt":null`，字符串匹配 `completed` 会误判。 > mimosa 的调用不依赖 zcode 插件体系：bridge 用自带极简 stdio MCP client 直接 spawn mimosa 的 `server.js`（env `ZCODE_PLUGIN_ROOT=`、`MIMOSA_ENGINE=native`，cwd=被扫项目）。mimosa 快扫是确定性规则引擎、零 LLM 流量，故不走 review 文件锁。 > > 实测备注（2026-08-08，GC-8G）：① 独立调用时 mimosa 也会在被扫项目写一个小会话状态文件（`.mimosa/hook-state/sess_*.continue.json`，约 200 字节，无害）——即 bridge 自身的代码路径对被扫目录只读，但 mimosa 引擎会落这个状态文件，说"完全只读"不准确；② 从非登录 shell（systemd unit、cron、`sudo -u` 直调）启动时 PATH 可能不含 `~/.local/bin`，需显式 `export PATH="$HOME/.local/bin:$PATH"` 否则找不到 `zcode`。 > -> 并发与阻塞边界（狗食 review P2-4/P2-5）：mimosa 预扫**不在** review 文件锁内（确定性引擎无 LLM 限流问题），只有 zcode 复核阶段持锁——并发扫同一项目时 mimosa 的 hook-state 文件各写各的会话，无冲突。最坏阻塞时长估算：锁等待 300s + 单次调用 `ZCODE_BRIDGE_REVIEW_TIMEOUT`（默认 300s）×（1 + `ZCODE_BRIDGE_MAX_RETRIES` 默认 3）+ 限流退避，极端情况单次 tool 调用可阻塞约 20 分钟，调用方应把 MCP 超时设到相应量级。 +> 并发与阻塞边界（狗食 review P2-4/P2-5）：mimosa 预扫**不在** review 文件锁内（确定性引擎无 LLM 限流问题），只有 zcode 复核阶段持锁——并发扫同一项目时 mimosa 的 hook-state 文件各写各的会话，无冲突。最坏阻塞时长估算：锁等待 300s + 单次调用 `ZCODE_BRIDGE_REVIEW_TIMEOUT`（默认 300s）×（1 + `ZCODE_BRIDGE_MAX_RETRIES` 默认 3）+ 限流退避，极端情况单次 tool 调用可阻塞约 20 分钟；depth=deep 时前面还要再加 mimosa 异步扫描预算（`ZCODE_BRIDGE_MIMOSA_DEEP_TIMEOUT` 默认 900s）。调用方应把 MCP 超时设到相应量级。 ### 聚焦审查 prompt 建议 diff --git a/packages/agent-help/zcode-agent-help b/packages/agent-help/zcode-agent-help index b2d892d..23f0df3 100755 --- a/packages/agent-help/zcode-agent-help +++ b/packages/agent-help/zcode-agent-help @@ -521,7 +521,7 @@ ECOSYSTEM = { "tools_exposed": [ {"name": "get_zcode_capabilities", "description": "返回 zcode 能力清单 (调 agent-help)"}, {"name": "zcode_review", "description": "调 zcode 审查代码 (yolo+写工具物理禁用: 免授权只读)"}, - {"name": "zcode_security_review", "description": "mimosa 规则引擎预扫 + zcode 只读复核 (安全专项)"}, + {"name": "zcode_security_review", "description": "mimosa 预扫 (normal 快扫/deep 业务逻辑深扫) + zcode 只读复核 (安全专项)"}, ], "config": "~/.zcode/cli/config.json 的 mcp.servers.zcode-mcp", "use_case": "让 MCP client (zcode 自身/Claude Code/Cursor) 标准化调用 zcode", diff --git a/packages/mcp-server/zcode-mcp-server b/packages/mcp-server/zcode-mcp-server index 8dd9e0a..1316fc6 100755 --- a/packages/mcp-server/zcode-mcp-server +++ b/packages/mcp-server/zcode-mcp-server @@ -261,6 +261,10 @@ REVIEW_DISALLOWED_TOOLS = ( "mcp__node_repl__js mcp__node_repl__js_reset mcp__node_repl__js_add_node_module_dir" ) +# deep 轮询: status 单次响应超时的最大连续容忍次数 (复审 P1-3: +# 不设上限时每次容忍最长白等 client 超时 120s, 会显著推迟总预算触发) +_MAX_STATUS_TIMEOUTS = 5 + def _find_mimosa_root(): """定位 mimosa 插件根目录 (含 payload/dist/mcp/server.js 的那层)。 @@ -362,15 +366,50 @@ class MimosaMcpClient: pass +def _mimosa_result_body(result): + """提取 mimosa MCP tool 响应的文本正文 (content 里所有 text 块拼接)。""" + return "\n".join( + c.get("text", "") for c in result.get("content", []) + if isinstance(c, dict) and c.get("type") == "text" and c.get("text")) + + +def _read_mimosa_findings(scan_dir_str): + """按 scanDir 回读 /findings.json 的 findings list。失败/越界返回 []。 + + 狗食 review P1-1 + 复审 P1-A: scanDir 是不可信输入 (子进程产物经文本协议 + 传回, 属信任边界外) — 目录与 findings.json resolve 后都必须落在扫描历史根 + (ZCODE_BRIDGE_MIMOSA_SCAN_ROOT, 默认 ~/.mimosa/security-scans) 之下, + 防路径注入 / symlink 绕过导致任意文件回读喂给 LLM。 + """ + scan_root = Path(os.environ.get( + "ZCODE_BRIDGE_MIMOSA_SCAN_ROOT", + str(Path.home() / ".mimosa" / "security-scans"))) + try: + scan_dir = Path(scan_dir_str).resolve() + resolved_root = scan_root.resolve() + if not (scan_dir == resolved_root or resolved_root in scan_dir.parents): + log(f"⚠ scanDir 越界 (不在 {resolved_root} 下), 跳过回读: {scan_dir_str}") + return [] + real_file = (scan_dir / "findings.json").resolve() + if not (resolved_root in real_file.parents or real_file.parent == resolved_root): + log(f"⚠ findings.json 解析后越界 (疑似 symlink), 跳过回读: {real_file}") + return [] + with open(real_file) as f: + return json.load(f).get("findings", []) + except Exception as e: + log(f"⚠ findings.json 读取失败 ({e}), 仅用摘要文本") + return [] + + def _mimosa_quick_scan(root, scan_path): """调 mimosa security_scan (同步快扫), 返回 (摘要文本, findings list)。失败抛异常。 security_scan 的 MCP 响应只有 Markdown 摘要 (scanId/scanDir/seal/计数), 全量 findings 需按摘要里的 scanDir 回读 /findings.json (schemaVersion: mimosa-security-scan-findings/v1)。depth=normal 跳过 - 业务逻辑投研, 秒级; deep 明显更慢, 未接 (需要时应走 start/status 异步对)。 + 业务逻辑投研, 秒级; deep 走 _mimosa_deep_scan 异步管线。 """ - timeout = max(30, _env_int("ZCODE_BRIDGE_MIMOSA_TIMEOUT", 180)) + timeout = max(30, _env_int("ZCODE_BRIDGE_MIMOSA_TIMEOUT", 180, maximum=3600)) client = MimosaMcpClient(root, cwd=scan_path, timeout=timeout) try: client.initialize() @@ -378,42 +417,130 @@ def _mimosa_quick_scan(root, scan_path): "security_scan", {"project": scan_path, "depth": "normal"}) finally: client.close() - texts = [c.get("text", "") for c in result.get("content", []) - if isinstance(c, dict) and c.get("type") == "text"] - body = "\n".join(t for t in texts if t) + body = _mimosa_result_body(result) if result.get("isError"): raise RuntimeError(f"security_scan 返回错误: {body[:500]}") findings = [] m = re.search(r"scanDir:\s*`([^`]+)`", body) if m: - # 狗食 review P1-1: scanDir 是不可信输入 (子进程产物经文本协议传回, - # 属信任边界外) — 必须校验落在 mimosa 扫描历史根之下, 防路径注入 - # 导致任意文件回读喂给 LLM。校验失败降级为仅用摘要文本。 - scan_root = Path(os.environ.get( - "ZCODE_BRIDGE_MIMOSA_SCAN_ROOT", - str(Path.home() / ".mimosa" / "security-scans"))) - try: - scan_dir = Path(m.group(1)).resolve() - resolved_root = scan_root.resolve() - if scan_dir == resolved_root or resolved_root in scan_dir.parents: - # 复审 P1-A: findings.json 本身 resolve 后也须在根下, - # 防"目录合法但 findings.json 是指向外部的 symlink"绕过 - real_file = (scan_dir / "findings.json").resolve() - if resolved_root in real_file.parents or real_file.parent == resolved_root: - with open(real_file) as f: - findings = json.load(f).get("findings", []) - else: - log(f"⚠ findings.json 解析后越界 (疑似 symlink), 跳过回读: {real_file}") - else: - log(f"⚠ scanDir 越界 (不在 {resolved_root} 下), 跳过回读: {m.group(1)}") - except Exception as e: - log(f"⚠ findings.json 读取失败 ({e}), 仅用摘要文本") + findings = _read_mimosa_findings(m.group(1)) else: log("⚠ mimosa 摘要中未解析到 scanDir, 仅用摘要文本") return body, findings +def _parse_scan_job(result): + """解析异步扫描四件 (start/status/cancel/resume) 的响应, 返回 job dict。 + + 响应 content[0].text 是嵌套 JSON 字符串 + (schemaVersion: mimosa-mcp-security-scan-job/v1), 不是 Markdown。 + """ + body = _mimosa_result_body(result) + if result.get("isError"): + raise RuntimeError(f"mimosa 异步扫描返回错误: {body[:500]}") + if not body: + raise RuntimeError("mimosa 异步扫描返回空响应") + try: + payload = json.loads(body) + except json.JSONDecodeError as e: + raise RuntimeError(f"mimosa 异步响应非 JSON (前80字符): {body[:80]!r}") from e + job = payload.get("job") + if not isinstance(job, dict): + raise RuntimeError(f"mimosa 异步响应缺 job 字段: {body[:200]}") + return job + + +def _mimosa_deep_scan(root, scan_path, focus_files=None): + """deep 深扫 (含业务逻辑投研): start → status 轮询 → completed 后回读 findings。 + + 实测 (mimosa 1.0.3, GC-8G): 400 文件项目 ~13s, 秒级到十几秒量级; + 纯 native 引擎零 LLM 零网络 (evidenceBoundary: static_only_no_runtime_execution)。 + 注意: 完成判定必须 parse JSON 看 job.status — running 态也含 + "completedAt":null, 字符串匹配 "completed" 会误判。 + focus_files 只是 deep 业务逻辑复核的优先级提示, 静态引擎永远全量扫 + (实测), 不能当"只扫这些文件"用。 + 超时/失败抛异常; 超时会 best-effort cancel 后台 job。 + """ + timeout = max(60, _env_int("ZCODE_BRIDGE_MIMOSA_DEEP_TIMEOUT", 900, maximum=7200)) + poll = max(1, _env_int("ZCODE_BRIDGE_MIMOSA_POLL_INTERVAL", 2, maximum=60)) + client = MimosaMcpClient(root, cwd=scan_path, timeout=120) + job_id = None + terminal = False # 到达终态 (completed/failed/cancelled) 则 finally 不再 cancel + status_timeouts = 0 + t0 = time.time() # 总预算含 start 开销 (狗食 review P0-1) + try: + client.initialize() + start_args = {"project": scan_path, "depth": "deep"} + if focus_files: + start_args["focusFiles"] = list(focus_files) + job = _parse_scan_job(client.call_tool("security_scan_start", start_args)) + job_id = job.get("jobId") + if not job_id: + raise RuntimeError("security_scan_start 未返回 jobId") + log(f"mimosa deep 扫描已启动: {job_id}") + + while True: + if time.time() - t0 > timeout: + raise TimeoutError( + f"mimosa deep 扫描超时 ({timeout}s), jobId={job_id}") + time.sleep(poll) + try: + # try 只包 call_tool (复审 P1-1: 范围收窄, 防未来 _parse_scan_job + # 内部引入 TimeoutError 被误吞成"继续轮询") + raw = client.call_tool("security_scan_status", {"jobId": job_id}) + except TimeoutError as e: + # 单次 status 响应超时 (mimosa 偶发卡顿/GC): 视同 running 继续, + # 由总预算兜底; 直接冒泡会留孤儿 job (狗食 review P0-2) + status_timeouts += 1 + log(f"⚠ status 轮询单次超时 ({status_timeouts}/" + f"{_MAX_STATUS_TIMEOUTS}), 视同 running 继续: {e}") + if status_timeouts >= _MAX_STATUS_TIMEOUTS: + # 复审 P1-3: 连续超时过多会显著推迟总预算触发, 设上限 + raise RuntimeError( + f"mimosa deep status 连续 {status_timeouts} 次超时, " + f"放弃轮询 (jobId={job_id})") from e + continue + status_timeouts = 0 + job = _parse_scan_job(raw) + status = job.get("status") + if status == "completed": + terminal = True + break + if status in ("failed", "cancelled"): + terminal = True # 复审 P1-2: 终态无需再 cancel + err = job.get("error") or {} + raise RuntimeError( + f"mimosa deep 扫描 {status}: " + f"{err.get('message') or err or '未知错误'} (jobId={job_id})") + + result = job.get("result") or {} + scan_dir = result.get("scanDir") + dep = result.get("dependencySummary") or {} + summary = ( + f"**Mimosa deep security scan (异步 job {job_id})**\n" + f"- scanId: `{result.get('scanId')}`\n" + f"- scanDir: `{scan_dir}`\n" + f"- seal: `{result.get('seal')}`\n" + f"- findings: {result.get('findingCount')}\n" + f"- business-logic hypotheses: {len(result.get('hypotheses') or [])}\n" + f"- dependency: {dep.get('packagesScanned', 0)} package(s) " + f"(completion={dep.get('completion') or 'unknown'})" + ) + findings = _read_mimosa_findings(scan_dir) if scan_dir else [] + return summary, findings + finally: + # 狗食 review P0-3: start 之后任何未到终态的退出路径 (解析异常/ + # 超时/调用方中断) 统一 best-effort cancel, 不留孤儿 job; + # cancel 失败也要留诊断痕迹 (P1-5); 已到终态不重复 cancel (复审 P1-2) + if job_id and not terminal: + try: + client.call_tool("security_scan_cancel", {"jobId": job_id}) + except Exception as e: + log(f"⚠ best-effort cancel 失败 (jobId={job_id}): {e}") + client.close() + + def _compact_findings(findings): """findings 投影到复核所需字段 (控制附件体积; 同类多 occurrence 逐条保留)。""" out = [] @@ -494,9 +621,10 @@ TOOLS = [ { "name": "zcode_security_review", "description": ( - "安全专项审查,两阶段: ① mimosa 确定性规则引擎全仓快扫 (零 LLM 流量," + "安全专项审查,两阶段: ① mimosa 确定性规则引擎扫描 (零 LLM 流量," "覆盖面兜底) → ② headless ZCode 拿着 findings 逐条核实 (确认真漏洞/" "排除误报/给出攻击路径与修复建议),并可发现清单之外的问题。" + "depth=normal 秒级快扫 (默认); depth=deep 含业务逻辑投研, 走异步任务, 稍慢。" "zcode 同样以 yolo+写工具物理禁用运行,只输出报告不动代码。" "需要本机装有 mimosa 插件 (或设 ZCODE_BRIDGE_MIMOSA_ROOT 指向插件根)。" ), @@ -507,6 +635,18 @@ TOOLS = [ "type": "string", "description": "要扫描审查的项目目录,默认当前目录", }, + "depth": { + "type": "string", + "enum": ["normal", "deep"], + "description": "扫描深度: normal=规则快扫 (默认, 秒级); deep=含业务逻辑投研 (异步, 更彻底)", + "default": "normal", + }, + "focus_files": { + "type": "array", + "items": {"type": "string"}, + "description": "可选, 仅 deep: 优先复核的项目内文件 (如本次改动的文件)。" + "注意是优先级提示不是过滤器, 静态引擎仍全量扫", + }, "focus": { "type": "string", "description": "可选,额外审查重点,如 '重点关注注入与鉴权'", @@ -568,7 +708,7 @@ def _run_zcode_headless(cmd, env, timeout): """ # 有限重试配置 (issue #3 子项3c): 仅对 provider 限流错误重试, 退避指数。 # clamp 到 >=0, 防止 ZCODE_BRIDGE_MAX_RETRIES=-1 导致不执行 zcode (Codex P2) - max_retries = max(0, _env_int("ZCODE_BRIDGE_MAX_RETRIES", 3)) + max_retries = max(0, _env_int("ZCODE_BRIDGE_MAX_RETRIES", 3, maximum=10)) last_err = None try: @@ -619,7 +759,7 @@ def _run_zcode_headless(cmd, env, timeout): def _review_timeout(): """review 单次 zcode 调用超时 (秒), ZCODE_BRIDGE_REVIEW_TIMEOUT 可配。""" - return max(30, _env_int("ZCODE_BRIDGE_REVIEW_TIMEOUT", 300)) + return max(30, _env_int("ZCODE_BRIDGE_REVIEW_TIMEOUT", 300, maximum=3600)) def _write_temp(text, prefix): @@ -685,15 +825,35 @@ def tool_zcode_review(args): def tool_zcode_security_review(args): """mimosa 规则引擎预扫 + zcode 只读复核 的安全专项管线。 - ① mimosa security_scan (确定性规则, 零 LLM 流量) 全仓快扫 → findings 清单, - 解决"AI 自由探索覆盖面随缘"的问题; + ① mimosa security_scan (确定性规则, 零 LLM 流量) 扫描 → findings 清单, + 解决"AI 自由探索覆盖面随缘"的问题; depth=normal 秒级快扫, + depth=deep 走异步 start/status 管线, 含业务逻辑投研; ② findings 作附件喂给 zcode, 逐条核实 (确认/误报/存疑 + 攻击路径 + 修复建议), 读工具全开可查证上下文, 写工具物理禁用保证只审不修。 """ scan_path = args.get("path") or args.get("cwd") or os.getcwd() focus = args.get("focus", "") - cwd = args.get("cwd") or scan_path + depth = args.get("depth", "normal") + focus_files = args.get("focus_files") or [] + if depth not in ("normal", "deep"): + return {"content": [{"type": "text", + "text": f"非法 depth: {depth!r} (只支持 normal|deep)"}], + "isError": True} + # focus_files 来自 MCP client (信任边界外): 类型校验 + 长度上限 + # (狗食 review P1-4: list("app.py") 会静默炸成单字符列表) + if not isinstance(focus_files, list): + return {"content": [{"type": "text", + "text": "focus_files 必须是字符串数组"}], + "isError": True} + if len(focus_files) > 200: + log(f"⚠ focus_files 超上限 ({len(focus_files)} > 200), 截断") + focus_files = focus_files[:200] + # 元素级过滤: 非字符串元素丢弃而非 stringify (复审 P2-1) + valid_files = [f for f in focus_files if isinstance(f, str)] + if len(valid_files) != len(focus_files): + log(f"⚠ focus_files 丢弃 {len(focus_files) - len(valid_files)} 个非字符串元素") + focus_files = valid_files if not os.path.isdir(scan_path): # 狗食 review P2-8: 先校验, 错误信息明确 return {"content": [{"type": "text", "text": f"扫描目录不存在或不是目录: {scan_path}"}], @@ -706,16 +866,19 @@ def tool_zcode_security_review(args): "指向插件根目录 (含 payload/ 的那层);也可改用 zcode_review 做纯 AI 审查。" )}], "isError": True} - # ① mimosa 快扫 (不走 review 文件锁: 确定性引擎无 LLM 限流问题) - log(f"mimosa 快扫: {scan_path} (root={root})") + # ① mimosa 扫描 (不走 review 文件锁: 确定性引擎无 LLM 限流问题) + log(f"mimosa {depth} 扫描: {scan_path} (root={root})") try: - summary_text, findings = _mimosa_quick_scan(root, scan_path) + if depth == "deep": + summary_text, findings = _mimosa_deep_scan(root, scan_path, focus_files) + else: + summary_text, findings = _mimosa_quick_scan(root, scan_path) except Exception as e: return {"content": [{"type": "text", "text": f"mimosa 扫描失败: {e}"}], "isError": True} # 附件 = 扫描摘要 + 结构化 findings (投影到复核所需字段) - attachment = f"# mimosa 扫描摘要\n{summary_text}\n" + attachment = f"# mimosa 扫描摘要 (depth={depth})\n{summary_text}\n" if findings: compact = _compact_findings(findings) attachment += ( @@ -731,7 +894,8 @@ def tool_zcode_security_review(args): try: tmp_path = _write_temp(attachment, "zcode-mimosa-findings-") prompt = ( - "你是资深安全审查专家。附件是安全扫描器 (mimosa,确定性规则引擎) 对本项目的扫描结果。\n" + f"你是资深安全审查专家。附件是安全扫描器 (mimosa,确定性规则引擎,depth={depth})" + "对本项目的扫描结果。\n" "任务:\n" "1. 逐条核实每个 finding: 用只读工具 (Read/Grep/Glob) 阅读相关代码,判定为" "【确认漏洞】/【误报】/【存疑】,给出依据 (可利用的攻击路径,或为什么不可利用)。\n" @@ -744,7 +908,7 @@ def tool_zcode_security_review(args): if focus: prompt += f"\n额外审查重点: {focus}。" - cmd = _build_review_cmd(prompt, cwd) + cmd = _build_review_cmd(prompt, args.get("cwd") or scan_path) cmd += ["--attach", tmp_path] log("调用 zcode 复核 mimosa findings (yolo+只读黑名单)") @@ -758,12 +922,16 @@ def tool_zcode_security_review(args): pass -def _env_int(name, default): - """从环境变量读整数, 失败用默认值。""" +def _env_int(name, default, maximum=None): + """从环境变量读整数, 失败用默认值; maximum 给上界 (复审 R2 P2-5: + 之前只靠调用点 max() 兜下限, env 误设天文数字没有防线)。""" try: - return int(os.environ.get(name, str(default))) + v = int(os.environ.get(name, str(default))) except (TypeError, ValueError): return default + if maximum is not None: + v = min(v, maximum) + return v TOOL_HANDLERS = { diff --git a/skills/zcode-bridge-guide/SKILL.md b/skills/zcode-bridge-guide/SKILL.md index c951d52..83fe69a 100644 --- a/skills/zcode-bridge-guide/SKILL.md +++ b/skills/zcode-bridge-guide/SKILL.md @@ -338,10 +338,12 @@ MCP server 暴露三个标准 MCP tool，供 Claude Code / Cursor 等 MCP client ### `zcode_security_review` 参数 - `path`：要扫描审查的项目目录（默认当前目录） +- `depth`：`normal`（默认，同步快扫秒级）/ `deep`（含业务逻辑投研，异步 start/status 管线，400 文件项目 ~13s） +- `focus_files`：可选，仅 deep：优先复核的项目内文件（如本次改动的文件）。**是优先级提示不是过滤器**，静态引擎仍全量扫 - `focus`：可选，额外审查重点（如 "重点关注注入与鉴权"） - `cwd`：zcode 工作目录（默认与 path 相同） -> 两阶段流程：① mimosa `security_scan`（确定性规则引擎，零 LLM 流量）全仓快扫出 findings；② findings 作 `--attach` 附件喂 zcode 逐条核实（确认漏洞/误报/存疑 + 攻击路径 + 修复建议），并可发现清单之外的问题。mimosa 定位：`ZCODE_BRIDGE_MIMOSA_ROOT` 优先，否则探测 `~/.local/share/mimosa/*` 与 `~/.zcode/cli/plugins/cache/*/mimosa/*`；扫描超时 `ZCODE_BRIDGE_MIMOSA_TIMEOUT`（默认 180s）。 +> 两阶段流程：① mimosa 扫描（确定性规则引擎，零 LLM 流量零网络）出 findings；② findings 作 `--attach` 附件喂 zcode 逐条核实（确认漏洞/误报/存疑 + 攻击路径 + 修复建议），并可发现清单之外的问题。mimosa 定位：`ZCODE_BRIDGE_MIMOSA_ROOT` 优先，否则探测 `~/.local/share/mimosa/*` 与 `~/.zcode/cli/plugins/cache/*/mimosa/*`；超时 env：normal 档 `ZCODE_BRIDGE_MIMOSA_TIMEOUT`（默认 180s），deep 档 `ZCODE_BRIDGE_MIMOSA_DEEP_TIMEOUT`（默认 900s）+ 轮询间隔 `ZCODE_BRIDGE_MIMOSA_POLL_INTERVAL`（默认 2s）。 --- diff --git a/tests/test_security_review.py b/tests/test_security_review.py index 6488799..64fca88 100644 --- a/tests/test_security_review.py +++ b/tests/test_security_review.py @@ -69,7 +69,9 @@ class _EnvGuard(unittest.TestCase): """保存/恢复本文件用到的环境变量。""" ENV_KEYS = ("ZCODE_BRIDGE_REVIEW_LOCK", "ZCODE_BRIDGE_MIMOSA_ROOT", - "ZCODE_BRIDGE_REVIEW_TIMEOUT", "ZCODE_BRIDGE_MIMOSA_SCAN_ROOT") + "ZCODE_BRIDGE_REVIEW_TIMEOUT", "ZCODE_BRIDGE_MIMOSA_SCAN_ROOT", + "ZCODE_BRIDGE_MIMOSA_DEEP_TIMEOUT", + "ZCODE_BRIDGE_MIMOSA_POLL_INTERVAL") def setUp(self): self._saved = {k: os.environ.get(k) for k in self.ENV_KEYS} @@ -408,8 +410,14 @@ def _patch_common(self, summary="摘要: findings 1", "scan": mod._mimosa_quick_scan, "run": mod.subprocess.run, } - mod._find_mimosa_root = lambda: "/fake/mimosa" - mod._mimosa_quick_scan = lambda root, path: (summary, findings) + def fake_find_root(): + return "/fake/mimosa" + + def fake_scan(root, path): + return (summary, findings) + + mod._find_mimosa_root = fake_find_root + mod._mimosa_quick_scan = fake_scan os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" return mod, saved, proj @@ -422,7 +430,11 @@ def test_sr0_mimosa_missing_clean_error(self): """SR0: mimosa 未安装 → 明确报错并建议 fallback (不崩)""" mod = self.mod saved = mod._find_mimosa_root - mod._find_mimosa_root = lambda: None + + def no_mimosa(): + return None + + mod._find_mimosa_root = no_mimosa try: result = mod.tool_zcode_security_review({"path": "/tmp"}) finally: @@ -509,5 +521,312 @@ def test_sr3_findings_tmpfile_cleaned(self): self.assertFalse(os.path.exists(attach_path), "临时 findings 文件应被清理") +def _make_deep_stub(statuses, scan_dir=None): + """构造 deep 异步管线 stub 类, 返回 (StubClass, calls_log)。 + + 每次调用生成全新的类与调用记录 — 不用类变量收集实例 + (狗食 review R1 P2-5: 类变量全局状态在并行/泄漏场景下脆弱)。 + """ + calls_log = [] + + class Stub: + def __init__(self, root, cwd, timeout=120): + self.calls = calls_log + self._status_idx = 0 + + def initialize(self): + pass + + def call_tool(self, name, arguments): + self.calls.append((name, arguments)) + if name == "security_scan_start": + job = {"jobId": "job-1", "status": "running"} + elif name == "security_scan_status": + st = statuses[min(self._status_idx, len(statuses) - 1)] + self._status_idx += 1 + job = {"jobId": "job-1", "status": st} + if st == "completed": + job["result"] = {"scanId": "s1", "scanDir": scan_dir, + "seal": "sha256:x", "findingCount": 1, + "hypotheses": [], "dependencySummary": {}} + if st == "failed": + job["error"] = {"message": "engine exploded"} + elif name == "security_scan_cancel": + job = {"jobId": "job-1", "status": "cancel_requested"} + else: + raise AssertionError(f"未预期的 tool: {name}") + text = json.dumps( + {"schemaVersion": "mimosa-mcp-security-scan-job/v1", "job": job}) + return {"content": [{"type": "text", "text": text}]} + + def close(self): + pass + + return Stub, calls_log + + +class TestParseScanJob(unittest.TestCase): + """_parse_scan_job: 异步响应 (嵌套 JSON 字符串) 解析""" + + @classmethod + def setUpClass(cls): + cls.mod = _load_mcp_module() + + def _result(self, text=None, is_error=False): + r = {"content": [{"type": "text", "text": text}] if text is not None else []} + if is_error: + r["isError"] = True + return r + + def test_pj0_valid(self): + text = json.dumps({"schemaVersion": "mimosa-mcp-security-scan-job/v1", + "job": {"jobId": "j1", "status": "running"}}) + job = self.mod._parse_scan_job(self._result(text)) + self.assertEqual(job["jobId"], "j1") + + def test_pj1_is_error_raises(self): + with self.assertRaises(RuntimeError): + self.mod._parse_scan_job(self._result("boom", is_error=True)) + + def test_pj2_non_json_raises(self): + with self.assertRaises(RuntimeError) as ctx: + self.mod._parse_scan_job(self._result("not json at all")) + self.assertIn("非 JSON", str(ctx.exception)) + + def test_pj3_missing_job_raises(self): + with self.assertRaises(RuntimeError) as ctx: + self.mod._parse_scan_job(self._result(json.dumps({"other": 1}))) + self.assertIn("job", str(ctx.exception)) + + def test_pj4_empty_raises(self): + with self.assertRaises(RuntimeError): + self.mod._parse_scan_job(self._result(None)) + + +class TestMimosaDeepScan(_EnvGuard): + """_mimosa_deep_scan: start/status 轮询/回读/失败/超时""" + + @classmethod + def setUpClass(cls): + cls.mod = _load_mcp_module() + + def _run_deep(self, stub_cls, focus_files=None): + mod = self.mod + saved_client = mod.MimosaMcpClient + saved_sleep = mod.time.sleep + mod.MimosaMcpClient = stub_cls + mod.time.sleep = lambda s: None + try: + return mod._mimosa_deep_scan("/fake/root", "/fake/proj", focus_files) + finally: + mod.MimosaMcpClient = saved_client + mod.time.sleep = saved_sleep + + def _make_scan_dir(self): + import tempfile + scan_root = tempfile.mkdtemp(prefix="mimosa-scans-") + self.addCleanup(lambda: __import__("shutil").rmtree(scan_root, ignore_errors=True)) + scan_dir = os.path.join(scan_root, "project-x", "scan-1") + os.makedirs(scan_dir) + with open(os.path.join(scan_dir, "findings.json"), "w") as f: + json.dump({"findings": [ + {"identity": {"publicClass": "sql-injection"}, + "severity": "high", "cwe": ["CWE-89"], + "location": {"path": "app.py", "line": 11}, + "title": "SQL 注入", "message": "拼接查询"}]}, f) + os.environ["ZCODE_BRIDGE_MIMOSA_SCAN_ROOT"] = scan_root + return scan_dir + + def test_ds0_happy_path(self): + """DS0: start→running→completed, findings 回读, focusFiles 透传""" + stub, calls = _make_deep_stub(["running", "completed"], + scan_dir=self._make_scan_dir()) + summary, findings = self._run_deep(stub, focus_files=["app.py"]) + self.assertIn("deep", summary) + self.assertIn("job-1", summary) + self.assertEqual(len(findings), 1) + start_call = [a for n, a in calls if n == "security_scan_start"][0] + self.assertEqual(start_call["depth"], "deep") + self.assertEqual(start_call["focusFiles"], ["app.py"]) + status_calls = [n for n, _ in calls if n == "security_scan_status"] + self.assertEqual(len(status_calls), 2, "running 一次 + completed 一次") + + def test_ds1_failed_raises_with_message(self): + """DS1: status=failed → 抛错带 error.message; 终态不重复 cancel (P1-2)""" + stub, calls = _make_deep_stub(["running", "failed"]) + with self.assertRaises(RuntimeError) as ctx: + self._run_deep(stub) + self.assertIn("engine exploded", str(ctx.exception)) + self.assertIn("failed", str(ctx.exception)) + names = [n for n, _ in calls] + self.assertNotIn("security_scan_cancel", names, "failed 终态不应再 cancel") + + def test_ds2_timeout_cancels_job(self): + """DS2: 一直 running + 超时 → TimeoutError 且 finally 统一 cancel (P0-3)""" + mod = self.mod + stub, calls = _make_deep_stub(["running"]) + + saved_client = mod.MimosaMcpClient + saved_sleep = mod.time.sleep + saved_time = mod.time.time + mod.MimosaMcpClient = stub + mod.time.sleep = lambda s: None + # 单调递增无界 fake clock (P0-1 修复 + R2 P2-2: itertools.count 无上限) + import itertools + real_t = saved_time() + ticks = itertools.count(0, 5000) # 每次调用 +5000s → 首轮即超 900s 预算 + mod.time.time = lambda: real_t + next(ticks) + try: + with self.assertRaises(TimeoutError): + mod._mimosa_deep_scan("/fake/root", "/fake/proj") + finally: + mod.MimosaMcpClient = saved_client + mod.time.sleep = saved_sleep + mod.time.time = saved_time + names = [n for n, _ in calls] + self.assertIn("security_scan_cancel", names, "超时应在 finally 统一 cancel") + + def test_ds3_no_jobid_raises(self): + """DS3: start 响应缺 jobId → 明确报错 (patch sleep 保持与其他用例一致)""" + mod = self.mod + + class NoJobClient: + def __init__(self, root, cwd, timeout=120): + pass + + def initialize(self): + pass + + def call_tool(self, name, arguments): + assert name == "security_scan_start" + return {"content": [{"type": "text", "text": json.dumps( + {"job": {"status": "running"}})}]} + + def close(self): + pass + + saved_client = mod.MimosaMcpClient + saved_sleep = mod.time.sleep + mod.MimosaMcpClient = NoJobClient + mod.time.sleep = lambda s: None + try: + with self.assertRaises(RuntimeError) as ctx: + mod._mimosa_deep_scan("/fake/root", "/fake/proj") + self.assertIn("jobId", str(ctx.exception)) + finally: + mod.MimosaMcpClient = saved_client + mod.time.sleep = saved_sleep + + +class TestSecurityReviewDepth(_EnvGuard): + """tool_zcode_security_review 的 depth/focus_files 参数""" + + @classmethod + def setUpClass(cls): + cls.mod = _load_mcp_module() + + def _patch(self): + import tempfile + mod = self.mod + proj = tempfile.mkdtemp(prefix="zcode-scan-proj-") + self.addCleanup(lambda: __import__("shutil").rmtree(proj, ignore_errors=True)) + saved = { + "find_root": mod._find_mimosa_root, + "quick": mod._mimosa_quick_scan, + "deep": mod._mimosa_deep_scan, + "run": mod.subprocess.run, + } + + def fake_find_root(): + return "/fake/mimosa" + + mod._find_mimosa_root = fake_find_root + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + return mod, saved, proj + + def _restore(self, mod, saved): + mod._find_mimosa_root = saved["find_root"] + mod._mimosa_quick_scan = saved["quick"] + mod._mimosa_deep_scan = saved["deep"] + mod.subprocess.run = saved["run"] + + def test_dp0_invalid_depth_rejected(self): + """DP0: 非法 depth → 明确报错""" + mod = self.mod + result = mod.tool_zcode_security_review( + {"path": "/tmp", "depth": "turbo"}) + self.assertTrue(result.get("isError")) + self.assertIn("depth", result["content"][0]["text"]) + + def test_dp0b_focus_files_must_be_list(self): + """DP0b: focus_files 传字符串 → 明确拒绝 (狗食 review P1-4: + list("app.py") 会静默炸成单字符列表)""" + mod = self.mod + result = mod.tool_zcode_security_review( + {"path": "/tmp", "depth": "deep", "focus_files": "app.py"}) + self.assertTrue(result.get("isError")) + self.assertIn("focus_files", result["content"][0]["text"]) + + def test_dp1_default_is_normal(self): + """DP1: 不传 depth → 走 normal 快扫, 不碰 deep""" + mod, saved, proj = self._patch() + called = {"quick": 0, "deep": 0} + + def fake_quick(r, p): + called["quick"] += 1 + return ("摘要", []) + + def deep_should_not_run(r, p, f=None): + called["deep"] += 1 + return ("", []) + + mod._mimosa_quick_scan = fake_quick + mod._mimosa_deep_scan = deep_should_not_run + mod.subprocess.run = lambda *a, **kw: _FakeCompletedProcess(0, "OK", "") + try: + result = mod.tool_zcode_security_review({"path": proj}) + finally: + self._restore(mod, saved) + self.assertNotIn("isError", result) + self.assertEqual(called["quick"], 1) + self.assertEqual(called["deep"], 0) + + def test_dp2_deep_routes_with_focus_files(self): + """DP2: depth=deep → 走异步管线, focus_files 透传, 附件标注 depth=deep""" + mod, saved, proj = self._patch() + captured = {} + + def quick_should_not_run(r, p): + raise AssertionError("depth=deep 不应走 normal 快扫") + + mod._mimosa_quick_scan = quick_should_not_run + + def fake_deep(r, p, focus_files=None): + captured["focus_files"] = focus_files + return ("deep 摘要", [{"identity": {"publicClass": "x"}, + "severity": "high", "cwe": [], + "location": {"path": "a.py", "line": 1}, + "title": "t", "message": "m"}]) + + mod._mimosa_deep_scan = fake_deep + + def fake_run(cmd, *a, **kw): + captured["cmd"] = cmd + return _FakeCompletedProcess( + 0, json.dumps({"response": "deep 报告"}, ensure_ascii=False), "") + + mod.subprocess.run = fake_run + try: + result = mod.tool_zcode_security_review( + {"path": proj, "depth": "deep", "focus_files": ["a.py"]}) + finally: + self._restore(mod, saved) + self.assertNotIn("isError", result) + self.assertEqual(result["content"][0]["text"], "deep 报告") + self.assertEqual(captured["focus_files"], ["a.py"]) + prompt = captured["cmd"][captured["cmd"].index("--prompt") + 1] + self.assertIn("depth=deep", prompt) + + if __name__ == "__main__": unittest.main(verbosity=2)