fix(cr-agent): PR#201审查修复(沙箱逃逸/超时decode/落库孤儿+8W+2S)

3 Critical:
- 沙箱路径逃逸: base.py 新增 _validate_script_path(realpath+commonpath),
  local/cube 执行前校验, 逃逸返回 PathEscapeError
- local.py 超时分支 text=True 时 e.stdout 为 str 仍调 .decode 致
  AttributeError 崩溃 → 按类型分支处理
- pipeline start_task 失败降级后仍无条件 store.save 致外键孤儿
  → store=None + save 前判空

8 Warning + 2 Suggestion:
- W1 ast 跨 hunk 污点丢失 → visitor 提到文件级, 跨 hunk 复用 tainted
- W2 LLM response.choices 未判空 → IndexError 防护
- W3 重试 "5" in msg 子串过宽 → _is_retryable_error(status_code/异常/5xx正则)
- W4 脱敏幂等短路致"部分脱敏后跳过剩余明文" → 移除短路 + 末尾回扫自检
- W5 allowed_executables 未生效 → allow 前首词白名单校验(fail-closed)
- W6 test_force_secret_output 断言与脱敏矛盾 → 真实格式密钥 + 脱敏断言
- W7 SANDBOX_MEMORY_MB 未接线 → host_config.Memory(字节)
- W8 INSERT OR IGNORE 静默吞冲突 → cursor.rowcount 统计忽略条数
- S1 sandbox_duration 恒等于 total → 累加各 run.duration_ms
- S2 dry_run split(":") 切错 Windows 盘符 → rpartition

验证: flake8 全绿(未加 ignore) + 190 测试全过无回归

Author: wzxsdf

examples/skills_code_review_agent/agent/ast_analyzer.py

@@ -264,6 +264,10 @@ def analyze(files: list[DiffFile]) -> list[Finding]:
         if not diff_file.path.endswith('.py'):
             continue
 
+        # W1 修复: visitor 提到文件级，跨 hunk 复用 tainted 集合，捕获跨 hunk 污点传播
+        # （局限：diff 仅含 added 行，函数体可能截断；跨 hunk 污点累积可部分缓解漏报）
+        visitor = _Visitor(diff_file.path)
+
         # 处理每个 hunk
         for hunk in diff_file.hunks:
             if not hunk.added:
@@ -281,8 +285,7 @@ def analyze(files: list[DiffFile]) -> list[Finding]:
                 findings.extend(_analyze_with_fallback(diff_file.path, hunk))
                 continue
 
-            # 创建 visitor 并分析
-            visitor = _Visitor(diff_file.path)
+            # 复用文件级 visitor（tainted 跨 hunk 累积），分析当前 hunk
             visitor.visit(tree)
 
             # 转换 findings
@@ -305,6 +308,9 @@ def analyze(files: list[DiffFile]) -> list[Finding]:
                                   bucket=Bucket.FINDINGS)
                 findings.append(finding)
 
+            # 清空当前 hunk 的 findings，保留 tainted（跨 hunk 污点累积）
+            visitor.findings = []
+
     return findings
 
 

examples/skills_code_review_agent/agent/llm_layer.py

@@ -14,6 +14,7 @@
 
 import json
 import os
+import re
 import time
 
 # 复用现有模型
@@ -74,6 +75,31 @@ def _get_llm_config() -> dict:
     return {"api_key": api_key, "base_url": base_url, "model_name": model_name}
 
 
+def _is_retryable_error(e: Exception) -> bool:
+    """判断异常是否值得重试（5xx/超时/上游错误）。
+
+    W3 修复：精确判定，避免 "5" in msg 误匹配 "500 tokens"/"line 5"。
+    优先级：status_code 属性(5xx) → TimeoutError → 异常名含 timeout →
+    msg 含 upstream → msg 精确匹配 5xx 状态码。
+    """
+    status = getattr(e, "status_code", None)
+    if status is not None:
+        try:
+            return 500 <= int(status) < 600
+        except (TypeError, ValueError):
+            pass
+    if isinstance(e, TimeoutError):
+        return True
+    name = type(e).__name__.lower()
+    if "timeout" in name:
+        return True
+    msg = str(e).lower()
+    if "upstream" in msg:
+        return True
+    # 精确匹配 5xx（前后非数字），避免 "500 tokens"/"line 5" 误命中
+    return bool(re.search(r'(?:^|\D)5\d{2}(?:\D|$)', msg))
+
+
 def _findings_to_json(findings: list[Finding]) -> str:
     """将 findings 转换为 JSON 格式供 LLM 分析"""
     findings_data = []
@@ -160,7 +186,10 @@ def _call_llm_with_openai_client(findings: list[Finding]) -> list[dict]:
                     temperature=0.1,
                     max_tokens=2000)  # 降低到2000减少400错误
 
