Improve skills code review agent governance

Author: xyxhhhhh

examples/skills_code_review_agent/DESIGN.md

@@ -1,5 +1,5 @@
 # 方案设计
 
-本示例以 `code-review` Skill 封装 `SKILL.md`、规则、脚本和 Filter 配置。Agent 可读取 diff、patch、git 工作区、文件列表或 fixture，解析变更文件、hunk、上下文和行号。主审查流水线会审计 Skill 文件哈希并加载规则；同时提供 `--skill-smoke` 和测试用例，真实调用 SDK 的 `skill_load(skill_name="code-review")` 与 `skill_run` 执行 `scripts/diff_summary.py`，证明该 Skill 能被 tRPC-Agent 原生工具链加载和运行。无模型 Key 时使用确定性规则、AST/taint 辅助和 fake sandbox 完成 dry-run；生产接入 Container 或 Cube/E2B workspace，`local` 仅作开发 fallback。
+本示例以 `code-review` Skill 封装 `SKILL.md`、规则、脚本和 Filter 配置。Agent 可读取 diff、patch、git 工作区、文件列表或 fixture，解析变更文件、hunk、上下文和行号。主审查流水线会审计 Skill 文件哈希并加载规则；普通评审不在沙箱策略外执行 SDK `skill_run`，`--skill-smoke` 可单独调用 `skill_load(skill_name="code-review")` 与 `skill_run` 验证原生 Skill 链路。无模型 Key 时使用确定性规则、AST/taint 辅助和 fake sandbox 完成 dry-run；CLI 与 `run_review()` API 默认生产沙箱为 Container，也可接入 Cube/E2B workspace，`local` 仅作开发 fallback。
 
-规则覆盖安全、异步错误、资源泄漏、测试缺失、敏感信息、数据库事务和连接生命周期。沙箱请求先经过 Filter，拦截高风险命令、敏感路径、非白名单网络和超预算执行；`deny` 与 `needs_human_review` 写入报告和数据库，不能执行。默认 `python:3-slim` 验证容器隔离链路，`Dockerfile.scanners` 提供带 `bandit`、`ruff`、`detect-secrets` 的镜像，用于实际离线 scanner 执行。SQLite 保存 task、输入、sandbox run、Filter 决策、finding、监控和报告，并通过 `ReviewStore` 保留 SQL 后端扩展点。报告输出 JSON/Markdown，高置信进入 findings，中置信进入 warnings，低置信进入人工复核；同一文件、行号、类别只保留最高优先级项。所有 diff、stdout、stderr、Filter reason 和 finding evidence 在落库前脱敏，并记录耗时、工具调用、拦截、异常、严重级别分布和去重数量，便于后续回放和审计。
+规则覆盖安全、异步错误、资源泄漏、测试缺失、敏感信息、数据库事务和连接生命周期。沙箱请求先经过 Filter，拦截高风险命令、敏感路径、声明型网络 scanner 与测试命令 URL/domain 的非白名单网络访问、以及超预算执行；`deny` 与 `needs_human_review` 写入报告和数据库，不能执行。默认 `python:3-slim` 验证容器隔离链路，`Dockerfile.scanners` 提供带 `bandit`、`ruff`、`detect-secrets` 的镜像，用于实际离线 scanner 执行；scanner 物化 diff 文件时会拒绝路径逃逸。SQLite 保存 task、输入、sandbox run、Filter 决策、finding、监控和报告，并通过 `ReviewStore` 保留 SQL 后端扩展点。报告输出 JSON/Markdown，高置信进入 findings，中置信进入 warnings，低置信进入人工复核；同一文件、行号、类别只保留最高优先级项。所有 diff、stdout、stderr、Filter reason 和 finding evidence 在落库前脱敏，stdout/stderr 共享输出总预算，并记录耗时、工具调用、拦截、异常、严重级别分布和去重数量，便于后续回放和审计。

examples/skills_code_review_agent/README.md

@@ -6,6 +6,7 @@
 import re
 from dataclasses import dataclass
 from pathlib import Path
+from urllib.parse import urlparse
 
 from .models import DiffInput
 from .models import FilterDecision
@@ -37,6 +38,15 @@
     r":\(\)\s*\{",
 )
 
