From b8ff37b81a2cb44c2d8709cbfecb7d0a9816cc16 Mon Sep 17 00:00:00 2001 From: 6tizer <6tizer@gmail.com> Date: Sat, 8 Aug 2026 13:02:28 -0400 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=95=B4=E4=BD=93=20deep=20review=20?= =?UTF-8?q?=E5=85=A8=E9=87=8F=E4=BF=AE=E5=A4=8D=20=E2=80=94=20=E6=97=A0=20?= =?UTF-8?q?P0/P1,=20P2=20=E6=B8=85=E9=9B=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 五路 zcode deep review (security deep + 四组件) 共 15 P1 + 31 P2, 复审 又抓 2 P1 + 7 P2, 全部修复: mcp-server: - MimosaMcpClient._send 写超时 (select 可写性检查, 防 server 异常时永久阻塞) - files 200 上限+元素类型过滤 / code 体积上限 ZCODE_BRIDGE_CODE_MAX / stdout 截断 ZCODE_BRIDGE_MAX_OUTPUT (先提取 response 再截断, 保 JSON 结构) - has_error 判定: stderr 命中错误词时仅当 stdout 不可解析出 --json 结果 才判错误 (修误杀成功 + 防漏判错误回显); _parse_result_json 容忍前置警告行 - -32603 改通用文案防内部路径泄露; provider_error 副本对齐权威版 quota 模式 - batch 拒收回数组形式; _git 超时分档; O_NOFOLLOW 防 TOCTOU; node 缺失明确提示 acp-bridge: - _next_id 改 itertools.count 原子化; stdout 写口收编 _write_stdout + 锁 - pending_turns try/finally 兜底清理; 轮询连续失败探测后端死进程立即报错 - _fetch_last_reply 重试间隙 drain inbox; 事件流超时返回 -32603 (契约变更, README 限制 #9 记录); _wait_for_turn_idle 退避+drain; close() kill 后收尸 - session 状态 FIFO 上限淘汰; _redact_secret 补 hex/uuid; register 覆盖告警 - ZCODE_ACP_DEFAULT_MODE 可收紧默认 yolo + 启动显眼告警 agent-help / shared: - _safe_host 与权威版逐字对齐 (残留诊断不再与注入行为矛盾); --section 越界报错; creds 空时显示 env 残留; 异常捕获放宽; 副本同步约定列入 agent-help + C11/C12 对拍测试 docs: - README eval 示例注入修复 (shlex.quote, 实测注入被阻断); client 可信假设 声明; 新 env 表; 事件超时契约限制 #9 测试: 全套件 10 文件全绿 (新增约 60 用例); ruff 全过 --- README.md | 23 +- packages/acp-bridge/zcode-acp-bridge | 367 ++++++++++++++++++--------- packages/agent-help/zcode-agent-help | 130 +++++++--- packages/mcp-server/zcode-mcp-server | 182 ++++++++++--- shared/credentials.py | 11 +- tests/test_app_server_methods.py | 223 ++++++++++++++++ tests/test_credentials.py | 272 ++++++++++++++++++++ tests/test_event_translator.py | 82 ++++++ tests/test_mcp_protocol.py | 67 ++++- tests/test_mcp_retry_lock.py | 135 ++++++++++ tests/test_polling_failure.py | 65 +++++ tests/test_security_review.py | 268 +++++++++++++++++++ 12 files changed, 1616 insertions(+), 209 deletions(-) diff --git a/README.md b/README.md index 3f9157f..06386ea 100644 --- a/README.md +++ b/README.md @@ -60,21 +60,24 @@ ln -s /opt/ZCode/app/resources/glm/zcode.cjs ~/.local/bin/zcode zcode() { local cfg="$HOME/.zcode/v2/config.json" # 从配置文件动态读取凭证注入环境变量 + # 值一律经 shlex.quote 转义后再交给 eval，防 config 值含 shell 元字符时命令注入 eval "$(python3 -c " -import json +import json, shlex c=json.load(open('$cfg')) for k,v in c['provider'].items(): if v.get('enabled'): o=v['options'] - print(f'export ZCODE_MODEL=\"{next(iter(v.get(\"models\",{}))) or \"GLM-5.2\"}\"') - print(f'export ZCODE_BASE_URL=\"{o.get(\"baseURL\",\"\")}\"') - print(f'export ANTHROPIC_API_KEY=\"{o.get(\"apiKey\",\"\")}\"') + print('export ZCODE_MODEL=' + shlex.quote(next(iter(v.get('models',{}))) or 'GLM-5.2')) + print('export ZCODE_BASE_URL=' + shlex.quote(o.get('baseURL',''))) + print('export ANTHROPIC_API_KEY=' + shlex.quote(o.get('apiKey',''))) break ")" command zcode "$@" } ``` +> ⚠️ **安全说明**（整体 review 安全新发现 1）：`eval` + 未转义插值是危险模板——若 `config.json` 的值含 `"`、`` ` ``、`$` 等元字符会被 `eval` 执行，所以上面示例对每个值都做了 `shlex.quote`。更稳妥的做法是**不复用这段 shell 函数**，直接用本项目的 `shared/credentials.py`（三个组件已内置，纯 `json.load` 读 config，不走 eval/shell），或参考 `--print-injected-env` 诊断输出手动 export。 + ### 使用三个组件 ```bash @@ -116,6 +119,8 @@ zcode --prompt "继续" --resume sess_xxxx > **只读原理（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 层另有「只审不修」职责约束（不修改文件、不提议帮忙修复）作双保险。 +> **client 可信假设**：`zcode_review` 等 tool 的 `path`/`cwd`/`files` 参数**不做沙箱限制**（可让 ZCode 扫描本机任意目录）。本 server 是本地 stdio 桥，**假设 MCP client 可信**；若要用于远程/多租户部署，需自行在调用侧加路径白名单（整体 review P2 文档项）。 + ### ACP bridge（`zcode-acp-bridge`） 实现 ACP 协议的子集，把 ZCode 的私有协议翻译成标准 ACP： @@ -245,6 +250,10 @@ ZCODE_BASE_URL=https://api.z.ai/api/anthropic ./packages/mcp-server/zcode-mcp-se | **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） | +| **code 参数体积上限** | `zcode_review` 的 `code` 参数超过上限即截断，防超大内联代码撑爆调用 | `ZCODE_BRIDGE_CODE_MAX`（默认 500KB） | +| **zcode 输出体积上限** | zcode stdout 输出超过上限即截断 | `ZCODE_BRIDGE_MAX_OUTPUT`（默认 10MB） | + +> **内存峰值取舍**（整体 review P2-1）：zcode 子进程用 `capture_output` 全量缓冲输出，`ZCODE_BRIDGE_MAX_OUTPUT` 是**事后截断**——内存峰值仍约为完整输出的一倍（`text=True` 解码再翻一倍），上限（100MB）只是给失控场景兜底，不是流式背压。审查超大项目时建议拆分文件/目录分批调用，而不是把该值调大硬扛。 注意：zcode 内部已有自己的指数退避重试（`_retryWithExponentialBackoff`），MCP 层的重试是补充，默认保守（max 3）。 @@ -257,6 +266,9 @@ ZCODE_BASE_URL=https://api.z.ai/api/anthropic ./packages/mcp-server/zcode-mcp-se | `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） | +| `ZCODE_BRIDGE_MIMOSA_RECV_TIMEOUT` | mimosa stdio 单次响应（一问一答）超时（默认 120s） | + +ACP bridge 侧另有一个 env（不在上两表，仅 ACP 用）：`ZCODE_ACP_DEFAULT_MODE` —— `session/new` 的默认权限模式，默认 `yolo`，可设 `build` 收紧（详见下方「重要限制」#7）。 **depth 两档**（2026-08-08 接入，mimosa 1.0.3 实测）： @@ -336,8 +348,9 @@ cp -r skills/zcode-bridge-guide ~/.zcode/skills/ 4. **diff 无内容**：ZCode 协议层不暴露 oldText/newText，只能列文件名。 5. **GLM-5.2 无推理输出**：思考过程（agent_thought_chunk）在 GLM-5.2 下不触发，需 GLM-5-Turbo。 6. **TUI 不可用**：0.16.1 起 CLI 帮助虽列出 `tui` 命令（无参数即进入 TUI），但独立终端实测仍报错（`Cannot find package '@zcode/tui'`），仅 headless 模式可用。 -7. **⚠️ ACP bridge 默认 `mode=yolo`（权限风险）**：为避免工具调用 turn 卡在权限确认，ACP bridge 的 `session/new` 强制以 `mode=yolo` 创建会话（见 `zcode-acp-bridge` 的 `_on_session_new`）。这意味着任意 prompt 都可能触发**无确认的文件修改和命令执行**。作为编辑器集成时请知悉此风险；如需更安全的 `build` 模式（带权限确认），需自行修改并实现 ACP↔ZCode 的 permission 转发（本项目 P4b 未实现）。 +7. **⚠️ ACP bridge 默认 `mode=yolo`（权限风险）**：为避免工具调用 turn 卡在权限确认，ACP bridge 的 `session/new` 强制以 `mode=yolo` 创建会话（见 `zcode-acp-bridge` 的 `_on_session_new`）。这意味着任意 prompt 都可能触发**无确认的文件修改和命令执行**。作为编辑器集成时请知悉此风险；现可用 `ZCODE_ACP_DEFAULT_MODE=build` 收紧默认值，且 bridge 启动日志（stderr）会对当前默认 mode 打显眼告警。更完整的方案是实现 ACP↔ZCode 的 permission 转发（本项目 P4b 未实现）。 8. **⚠️ Provider 管理方法涉及 apiKey**：`workspace/upsertModelProvider`、`workspace/updateProviderRegistry` 的 `provider`/`registry` 参数会携带 `apiKey`（可能为 `{source:"inline", value:"sk-..."}` 明文）。ACP bridge 仅整体透传给 ZCode 后端、不读取也不在日志打印其明文；但调用方应自行确保传输通道（stdio）可信，并避免在日志中回显原始参数。 +9. **⚠️ 事件模式 turn 超时契约（2026-08-08 起）**：`session/prompt` 在事件模式下若 turn 已启动但 120s 未收到完成信号，返回 **JSON-RPC 错误 `-32603`（"事件流超时"）**，而**不是**正常 `stopReason=max_turn_requests`——后者只保留给"turn 从未启动"的场景。ACP client 侧应按此区分「卡死」与「真的太长」（整体 review P1 + 复审 P1-B 的契约变更）。 ## 项目结构 diff --git a/packages/acp-bridge/zcode-acp-bridge b/packages/acp-bridge/zcode-acp-bridge index 8f9391f..31d8b26 100755 --- a/packages/acp-bridge/zcode-acp-bridge +++ b/packages/acp-bridge/zcode-acp-bridge @@ -73,15 +73,19 @@ CLI 0.16.1 适配 (依据 docs/upgrade-0.16.1-spec.md, 全部经实测复核): 依赖: 仅 Python3 标准库 """ +import itertools import json import os import queue +import re import subprocess import sys import threading import time import uuid +from datetime import datetime, timezone from pathlib import Path +from urllib.parse import urlparse # ============================================================ # 配置 @@ -91,6 +95,16 @@ BRIDGE_INFO = {"name": "zcode-acp-bridge", "title": "ZCode (GLM-5.2)", "version" ZCODE_BIN = os.environ.get("ZCODE_BIN", "zcode") ZCODE_CREDS_PATH = Path.home() / ".zcode" / "v2" / "config.json" +# 整体 review 安全发现 (mimosa 新发现2): session/new 未显式指定 mode 时的默认权限 +# 模式。默认 yolo (历史行为: 跳过一切工具权限确认, 任意 prompt 都可能触发无确认的 +# 文件修改/命令执行); 注重安全的环境可设 ZCODE_ACP_DEFAULT_MODE=build 等收紧默认值, +# 无需改代码。启动日志会对 yolo 默认值打出显眼告警 (见 run())。 +DEFAULT_ACP_MODE = os.environ.get("ZCODE_ACP_DEFAULT_MODE") or "yolo" + +# 整体 review P2-1/P2-2: differ 与 state 投影按 session 累积, 而 ACP baseline 没有 +# session 关闭钩子可挂清理, 设软上限 FIFO 淘汰最旧条目, 防长驻进程内存单调增长 +_MAX_SESSION_STATES = 64 + # 0.16+ server→client 反向调用 session/requestRuntimePreferences 的应答体 # (schema 实测自 bundle zod): bridge 无这些客户端能力, 全部关闭 RUNTIME_PREFERENCES = { @@ -113,6 +127,14 @@ def load_zcode_credentials(): try: with open(ZCODE_CREDS_PATH) as f: cfg = json.load(f) + except FileNotFoundError: + return {} # 未配置凭证文件: 正常静默降级 (走纯环境变量, 不算故障) + except Exception as e: + # 整体 review P2-4: 文件存在但权限错误/JSON 损坏 → 显式警告, + # 让用户能区分「没配凭证」与「凭证文件损坏」(此前一律打成普通 log 难察觉) + log(f"⚠ 凭证文件存在但读取失败 ({ZCODE_CREDS_PATH}): {e}; 回退纯环境变量") + return {} + try: for _pid, p in cfg.get("provider", {}).items(): if p.get("enabled"): opts = p.get("options", {}) @@ -123,7 +145,7 @@ def load_zcode_credentials(): "ANTHROPIC_API_KEY": opts.get("apiKey", ""), } except Exception as e: - log(f"读取凭证失败: {e}") + log(f"⚠ 凭证文件结构异常 ({ZCODE_CREDS_PATH}): {e}; 回退纯环境变量") return {} @@ -146,21 +168,20 @@ def _merge_env_with_creds(creds): if env_bu and config_bu and env_bu != config_bu: all_hosts = set() try: - from urllib.parse import urlparse as _up with open(ZCODE_CREDS_PATH) as _f: _cfg = json.load(_f) for _p in _cfg.get("provider", {}).values(): _bu = _p.get("options", {}).get("baseURL", "") if _bu: try: - _h = _up(_bu) + _h = urlparse(_bu) if _h.hostname: all_hosts.add(f"{_h.scheme}://{_h.netloc}" if _h.scheme else _h.netloc) except Exception: pass - _eh = _up(env_bu) + _eh = urlparse(env_bu) env_host = f"{_eh.scheme}://{_eh.netloc}" if (_eh.scheme and _eh.hostname) else None - _ch = _up(config_bu) + _ch = urlparse(config_bu) config_host = f"{_ch.scheme}://{_ch.netloc}" if (_ch.scheme and _ch.hostname) else None except Exception: env_host = config_host = None @@ -324,6 +345,12 @@ class ZCodeBackend: def register_event_listener(self, zcode_sid, listener): """注册事件监听器, 之后该 sid 的 session/event 会路由到它。""" with self._listeners_lock: + # 整体 review P2-8: 同 sid 覆盖注册会静默丢前一个 listener (其事件 + # 从此无人消费)。当前调用方保证同 sid 串行 + finally 注销, 正常不触发; + # 打告警防御未来引入同 session 并发 prompt 时丢事件难以排查 + if zcode_sid in self._event_listeners: + log(f"⚠ register_event_listener: sid={zcode_sid} 已有 listener, " + f"覆盖注册 (前一个 listener 的事件将无人消费)") self._event_listeners[zcode_sid] = listener def unregister_event_listener(self, zcode_sid): @@ -406,6 +433,11 @@ class ZCodeBackend: self.proc.wait(timeout=3) except Exception: self.proc.kill() + # 整体 review P2-7: kill 后补 wait 收尸, 防子进程变僵尸残留 + try: + self.proc.wait(timeout=2) + except Exception: + pass # ---------- 0.16+ 协议探测 / 反向调用应答 / 状态投影 ---------- @@ -521,8 +553,10 @@ class ZCodeBackend: """ msg_id = msg.get("id") method = msg.get("method") - # 本方法在 reader 线程内执行: send() 失败只记日志不抛出 — reader 循环的 - # except 会 break 并把 backend 标记 dead, 一次应答异常不能带走整个 reader。 + # 本方法在 reader 线程内执行: send() 失败只记日志、不抛出也不 break — 这是 + # 设计意图 (reader 存活优先, 一次应答异常不能带走整个 reader)。代价是本次 + # 应答丢失后 server 侧该请求会空等到它自己超时 (桥无法挽救), 但 reader 继续 + # 分发后续帧, 其他请求不受影响 (整体 review P2-6: 注释与实际行为对齐)。 if method == "session/requestRuntimePreferences": scope = msg.get("params", {}).get("scope", "?") log(f" ← 应答 session/requestRuntimePreferences (scope={scope})") @@ -549,6 +583,14 @@ class ZCodeBackend: if not sid or not isinstance(patch, dict): return with self._state_lock: + # 整体 review P2-2: 投影跨 session 无清理 (reset_projection 只管当前 + # sid 的每 turn 入口), 且无 session 关闭钩 → 超上限 FIFO 淘汰最旧条目 + if sid not in self._state_projections \ + and len(self._state_projections) >= _MAX_SESSION_STATES: + evicted = next(iter(self._state_projections)) + self._state_projections.pop(evicted, None) + log(f"⚠ state 投影超上限 ({_MAX_SESSION_STATES}), " + f"淘汰最旧 session: {evicted}") self._state_projections.setdefault(sid, {}).update(patch) def get_projection(self, zcode_sid): @@ -1083,16 +1125,21 @@ class ACPBridge: self.session_map = {} # 当前进行中的 prompt: {acp_prompt_id: {zcode_sid, cancelled, perms_responses}} self.pending_turns = {} - # id 基数提到 10_000_000, 避免与 zcode 主动请求的 id 空间冲突 - self._msg_counter = 10_000_000 + # id 基数提到 10_000_000, 避免与 zcode 主动请求的 id 空间冲突。 + # 整体 review P1-1: 用 itertools.count 取代"读-改-写"的 += 1 — next() 在 + # CPython 下原子, 消除未来跨线程调用 _next_id 的 id 重复竞态 (当前虽全部 + # 串行在主线程, 但原写法无护栏) + self._id_counter = itertools.count(10_000_001) + # 整体 review P1-2: stdout 协议帧写锁 — 全文件 stdout 写收进 _write_stdout(), + # 防未来后台线程写帧时与主循环产生交错的半截 JSON (ACP client 会解析失败) + self._stdout_lock = threading.Lock() # stdin 消息队列: 后台线程把每行 stdin 入队, 主循环/轮询循环从队列取 # 这解决了 run() 单线程阻塞导致 cancel/permission 响应读不到的问题 self._inbox = queue.Queue() self._stdin_thread = None def _next_id(self): - self._msg_counter += 1 - return self._msg_counter + return next(self._id_counter) # 0.16 已删方法 (实测后端返 -32601 Method not found, 规格书 §2) # 0.16 已删方法清单 — 与 README/SKILL 的「0.16 已移除」表格标注双处维护, @@ -1106,6 +1153,22 @@ class ACPBridge: """后端是否为 0.16+ 新协议 (ZCodeBackend 启动探测缓存; 测试桩无该属性按旧协议)。""" return getattr(self.backend, "protocol_mode", "legacy") == "v16" + def _backend_proc_dead(self): + """探测 backend 子进程是否已退出 (proc.poll() 非 None = 已死)。 + + 整体 review P1-3 的收敛探测: 轮询连续无响应时用于区分「后端 hang 但活着」 + (继续等) 与「子进程已死」(确定性故障, 立即报错)。测试桩无 proc 属性时按 + 「无法判定」返回 False (不提前收敛, 保持原有等满超时的行为)。 + """ + proc = getattr(self.backend, "proc", None) + poll = getattr(proc, "poll", None) + if not callable(poll): + return False + try: + return poll() is not None + except Exception: + return False + def _removed_method_error(self, msg_id, zcode_method, resp): """把已删方法的后端 -32601 映射为明确文案 (规格书 §7)。 @@ -1238,9 +1301,11 @@ class ACPBridge: """ACP session/new → zcode session/create""" self.ensure_backend() cwd = params.get("cwd") or os.getcwd() - # 显式 mode 透传; 默认 yolo: 跳过工具权限确认, 否则 build 模式下工具调用 - # turn 会卡在 pendingPermissions (0.15 实测; 0.16 stdio 下工具自动执行, 参数被忽略) - mode = params.get("mode") or "yolo" + # 显式 mode 透传; 缺省用 DEFAULT_ACP_MODE (默认 yolo, 可用环境变量 + # ZCODE_ACP_DEFAULT_MODE 收紧 — 整体 review 安全发现): yolo 跳过工具权限确认, + # 否则 build 模式下工具调用 turn 会卡在 pendingPermissions (0.15 实测; + # 0.16 stdio 下工具自动执行, 参数被忽略) + mode = params.get("mode") or DEFAULT_ACP_MODE log(f"session/new: cwd={cwd}, mode={mode}") zc_id = self._next_id() @@ -1498,15 +1563,22 @@ class ACPBridge: error.message 里, 原样返回会泄漏。这里用正则遮蔽常见密钥形态: - sk- 开头的 OpenAI/Anthropic 风格 key 前缀 - JSON 里 "value":"..." / "apiKey":{...} 的敏感值 + - 纯 hex (32+) / uuid 形态的无前缀 key (整体 review P2-3 纵深防御: + provider 用这类 apiKey 且被错误消息原样回显时也能遮蔽) 保留非敏感部分以便排查。 """ - import re if not text: return text s = re.sub(r'sk-[A-Za-z0-9_-]{6,}', 'sk-***', text) s = re.sub(r'("value"\s*:\s*")([^"]{4,})(")', r'\1***\3', s) s = re.sub(r'("apiKey"\s*:\s*)("?\{[^}]*\}"?|"[^"]{4,}")', r'\1***', s) + # uuid (8-4-4-4-12) 与 32+ 位纯 hex 是常见无前缀 key 形态, 一并遮蔽。 + # 取舍说明 (复审 P2-5): 32+ hex 也可能命中 git SHA 等合法内容 — 但本函数 + # 只用于错误消息回显路径, 宁可多遮蔽也不可漏 key。 + s = re.sub(r'\b[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}\b', + '***', s) + s = re.sub(r'\b[0-9a-fA-F]{32,}\b', '***', s) return s # ---------- session 级新方法 (0.15.0+) ---------- @@ -1874,8 +1946,7 @@ class ACPBridge: cancel_id = req.get("id") cancel_resp = self._on_prompt_enhance_cancel(cancel_id, req.get("params", {})) if cancel_id is not None: - sys.stdout.write(json.dumps(cancel_resp, ensure_ascii=False) + "\n") - sys.stdout.flush() + self._write_stdout(cancel_resp) # P1-2: 统一加锁写口 continue # 非 cancel: 放回尾部, 不吞掉其他带外消息 (顺序保持) self._inbox.put(line) @@ -2077,7 +2148,6 @@ class ACPBridge: if not ts: return None try: - from datetime import datetime, timezone return datetime.fromtimestamp(ts / 1000, tz=timezone.utc).isoformat() except Exception: return None @@ -2101,8 +2171,13 @@ class ACPBridge: """ t0 = time.time() if probe_method: - # 用探测调用重试, 直到不再报 "prompt is running" + # 用探测调用重试, 直到不再报 "prompt is running"。 + # 整体 review P2-5: (a) 固定 2s 改指数退避 (1s 起倍增, 8s 封顶), + # 防高频探测打后端; (b) 等待期间每轮 drain inbox — 主循环串行, 不 + # drain 则此时到达的 session/cancel 会卡到等完才被处理 + backoff = 1.0 while time.time() - t0 < timeout: + self._drain_inbox() probe_id = self._next_id() resp, _ = self.backend.request(probe_id, probe_method, {"sessionId": zcode_sid, "action": "show"}, @@ -2113,16 +2188,18 @@ class ACPBridge: err_msg = resp.get("error", {}).get("message", "") # timeout 或 "prompt is running" 都继续等; 其他错误视为释放 if "prompt is running" in err_msg or "timeout" in err_msg.lower(): - time.sleep(2) + time.sleep(backoff) + backoff = min(backoff * 2, 8.0) continue log(f" [probe] 非锁错误, 视为释放: {err_msg[:50]}") return True log(f" [probe] 等待超时 ({timeout}s), lock 可能仍占用") return False else: - # fallback: 轮询 status + # fallback: 轮询 status (同样每轮 drain inbox 让 cancel 可响应, P2-5) monitor = TurnMonitor(self.backend, zcode_sid, self) while time.time() - t0 < timeout: + self._drain_inbox() proj = monitor.poll_once() if proj and proj.get("status") == "idle": time.sleep(1) @@ -2166,105 +2243,107 @@ class ACPBridge: log(f"session/prompt: sid={acp_sid[:20]}..., text={prompt_text[:40]!r}") - # 注册 pending turn (供 cancel 查询) + # 注册 pending turn (供 cancel 查询)。 + # 整体 review P1-5: try/finally 保证任何退出路径 (含中途异常) 都摘掉 + # pending 条目 — 此前各 return 前裸 del, 异常路径会留下僵尸条目, 长驻 + # 运行累积内存且干扰后续 cancel 的 turn 匹配 turn = {"zcode_sid": zcode_sid, "cancelled": False} self.pending_turns[msg_id] = turn + try: + # zcode review P1-1 防御: 新 turn 入口清该 session 的 state.updated 状态投影, + # 防跨 turn 陈旧值误导 _run_event_turn 停滞检查 (核实结论: 该竞态在 0.16.1 + # 不可达, 此处为零成本防御, 详见 ZCodeBackend.reset_projection docstring)。 + # getattr 兜底: 测试桩 (FakeBackend) 无此方法。 + _reset_proj = getattr(self.backend, "reset_projection", None) + if callable(_reset_proj): + _reset_proj(zcode_sid) + + # 建立 baseline: send 前先记录已有消息 id, 让 differ 只产出本轮新增 (修复 P1.1) + # differ 跨 turn 持久化 (按 session 存), 避免多轮/resume 误发历史 + differ = self._get_or_create_differ(zcode_sid) + # 先喂一次当前 messages, 让 _seen_message_ids 标记所有历史消息为"已见" + baseline_msgs = self._fetch_messages(zcode_sid) + if baseline_msgs: + differ.mark_seen(baseline_msgs) + chunk_msg_id = f"msg_{uuid.uuid4().hex[:12]}" + + # ===== 双模式: 优先事件驱动, 失败降级轮询 (轮询仅限旧协议模式) ===== + # 新版 ZCode (0.14.8+) 支持 session/subscribe 推送流式事件 (model.streaming)。 + # 事件驱动模式下: model.streaming → 逐段 TextDelta (真流式); + # tool.updated → 实时工具调用状态。 + # 降级模式 (旧协议 / subscribe 失败): 轮询 session/read, turn 结束整段发 (伪流式)。 + # 0.16+ 新协议模式不降级: subscribe 是其必备事件通道, 失败即报错 (规格书 §4)。 + + # ⚠️ 时序关键: 必须先 subscribe + register listener, 再 session/send。 + # 否则 send 和 register 之间的 session/event 会被 reader 线程丢弃, + # 短 turn 的 turn.completed 可能丢失导致 120s 超时。 + listener = EventStreamListener(self.backend, zcode_sid) + sub_id = self._next_id() + sub_result = listener.subscribe(sub_id) + + if sub_result is not None: + # 事件驱动模式: 先注册 listener (确保不丢事件), 再 send。 + # register/unregister 接口只有真实 ZCodeBackend 有, getattr 兜底 + # (测试桩没有, 此时事件经 request() 第二返回值喂入, 见下)。 + _reg = getattr(self.backend, "register_event_listener", None) + if callable(_reg): + _reg(zcode_sid, listener) + _unreg = getattr(self.backend, "unregister_event_listener", None) + try: + # 现在 send prompt (listener 已注册, 不会丢事件) + zc_id = self._next_id() + resp, send_events = self.backend.request( + zc_id, "session/send", + {"sessionId": zcode_sid, "content": prompt_text}, + timeout=10) + if "error" in resp: + return self._error(msg_id, -32603, + f"zcode send failed: {resp['error'].get('message', '')}") + if not resp.get("result", {}).get("accepted", False): + return self._error(msg_id, -32603, + "zcode send 未被接受 (无 accepted 响应)") - # zcode review P1-1 防御: 新 turn 入口清该 session 的 state.updated 状态投影, - # 防跨 turn 陈旧值误导 _run_event_turn 停滞检查 (核实结论: 该竞态在 0.16.1 - # 不可达, 此处为零成本防御, 详见 ZCodeBackend.reset_projection docstring)。 - # getattr 兜底: 测试桩 (FakeBackend) 无此方法。 - _reset_proj = getattr(self.backend, "reset_projection", None) - if callable(_reset_proj): - _reset_proj(zcode_sid) - - # 建立 baseline: send 前先记录已有消息 id, 让 differ 只产出本轮新增 (修复 P1.1) - # differ 跨 turn 持久化 (按 session 存), 避免多轮/resume 误发历史 - differ = self._get_or_create_differ(zcode_sid) - # 先喂一次当前 messages, 让 _seen_message_ids 标记所有历史消息为"已见" - baseline_msgs = self._fetch_messages(zcode_sid) - if baseline_msgs: - differ.mark_seen(baseline_msgs) - chunk_msg_id = f"msg_{uuid.uuid4().hex[:12]}" - - # ===== 双模式: 优先事件驱动, 失败降级轮询 (轮询仅限旧协议模式) ===== - # 新版 ZCode (0.14.8+) 支持 session/subscribe 推送流式事件 (model.streaming)。 - # 事件驱动模式下: model.streaming → 逐段 TextDelta (真流式); - # tool.updated → 实时工具调用状态。 - # 降级模式 (旧协议 / subscribe 失败): 轮询 session/read, turn 结束整段发 (伪流式)。 - # 0.16+ 新协议模式不降级: subscribe 是其必备事件通道, 失败即报错 (规格书 §4)。 - - # ⚠️ 时序关键: 必须先 subscribe + register listener, 再 session/send。 - # 否则 send 和 register 之间的 session/event 会被 reader 线程丢弃, - # 短 turn 的 turn.completed 可能丢失导致 120s 超时。 - listener = EventStreamListener(self.backend, zcode_sid) - sub_id = self._next_id() - sub_result = listener.subscribe(sub_id) - - if sub_result is not None: - # 事件驱动模式: 先注册 listener (确保不丢事件), 再 send。 - # register/unregister 接口只有真实 ZCodeBackend 有, getattr 兜底 - # (测试桩没有, 此时事件经 request() 第二返回值喂入, 见下)。 - _reg = getattr(self.backend, "register_event_listener", None) - if callable(_reg): - _reg(zcode_sid, listener) - _unreg = getattr(self.backend, "unregister_event_listener", None) - try: - # 现在 send prompt (listener 已注册, 不会丢事件) + # request() 第二返回值携带 send 期间到达的事件 (测试 seam; 真实 + # backend 恒为 [], 真实事件经 reader → 已注册 listener 到达) + for ev in send_events or []: + listener.handle_event(ev) + + result = self._run_event_turn( + listener, acp_sid, zcode_sid, msg_id, turn, chunk_msg_id, differ) + finally: + if callable(_unreg): + _unreg(zcode_sid) + return result + else: + # 0.16+ 新协议模式 (实锤判定): 轮询降级为旧协议专用, subscribe 失败直接报错 + if self._backend_v16() and getattr(self.backend, "protocol_confirmed", True): + return self._error(msg_id, -32603, + "zcode session/subscribe 失败 (0.16+ 必须走事件订阅; " + "轮询降级仅限旧协议模式)") + # zcode review P1-3 修复: "v16" 若只是探测超时兜底判定 (protocol_confirmed + # =False, 无实锤), subscribe 失败本身就是旧协议的实锤 (真 0.16 server 的 + # subscribe 不会失败) → 推翻探测结果, 回退 legacy 走原有轮询降级 + # (与 _detect_protocol docstring「新协议代码路径对 0.15 亦兼容」对齐)。 + if self._backend_v16(): + self.backend.protocol_mode = "legacy" + self.backend.protocol_confirmed = True + log(" ⚠ subscribe 失败, 推翻探测超时的 v16 兜底判定 → 回退旧协议轮询") + # 降级: 轮询模式 — send 在轮询循环前发 zc_id = self._next_id() - resp, send_events = self.backend.request( - zc_id, "session/send", - {"sessionId": zcode_sid, "content": prompt_text}, - timeout=10) + resp, _ = self.backend.request(zc_id, "session/send", + {"sessionId": zcode_sid, "content": prompt_text}, + timeout=10) if "error" in resp: - del self.pending_turns[msg_id] return self._error(msg_id, -32603, f"zcode send failed: {resp['error'].get('message', '')}") if not resp.get("result", {}).get("accepted", False): - del self.pending_turns[msg_id] return self._error(msg_id, -32603, "zcode send 未被接受 (无 accepted 响应)") - - # request() 第二返回值携带 send 期间到达的事件 (测试 seam; 真实 - # backend 恒为 [], 真实事件经 reader → 已注册 listener 到达) - for ev in send_events or []: - listener.handle_event(ev) - - result = self._run_event_turn( - listener, acp_sid, zcode_sid, msg_id, turn, chunk_msg_id, differ) - finally: - if callable(_unreg): - _unreg(zcode_sid) - return result - else: - # 0.16+ 新协议模式 (实锤判定): 轮询降级为旧协议专用, subscribe 失败直接报错 - if self._backend_v16() and getattr(self.backend, "protocol_confirmed", True): - del self.pending_turns[msg_id] - return self._error(msg_id, -32603, - "zcode session/subscribe 失败 (0.16+ 必须走事件订阅; " - "轮询降级仅限旧协议模式)") - # zcode review P1-3 修复: "v16" 若只是探测超时兜底判定 (protocol_confirmed - # =False, 无实锤), subscribe 失败本身就是旧协议的实锤 (真 0.16 server 的 - # subscribe 不会失败) → 推翻探测结果, 回退 legacy 走原有轮询降级 - # (与 _detect_protocol docstring「新协议代码路径对 0.15 亦兼容」对齐)。 - if self._backend_v16(): - self.backend.protocol_mode = "legacy" - self.backend.protocol_confirmed = True - log(" ⚠ subscribe 失败, 推翻探测超时的 v16 兜底判定 → 回退旧协议轮询") - # 降级: 轮询模式 — send 在轮询循环前发 - zc_id = self._next_id() - resp, _ = self.backend.request(zc_id, "session/send", - {"sessionId": zcode_sid, "content": prompt_text}, - timeout=10) - if "error" in resp: - del self.pending_turns[msg_id] - return self._error(msg_id, -32603, - f"zcode send failed: {resp['error'].get('message', '')}") - if not resp.get("result", {}).get("accepted", False): - del self.pending_turns[msg_id] - return self._error(msg_id, -32603, "zcode send 未被接受 (无 accepted 响应)") - log(" ⟳ 降级轮询模式 (事件流不可用)") - return self._run_polling_turn( - acp_sid, zcode_sid, msg_id, turn, chunk_msg_id, differ) + log(" ⟳ 降级轮询模式 (事件流不可用)") + return self._run_polling_turn( + acp_sid, zcode_sid, msg_id, turn, chunk_msg_id, differ) + finally: + # P1-5: 正常返回/异常上抛都必执行; turn 循环内部先 del 过也无害 (幂等) + self.pending_turns.pop(msg_id, None) def _run_event_turn(self, listener, acp_sid, zcode_sid, msg_id, turn, chunk_msg_id, differ): """事件驱动 turn 循环: 消费 session/event 推送, 翻译成 ACP 事件实时发。 @@ -2395,6 +2474,13 @@ class ACPBridge: # 超时 del self.pending_turns[msg_id] log("session/prompt 超时: 事件模式下 120s 未完成") + if translator.turn_started and not translator.turn_done: + # 整体 review P2-5 (tests 报告): turn 已启动但 120s 未收到完成信号, + # 说明事件流断了 (turn.completed/turn.failed 丢失), 属异常而非 + # 「turn 真的太长」— 返回错误, 不再用正常 stopReason 让 client 误判 + return self._error(msg_id, -32603, + "zcode 事件流超时: turn 已启动但 120s 未收到完成信号 " + "(turn.completed 丢失或事件流中断)") return {"jsonrpc": "2.0", "id": msg_id, "result": {"stopReason": "max_turn_requests"}} @@ -2426,6 +2512,15 @@ class ACPBridge: consecutive_none += 1 if consecutive_none >= 10: # 连续 ~5s 无响应 log(f" ⚠ 连续 {consecutive_none} 次 poll 失败") + # 整体 review P1-3: 连续无响应超阈值 (20 次 ≈ 10s) 时探测后端 + # 子进程存活; 已死是确定性故障, 立即报错返回, 不让调用方空等 + # 耗满 120s 预算 (子进程 hang 但活着则继续等, 行为不变) + if consecutive_none >= 20 and self._backend_proc_dead(): + del self.pending_turns[msg_id] + log(" ✗ zcode 后端进程已退出, 轮询提前收敛 (不再空等 120s)") + return self._error( + msg_id, -32603, + "zcode 后端进程已退出 (连续 20 次 poll 无响应)") time.sleep(0.5) continue consecutive_none = 0 @@ -2487,9 +2582,12 @@ class ACPBridge: return {"jsonrpc": "2.0", "id": msg_id, "result": {"stopReason": "max_turn_requests"}} - # turn 正常完成。differ 已发本轮新增 text; 若本轮无新增 text, 兜底发最后一条 assistant 回复 + # turn 正常完成。differ 已发本轮新增 text; 若本轮无新增 text, 兜底发最后一条 assistant 回复。 + # 整体 review P1-4: 删掉此处两处 emitted_text_this_turn=False 冗余自赋值 — + # "本轮"标记的 reset 单点在 _get_or_create_differ 的 reset_turn() (每 turn + # 入口必调), 这里把一个已是 False 的字段再赋 False 是空操作, 且会掩盖 + # 「忘了在入口 reset」的未来回归 if not differ.emitted_text_this_turn: - differ.emitted_text_this_turn = False # reset reply_text = self._fetch_last_reply(zcode_sid, differ) if reply_text: self._send_acp_notification("session/update", { @@ -2502,7 +2600,6 @@ class ACPBridge: }) log(f"session/prompt 完成: stopReason=end_turn, 回复 {len(reply_text)} 字 (兜底)") else: - differ.emitted_text_this_turn = False # reset for next turn log("session/prompt 完成: stopReason=end_turn, text 已随事件发送") return {"jsonrpc": "2.0", "id": msg_id, @@ -2514,13 +2611,21 @@ class ACPBridge: """获取或创建 session 级别的 ProjectionDiffer (跨 turn 持久化)。 修复 P1.1: differ 按 session 存, 不每 turn 新建, 避免多轮/resume 误发历史。 + 整体 review P2-1: ACP baseline 无 session 关闭钩, differ 只增不减 → 超 + _MAX_SESSION_STATES 上限时 FIFO 淘汰最旧条目 (代价: 该 session 再 prompt + 时会重建 baseline, 可接受), 防长驻进程内存单调增长。 """ if not hasattr(self, "_differs"): self._differs = {} if zcode_sid not in self._differs: + if len(self._differs) >= _MAX_SESSION_STATES: + evicted = next(iter(self._differs)) + self._differs.pop(evicted, None) + log(f"⚠ differ 超上限 ({_MAX_SESSION_STATES}), " + f"淘汰最旧 session: {evicted}") self._differs[zcode_sid] = ProjectionDiffer() d = self._differs[zcode_sid] - d.reset_turn() # 重置"本轮"标记 + d.reset_turn() # 重置"本轮"标记 (emitted_text_this_turn 的唯一 reset 点) return d def _fetch_messages(self, zcode_sid): @@ -2594,12 +2699,17 @@ class ACPBridge: resp, _ = self.backend.request(zc_id, "session/messages", {"sessionId": zcode_sid}, timeout=10) if "error" in resp or "result" not in resp: + # 整体 review P1-6: 重试间隙 drain inbox — 主循环串行, 此处阻塞 + # 期间到达的 session/cancel 不 drain 就会卡到重试结束才生效 + self._drain_inbox() time.sleep(0.4) continue messages = resp["result"].get("messages", []) # 找最后一条 differ 未见的 assistant 消息 for m in reversed(messages): - info = m.get("info", {}) + if not isinstance(m, dict): + continue # 整体 review P2-9: 防御非 dict 元素 (与 diff() 风格一致) + info = m.get("info", {}) if isinstance(m.get("info"), dict) else {} if info.get("role") != "assistant": continue # 若传了 differ, 跳过已见消息 (避免兜底旧回复) @@ -2613,6 +2723,7 @@ class ACPBridge: text = "\n".join(texts) if text.strip(): # 拿到非空且未见的文本才返回 return text + self._drain_inbox() # P1-6: 同上, 重试间隙让 cancel 可响应 time.sleep(0.4) # 还没数据, 等一会再试 return "" @@ -2730,11 +2841,19 @@ class ACPBridge: # ---------- 工具方法 ---------- + def _write_stdout(self, obj): + """把一帧 JSON 写进 stdout 协议流 (整体 review P1-2: 全文件唯一 stdout 写口)。 + + 加 _stdout_lock: 主循环响应 / notification / enhance drain 的写帧都走这里, + 防未来任何后台线程写 stdout 时产生交错的半截 JSON 帧 (ACP client 解析即失败)。 + """ + with self._stdout_lock: + sys.stdout.write(json.dumps(obj, ensure_ascii=False) + "\n") + sys.stdout.flush() + def _send_acp_notification(self, method, params): """向 ACP client 发 notification (无 id)""" - msg = {"jsonrpc": "2.0", "method": method, "params": params} - sys.stdout.write(json.dumps(msg, ensure_ascii=False) + "\n") - sys.stdout.flush() + self._write_stdout({"jsonrpc": "2.0", "method": method, "params": params}) def _error(self, msg_id, code, message): return {"jsonrpc": "2.0", "id": msg_id, @@ -2783,6 +2902,10 @@ class ACPBridge: def run(self): """主循环: 后台线程读 stdin 入队, 主循环从队列取消息处理""" log(f"启动 zcode-acp-bridge, ACP protocol v{ACP_PROTOCOL_VERSION}, agent={BRIDGE_INFO['name']}") + if DEFAULT_ACP_MODE == "yolo": + # 整体 review 安全发现: 启动即显眼标注当前默认 mode 的权限风险 + log("⚠️ 默认权限模式 mode=yolo: 任意 session/prompt 触发的文件修改与命令 " + "执行均无确认; 注重安全可设 ZCODE_ACP_DEFAULT_MODE=build 等收紧默认值") self._stdin_thread = threading.Thread(target=self._stdin_reader, daemon=True) self._stdin_thread.start() while True: @@ -2803,13 +2926,11 @@ class ACPBridge: try: resp = self.handle_acp(req) if resp is not None: - sys.stdout.write(json.dumps(resp, ensure_ascii=False) + "\n") - sys.stdout.flush() + self._write_stdout(resp) # P1-2: 统一加锁写口 except Exception as e: log(f"处理异常: {e}") err = self._error(req.get("id"), -32603, str(e)) - sys.stdout.write(json.dumps(err, ensure_ascii=False) + "\n") - sys.stdout.flush() + self._write_stdout(err) # P1-2: 统一加锁写口 log("stdin 关闭, 退出") if self.backend: self.backend.close() diff --git a/packages/agent-help/zcode-agent-help b/packages/agent-help/zcode-agent-help index b1080cd..f87888e 100755 --- a/packages/agent-help/zcode-agent-help +++ b/packages/agent-help/zcode-agent-help @@ -26,6 +26,7 @@ import os import subprocess import sys from pathlib import Path +from urllib.parse import urlparse ZCODE_BIN = os.environ.get("ZCODE_BIN", "zcode") @@ -68,44 +69,65 @@ def _mask_key(key): return f"{key[:4]}...{key[-4:]}" -def print_injected_env(): +def _safe_host(url): + """提取 URL 的 host (scheme://host[:port]), 解析失败返回 None。 + + 与 shared/credentials.py._safe_host 逐字对齐 (整体 review P1-1), + 含"无 scheme 补 https://"分支 — 保证本工具的残留诊断与 merge 的真实判定一致。 + """ + try: + p = urlparse(url) + if not p.hostname: + return None + return f"{p.scheme}://{p.netloc}" if p.scheme else f"https://{p.netloc}" + except Exception: + return None + + +def print_injected_env(config_path=None): """打印"如果现在注入凭证, 会用哪些值"(脱敏), 用于不改 config 诊断 model/baseURL 来源。 显示每个 key 的来源 (config vs 显式 env) + 最终值 (apiKey 脱敏), 体现"显式 env 优先"的合并语义 (issue #3 子项2)。 + + Args: + config_path: 可选, 自定义配置文件路径 (测试用, 整体 review P2-3)。 + 默认 ZCODE_CREDS_PATH。 """ - creds = _load_creds_internal() + path = config_path or ZCODE_CREDS_PATH + creds = _load_creds_internal(path) + keys = ["ZCODE_MODEL", "ZCODE_BASE_URL", "ANTHROPIC_API_KEY"] if not creds: - print(f"❌ 未从 {ZCODE_CREDS_PATH} 读取到凭证 (config 无 enabled provider 或文件缺失)") + # 整体 review P1-5: config 无 enabled provider 时不直接退出 — env 里的非空值 + # 仍会被 merge 原样透传给 zcode 子进程, 必须脱敏打印出来, 否则用户无从知晓。 + env_hits = {k: v for k in keys if (v := os.environ.get(k))} + if not env_hits: + print(f"❌ 未从 {path} 读取到凭证 (config 无 enabled provider 或文件缺失)") + return 0 + print("=== 凭证注入诊断 (显式非空 env 优先于 config; 空串视为未设置) ===") + print(f"config 源: {path}") + print("⚠ config 无 enabled provider, 将直接使用 env 值:") + for k, v in env_hits.items(): + shown = _mask_key(v) if k == "ANTHROPIC_API_KEY" else v + print(f" {k}: {shown} (来自 env)") return 0 # 读 config 里所有 provider 的 baseURL (残留检测用, host 集合) + # 整体 review P1-1: 用与权威版逐字对齐的 _safe_host 构造, 消除诊断与真实注入的漂移 all_config_hosts = set() try: - from urllib.parse import urlparse as _up - with open(ZCODE_CREDS_PATH) as _f: + with open(path) as _f: _cfg = json.load(_f) for _p in _cfg.get("provider", {}).values(): _bu = _p.get("options", {}).get("baseURL", "") if _bu: - try: - _h = _up(_bu) - if _h.hostname: - all_config_hosts.add(f"{_h.scheme}://{_h.netloc}" if _h.scheme else _h.netloc) - except Exception: - pass + _h = _safe_host(_bu) + if _h: + all_config_hosts.add(_h) except Exception: pass - def _host(u): - try: - h = _up(u) - return f"{h.scheme}://{h.netloc}" if (h.scheme and h.hostname) else None - except Exception: - return None - - keys = ["ZCODE_MODEL", "ZCODE_BASE_URL", "ANTHROPIC_API_KEY"] print("=== 凭证注入诊断 (显式非空 env 优先于 config; 空串视为未设置) ===") - print(f"config 源: {ZCODE_CREDS_PATH}") + print(f"config 源: {path}") for k in keys: config_val = creds.get(k, "") env_val = os.environ.get(k) @@ -115,8 +137,8 @@ def print_injected_env(): print(f" config → {('脱敏 ' + _mask_key(config_val)) if k == 'ANTHROPIC_API_KEY' and config_val else (config_val or '(空)')}") # ZCODE_BASE_URL 特殊: 检测 App 注入的残留官方 endpoint (host 匹配) if k == "ZCODE_BASE_URL" and config_val and env_val != config_val: - _eh = _host(env_val) - _ch = _host(config_val) + _eh = _safe_host(env_val) + _ch = _safe_host(config_val) if _eh and _ch and _eh != _ch and _eh in all_config_hosts: print(f" env → {shown} 🚫 残留! 是 config 里另一个 provider 的 endpoint") print(f" → bridge 会自动用 config 值: {config_val}") @@ -141,6 +163,9 @@ def print_injected_env(): OVERVIEW = { "tool": "zcode", "version_described": "0.16.1 (App 3.6.5)", + # 整体 review P2-6: version_described 是"静态清单基于该版本实测"的意思; + # 实际安装版本以 environment.zcode_version (动态探测) 为准, 两者可能不一致。 + "tested_against": "0.16.1 (App 3.6.5)", "what_it_is": ( "ZCode 是智谱(Z.AI)的 Agentic Coding CLI, 由 GLM 系列模型驱动。" "它能自主完成代码编写/审查/调试/分析等任务, 可调用工具(读写文件、" @@ -592,15 +617,25 @@ ECOSYSTEM = { # 动态探测: 实时调用 zcode 获取当前环境信息 # ============================================================ def _run_zcode_json(args): + """调 zcode 子命令并解析 --json 输出; 探测失败返回 None 并打 stderr 提示。 + + 整体 review P1-4: 异常捕获放宽为 (SubprocessError, OSError, ValueError) — + SubprocessError 含 TimeoutExpired, OSError 含 FileNotFoundError/PermissionError + (二进制存在但不可执行), ValueError 含 JSONDecodeError。提示走 stderr, + 不污染 --json 的 stdout。 + """ try: result = subprocess.run( [ZCODE_BIN] + args + ["--json"], capture_output=True, text=True, timeout=15, ) if result.returncode != 0: + print(f"⚠ 探测失败: {ZCODE_BIN} {' '.join(args)} 退出码 {result.returncode}", + file=sys.stderr) return None return json.loads(result.stdout) - except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError): + except (subprocess.SubprocessError, OSError, ValueError) as e: + print(f"⚠ 探测失败: {ZCODE_BIN} {' '.join(args)}: {e}", file=sys.stderr) return None @@ -610,7 +645,9 @@ def _get_version(): [ZCODE_BIN, "version"], capture_output=True, text=True, timeout=10 ) return result.stdout.strip() or "unknown" - except Exception: + except (subprocess.SubprocessError, OSError, ValueError) as e: + # 整体 review P1-4: 与 _run_zcode_json 同一捕获集合 + print(f"⚠ 版本探测失败: {e}", file=sys.stderr) return "unknown" @@ -633,7 +670,11 @@ def discover_environment(): cmds_data = _run_zcode_json(["commands", "list"]) if cmds_data and isinstance(cmds_data, dict): cmds = cmds_data.get("commands") or cmds_data.get("data") or [] - env["custom_commands"] = [c.get("name", str(c)) if isinstance(c, dict) else str(c) for c in cmds] + # 整体 review P2-2: 守卫 cmds 非 list (dict/str 会被迭代出垃圾数据) + if isinstance(cmds, list): + env["custom_commands"] = [c.get("name", str(c)) if isinstance(c, dict) else str(c) for c in cmds] + else: + env["custom_commands"] = [] else: env["custom_commands"] = [] @@ -668,7 +709,8 @@ def build_full(): def print_pretty(cap): - v = cap["environment"]["zcode_version"] + # 整体 review P2-4: environment 是动态探测结果, 对缺 key 做 .get() 防御 + v = cap.get("environment", {}).get("zcode_version", "unknown") print(f"╔══ ZCode v{v} — 开放生态能力说明书 ══╗\n") print(cap["what_it_is"], "\n") @@ -712,11 +754,11 @@ def print_pretty(cap): f"({eco['session_storage']['location'][:40]}...)\n") # environment - env = cap["environment"] + env = cap.get("environment", {}) print("▼ 当前环境 (动态探测)") - print(f" zcode 版本: {env['zcode_version']}") - print(f" 已装 skill: {env['skills_count']} 个") - print(f" 已装组件: {[k for k,v in env['installed_components'].items() if v] or '无'}") + print(f" zcode 版本: {env.get('zcode_version', 'unknown')}") + print(f" 已装 skill: {env.get('skills_count', 0)} 个") + print(f" 已装组件: {[k for k, v in env.get('installed_components', {}).items() if v] or '无'}") def main(): @@ -731,19 +773,25 @@ def main(): if "--section" in args: idx = args.index("--section") - if idx + 1 < len(args): - section = args[idx + 1] - # 静态 section 直接查 SECTIONS; environment 是动态探测的, 单独处理 - if section == "environment": - data = discover_environment() - else: - data = SECTIONS.get(section) - if data is not None: - print(json.dumps(data, indent=2, ensure_ascii=False)) - return 0 - print(f"未知 section: {section}\n可用: {', '.join(list(SECTIONS.keys()) + ['environment'])}", + # 整体 review P1-3: 末尾无值, 或后随另一个 flag (误把 --xxx 当 section 名), + # 按用法错误处理并 return 1, 不静默落到全量输出 (脚本调用方会误以为拿到了某节)。 + if idx + 1 >= len(args) or args[idx + 1].startswith("--"): + print(f"用法错误: --section 需要跟一个 section 名\n" + f"可用: {', '.join(list(SECTIONS.keys()) + ['environment'])}", file=sys.stderr) return 1 + section = args[idx + 1] + # 静态 section 直接查 SECTIONS; environment 是动态探测的, 单独处理 + if section == "environment": + data = discover_environment() + else: + data = SECTIONS.get(section) + if data is not None: + print(json.dumps(data, indent=2, ensure_ascii=False)) + return 0 + print(f"未知 section: {section}\n可用: {', '.join(list(SECTIONS.keys()) + ['environment'])}", + file=sys.stderr) + return 1 cap = build_full() if "--pretty" in args: diff --git a/packages/mcp-server/zcode-mcp-server b/packages/mcp-server/zcode-mcp-server index 43d4d88..0015851 100755 --- a/packages/mcp-server/zcode-mcp-server +++ b/packages/mcp-server/zcode-mcp-server @@ -22,10 +22,12 @@ import json import os import re import select +import shutil import subprocess import sys import time from pathlib import Path +from urllib.parse import urlparse # zcode-agent-help 的路径 (能力发现的数据源) # 优先解析仓库相邻路径 (../agent-help/zcode-agent-help), 让"从仓库直接运行"读到同版本; @@ -93,6 +95,22 @@ def load_zcode_credentials(): return {} +def _safe_host(url): + """提取 URL 的 host (scheme://host[:port]), 解析失败返回 None。 + + 与 shared/credentials.py._safe_host 权威版逐字对齐 (整体 review P2-3): + 有 netloc 但缺 scheme (如 protocol-relative //host/path) 时补 https://, + 内嵌旧副本此时返回 None, 会让残留检测静默跳过。 + """ + try: + p = urlparse(url) + if not p.hostname: + return None + return f"{p.scheme}://{p.netloc}" if p.scheme else f"https://{p.netloc}" + except Exception: + return None + + def _merge_env_with_creds(creds): """合并 config 凭证与环境变量: 显式非空 env 优先, 空串视为未设置; 且自动检测/自愈 App 注入的残留 baseURL (issue #3)。 @@ -112,23 +130,17 @@ def _merge_env_with_creds(creds): if env_bu and config_bu and env_bu != config_bu: all_hosts = set() try: - from urllib.parse import urlparse as _up _cfg_path = Path.home() / ".zcode" / "v2" / "config.json" with open(_cfg_path) as _f: _cfg = json.load(_f) for _p in _cfg.get("provider", {}).values(): _bu = _p.get("options", {}).get("baseURL", "") if _bu: - try: - _h = _up(_bu) - if _h.hostname: - all_hosts.add(f"{_h.scheme}://{_h.netloc}" if _h.scheme else _h.netloc) - except Exception: - pass - _eh = _up(env_bu) - env_host = f"{_eh.scheme}://{_eh.netloc}" if (_eh.scheme and _eh.hostname) else None - _ch = _up(config_bu) - config_host = f"{_ch.scheme}://{_ch.netloc}" if (_ch.scheme and _ch.hostname) else None + _h = _safe_host(_bu) # 与权威版同用 _safe_host (P2-3) + if _h: + all_hosts.add(_h) + env_host = _safe_host(env_bu) + config_host = _safe_host(config_bu) except Exception: env_host = config_host = None if env_host and config_host and env_host != config_host and env_host in all_hosts: @@ -160,9 +172,11 @@ def _parse_provider_error(text): # 配额耗尽 (不可短期重试) — 用精确的 quota/credit/配额 词, 不用模糊的 exceeded.*limit # (会误吃 "rate limit exceeded")。中文收紧为带限定词。Codex review P1#2。 + # insufficient 一条与权威版对齐 (整体 review P1-4: 拆成 credit/balance 两条 + # 漏了 "insufficient quota" 形态, 副本漂移未同步) quota_pats = [r"\bquota\b", r"配额(不足|耗尽|用完|超限)", r"额度(不足|耗尽|用完)", r"余额不足", - r"insufficient.*credit", r"insufficient.*balance", r"\bcredit\b.*exhaust"] + r"insufficient.*(quota|credit|balance)", r"\bcredit\b.*exhaust"] if any(re.search(p, lower) for p in quota_pats): return {"is_rate_limit": False, "is_quota": True, "retry_after_sec": retry_after, "error_kind": "quota"} @@ -321,6 +335,12 @@ class MimosaMcpClient: server_js = Path(root) / "payload" / "dist" / "mcp" / "server.js" if not server_js.exists(): raise FileNotFoundError(f"mimosa MCP server 不存在: {server_js}") + # 整体 review P2-6: 提前检查 node, 缺则给明确提示 (否则 Popen 抛 + # 裸 FileNotFoundError, 被上层包成"mimosa 扫描失败", 排查困难) + if not shutil.which("node"): + raise FileNotFoundError( + "未找到 node 可执行文件, 请先安装 Node.js " + "(mimosa MCP server 依赖 node 运行)") env = dict(os.environ) env["ZCODE_PLUGIN_ROOT"] = str(root) env.setdefault("MIMOSA_ENGINE", "native") @@ -333,6 +353,13 @@ class MimosaMcpClient: self._id = 0 def _send(self, msg): + # 写超时 (整体 review P1-1, 与 _recv 对称): server 卡住不读 stdin 时, + # pipe 缓冲写满后裸 write 会永久阻塞, 且无上层超时可打断 → 先 select + # 查可写性, 超时抛 TimeoutError + fd = self._proc.stdin.fileno() + _, writable, _ = select.select([], [fd], [], self.timeout) + if not writable: + raise TimeoutError(f"mimosa MCP 写入超时 ({self.timeout}s)") self._proc.stdin.write(json.dumps(msg, ensure_ascii=False) + "\n") self._proc.stdin.flush() @@ -414,7 +441,11 @@ def _read_mimosa_findings(scan_dir_str): 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: + # O_NOFOLLOW (整体 review P2-1): 关掉 resolve 与 open 之间的 TOCTOU 窗口 + # — 检查后目标文件被替换成 symlink 时 open 直接失败; 静态 symlink 仍由 + # 上面的 resolve 越界检查拦截, 两者互补。 + fd = os.open(real_file, os.O_RDONLY | os.O_NOFOLLOW) + with os.fdopen(fd) as f: return json.load(f).get("findings", []) except Exception as e: log(f"⚠ findings.json 读取失败 ({e}), 仅用摘要文本") @@ -484,7 +515,10 @@ def _mimosa_deep_scan(root, scan_path, focus_files=None): """ 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) + # 单次一问一答 (recv/send) 超时独立可配 (整体 review P2-4: 原先硬编码 120, + # 大项目 mimosa GC 卡顿偶超 120s 会白累计 status 超时计数) + recv_timeout = max(1, _env_int("ZCODE_BRIDGE_MIMOSA_RECV_TIMEOUT", 120, maximum=600)) + client = MimosaMcpClient(root, cwd=scan_path, timeout=recv_timeout) job_id = None terminal = False # 到达终态 (completed/failed/cancelled) 则 finally 不再 cancel status_timeouts = 0 @@ -581,11 +615,15 @@ def _compact_findings(findings): # ============================================================ # PR 审查: git diff 计算 + base 自动探测 # ============================================================ -def _git(repo, *git_args): - """跑 git 子命令, 返回 (returncode, stdout, stderr)。不抛异常, 调用方判。""" +def _git(repo, *git_args, timeout=15): + """跑 git 子命令, 返回 (returncode, stdout, stderr)。不抛异常, 调用方判。 + + timeout 分档 (整体 review P2-6): 元数据类 (rev-parse/symbolic-ref) 15s 充裕; + diff 类在超大仓/慢磁盘下可能超 30s, 由 _pr_diff 传 timeout=60。 + """ try: r = subprocess.run(["git", "-C", repo, *git_args], - capture_output=True, text=True, timeout=30) + capture_output=True, text=True, timeout=timeout) return r.returncode, r.stdout, r.stderr except (OSError, subprocess.TimeoutExpired) as e: return 128, "", str(e) @@ -616,11 +654,12 @@ def _pr_diff(repo, base, head): """ range_spec = f"{base}...{head}" rc, files_out, err = _git(repo, "diff", "--name-only", "--end-of-options", - range_spec, "--") + range_spec, "--", timeout=60) if rc != 0: raise RuntimeError(f"git diff {range_spec} 失败: {err.strip()[:200]}") changed = [ln.strip() for ln in files_out.splitlines() if ln.strip()] - rc, diff_text, err = _git(repo, "diff", "--end-of-options", range_spec, "--") + rc, diff_text, err = _git(repo, "diff", "--end-of-options", range_spec, "--", + timeout=60) if rc != 0: raise RuntimeError(f"git diff {range_spec} 失败: {err.strip()[:200]}") return changed, diff_text @@ -838,14 +877,29 @@ def tool_get_zcode_capabilities(args): return {"content": [{"type": "text", "text": result.stdout}]} +def _parse_result_json(output): + """从 zcode --json stdout 解析结果 JSON, 返回 dict 或 None。 + + 容忍前置非 JSON 行 (实测 zcode 会先打 "AI SDK Warning System" 等警告 + 再输出 JSON) 与尾随杂项; 逐个 "{" 位置尝试 raw_decode。 + """ + text = (output or "").strip() + start = text.find("{") + decoder = json.JSONDecoder() + while start != -1: + try: + data, _end = decoder.raw_decode(text[start:]) + return data if isinstance(data, dict) else None + except json.JSONDecodeError: + start = text.find("{", start + 1) + return None + + def _extract_response(output): - """--json 模式下 zcode stdout 是 JSON, 提取 response 字段; 解析失败原样返回。""" - try: - data = json.loads((output or "").strip()) - if isinstance(data, dict) and isinstance(data.get("response"), str): - return data["response"] - except (json.JSONDecodeError, ValueError): - pass + """--json 模式下提取 zcode stdout 的 response 字段; 无法识别时原样返回。""" + data = _parse_result_json(output) + if data is not None and isinstance(data.get("response"), str): + return data["response"] return output @@ -884,7 +938,18 @@ def _run_zcode_headless(cmd, env, timeout): # 成限流而白重试, 甚至把成功结果截断成报错)。zcode headless 错误 # 常在 stderr (APICallError 堆栈), stdout 是审查结论正文。 err_info = _parse_provider_error(result.stderr or "") - has_error = result.returncode != 0 or (result.stderr and err_info["error_kind"] != "unknown") + # has_error 判定 (整体 review P1-1 + 复审 P1-A): stderr 匹配 + # 错误关键词时, 仅当 stdout 不是合法 --json 结果才判错误。 + # zcode 常往 stderr 打 Node deprecation 警告, 恰好含 "Unauthorized" + # 等词时会误杀成功调用; 但"stdout 非空"也可能是错误回显文本, + # 所以判据是"能否解析出 response 结果"而非"非空"。 + stderr_matches = bool(result.stderr) \ + and err_info["error_kind"] != "unknown" + has_error = result.returncode != 0 or ( + stderr_matches and _parse_result_json(output) is None) + if stderr_matches and not has_error and result.returncode == 0: + log(f"⚠ stderr 含 {err_info['error_kind']} 关键词但 stdout " + "是合法结果, 按成功处理 (便于事后排查边界)") if has_error and err_info["is_rate_limit"] and attempt < max_retries: # 限流 → 退避后重试 (优先用 retry-after, 否则指数 + jitter) @@ -902,8 +967,18 @@ def _run_zcode_headless(cmd, env, timeout): return {"content": [{"type": "text", "text": f"zcode 调用失败{suffix}: {err_text[:500]}"}], "isError": True} - # 成功 (--json 输出提取 response; 非 JSON 原样返回) - return {"content": [{"type": "text", "text": _extract_response(output)}]} + # 成功: 先提取 response 再截断 (复审 P2-1: 先截断会破坏 + # JSON 结构导致提取失败)。体积上限 ZCODE_BRIDGE_MAX_OUTPUT + # (整体 review P2-1): 超大审查/verbose 输出可达几十 MB。 + text = _extract_response(output) + max_output = max(10_000, _env_int( + "ZCODE_BRIDGE_MAX_OUTPUT", 10_000_000, maximum=100_000_000)) + if len(text) > max_output: + log(f"⚠ zcode 输出超上限 ({len(text)} > {max_output} 字符), 截断") + text = text[:max_output] + ( + f"\n\n[... 输出超 {max_output} 字符已截断; " + "超大审查建议拆分文件分批进行 ...]") + return {"content": [{"type": "text", "text": text}]} except subprocess.TimeoutExpired: return {"content": [{"type": "text", "text": f"zcode review 超时 ({timeout}s)"}], @@ -950,6 +1025,37 @@ def tool_zcode_review(args): focus = _clamp_focus(args.get("focus")) or "全面审查: 安全、正确性、可维护性" cwd = args.get("cwd") + # files 来自 MCP client (信任边界外): 类型校验 + 200 上限 (整体 review P1-2: + # 无上限时上万个路径拼 argv 会撞 ARG_MAX; 与 focus_files 的处理对齐) + if not isinstance(files, list): + return {"content": [{"type": "text", + "text": "files 必须是字符串数组"}], + "isError": True} + if len(files) > 200: + log(f"⚠ files 超上限 ({len(files)} > 200), 截断") + files = files[:200] + # 元素级过滤: 非字符串元素丢弃 (复审 P2-3: argv 含非 str 会抛 TypeError) + valid_files = [f for f in files if isinstance(f, str)] + if len(valid_files) != len(files): + log(f"⚠ files 丢弃 {len(files) - len(valid_files)} 个非字符串元素") + files = valid_files + + # code 体积上限 (整体 review P1-3: 无上限时超大内联代码造成磁盘/内存放大; + # 与 PR diff 的 ZCODE_BRIDGE_PR_DIFF_MAX 防护对称) + if code: + code_max = max(10_000, _env_int("ZCODE_BRIDGE_CODE_MAX", 500_000, + maximum=5_000_000)) + code_bytes = code.encode("utf-8") + if len(code_bytes) > code_max: + log(f"⚠ code 超上限 ({len(code_bytes)} > {code_max} 字节), 截断") + cut = code_bytes[:code_max] + nl = cut.rfind(b"\n") # 退到完整行, 防半行误导; 容错解码防多字节切半 + if nl > 0: + cut = cut[:nl] + code = cut.decode("utf-8", errors="ignore") + ( + f"\n\n[... code 超 {code_max} 字节已截断; " + "超大代码建议改用 files 传文件路径 ...]") + # 构造 prompt — "只审不修"职责钉死在文本层, 写工具在工具集层物理禁用, # 双保险防 "审着审着顺手改了" (plan→build 惯性, 2026-08-08 重构动因)。 prompt = ( @@ -1349,8 +1455,10 @@ def handle_modern_request(req): # isError 也是 complete (工具执行错误走 isError, 非协议错误) return make_response(msg_id, _modern_result(result)) except Exception as e: + # 整体 review P1-5: str(e) 可能含内部路径/用户名, 不外泄给 client, + # 详细错误只走 log (stderr) log(f"tool {tool_name} 异常: {e}") - return make_error(msg_id, -32603, f"tool 执行异常: {e}") + return make_error(msg_id, -32603, "内部错误, 详见服务端日志") if method is not None: return make_error(msg_id, -32601, f"未知 method: {method}") @@ -1403,8 +1511,10 @@ def handle_request(req): result = TOOL_HANDLERS[tool_name](arguments) return make_response(msg_id, result) except Exception as e: + # 整体 review P1-5: str(e) 可能含内部路径/用户名, 不外泄给 client, + # 详细错误只走 log (stderr) log(f"tool {tool_name} 异常: {e}") - return make_error(msg_id, -32603, f"tool 执行异常: {e}") + return make_error(msg_id, -32603, "内部错误, 详见服务端日志") # 未知方法 if method is not None: @@ -1434,9 +1544,10 @@ def main(): # ≥2025-06-18 的 server 必须拒绝) 和其他非 dict 形态, 统一 -32600。 if not isinstance(req, dict): log("收到非 JSON-RPC 对象消息 (batch?), 拒绝") - send_message(make_error( - None, -32600, - "Invalid Request: 仅接受单个 JSON-RPC 对象")) + err = make_error(None, -32600, "Invalid Request: 仅接受单个 JSON-RPC 对象") + # batch 输入按 JSON-RPC 规范回 batch 数组 (整体 review P2-5); + # 其他裸值仍回单个 error 对象 + send_message([err] if isinstance(req, list) else err) continue # ---- 纪元路由 (dual-era, 逐请求判定, 无会话状态) ---- @@ -1463,8 +1574,9 @@ def main(): if resp is not None: send_message(resp) except Exception as e: + # 整体 review P1-5: 同 tools/call 兜底, 不外泄内部细节 log(f"处理请求异常: {e}") - send_message(make_error(req.get("id"), -32603, str(e))) + send_message(make_error(req.get("id"), -32603, "内部错误, 详见服务端日志")) log("stdin 关闭, 退出") diff --git a/shared/credentials.py b/shared/credentials.py index a9b60c0..bb0906b 100644 --- a/shared/credentials.py +++ b/shared/credentials.py @@ -11,8 +11,12 @@ 设计说明: 为了保持"单文件可独立运行"的特性 (用户复制一个文件就能用), - mcp-server 和 acp-bridge 各自内嵌了一份本逻辑的副本 (标注"源自此处")。 - 本文件是权威实现; 修改凭证逻辑时, 请同步更新两处副本。 + mcp-server、acp-bridge 和 agent-help 各自内嵌了一份本逻辑的副本 (标注"源自此处")。 + 本文件是权威实现; 修改凭证逻辑时, 请同步更新三处副本 (整体 review P1-2): + packages/mcp-server/zcode-mcp-server (load_zcode_credentials / _merge_env_with_creds) + packages/acp-bridge/zcode-acp-bridge (load_zcode_credentials / _merge_env_with_creds) + packages/agent-help/zcode-agent-help (_load_creds_internal / _safe_host) + tests/test_credentials.py 的 C11/C12 同步测试会断言副本与权威版行为一致。 用法: from shared.credentials import load_zcode_credentials, merge_env_with_creds @@ -194,6 +198,7 @@ def merge_env_with_creds(creds=None, environ=None, config_path=None, warn=None): print(f" ZCODE_MODEL: {creds.get('ZCODE_MODEL')}") print(f" ZCODE_BASE_URL: {creds.get('ZCODE_BASE_URL')}") key = creds.get("ANTHROPIC_API_KEY", "") - print(f" ANTHROPIC_API_KEY: {key[:8]}...{key[-4:]}" if len(key) > 12 else " (短或空)") + # 脱敏位数与 zcode-agent-help._mask_key 对齐 (统一前 4 后 4, 整体 review P2-5) + print(f" ANTHROPIC_API_KEY: {key[:4]}...{key[-4:]}" if len(key) > 12 else " (短或空)") else: print(f"❌ 未从 {ZCODE_CREDS_PATH} 读取到凭证") diff --git a/tests/test_app_server_methods.py b/tests/test_app_server_methods.py index b4c0b21..ebce641 100644 --- a/tests/test_app_server_methods.py +++ b/tests/test_app_server_methods.py @@ -41,6 +41,8 @@ import json import os import queue +import subprocess +import tempfile import threading import time import types @@ -1264,6 +1266,227 @@ def test_z1_unknown_method_32601(self): resp = self._call(bridge, "session/nonexistent", {"sessionId": "sess_x"}) self._assert_error_code(resp, -32601) + # ---------- RV: 整体 deep review (zcode 狗食) 修复回归锚 ---------- + def test_rv1_next_id_monotonic_and_threadsafe(self): + """RV1 (P1-1): _next_id 基于 itertools.count — 单调递增且并发取 id 不重复""" + bridge, _ = self._new_bridge() + first = bridge._next_id() + self.assertGreaterEqual(first, 10_000_001, + "id 空间从 10_000_001 起 (避开 server 的 id 空间)") + self.assertEqual(bridge._next_id(), first + 1) + # 并发冒烟: 两线程各取 500 个 id, 全集不得重复 (count 的 next() 原子) + got = [] + lock = threading.Lock() + + def _grab(): + local = [bridge._next_id() for _ in range(500)] + with lock: + got.extend(local) + + threads = [threading.Thread(target=_grab) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(len(set(got)), len(got), "并发取 id 不得重复") + + def test_rv2_write_stdout_single_locked_frame(self): + """RV2 (P1-2): stdout 全收进 _write_stdout 加锁写口, 输出完整单行 JSON 帧""" + bridge, _ = self._new_bridge() + self.assertTrue(hasattr(bridge, "_stdout_lock"), "bridge 应持有 stdout 写锁") + out = io.StringIO() + with contextlib.redirect_stdout(out): + bridge._send_acp_notification("session/update", {"sessionId": "s"}) + line = out.getvalue() + self.assertTrue(line.endswith("\n") and "\n" not in line[:-1], + "一帧必须是单行 JSON (半截/交错帧会让 client 解析失败)") + frame = json.loads(line) + self.assertEqual(frame["method"], "session/update") + self.assertNotIn("id", frame, "notification 无 id") + + def test_rv3_pending_turns_cleaned_on_exception(self): + """RV3 (P1-5): prompt 中途异常也必须摘掉 pending turn (try/finally 兜底)""" + class _BoomBackend: + """session/messages 抛异常 (模拟 baseline 拉取失败) 的最小桩""" + + def request(self, msg_id, method, params=None, timeout=30): + if method == "session/messages": + raise RuntimeError("boom (模拟中途异常)") + return {"result": {"ok": True}}, [] + + def send(self, msg): + pass + + bridge = self.Bridge() + bridge.backend = _BoomBackend() + bridge.session_map["acp_rv3"] = "sess_rv3" + with self.assertRaises(RuntimeError): + self._call(bridge, "session/prompt", + {"sessionId": "acp_rv3", "prompt": "hi"}) + self.assertEqual(bridge.pending_turns, {}, + "异常路径不得留下僵尸 pending turn (P1-5)") + + def test_rv4_differs_cap_fifo_eviction(self): + """RV4 (P2-1): differ 超 _MAX_SESSION_STATES 上限 FIFO 淘汰最旧 session""" + bridge, _ = self._new_bridge() + cap = self.mod._MAX_SESSION_STATES + for i in range(cap + 3): + bridge._get_or_create_differ(f"sess_{i}") + self.assertEqual(len(bridge._differs), cap, "differ 总数不得超上限") + self.assertNotIn("sess_0", bridge._differs, "最旧条目应被 FIFO 淘汰") + self.assertIn(f"sess_{cap + 2}", bridge._differs, "最新条目必须保留") + + def test_rv5_state_projection_cap_fifo_eviction(self): + """RV5 (P2-2): state 投影超上限同样 FIFO 淘汰 (跨 session 无清理的兜底)""" + backend = self._bare_backend() + cap = self.mod._MAX_SESSION_STATES + for i in range(cap + 3): + backend._merge_state_patch({"sessionId": f"sess_{i}", + "patch": {"status": "running"}}) + self.assertEqual(len(backend._state_projections), cap, + "投影总数不得超上限") + self.assertIsNone(backend.get_projection("sess_0"), "最旧投影应被淘汰") + self.assertIsNotNone(backend.get_projection(f"sess_{cap + 2}"), + "最新投影必须保留") + + def test_rv6_redact_secret_hex_and_uuid(self): + """RV6 (P2-3): 纯 hex (32+) / uuid 形态的无前缀 key 也脱敏""" + hex_key = "0123456789abcdef" * 4 # 64 位 hex + uuid_key = "123e4567-e89b-12d3-a456-426614174000" + out = self.Bridge._redact_secret( + f"provider error: key={hex_key} alt={uuid_key} path=/tmp/ok") + self.assertNotIn(hex_key, out, "64 位 hex key 应被遮蔽") + self.assertNotIn(uuid_key, out, "uuid 形态 key 应被遮蔽") + self.assertIn("path=/tmp/ok", out, "非敏感部分保留") + # 短十六进制不误伤 (如 git 短 sha / 普通单词) + self.assertIn("abc1234", self.Bridge._redact_secret("commit abc1234")) + + def test_rv7_close_kill_reaps_zombie(self): + """RV7 (P2-7): close() 的 kill() 后补 wait(timeout=2) 收尸防僵尸""" + calls = [] + + class _Proc: + def terminate(self): + calls.append("terminate") + + def wait(self, timeout=None): + calls.append(("wait", timeout)) + if timeout == 3: + raise subprocess.TimeoutExpired("zcode", 3) + + def kill(self): + calls.append("kill") + + backend = self.mod.ZCodeBackend.__new__(self.mod.ZCodeBackend) + backend.proc = _Proc() + backend.close() + self.assertEqual(calls, ["terminate", ("wait", 3), "kill", ("wait", 2)], + "kill 后必须再 wait 收尸, 防子进程变僵尸") + + def test_rv8_register_listener_overwrite_warns(self): + """RV8 (P2-8): 同 sid 重复注册 listener 打告警 (防静默丢事件)""" + backend = self._bare_backend() + l1, l2 = object(), object() + captured = io.StringIO() + with contextlib.redirect_stderr(captured): + backend.register_event_listener("sess_rv8", l1) + backend.register_event_listener("sess_rv8", l2) + self.assertIn("覆盖", captured.getvalue(), + "同 sid 覆盖注册必须打告警") + self.assertIs(backend._event_listeners["sess_rv8"], l2, + "覆盖后路由到新 listener (当前行为保留, 只是不再静默)") + + def test_rv9_wait_for_turn_idle_backoff_and_drain(self): + """RV9 (P2-5): goal 等待循环指数退避 (1s→2s) 且每轮 drain inbox""" + bridge, fake = self._new_bridge({ + "session/goal": {"response": [ + {"error": {"message": "Cannot manage goals while a prompt is running"}}, + {"error": {"message": "Cannot manage goals while a prompt is running"}}, + {"result": {"ok": True}}, + ]}, + }) + bridge.session_map["sess_rv9"] = "sess_rv9" + turn = {"zcode_sid": "sess_rv9", "cancelled": False} + bridge.pending_turns[7] = turn + bridge._inbox.put(json.dumps({"method": "session/cancel", + "params": {"sessionId": "sess_rv9"}})) + sleeps = [] + real_sleep = time.sleep + time.sleep = lambda s: sleeps.append(s) + try: + ok = bridge._wait_for_turn_idle("sess_rv9", timeout=60, + probe_method="session/goal") + finally: + time.sleep = real_sleep + self.assertTrue(ok, "第三次探测成功应返回 True") + self.assertEqual(sleeps[:2], [1.0, 2.0], "探测重试应指数退避 (1s→2s, 封顶 8s)") + self.assertTrue(turn["cancelled"], "等待期间应 drain inbox 让 cancel 生效") + + def test_rv10_credentials_missing_vs_corrupt(self): + """RV10 (P2-4): 凭证文件缺失静默降级 {}; 存在但损坏 → 显式警告文案""" + mod = self.mod + orig_path = mod.ZCODE_CREDS_PATH + bad_path = None + try: + # 缺失: 静默降级, 不打损坏警告 + mod.ZCODE_CREDS_PATH = "/nonexistent/dir/config.json" + captured = io.StringIO() + with contextlib.redirect_stderr(captured): + creds = mod.load_zcode_credentials() + self.assertEqual(creds, {}, "未配置文件应静默降级为空 dict") + self.assertNotIn("凭证文件存在但读取失败", captured.getvalue(), + "未配置属正常降级, 不得打损坏警告") + # 损坏: 返回 {} 但打显式警告 (区别于「没配凭证」) + with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f: + f.write("{not valid json") + bad_path = f.name + mod.ZCODE_CREDS_PATH = bad_path + captured = io.StringIO() + with contextlib.redirect_stderr(captured): + creds = mod.load_zcode_credentials() + self.assertEqual(creds, {}) + self.assertIn("凭证文件存在但读取失败", captured.getvalue(), + "文件存在但损坏必须显式警告") + finally: + mod.ZCODE_CREDS_PATH = orig_path + if bad_path: + os.unlink(bad_path) + + def test_rv11_default_mode_env_override(self): + """RV11 (安全发现): ZCODE_ACP_DEFAULT_MODE 覆盖 session/new 缺省 mode, + 默认仍 yolo, 显式 mode 优先级最高""" + self.assertEqual(self.mod.DEFAULT_ACP_MODE, "yolo", + "缺省默认仍是 yolo (历史行为不变)") + # 无 env 时 session/new 缺省 mode=yolo + bridge, fake = self._new_bridge() + self._call(bridge, "session/new", {"cwd": "/p"}) + create = [c for c in fake.calls if c["method"] == "session/create"] + self.assertEqual(create[0]["params"].get("mode"), "yolo") + + old = os.environ.get("ZCODE_ACP_DEFAULT_MODE") + os.environ["ZCODE_ACP_DEFAULT_MODE"] = "build" + try: + mod2 = _load_bridge_module() # 重新 exec 模块让 env 生效 + self.assertEqual(mod2.DEFAULT_ACP_MODE, "build") + bridge2 = mod2.ACPBridge() + fake2 = FakeBackend() + bridge2.backend = fake2 + bridge2.handle_acp({"jsonrpc": "2.0", "id": 1, "method": "session/new", + "params": {"cwd": "/p"}}) + create2 = [c for c in fake2.calls if c["method"] == "session/create"] + self.assertEqual(create2[0]["params"].get("mode"), "build", + "env 覆盖后缺省 mode 应取 env 值") + bridge2.handle_acp({"jsonrpc": "2.0", "id": 2, "method": "session/new", + "params": {"cwd": "/p", "mode": "plan"}}) + create3 = [c for c in fake2.calls if c["method"] == "session/create"] + self.assertEqual(create3[1]["params"].get("mode"), "plan", + "显式 mode 优先于 env 默认值") + finally: + if old is None: + os.environ.pop("ZCODE_ACP_DEFAULT_MODE", None) + else: + os.environ["ZCODE_ACP_DEFAULT_MODE"] = old + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/tests/test_credentials.py b/tests/test_credentials.py index 3733ab4..14b84fa 100644 --- a/tests/test_credentials.py +++ b/tests/test_credentials.py @@ -18,14 +18,19 @@ 依赖: 仅 Python 标准库 + shared/credentials.py """ +import contextlib +import io import json import os import sys import tempfile +import types import unittest +from unittest import mock sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "shared")) from credentials import ( # noqa: E402 + _safe_host, is_stale_env_base_url, load_zcode_credentials, merge_env_with_creds, @@ -293,5 +298,272 @@ def test_c10g_no_config_path_still_safe(self): self.assertEqual(merged["ZCODE_BASE_URL"], "https://self-hosted.test") +# ============================================================ +# C11/C12: 内嵌副本与 shared 权威版的同步测试 (整体 review P2-1) +# mcp-server / agent-help 为"单文件可独立运行"各内嵌了一份凭证逻辑副本, +# 这里断言它们与 shared/credentials.py 权威实现在同一 fixture 上行为一致, 防漂移。 +# ============================================================ +MCP_SERVER_PATH = os.path.join( + os.path.dirname(__file__), "..", "packages", "mcp-server", "zcode-mcp-server" +) +AGENT_HELP_PATH = os.path.join( + os.path.dirname(__file__), "..", "packages", "agent-help", "zcode-agent-help" +) + +_CRED_KEYS = ("ZCODE_MODEL", "ZCODE_BASE_URL", "ANTHROPIC_API_KEY") + + +def _load_single_file_module(path, name): + """exec 加载无后缀单文件组件 (去掉 __main__ 块), 与 test_mcp_protocol 同模式。""" + mod = types.ModuleType(name) + mod.__file__ = path + with open(path) as f: + code = f.read() + code_no_main = code.split('if __name__ == "__main__":')[0] + exec(code_no_main, mod.__dict__) + return mod + + +def _isolated_env(test_case, home, **env): + """patch os.environ: 移除三个凭证 key + HOME 指向隔离目录, 叠加 env 指定值。""" + new = {k: v for k, v in os.environ.items() if k not in _CRED_KEYS} + new["HOME"] = home + new.update(env) + p = mock.patch.dict(os.environ, new, clear=True) + p.start() + test_case.addCleanup(p.stop) + + +def _write_isolated_config(home, enabled_url="https://api.z.ai/api/anthropic", + stale_url="https://zcode.z.ai/api/v1/zcode-plan/anthropic"): + """在隔离 HOME 里写两 provider 的 config (enabled + 一个 disabled 的 stale), 返回路径。""" + cfg_dir = os.path.join(home, ".zcode", "v2") + os.makedirs(cfg_dir, exist_ok=True) + cfg_path = os.path.join(cfg_dir, "config.json") + with open(cfg_path, "w") as f: + json.dump({ + "provider": { + "builtin:zai-coding-plan": { + "enabled": True, + "options": {"baseURL": enabled_url, "apiKey": "sk-good-key-123456"}, + "models": {"GLM-5.2": {}}, + }, + "builtin:zai-start-plan": { + "enabled": False, + "options": {"baseURL": stale_url, "apiKey": "jwt-stale"}, + "models": {"GLM-5.2": {}}, + }, + } + }, f) + return cfg_path + + +class TestMcpServerSync(unittest.TestCase): + """C11: mcp-server 内嵌副本 (load_zcode_credentials / _merge_env_with_creds) + 与 shared 权威版行为一致 (整体 review P2-1)。 + + mcp-server 副本用 Path.home() 定位 config、直读 os.environ (均不接受参数), + 故用 HOME 环境变量隔离 + patch os.environ 注入场景。 + """ + + @classmethod + def setUpClass(cls): + cls.mcp = _load_single_file_module(MCP_SERVER_PATH, "zcode_mcp_server") + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + self.cfg_path = _write_isolated_config(self._tmp.name) + _isolated_env(self, self._tmp.name) # 基线: 无凭证 env + + def _assert_merge_parity(self, scenario): + """对拍: mcp 副本与权威版 merge 后, 三个凭证 key 完全一致。""" + mcp_merged = self.mcp._merge_env_with_creds(self.mcp.load_zcode_credentials()) + ref_merged = merge_env_with_creds( + load_zcode_credentials(config_path=self.cfg_path), + dict(os.environ), config_path=self.cfg_path) + for k in _CRED_KEYS: + self.assertEqual(mcp_merged.get(k, ""), ref_merged.get(k, ""), + f"{scenario}: {k} 在 mcp-server 副本与权威版间不一致") + return ref_merged + + def test_c11a_load_parity(self): + """C11a: load_zcode_credentials 副本与权威版读同一 config 结果一致""" + self.assertEqual(self.mcp.load_zcode_credentials(), + load_zcode_credentials(config_path=self.cfg_path)) + + def test_c11b_merge_no_env(self): + """C11b: 无凭证 env → 双份都用 config 值""" + merged = self._assert_merge_parity("无 env") + self.assertEqual(merged["ZCODE_MODEL"], "GLM-5.2") + self.assertEqual(merged["ZCODE_BASE_URL"], "https://api.z.ai/api/anthropic") + + def test_c11c_empty_env_not_override(self): + """C11c: 空串 env 不覆盖 (双份一致)""" + _isolated_env(self, self._tmp.name, ZCODE_MODEL="") + merged = self._assert_merge_parity("空串 env") + self.assertEqual(merged["ZCODE_MODEL"], "GLM-5.2") + + def test_c11d_nonempty_env_overrides(self): + """C11d: 非空 env 覆盖 (双份一致)""" + _isolated_env(self, self._tmp.name, ZCODE_MODEL="GLM-5-Turbo") + merged = self._assert_merge_parity("非空 env") + self.assertEqual(merged["ZCODE_MODEL"], "GLM-5-Turbo") + + def test_c11e_stale_env_self_heals(self): + """C11e: 残留 env (config 另一 provider 的 endpoint) → 双份都自愈为 enabled 值""" + _isolated_env(self, self._tmp.name, ZCODE_BASE_URL="https://zcode.z.ai") + merged = self._assert_merge_parity("残留 env") + self.assertEqual(merged["ZCODE_BASE_URL"], "https://api.z.ai/api/anthropic") + + def test_c11f_custom_endpoint_respected(self): + """C11f: 自建 endpoint (不在 config 任何 provider) → 双份都尊重 env""" + _isolated_env(self, self._tmp.name, ZCODE_BASE_URL="https://my-proxy.example.com") + merged = self._assert_merge_parity("自建 endpoint") + self.assertEqual(merged["ZCODE_BASE_URL"], "https://my-proxy.example.com") + + +class TestAgentHelp(unittest.TestCase): + """C12: agent-help 内嵌副本同步 + 本轮 review 修复的回归 (P1-1/3/4/5, P2-2/4/6)。 + + agent-help 是单文件可独立运行设计, 不 import shared, 靠这里的对拍防漂移。 + """ + + @classmethod + def setUpClass(cls): + cls.ah = _load_single_file_module(AGENT_HELP_PATH, "zcode_agent_help") + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.addCleanup(self._tmp.cleanup) + + def _run_main(self, argv): + """调 agent-help main(), 返回 (rc, stdout, stderr)。""" + out, err = io.StringIO(), io.StringIO() + with mock.patch.object(sys, "argv", ["zcode-agent-help"] + argv), \ + contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + rc = self.ah.main() + return rc, out.getvalue(), err.getvalue() + + def test_c12a_safe_host_parity(self): + """C12a: agent-help._safe_host 与权威 _safe_host 逐值一致 (整体 review P1-1)""" + cases = [ + "https://api.z.ai/api/anthropic", # 带路径 + "https://zcode.z.ai", # 根域名 (App 注入形态) + "//zcode.z.ai/api/v1/x", # 无 scheme → 补 https:// + "http://localhost:8080/path", # 带端口 + "ftp://example.com/x", # 非 http scheme + "zcode.z.ai/api/v1/x", # 无 // → hostname 解析不出 → None + "", "not a url", + ] + for u in cases: + self.assertEqual(self.ah._safe_host(u), _safe_host(u), f"_safe_host 漂移: {u!r}") + # 无 scheme 补 https:// 分支必须存在 (P1-1 核心) + self.assertEqual(self.ah._safe_host("//zcode.z.ai/x"), "https://zcode.z.ai") + + def test_c12b_section_missing_value(self): + """C12b: --section 末尾无值 → 用法错误 + return 1, 不静默打印全量 (P1-3)""" + rc, out, err = self._run_main(["--section"]) + self.assertEqual(rc, 1) + self.assertIn("用法错误", err) + self.assertEqual(out, "", "不应打印全量 JSON") + + def test_c12c_section_followed_by_flag(self): + """C12c: --section 后随另一个 flag → 同样按用法错误处理 (P1-3)""" + rc, out, err = self._run_main(["--section", "--pretty"]) + self.assertEqual(rc, 1) + self.assertIn("用法错误", err) + + def test_c12d_empty_creds_prints_env(self): + """C12d: config 无 enabled provider 但 env 有值 → 脱敏打印并标注 (P1-5)""" + _isolated_env(self, self._tmp.name, + ZCODE_BASE_URL="https://zcode.z.ai", + ANTHROPIC_API_KEY="sk-abcdef1234567890") + out = io.StringIO() + with contextlib.redirect_stdout(out): + rc = self.ah.print_injected_env(config_path="/nonexistent/config.json") + self.assertEqual(rc, 0) + text = out.getvalue() + self.assertIn("config 无 enabled provider, 将直接使用 env 值", text) + self.assertIn("https://zcode.z.ai", text) + self.assertIn("sk-a...7890", text, "apiKey 应按 4+4 脱敏") + self.assertNotIn("sk-abcdef1234567890", text, "不得泄露明文 key") + + def test_c12e_empty_creds_empty_env(self): + """C12e: config 空且 env 也空 → 维持原 ❌ 提示 (P1-5 不改变该路径)""" + _isolated_env(self, self._tmp.name) + out = io.StringIO() + with contextlib.redirect_stdout(out): + rc = self.ah.print_injected_env(config_path="/nonexistent/config.json") + self.assertEqual(rc, 0) + self.assertIn("❌", out.getvalue()) + + def test_c12f_probe_exception_safety(self): + """C12f: zcode 二进制不可执行 (PermissionError) 时探测兜底不崩 (P1-4)""" + err = io.StringIO() + saved_run = self.ah.subprocess.run + + def fake_run(*a, **kw): + raise PermissionError("cannot execute binary") + + self.ah.subprocess.run = fake_run + try: + with contextlib.redirect_stderr(err): + self.assertIsNone(self.ah._run_zcode_json(["skills", "list"])) + self.assertEqual(self.ah._get_version(), "unknown") + finally: + self.ah.subprocess.run = saved_run + self.assertIn("探测失败", err.getvalue(), "探测失败应有 stderr 提示") + + def test_c12g_commands_non_list_guard(self): + """C12g: commands list 返回 dict 型 commands → custom_commands=[] 不出垃圾 (P2-2)""" + saved_run = self.ah.subprocess.run + + class _R: + returncode = 0 + stdout = json.dumps({"commands": {"a": 1}}) # dict 而非 list + + self.ah.subprocess.run = lambda *a, **kw: _R() + try: + env = self.ah.discover_environment() + finally: + self.ah.subprocess.run = saved_run + self.assertEqual(env["custom_commands"], []) + + def test_c12h_pretty_incomplete_environment(self): + """C12h: print_pretty 对缺 key 的 environment 不 KeyError (P2-4)""" + saved_run = self.ah.subprocess.run + + def fake_run(*a, **kw): + raise OSError("no zcode") + + self.ah.subprocess.run = fake_run + try: + with contextlib.redirect_stderr(io.StringIO()): + cap = self.ah.build_full() + finally: + self.ah.subprocess.run = saved_run + cap["environment"] = {} # 模拟不完整探测结果 + out = io.StringIO() + with contextlib.redirect_stdout(out): + self.ah.print_pretty(cap) # 不应抛 KeyError + self.assertIn("unknown", out.getvalue()) + + def test_c12i_overview_tested_against(self): + """C12i: OVERVIEW 带 tested_against 语义注记 (P2-6)""" + self.assertIn("tested_against", self.ah.OVERVIEW) + self.assertIn("0.16.1", self.ah.OVERVIEW["tested_against"]) + + def test_c12j_stale_env_diagnosed(self): + """C12j: 残留 env 在 --print-injected-env 里被标注 🚫 (P1-1 行为级)""" + cfg_path = _write_isolated_config(self._tmp.name) + _isolated_env(self, self._tmp.name, ZCODE_BASE_URL="https://zcode.z.ai") + out = io.StringIO() + with contextlib.redirect_stdout(out): + rc = self.ah.print_injected_env(config_path=cfg_path) + self.assertEqual(rc, 0) + self.assertIn("🚫 残留", out.getvalue()) + + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/tests/test_event_translator.py b/tests/test_event_translator.py index 1fd7923..40b88d2 100644 --- a/tests/test_event_translator.py +++ b/tests/test_event_translator.py @@ -59,6 +59,7 @@ import os import threading +import time import types import unittest @@ -563,5 +564,86 @@ def send(self, msg): "失败文案应取 payload 的 error.message (实测载荷无 resultType)") +class _TimeoutScriptBackend: + """最小桩: 按 method 脚本化响应 (事件超时用例只需 session/read)""" + + def __init__(self, responses): + self.responses = responses + self.calls = [] + + def request(self, msg_id, method, params=None, timeout=30): + self.calls.append(method) + return self.responses.get(method, {"result": {}}), [] + + def send(self, msg): + pass + + +class TestEventTurnTimeout(unittest.TestCase): + """事件模式 120s 超时的语义 (整体 review tests 报告 P2-5): + + turn 已启动 (turn.started) 但事件流再未给完成信号 → -32603「事件流超时」, + 不再返回 max_turn_requests 正常 stopReason (否则 client 无法区分「turn 真的 + 太长」与「事件流卡死」); turn 从未启动则保持 max_turn_requests 原语义。 + + time.time 快进 (同 test_app_server_methods.py PM3 手法), 不真等 120s。 + """ + + @classmethod + def setUpClass(cls): + cls.mod = _load_bridge_module() + + def _run_event_turn_fast(self, bridge, first_event): + """直调 _run_event_turn; poll_event 首调用给 first_event 之后恒 None, + time.time 每次调用快进 1s 让 120s 超时瞬时触发。""" + zcode_sid = "sess_timeout" + bridge.session_map[zcode_sid] = zcode_sid + msg_id = 1 + turn = {"zcode_sid": zcode_sid, "cancelled": False, "perms_responses": {}} + bridge.pending_turns[msg_id] = turn + listener = self.mod.EventStreamListener(bridge.backend, zcode_sid) + sent = [False] + + def _poll(timeout=0.5): + if not sent[0] and first_event is not None: + sent[0] = True + return first_event + return None + + listener.poll_event = _poll # 实例级替换, 不真阻塞 0.5s + differ = bridge._get_or_create_differ(zcode_sid) + real_time = time.time + clock = [real_time()] + try: + time.time = lambda: (clock.__setitem__(0, clock[0] + 1.0), clock[0])[1] + return _run_with_guard(lambda: bridge._run_event_turn( + listener, zcode_sid, zcode_sid, msg_id, turn, + chunk_msg_id="chunk_timeout", differ=differ)) + finally: + time.time = real_time + + def test_et1_turn_started_stream_broken_is_error(self): + """ET1 (P2-5): turn.started 后事件流断 (120s 无完成信号) → -32603 事件流超时""" + bridge = self.mod.ACPBridge() + # 停滞检查的 session/read 一直说 running (turn 永不完成, 事件流实断) + bridge.backend = _TimeoutScriptBackend({ + "session/read": {"result": {"projection": {"status": "running"}}}, + }) + resp = self._run_event_turn_fast( + bridge, _session_event("turn.started", _turn_started())) + self.assertIn("error", resp, "事件流中断属异常, 不得返回正常 stopReason") + self.assertEqual(resp["error"]["code"], -32603) + self.assertIn("事件流超时", resp["error"]["message"]) + self.assertEqual(bridge.pending_turns, {}, "超时路径也应摘掉 pending turn") + + def test_et2_turn_never_started_keeps_stop_reason(self): + """ET2 (P2-5 边界): turn 从未启动 (无任何事件) → 保持 max_turn_requests 原语义""" + bridge = self.mod.ACPBridge() + bridge.backend = _TimeoutScriptBackend({}) + resp = self._run_event_turn_fast(bridge, None) + self.assertNotIn("error", resp) + self.assertEqual(resp["result"]["stopReason"], "max_turn_requests") + + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/tests/test_mcp_protocol.py b/tests/test_mcp_protocol.py index 5a428c3..147edfb 100644 --- a/tests/test_mcp_protocol.py +++ b/tests/test_mcp_protocol.py @@ -92,12 +92,15 @@ class TestMainLoopRejection(_MainLoopCase): """非对象消息的拒收行为""" def test_mr0_batch_array_rejected(self): - """MR0: JSON-RPC batch 数组 → -32600 (2025-06-18 起 server 必须拒绝)""" + """MR0: JSON-RPC batch 数组 → -32600, 且响应本身也是 batch 数组 + (2025-06-18 起 server 必须拒绝; 整体 review P2-5 规范对齐)""" responses = self._run_main([ json.dumps([{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}]), ]) self.assertEqual(len(responses), 1) - self.assertEqual(responses[0]["error"]["code"], -32600) + self.assertIsInstance(responses[0], list, "batch 拒收响应应回 batch 数组") + self.assertEqual(len(responses[0]), 1) + self.assertEqual(responses[0][0]["error"]["code"], -32600) def test_mr1_bare_value_rejected(self): """MR1: 裸值 (非 dict) → -32600""" @@ -168,6 +171,66 @@ def test_tl3_tool_naming_convention(self): f"{t['name']} 不符合命名规范") +class TestInternalErrorMessage(_MainLoopCase): + """-32603 兜底回通用文案, 不把内部路径/异常细节泄露给 client (整体 review P1-5)""" + + @staticmethod + def _boom(args): + raise RuntimeError("/Users/secret/internal-path exploded") + + def test_ie0_legacy_tools_call_generic(self): + """IE0: legacy tools/call handler 抛异常 → -32603 通用文案""" + mod = self.mod + saved = mod.TOOL_HANDLERS["zcode_review"] + mod.TOOL_HANDLERS["zcode_review"] = self._boom + try: + resp = mod.handle_request( + {"jsonrpc": "2.0", "id": 7, "method": "tools/call", + "params": {"name": "zcode_review", "arguments": {}}}) + finally: + mod.TOOL_HANDLERS["zcode_review"] = saved + self.assertEqual(resp["error"]["code"], -32603) + self.assertNotIn("secret", resp["error"]["message"], + "异常细节不应回给 client") + self.assertIn("服务端日志", resp["error"]["message"]) + + def test_ie1_modern_tools_call_generic(self): + """IE1: modern tools/call handler 抛异常 → -32603 通用文案""" + mod = self.mod + saved = mod.TOOL_HANDLERS["zcode_review"] + mod.TOOL_HANDLERS["zcode_review"] = self._boom + try: + req = json.loads(_modern_req( + "tools/call", rid=8, + params={"name": "zcode_review", "arguments": {}})) + resp = mod.handle_modern_request(req) + finally: + mod.TOOL_HANDLERS["zcode_review"] = saved + self.assertEqual(resp["error"]["code"], -32603) + self.assertNotIn("secret", resp["error"]["message"]) + self.assertIn("服务端日志", resp["error"]["message"]) + + def test_ie2_main_loop_generic(self): + """IE2: 主循环兜底 (handler 层之外炸) → -32603 通用文案""" + mod = self.mod + saved = mod.handle_request + + def exploding(req): + raise RuntimeError("secret-main-loop-detail") + + mod.handle_request = exploding + try: + responses = self._run_main([ + json.dumps({"jsonrpc": "2.0", "id": 3, "method": "tools/list"}), + ]) + finally: + mod.handle_request = saved + self.assertEqual(responses[0]["error"]["code"], -32603) + self.assertNotIn("secret-main-loop-detail", + responses[0]["error"]["message"]) + self.assertIn("服务端日志", responses[0]["error"]["message"]) + + def _modern_req(method, rid=1, params=None, version="2026-07-28"): """构造一条 modern 纪元请求 (带 _meta 信封)。""" p = dict(params or {}) diff --git a/tests/test_mcp_retry_lock.py b/tests/test_mcp_retry_lock.py index 6f0deef..383bde6 100644 --- a/tests/test_mcp_retry_lock.py +++ b/tests/test_mcp_retry_lock.py @@ -22,6 +22,7 @@ 依赖: 仅 Python 标准库 + zcode-mcp-server 模块 """ +import json import os import types import unittest @@ -344,6 +345,140 @@ def test_rt6_review_text_with_rate_keywords_not_misjudged(self): os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) else: os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + # ---------- RT7: stderr 警告词 + stdout 有结论 → 不误杀 (整体 review P1-1) ---------- + def test_rt7_stderr_warning_with_stdout_not_error(self): + """RT7: exit=0 + stdout 是合法 --json 结论 + stderr 含错误关键词 → 算成功 + + 整体 review P1-1 + 复审 P1-A: 判据是"stdout 可解析出 response 结果" + (非"非空") — zcode 常往 stderr 打 Node 警告/诊断, 恰好含 + "Unauthorized" 等词时, 旧规则会把成功审查误判为错误丢弃结论。 + """ + mod = self.mod + procs = [_FakeCompletedProcess( + returncode=0, + stdout=json.dumps({"response": "审查结论: 代码无问题"}, ensure_ascii=False), + stderr="(node:123) Warning: Unauthorized token refresh attempt, retried OK")] + old = os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + try: + calls, saved_run, real_sleep = self._patch(mod, procs) + try: + result = self._review(mod) + finally: + self._restore(mod, saved_run, real_sleep) + self.assertNotIn("isError", result, + "stderr 警告词 + stdout 合法结论不应误判为错误") + self.assertEqual(calls["n"], 1, "不应触发重试") + self.assertIn("审查结论", result["content"][0]["text"]) + finally: + if old is None: + os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + else: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + + def test_rt7b_nonjson_stdout_with_stderr_keyword_is_error(self): + """RT7b: stdout 非空但不是合法 JSON 结果 (错误回显文本) + stderr 命中 → 判错误 + + 复审 P1-A 反例: 收窄不能漏掉"stdout 有内容但实为错误回显"。 + """ + mod = self.mod + procs = [_FakeCompletedProcess( + returncode=0, + stdout="Error: provider returned an error page", # 非 JSON 的错误文本 + stderr="APICallError: Unauthorized")] + old = os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + try: + calls, saved_run, real_sleep = self._patch(mod, procs) + try: + result = self._review(mod) + finally: + self._restore(mod, saved_run, real_sleep) + self.assertTrue(result.get("isError"), + "stdout 非 JSON 结果 + stderr 命中应判错误") + finally: + if old is None: + os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + else: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + + def test_rt7c_leading_warning_line_tolerated(self): + """RT7c: stdout 先打 AI SDK 警告行再输出 JSON → 仍能提取 response (实测形态)""" + mod = self.mod + payload = ("AI SDK Warning System: To turn off warning logging...\n" + + json.dumps({"response": "正文结论"}, ensure_ascii=False)) + procs = [_FakeCompletedProcess(returncode=0, stdout=payload, stderr="")] + old = os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + try: + calls, saved_run, real_sleep = self._patch(mod, procs) + try: + result = self._review(mod) + finally: + self._restore(mod, saved_run, real_sleep) + self.assertNotIn("isError", result) + self.assertEqual(result["content"][0]["text"], "正文结论") + finally: + if old is None: + os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + else: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + + # ---------- RT8: stdout 体积上限截断 (整体 review P2-1) ---------- + def test_rt8_stdout_size_cap(self): + """RT8: 成功但 stdout 超 ZCODE_BRIDGE_MAX_OUTPUT → 截断 + 标注""" + mod = self.mod + procs = [_FakeCompletedProcess(0, "y" * 20000, "")] + old = os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + old_mo = os.environ.pop("ZCODE_BRIDGE_MAX_OUTPUT", None) + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + os.environ["ZCODE_BRIDGE_MAX_OUTPUT"] = "10000" + try: + calls, saved_run, real_sleep = self._patch(mod, procs) + try: + result = self._review(mod) + finally: + self._restore(mod, saved_run, real_sleep) + self.assertNotIn("isError", result) + text = result["content"][0]["text"] + self.assertIn("已截断", text) + self.assertLess(len(text), 11000, "截断后长度应在上限附近") + finally: + if old is None: + os.environ.pop("ZCODE_BRIDGE_REVIEW_LOCK", None) + else: + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = old + if old_mo is None: + os.environ.pop("ZCODE_BRIDGE_MAX_OUTPUT", None) + else: + os.environ["ZCODE_BRIDGE_MAX_OUTPUT"] = old_mo + + +class TestEmbeddedProviderError(unittest.TestCase): + """内嵌 _parse_provider_error 与 shared/provider_error.py 权威版对齐 + (整体 review P1-4: insufficient 拆成 credit/balance 两条, 漏 quota 形态)""" + + @classmethod + def setUpClass(cls): + cls.mod = _load_mcp_module() + + def test_ep0_insufficient_quota_recognized(self): + """EP0: 'insufficientQuota' (无词边界, \bquota\b 吃不到) → quota""" + r = self.mod._parse_provider_error("Your plan has insufficientQuota") + self.assertTrue(r["is_quota"], "insufficient.*quota 形态应判为配额错误") + self.assertEqual(r["error_kind"], "quota") + + def test_ep1_insufficient_credit_balance(self): + """EP1: insufficient credit / balance → quota (对齐后不回归)""" + for text in ("insufficient credit", "insufficient balance"): + r = self.mod._parse_provider_error(text) + self.assertTrue(r["is_quota"], f"{text!r} 应判为配额错误") + + def test_ep2_rate_limit_exceeded_not_quota(self): + """EP2: 'rate limit exceeded' 仍判限流, 不被 quota 误吃 (对齐后语义不变)""" + r = self.mod._parse_provider_error("Rate limit exceeded, retry-after: 30") + self.assertTrue(r["is_rate_limit"]) + self.assertFalse(r["is_quota"]) if __name__ == "__main__": diff --git a/tests/test_polling_failure.py b/tests/test_polling_failure.py index 221887d..63d815e 100644 --- a/tests/test_polling_failure.py +++ b/tests/test_polling_failure.py @@ -24,6 +24,7 @@ 依赖: 仅 Python 标准库 + acp-bridge 模块 """ +import json import os import threading import types @@ -217,6 +218,70 @@ def test_pf3_subscribe_failure_falls_back_to_polling(self): self.assertIn("deliveryKind", sub_params, "subscribe 必传 deliveryKind (规格书 §4), 即使本次降级") + def test_pf4_dead_backend_poll_converges_early(self): + """PF4 (整体 review P1-3): 连续 20 次 poll 无响应且子进程已死 → 立即 -32603 + + 此前 consecutive_none 只 log 不收敛, 子进程已死这种确定性故障会让调用方 + 空等耗满 120s 预算; 修复后超阈值探测 proc.poll(), 已死立即报错。 + """ + bridge = self._new_bridge({ + # session/read 永远 error → projection 恒 None + "session/read": {"error": {"message": "connection reset"}}, + }) + # 子进程已退出 (poll() 返回退出码, 非 None) + bridge.backend.proc = types.SimpleNamespace(poll=lambda: 1) + resp = self._run_polling(bridge) + self.assertIn("error", resp) + self.assertEqual(resp["error"]["code"], -32603) + self.assertIn("已退出", resp["error"]["message"]) + reads = bridge.backend.methods_called().count("session/read") + self.assertEqual(reads, 20, + "应在第 20 次连续失败时探测到进程已死并提前收敛, 不等 120s") + + def test_pf5_live_backend_no_early_converge(self): + """PF5 (P1-3 边界): 连续 poll 失败但子进程活着 → 不提前收敛, 走原语义""" + bridge = self._new_bridge({ + "session/read": {"error": {"message": "timeout"}}, + }) + # 子进程活着 (poll() 返回 None): hang 而非死, 保持等满超时的原行为 + bridge.backend.proc = types.SimpleNamespace(poll=lambda: None) + resp = self._run_polling(bridge) + self.assertIn("error", resp) + self.assertIn("未启动", resp["error"]["message"], + "子进程活着时不提前收敛, 耗满 120s 后走「turn 未启动」原语义") + + def test_pf6_fetch_last_reply_drains_inbox_cancel(self): + """PF6 (整体 review P1-6): _fetch_last_reply 重试间隙 drain inbox + + 兜底取回复最多 4 次重试 (间隔 sleep), 此前这期间不 drain, 到达的 + session/cancel 会卡住; 修复后重试间隙 drain, cancel 可生效。 + """ + bridge = self._new_bridge({ + # session/messages 始终 error → 走满 4 次重试 + "session/messages": {"error": {"message": "boom"}}, + }) + zcode_sid = "sess_pf6" + bridge.session_map[zcode_sid] = zcode_sid + turn = {"zcode_sid": zcode_sid, "cancelled": False, "perms_responses": {}} + bridge.pending_turns[99] = turn + bridge._inbox.put(json.dumps({"method": "session/cancel", + "params": {"sessionId": zcode_sid}})) + out = bridge._fetch_last_reply(zcode_sid) + self.assertEqual(out, "", "始终 error → 重试耗尽返回空串") + self.assertTrue(turn["cancelled"], "重试间隙应 drain inbox 让 cancel 生效") + + def test_pf7_fetch_last_reply_skips_nondict_message(self): + """PF7 (整体 review P2-9): messages 混入非 dict 元素 → 跳过不炸 (isinstance 守卫)""" + bridge = self._new_bridge({ + "session/messages": {"result": {"messages": [ + "garbage", 42, + {"info": {"role": "assistant"}, + "parts": [{"type": "text", "text": "最终回复"}]}, + ]}}, + }) + out = bridge._fetch_last_reply("sess_pf7") + self.assertEqual(out, "最终回复", "非 dict 元素跳过, 照常取到最后回复") + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/tests/test_security_review.py b/tests/test_security_review.py index 125f161..01317b4 100644 --- a/tests/test_security_review.py +++ b/tests/test_security_review.py @@ -72,6 +72,8 @@ class _EnvGuard(unittest.TestCase): "ZCODE_BRIDGE_REVIEW_TIMEOUT", "ZCODE_BRIDGE_MIMOSA_SCAN_ROOT", "ZCODE_BRIDGE_MIMOSA_DEEP_TIMEOUT", "ZCODE_BRIDGE_MIMOSA_POLL_INTERVAL", + "ZCODE_BRIDGE_MIMOSA_RECV_TIMEOUT", + "ZCODE_BRIDGE_CODE_MAX", "ZCODE_BRIDGE_MAX_OUTPUT", "ZCODE_BRIDGE_PR_DIFF_MAX") def setUp(self): @@ -149,6 +151,63 @@ def test_rc4_timeout_env_configurable(self): os.environ["ZCODE_BRIDGE_REVIEW_TIMEOUT"] = "1" # clamp 下限 30 self.assertEqual(mod._review_timeout(), 30) + def test_rc5_files_capped_at_200(self): + """RC5: files 超 200 → 截断 (整体 review P1-2: 防 argv 撞 ARG_MAX)""" + cmd, result = self._capture_cmd( + files=[f"f{i}.py" for i in range(250)]) + self.assertNotIn("isError", result) + attaches = [cmd[i + 1] for i, v in enumerate(cmd) + if v == "--attach" and i + 1 < len(cmd)] + # 200 个 files + 1 个 code 临时文件 (_capture_cmd 默认带 code) + self.assertEqual(len(attaches), 201) + self.assertIn("f199.py", attaches) + self.assertNotIn("f200.py", attaches, "第 201 个起应被截断") + + def test_rc6_files_must_be_list(self): + """RC6: files 传字符串 → 明确拒绝 (与 focus_files 对齐; + 否则 list("app.py") 静默炸成单字符列表)""" + mod = self.mod + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + called = {"n": 0} + saved = mod.subprocess.run + + def fake_run(*a, **kw): + called["n"] += 1 + return _FakeCompletedProcess(0, "OK", "") + + mod.subprocess.run = fake_run + try: + result = mod.tool_zcode_review({"files": "app.py"}) + finally: + mod.subprocess.run = saved + self.assertTrue(result.get("isError")) + self.assertIn("files", result["content"][0]["text"]) + self.assertEqual(called["n"], 0, "非法 files 不应到达 zcode") + + def test_rc7_code_size_cap_truncated(self): + """RC7: code 超 ZCODE_BRIDGE_CODE_MAX → 截断 + 标注 (整体 review P1-3)""" + mod = self.mod + os.environ["ZCODE_BRIDGE_CODE_MAX"] = "10000" + os.environ["ZCODE_BRIDGE_REVIEW_LOCK"] = "0" + captured = {} + + def fake_run(cmd, *a, **kw): + # 临时文件在 finally 才清理, fake_run 内还能读到 + p = cmd[cmd.index("--attach") + 1] + with open(p) as f: + captured["content"] = f.read() + return _FakeCompletedProcess(0, "OK", "") + + saved = mod.subprocess.run + mod.subprocess.run = fake_run + try: + result = mod.tool_zcode_review({"code": "x" * 20000}) + finally: + mod.subprocess.run = saved + self.assertNotIn("isError", result) + self.assertIn("已截断", captured["content"]) + self.assertLess(len(captured["content"]), 11000, "截断后应在上限附近") + class TestExtractResponse(unittest.TestCase): """--json 输出的 response 提取""" @@ -248,6 +307,86 @@ def test_mc1_close_idempotent(self): client.close() client.close() + def test_mc2_send_write_timeout(self): + """MC2: stdin 不可写超 select 超时 → TimeoutError (整体 review P1-1: + server 卡住不读 stdin 时裸 write 会永久阻塞)""" + mod = self.mod + + class _FakeStdin: + def fileno(self): + return 1 + + def write(self, s): + raise AssertionError("不可写时不应真的 write") + + def flush(self): + pass + + class _FakeProc: + stdin = _FakeStdin() + + client = object.__new__(mod.MimosaMcpClient) + client._proc = _FakeProc() + client.timeout = 5 + saved = mod.select.select + mod.select.select = lambda r, w, x, t: ([], [], []) # 永不可写 + try: + with self.assertRaises(TimeoutError): + client._send({"jsonrpc": "2.0", "id": 1}) + finally: + mod.select.select = saved + + def test_mc3_send_writes_line_when_writable(self): + """MC3: 可写时正常写入单行 JSON (写超时不应影响正常路径)""" + mod = self.mod + buf = [] + + class _FakeStdin: + def fileno(self): + return 1 + + def write(self, s): + buf.append(s) + + def flush(self): + pass + + class _FakeProc: + stdin = _FakeStdin() + + client = object.__new__(mod.MimosaMcpClient) + client._proc = _FakeProc() + client.timeout = 5 + saved = mod.select.select + mod.select.select = lambda r, w, x, t: ([], [1], []) # 立即可写 + try: + client._send({"method": "ping", "备注": "中文"}) + finally: + mod.select.select = saved + self.assertEqual(len(buf), 1) + self.assertTrue(buf[0].endswith("\n")) + self.assertEqual(json.loads(buf[0])["method"], "ping") + self.assertIn("备注", buf[0], "ensure_ascii=False 应保留中文") + + def test_mc4_node_missing_clear_error(self): + """MC4: 系统无 node → __init__ 抛带安装提示的 FileNotFoundError + (整体 review P2-6: 否则裸 FileNotFoundError 被包成'扫描失败', 排查困难)""" + import tempfile + mod = self.mod + d = tempfile.mkdtemp(prefix="mimosa-root-") + self.addCleanup(lambda: __import__("shutil").rmtree(d, ignore_errors=True)) + os.makedirs(os.path.join(d, "payload", "dist", "mcp")) + open(os.path.join(d, "payload", "dist", "mcp", "server.js"), "w").close() + saved = mod.shutil.which + mod.shutil.which = lambda name: None + try: + with self.assertRaises(FileNotFoundError) as ctx: + mod.MimosaMcpClient(d, cwd="/tmp") + finally: + mod.shutil.which = saved + self.assertIn("node", str(ctx.exception)) + self.assertIn("Node.js", str(ctx.exception)) + class _StubMimosaClient: """_mimosa_quick_scan 的 stub (不起真子进程)。""" @@ -368,6 +507,38 @@ class SymlinkClient(_StubMimosaClient): body, findings = self._run_scan(SymlinkClient) self.assertEqual(findings, [], "symlink 指向根外的 findings.json 不应被读") + def test_ms2d_open_uses_nofollow(self): + """MS2d: 回读用 os.open + O_NOFOLLOW (整体 review P2-1: 关掉 resolve 与 + open 之间的 TOCTOU 窗口 — 检查后目标被换成 symlink 时 open 直接失败)""" + 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": [{"severity": "high", "title": "t"}]}, f) + os.environ["ZCODE_BRIDGE_MIMOSA_SCAN_ROOT"] = scan_root + + mod = self.mod + seen = {} + real_open = mod.os.open + + def spy_open(path, flags, *a, **kw): + seen["flags"] = flags + return real_open(path, flags, *a, **kw) + + class ScanDirClient(_StubMimosaClient): + response_text = f"**Mimosa**\n- scanDir: `{scan_dir}`\n- findings: 1" + + mod.os.open = spy_open + try: + body, findings = self._run_scan(ScanDirClient) + finally: + mod.os.open = real_open + self.assertEqual(len(findings), 1, "真实文件仍应正常回读") + self.assertTrue(seen.get("flags", 0) & os.O_NOFOLLOW, + "回读 open 应带 O_NOFOLLOW") + def test_ms3_compact_projection(self): """MS3: _compact_findings 投影保留复核所需字段""" findings = [{ @@ -718,6 +889,26 @@ def close(self): mod.MimosaMcpClient = saved_client mod.time.sleep = saved_sleep + def test_ds4_recv_timeout_env(self): + """DS4: deep client 单次一问一答超时读 ZCODE_BRIDGE_MIMOSA_RECV_TIMEOUT + (整体 review P2-4: 原硬编码 120; env 超 600 被 clamp)""" + def run_with(env_val): + captured = {} + stub, _ = _make_deep_stub(["completed"], + scan_dir=self._make_scan_dir()) + + class RecStub(stub): + def __init__(self, root, cwd, timeout=120): + captured["timeout"] = timeout + super().__init__(root, cwd, timeout=timeout) + + os.environ["ZCODE_BRIDGE_MIMOSA_RECV_TIMEOUT"] = env_val + self._run_deep(RecStub) + return captured["timeout"] + + self.assertEqual(run_with("42"), 42, "env=42 应透传") + self.assertEqual(run_with("9999"), 600, "env 超 600 应被 clamp") + class TestSecurityReviewDepth(_EnvGuard): """tool_zcode_security_review 的 depth/focus_files 参数""" @@ -1029,5 +1220,82 @@ def test_pr10_attachment_tmpfile_cleaned(self): self.assertFalse(os.path.exists(attach_path), "PR 附件临时文件应被清理") +class TestGitTimeouts(_EnvGuard): + """_git 超时分档 (整体 review P2-6): 元数据类 15s, diff 类 60s""" + + @classmethod + def setUpClass(cls): + cls.mod = _load_mcp_module() + + def test_gt0_metadata_15s_diff_60s(self): + """GT0: rev-parse 用默认 15s; _pr_diff 的两次 diff 都用 60s""" + mod = self.mod + seen = [] + + def fake_run(cmd, *a, **kw): + sub = cmd[3] # ["git", "-C", repo, , ...] + seen.append((sub, kw.get("timeout"))) + if sub == "diff" and "--name-only" in cmd: + return _FakeCompletedProcess(0, "a.py\n", "") + return _FakeCompletedProcess(0, "ok", "") + + saved = mod.subprocess.run + mod.subprocess.run = fake_run + try: + mod._git("/r", "rev-parse", "--verify", "main") + mod._pr_diff("/r", "main", "HEAD") + finally: + mod.subprocess.run = saved + self.assertEqual(seen[0], ("rev-parse", 15), "元数据类默认 15s") + diff_timeouts = [t for sub, t in seen if sub == "diff"] + self.assertEqual(diff_timeouts, [60, 60], "diff 类应 60s") + + +class TestEmbeddedCreds(_EnvGuard): + """内嵌凭证副本的 host 构造与 shared/credentials.py._safe_host 对齐 + (整体 review P2-3: 有 netloc 缺 scheme 时补 https://, 旧副本返回 None)""" + + ENV_KEYS = _EnvGuard.ENV_KEYS + ("ZCODE_BASE_URL", "HOME") + + @classmethod + def setUpClass(cls): + cls.mod = _load_mcp_module() + + def test_sh0_safe_host_parity(self): + """SH0: _safe_host 三种形态与权威版一致""" + sh = self.mod._safe_host + self.assertEqual(sh("https://a.example.com/api"), "https://a.example.com") + self.assertEqual(sh("//b.example.com/api"), "https://b.example.com", + "缺 scheme 有 netloc 应补 https://") + self.assertIsNone(sh("not-a-url/no-host")) + self.assertIsNone(sh("")) + + def test_sh1_stale_env_base_url_healed(self): + """SH1: env 残留 baseURL (protocol-relative) → 自愈用 config 值; + 修复前该形态 host 解析为 None, 残留检测静默跳过""" + import tempfile + mod = self.mod + home = tempfile.mkdtemp(prefix="zcode-home-") + self.addCleanup(lambda: __import__("shutil").rmtree(home, ignore_errors=True)) + cfg_dir = os.path.join(home, ".zcode", "v2") + os.makedirs(cfg_dir) + cfg = {"provider": { + "p-enabled": {"enabled": True, + "options": {"baseURL": "https://enabled.example.com/api", + "apiKey": "k"}, + "models": {"GLM-5.2": {}}}, + "p-old": {"enabled": False, + "options": {"baseURL": "https://stale.example.com/api"}, + "models": {}}, + }} + with open(os.path.join(cfg_dir, "config.json"), "w") as f: + json.dump(cfg, f) + os.environ["HOME"] = home # Path.home() 走 HOME env + os.environ["ZCODE_BASE_URL"] = "//stale.example.com/api" # 残留, 无 scheme + merged = mod._merge_env_with_creds(mod.load_zcode_credentials()) + self.assertEqual(merged["ZCODE_BASE_URL"], "https://enabled.example.com/api", + "残留 env baseURL 应被自愈为 enabled provider 的 config 值") + + if __name__ == "__main__": unittest.main(verbosity=2)