-                response_text = response.choices[0].message.content
+                # W2 修复: 空响应防护，避免 IndexError 导致全部裁决丢失
+                if not response.choices:
+                    raise Exception("LLM 返回空 choices（限流/内容过滤）")
+                response_text = response.choices[0].message.content or ""
                 verdicts = _parse_llm_response(response_text)
                 all_verdicts.extend(verdicts)
                 break  # 成功，跳出重试循环
@@ -171,8 +200,8 @@ def _call_llm_with_openai_client(findings: list[Finding]) -> list[dict]:
                 if "400" in error_msg or "bad request" in error_msg:
                     print("[LLM Layer] 400错误（prompt太大），不重试")
                     raise Exception(f"OpenAI API 400错误: {str(e)}") from e
-                # 超时/5xx/Upstream错误指数退避重试
-                elif attempt < max_retries and ("timeout" in error_msg or "5" in error_msg or "upstream" in error_msg):
+                # 超时/5xx/Upstream错误指数退避重试（W3 修复: 精确判定，避免 "5"/"timeout" 子串误匹配）
+                elif attempt < max_retries and _is_retryable_error(e):
                     wait_time = 2**attempt  # 指数退避：1s, 2s, 4s
                     print(f"[LLM Layer] 调用失败，{wait_time}秒后重试 {attempt + 1}/{max_retries}: {str(e)}")
                     time.sleep(wait_time)
@@ -357,7 +386,10 @@ def _call_llm_with_openai_client_supplementary(existing_findings: list[Finding],
                 temperature=0.1,
                 max_tokens=1500)  # 进一步降低到1500减少400错误
 
-            response_text = response.choices[0].message.content
+            # W2 修复: 空响应防护
+            if not response.choices:
+                raise Exception("LLM 补召回返回空 choices（限流/内容过滤）")
+            response_text = response.choices[0].message.content or ""
             new_findings = _parse_supplementary_findings(response_text)
             return new_findings
 