+NETWORK_COMMAND_PATTERNS = (
+    r"\b(pip|pip3|python(?:3)?\s+-m\s+pip)\s+install\b",
+    r"\b(npm|pnpm|yarn)\s+(install|add|ci)\b",
+    r"\b(go\s+get|cargo\s+install)\b",
+    r"\b(git)\s+(clone|fetch|pull|submodule\s+update)\b",
+    r"\b(ssh|scp|sftp|rsync)\b",
+    r"\b(apt-get|apt|apk|yum|dnf|brew)\s+(install|update|upgrade)\b",
+)
+
 
 @dataclass(slots=True)
 class SandboxRequest:
@@ -66,6 +76,7 @@ def __init__(
             sandbox_path_allowlist: tuple[str, ...] = ("scripts/", "work/"),
             sandbox_read_allowlist: tuple[str, ...] = ("scripts/", "work/", "repo/"),
             sandbox_write_allowlist: tuple[str, ...] = ("work/", ),
+            network_command_patterns: tuple[str, ...] = NETWORK_COMMAND_PATTERNS,
             schema_version: int = 1,
     ):
         self.network_policy = network_policy
@@ -77,6 +88,7 @@ def __init__(
         self.sandbox_path_allowlist = sandbox_path_allowlist
         self.sandbox_read_allowlist = sandbox_read_allowlist
         self.sandbox_write_allowlist = sandbox_write_allowlist
+        self.network_command_patterns = network_command_patterns
         self.schema_version = schema_version
         self.last_redaction_count = 0
 
@@ -102,6 +114,7 @@ def load(
             sandbox_path_allowlist=tuple(data.get("sandbox_path_allowlist", ("scripts/", "work/"))),
             sandbox_read_allowlist=tuple(data.get("sandbox_read_allowlist", ("scripts/", "work/", "repo/"))),
             sandbox_write_allowlist=tuple(data.get("sandbox_write_allowlist", ("work/", ))),
+            network_command_patterns=tuple(data.get("network_command_patterns", NETWORK_COMMAND_PATTERNS)),
             schema_version=int(data.get("schema_version", 1)),
         )
 
@@ -117,6 +130,7 @@ def audit(self) -> dict[str, object]:
             "sandbox_path_allowlist": list(self.sandbox_path_allowlist),
             "sandbox_read_allowlist": list(self.sandbox_read_allowlist),
             "sandbox_write_allowlist": list(self.sandbox_write_allowlist),
+            "network_command_patterns": list(self.network_command_patterns),
         }
 
     def evaluate(
@@ -241,6 +255,41 @@ def _evaluate_request(self, request: SandboxRequest) -> FilterDecision:
                 policy="high-risk-command",
                 severity="high",
             )
+        command_domains = _network_domains_in_command(command_to_check)
+        if command_domains:
+            if self.network_policy != "allowlist":
+                return self._decision(
+                    decision="needs_human_review",
+                    reason="Command references network domains while the active review policy denies network access: "
+                    f"{', '.join(command_domains)}.",
+                    command=command_to_check,
+                    path=request.script_path,
+                    policy="network-command",
+                    severity="high",
+                )
+            disallowed = sorted(set(command_domains) - set(self.allowed_network_domains))
+            if disallowed:
+                return self._decision(
+                    decision="needs_human_review",
+                    reason=f"Command references network domains outside the configured allowlist: "
+                    f"{', '.join(disallowed)}.",
+                    command=command_to_check,
+                    path=request.script_path,
+                    policy="network-command-allowlist",
+                    severity="high",
+                )
+        if not command_domains and _is_network_command(command_to_check, self.network_command_patterns):
+            return self._decision(
+                decision="needs_human_review",
+                reason=(
+                    "Command appears to require network access but does not declare reviewable allowlisted "
+                    "network domains."
+                ),
+                command=command_to_check,
+                path=request.script_path,
+                policy="network-command-implicit",
+                severity="high",
+            )
         return self._decision(
             decision="allow",
             reason=(
@@ -283,7 +332,20 @@ def _is_forbidden_path(path: str, markers: tuple[str, ...] = FORBIDDEN_PATH_MARK
 
 
 def _is_high_risk_command(command: str, patterns: tuple[str, ...] = HIGH_RISK_COMMAND_PATTERNS) -> bool:
-    return any(re.search(pattern, command) for pattern in patterns)
+    return any(re.search(pattern, command, re.IGNORECASE) for pattern in patterns)
+
+
+def _network_domains_in_command(command: str) -> tuple[str, ...]:
+    domains: set[str] = set()
+    for match in re.finditer(r"https?://[^\s'\"),]+", command, re.IGNORECASE):
+        parsed = urlparse(match.group(0))
+        if parsed.hostname:
+            domains.add(parsed.hostname.lower())
+    return tuple(sorted(domains))
+
+
+def _is_network_command(command: str, patterns: tuple[str, ...] = NETWORK_COMMAND_PATTERNS) -> bool:
+    return any(re.search(pattern, command, re.IGNORECASE) for pattern in patterns)
 
 
 def _path_is_allowed(path: str, allowlist: tuple[str, ...]) -> bool:

examples/skills_code_review_agent/agent/filter_policy.py

@@ -19,34 +19,61 @@
 async def code_review_tool(
     *,
     diff_file: str | None = None,
+    patch_file: str | None = None,
     repo_path: str | None = None,
+    file_list: str | None = None,
     fixture: str | None = None,
     output_dir: str | None = None,
     db_path: str | None = None,
     db_url: str | None = None,
-    dry_run: bool = True,
+    sandbox: str = "container",
+    dry_run: bool = False,
     container_image: str = "python:3-slim",
+    docker_path: str | None = None,
+    docker_base_url: str | None = None,
+    cube_template: str | None = None,
+    cube_api_url: str | None = None,
+    cube_api_key: str | None = None,
+    cube_sandbox_id: str | None = None,
+    timeout_sec: float = 5.0,
+    max_output_bytes: int = 12000,
+    filter_timeout_budget_sec: float = 30.0,
+    filter_max_output_bytes: int = 20000,
+    network_policy: str = "deny",
+    test_command: str | None = None,
+    custom_rule_script: str | None = None,
     include_network_scanners: bool = False,
+    max_diff_bytes: int = 2_000_000,
 ) -> dict[str, Any]:
     """Run the Skills code review pipeline as a FunctionTool-compatible callable."""
-    sandbox_runner = None
-    sandbox = "fake"
-    if not dry_run:
-        from .runtime_factory import create_container_sandbox_runner
-
-        sandbox = "container"
-        sandbox_runner = create_container_sandbox_runner(image=container_image)
+    effective_sandbox = "fake" if dry_run else sandbox
     report = await run_review(
         diff_file=Path(diff_file) if diff_file else None,
+        patch_file=Path(patch_file) if patch_file else None,
         repo_path=Path(repo_path) if repo_path else None,
+        file_list=Path(file_list) if file_list else None,
         fixture=fixture,
         output_dir=Path(output_dir) if output_dir else DEFAULT_OUTPUT_DIR,
         db_path=Path(db_path) if db_path else DEFAULT_DB_PATH,
         db_url=db_url,
         dry_run=dry_run,
-        sandbox=sandbox,
-        sandbox_runner=sandbox_runner,
+        sandbox=effective_sandbox,
+        container_image=container_image,
+        docker_path=docker_path,
+        docker_base_url=docker_base_url,
+        cube_template=cube_template,
+        cube_api_url=cube_api_url,
+        cube_api_key=cube_api_key,
+        cube_sandbox_id=cube_sandbox_id,
+        timeout_sec=timeout_sec,
+        max_output_bytes=max_output_bytes,
+        filter_timeout_budget_sec=filter_timeout_budget_sec,
+        filter_max_output_bytes=filter_max_output_bytes,
+        network_policy=network_policy,
+        test_command=test_command,
+        custom_rule_script=custom_rule_script,
         include_network_scanners=include_network_scanners,
+        max_diff_bytes=max_diff_bytes,
     )
     return {
         "task_id": report.task_id,
@@ -70,6 +97,7 @@ def create_code_review_skill_tool_set(*, workspace_runtime):
     repository = create_default_skill_repository(
         str(SKILL_DIR.parent),
         workspace_runtime=workspace_runtime,
+        enable_hot_reload=False,
         use_cached_repository=True,
     )
     return SkillToolSet(repository=repository, skill_stager=LinkSkillStager()), repository
@@ -88,10 +116,8 @@ def create_code_review_agent(*, model, skill_workspace_runtime=None):
 
     return LlmAgent(
         name="skills_code_review_agent",
-        description=(
-            "Automatic code review agent backed by Skills, sandbox execution, "
-            "Filter governance, and SQLite persistence."
-        ),
+        description=("Automatic code review agent backed by Skills, sandbox execution, "
+                     "Filter governance, and SQLite persistence."),
         model=model,
         instruction=INSTRUCTION,
         tools=tools,