@@ -367,8 +399,8 @@ def _call_llm_with_openai_client_supplementary(existing_findings: list[Finding],
             if "400" in error_msg or "bad request" in error_msg:
                 print("[LLM Layer] 补召回400错误（prompt太大），不重试")
                 raise Exception(f"OpenAI 补召回400错误: {str(e)}") from e
-            # 超时/5xx/Upstream错误指数退避重试
-            elif attempt < max_retries and ("timeout" in error_msg or "5" in error_msg or "upstream" in error_msg):
+            # 超时/5xx/Upstream错误指数退避重试（W3 修复: 精确判定）
+            elif attempt < max_retries and _is_retryable_error(e):
                 wait_time = 2**attempt  # 指数退避：1s, 2s, 4s
                 print(f"[LLM Layer] 补召回失败，{wait_time}秒后重试 {attempt + 1}/{max_retries}: {str(e)}")
                 time.sleep(wait_time)
@@ -495,20 +527,20 @@ def enhance(
         verdict_list = []
         for key, verdict_data in recorded_verdicts.items():
             # 解析 key: "rule_id:file:line"
-            parts = key.split(":")
-            if len(parts) == 3:
-                rule_id, file_path, line_str = parts
-                try:
-                    line = int(line_str)
-                    verdict_list.append({
-                        "rule_id": rule_id,
-                        "file": file_path,
-                        "line": line,
-                        "verdict": verdict_data["verdict"],
-                        "reason": verdict_data["reason"],
-                    })
-                except ValueError:
-                    continue
+            # S2 修复: rpartition/partition 切分，避免文件路径含 ":"（Windows 盘符）时切错
+            key_path, _, line_str = key.rpartition(":")
+            rule_id, _, file_path = key_path.partition(":")
+            try:
+                line = int(line_str)
+                verdict_list.append({
+                    "rule_id": rule_id,
+                    "file": file_path,
+                    "line": line,
+                    "verdict": verdict_data["verdict"],
+                    "reason": verdict_data["reason"],
+                })
+            except ValueError:
+                continue
 
         # 应用降噪裁决
         enhanced_findings = _apply_verdicts(findings, verdict_list)

examples/skills_code_review_agent/agent/pipeline.py

@@ -102,14 +102,17 @@ def run_review(diff_text: str,
     t0 = time.time()
 
     # 0. 修复隐患4: 先 start_task 创建 task_id（确保 save 时 review_tasks 记录存在）
+    # Critical 3 修复: start_task 失败时置 store=None，避免 save 对孤儿 task_id 落库
+    store = None
     try:
         store = ReviewStore()
         scope = f"diff-{len(diff_text)}chars" if diff_text else "empty"
         task_id = store.start_task(repo=repo, scope=scope)
     except Exception as e:
-        # start_task 失败时使用生成的 UUID（降级处理）
+        # start_task 失败时使用生成的 UUID（降级处理），并标记不落库
         print(f"[Pipeline] start_task 失败: {str(e)}")
         task_id = str(uuid.uuid4())
+        store = None
 
     # 1. 解析 diff
     files = parse_diff(diff_text)
@@ -198,7 +201,8 @@ def run_review(diff_text: str,
                 runs.append(failed_run)
 
     # 6. 构建监控摘要
-    t_sandbox = int((time.time() - t0) * 1000)
+    # S1 修复: sandbox_duration 累加各沙箱 run 实际耗时（而非等于 total_duration）
+    t_sandbox = sum(run.duration_ms for run in runs)
 
     # 合并所有 findings 用于监控
     all_findings = list(findings) + list(warnings) + list(needs_review)
@@ -229,11 +233,15 @@ def run_review(diff_text: str,
                           input_summary=_summary(diff_text))
 
     # 9. 落库（内含脱敏，使用已存在的 task_id）
-    try:
-        store.save(report)
-    except Exception as e:
-        # 落库失败不应中断报告生成
-        print(f"[Pipeline] 落库失败: {str(e)}")
+    # Critical 3 修复: store=None（start_task 失败）时跳过落库，避免外键孤儿
+    if store is not None:
+        try:
+            store.save(report)
+        except Exception as e:
+            # 落库失败不应中断报告生成
+            print(f"[Pipeline] 落库失败: {str(e)}")
+    else:
+        print(f"[Pipeline] 跳过落库（task 未注册，task_id={task_id}）")
 
     # 10. 写报告文件
     try:

examples/skills_code_review_agent/agent/redaction.py

@@ -123,8 +123,7 @@ def redact_text(text: str) -> tuple[str, int]:
     Returns:
         (脱敏后文本, 命中的密钥数量)
     """
-    # 重复脱敏保护：避免对已脱敏内容重复处理
-    if "[REDACTED_" in text:
+    if not text:
         return text, 0
 
     count = 0
@@ -150,6 +149,12 @@ def redact_text(text: str) -> tuple[str, int]:
                 text = text.replace(literal, "[REDACTED_ENTROPY]")
                 count += 1
 
+    # W4 修复: 回扫自检 —— 末尾对所有正则再扫一遍，捕获任何漏网明文残留
+    # （幂等：占位符 [REDACTED_*] 不会被同一正则重复匹配，故不会二次替换）
+    for pattern, replacement in SECRET_PATTERNS:
+        text, n = pattern.subn(replacement, text)
+        count += n
+
     return text, count
 
 

examples/skills_code_review_agent/filters/policy.py

@@ -112,5 +112,17 @@ def evaluate(self, command: str, ctx: dict) -> FilterDecision:
                 command_redacted=command[:80]
             )
 
-        # 5. 默认允许
+        # 5. 可执行文件白名单校验（命令首词必须在白名单内，fail-closed）
+        # W5 修复: allowed_executables 此前完全未生效，任何非高危命令均 allow
+        allowed = self.p.get("allowed_executables", [])
+        first_word = command.strip().split()[0] if command.strip() else ""
+        if allowed and first_word not in allowed:
+            return FilterDecision(
+                stage="pre_sandbox",
+                decision="deny",
+                reason=f"非白名单可执行文件 {first_word}",
+                command_redacted=command[:80]
+            )
+
+        # 6. 默认允许
         return FilterDecision(stage="pre_sandbox", decision="allow", reason="", command_redacted="")

examples/skills_code_review_agent/sandbox/base.py

@@ -8,6 +8,7 @@
 定义 SandboxProvider 抽象基类，所有沙箱实现都需要继承此类。
 """
 
+import os
 from abc import ABC, abstractmethod
 
 from agent.models import SandboxRun
@@ -93,3 +94,25 @@ def _redact_secrets(self, output: str) -> str:
         # 简单替换 sk- 开头的内容为 sk-REDACTED
         pattern = r'sk-[a-zA-Z0-9]{20,}'
         return re.sub(pattern, 'sk-REDACTED', output)
+
+    @staticmethod
+    def _validate_script_path(workspace: str, script: str) -> bool:
+        """校验脚本路径解析后仍在 workspace 内（防路径穿越逃逸）。
+
+        Trust-boundary 输入校验：即使上层 pipeline 已用约定脚本名，Local/Cube 后端
+        也独立校验 script 不含 "../" 等逃逸（纵深防御，绕过 Filter 时的最后闸门）。
+
+        Args:
+            workspace: 工作目录路径
+            script: 脚本文件名（相对 workspace）
+
+        Returns:
+            True 若脚本路径在 workspace 内；False（含跨盘符 ValueError）则拒绝
+        """
+        try:
+            ws_real = os.path.realpath(workspace)
+            script_real = os.path.realpath(os.path.join(workspace, script))
+            return os.path.commonpath([ws_real, script_real]) == ws_real
+        except ValueError:
+            # 不同盘符/无法求公共路径 → 视为逃逸，拒绝
+            return False

examples/skills_code_review_agent/sandbox/container.py

@@ -137,12 +137,17 @@ def run(
         try:
             # 资源限制（单向收紧）
             timeout_limit = _bounded_int("SANDBOX_TIMEOUT_SEC", timeout, 300)  # 最多 5 分钟
-            _bounded_int("SANDBOX_MEMORY_MB", 512, 1024)  # 默认 512MB，最多 1GB（预留给未来使用）
+            # W7 修复: memory 限制真正接线到 host_config（字节），避免环境变量形同虚设
+            memory_mb = _bounded_int("SANDBOX_MEMORY_MB", 512, 1024)  # 默认 512MB，最多 1GB
 
             # 创建容器客户端
             config = ContainerConfig(
                 image="skills-code-review-agent:latest",
-                host_config={"Binds": [f"{workspace}:/workspace:ro"]},  # 只读挂载
+                # 只读挂载 + 内存限制（Memory 单位为字节）
+                host_config={
+                    "Binds": [f"{workspace}:/workspace:ro"],
+                    "Memory": memory_mb * 1024 * 1024,
+                },
             )
 
             client = ContainerClient(config)

examples/skills_code_review_agent/sandbox/cube.py

@@ -71,6 +71,20 @@ def run(
             import subprocess
             script_path = os.path.join(workspace, script)
 
+            # Critical 1 加固：校验脚本路径未逃逸出 workspace（防路径穿越）
+            if not self._validate_script_path(workspace, script):
+                return SandboxRun(
+                    runtime="cube",
+                    script=script,
+                    status="failed",
+                    exit_code=1,
+                    stdout_redacted="",
+                    stderr_redacted=f"脚本路径逃逸出 workspace，拒绝执行: {script}",
+                    truncated=False,
+                    error_type="PathEscapeError",
+                    duration_ms=0,
+                )
+
             # 使用 subprocess 执行（简化实现）
             result = subprocess.run(
                 ["python", script_path],

examples/skills_code_review_agent/sandbox/fake.py

@@ -83,8 +83,8 @@ def run(
 
         # Trigger 3: force_secret_output → 模拟密钥泄露（Critical 1 加固：返回前脱敏）
         if "force_secret_output" in blob:
-            # 模拟明文密钥输出，但在返回前脱敏（纵深防御）
-            raw_stdout = "out sk-leaked-secret"
+            # 模拟明文密钥输出（真实格式 sk- + 20字母数字，确保被正则脱敏），返回前脱敏（纵深防御）
+            raw_stdout = "out sk-leakedsecret0123456789abcdef"
             stdout_redacted, _ = redact_text(raw_stdout)  # 应输出 "out [REDACTED_SK]"
             return SandboxRun(
                 runtime="fake",

examples/skills_code_review_agent/sandbox/local.py

@@ -47,6 +47,20 @@ def run(
 
         script_path = os.path.join(workspace, script)
 
+        # Critical 1 加固：校验脚本路径未逃逸出 workspace（防路径穿越）
+        if not self._validate_script_path(workspace, script):
+            return SandboxRun(
+                runtime="local",
+                script=script,
+                status="failed",
+                exit_code=1,
+                stdout_redacted="",
+                stderr_redacted=f"脚本路径逃逸出 workspace，拒绝执行: {script}",
+                truncated=False,
+                error_type="PathEscapeError",
+                duration_ms=0,
+            )
+
         try:
             # 执行脚本
             result = subprocess.run(
@@ -80,9 +94,19 @@ def run(
             # 超时处理
             duration_ms = int((time.time() - start_time) * 1000)
 
-            # 尝试获取部分输出
-            stdout = e.stdout.decode('utf-8', errors='ignore') if e.stdout else ""
-            stderr = e.stderr.decode('utf-8', errors='ignore') if e.stderr else ""
+            # 尝试获取部分输出（text=True 时 e.stdout/e.stderr 已是 str，不可再 .decode，Critical 2）
+            if isinstance(e.stdout, str):
+                stdout = e.stdout
+            elif e.stdout:
+                stdout = e.stdout.decode("utf-8", "ignore")
+            else:
+                stdout = ""
+            if isinstance(e.stderr, str):
+                stderr = e.stderr
+            elif e.stderr:
+                stderr = e.stderr.decode("utf-8", "ignore")
+            else:
+                stderr = ""
             stdout_redacted, _ = self._sanitize_output(stdout)
             stderr_redacted, _ = self._sanitize_output(stderr)