Skip to content

tools/safety: 新增 Tool 脚本安全防护机制#241

Open
Wsp030914 wants to merge 26 commits into
trpc-group:mainfrom
Wsp030914:task/tool-script-safety-guard
Open

tools/safety: 新增 Tool 脚本安全防护机制#241
Wsp030914 wants to merge 26 commits into
trpc-group:mainfrom
Wsp030914:task/tool-script-safety-guard

Conversation

@Wsp030914

Copy link
Copy Markdown

Tool、Skill、MCP Tool 和 CodeExecutor 在执行脚本时,可能产生危险文件操作、非预期网络访问、凭据泄漏、依赖环境变更和资源滥用等安全风险。

本次变更新增可选、策略驱动的 Tool Script Safety Guard,在 Python 脚本和 Bash 命令真正执行前完成静态扫描,并输出 allow、deny 或 needs_human_review 决策。

安全检查器覆盖危险文件操作、非白名单网络外连、进程和 Shell 命令、依赖安装、资源滥用以及敏感信息泄漏。用户可以通过 YAML 策略配置白名单域名、允许命令、禁止路径、最大执行超时和最大输出大小,无需修改代码。

本次变更提供 Tool Filter、Skill 和 MCP adapter,以及 CodeExecutor wrapper。被拒绝的执行会返回结构化报告并记录可审计事件;启用 OpenTelemetry 时,还会写入对应的安全 span attributes。

示例目录提供 12 个可独立扫描的安全、危险和人工复核样本。文档说明了规则体系、接入方式、扩展方法、误报、漏报、绕过风险,以及该机制不能替代沙箱隔离的原因。

已在 Python 3.12 Docker 环境完成验证:134 个安全测试全部通过,安全模块覆盖率为 94.28%,YAPF 和 flake8 检查通过。验收测试确认 12 个公开样本决策全部正确,三类强制风险检出率为 100%,安全样本误报率为 0%,500 行脚本扫描耗时小于 1 秒。

实现全部通过新增文件完成,不修改现有 FilterRunner 或其他核心执行文件。

Fixes #90

RELEASE NOTES: 新增可配置的脚本执行前安全检查、风险拦截和审计能力。

@github-actions

github-actions Bot commented Jul 26, 2026

Copy link
Copy Markdown

CLA Assistant Lite bot All contributors have signed the CLA ✍️ ✅

@codecov

codecov Bot commented Jul 26, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 99.76785% with 4 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (main@6a2f7f9). Learn more about missing BASE report.

Files with missing lines Patch % Lines
trpc_agent_sdk/tools/safety/_python_rules.py 99.17695% 4 Missing ⚠️
Additional details and impacted files
@@            Coverage Diff             @@
##             main        #241   +/-   ##
==========================================
  Coverage        ?   88.32551%           
==========================================
  Files           ?         492           
  Lines           ?       46880           
  Branches        ?           0           
==========================================
  Hits            ?       41407           
  Misses          ?        5473           
  Partials        ?           0           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

@Wsp030914

Copy link
Copy Markdown
Author

I have read the CLA Document and I hereby sign the CLA

Invalidate ambiguous Python bindings, recognize builtins.open, and conservatively scan unknown execution tools. Remove unrequested design documents from the PR.
@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

基于我对 pr.diff 以及相关仓库上下文(BaseFilter/FilterRunnerBashTool/WorkspaceExecTool/SkillRunTool/SkillExecTool 的参数定义、BaseCodeExecutor/CodeExecutionResult 的字段、MCPTool 返回 list[str])的分析,以下是我的审查结论。

发现的问题

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_integration.py:138-139ToolSafetyFilter._after 的输出截断对非 dict/str 返回值无效

    • finalize_responsetruncate_output 只处理 str 和含 stdout/stderr/output 键的 dict,对 list 等类型原样返回。而 MCPTool._run_async_impl 返回的是 list[str](见 trpc_agent_sdk/tools/mcp_tool/_mcp_tool.py),走流式 filter 链路时 _after 不会限制其大小,与 README“在 _after 阶段限制返回给 Agent 的 output 大小”的承诺不一致。建议在 truncate_output 中对 list 元素逐项截断,或文档明确仅 Bash 类工具受限。
  • trpc_agent_sdk/tools/safety/_integration.py:116-117_before 注入(:138-139):对 applicable=False 的工具仍可能改写其 timeout 参数

    • _timeoutargstimeout/timeout_sec 时即返回 timeout_arg_name_before 无条件执行 req[request.timeout_arg_name] = request.effective_timeout_seconds。若一个非执行类工具(applicable=False,无 command/code/script)恰好有名为 timeout 的自有参数,会被注入策略上限值而改变其语义。建议在注入前增加 request.applicable(或 request.payloads)判断。
  • examples/tool_safety_guard/mcp_server.py:212execute_command 使用固定 effective_timeout_seconds 忽略请求超时

    • ScriptScanRequest(..., effective_timeout_seconds=float(GUARD.policy.max_timeout_seconds), ...) 固定为策略上限,未读取调用方传入的超时,且实际子进程用 request.effective_timeout_seconds(=300s)作为 wall-clock。作为示例可接受,但与“注入有效 timeout”的总体设计不一致;建议从请求参数解析或注释说明。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_sanitizer.py:44-47_NAMED_SECRET_REtoken=/secret= 这类非引号赋值也会整体替换后续 token,可能将普通证据(如 token=abc 形式的配置示例)误判为密钥并打上 redacted=True。当前不影响安全性,但会降低 evidence 可读性;可考虑收紧为带引号或典型密钥长度/字符集的匹配。

总结

该 PR 整体设计稳健(fail-closed、审计脱敏、symlink/权限加固、严格 YAML 校验、并发锁),测试覆盖面广。未发现必须修复的 Critical 问题;存在 2 处与文档承诺不一致的边界缺陷(MCP/列表输出不截断、非执行工具 timeout 被改写),建议修复后再合入。

测试建议

  • 补充一个流式/list 返回值的 _after 截断测试:构造返回 list[str] 且单元素超过 max_output_bytes 的 handler,断言输出被限制(暴露当前 truncate_output 对 list 的缺口)。
  • 补充一个 applicable=Falseargstimeout 的工具用例,断言 reqtimeout 未被 filter 改写。

@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

确认:"r" in item[1:] 是区分大小写的,所以 -Rf 不会匹配。rm -Rf / 不会触发 FILE001 DENY。仅触发 POLICY002 REVIEW(rm 不在 allowed list 中)。真正的绕过。

我顺便确认一下 --recursive 的情况:GNU rm 接受 --recursive 以及 -R。代码处理了 --recursive。但 POSIX 也允许将 -R 作为短选项。已确认存在漏洞。

现在,我还需要确认是否存在针对 rm -Rf 的测试用例——答案是没有。很好。

现在让我确认一下 env-prefixed 的情况是否真的只产生了 REVIEW 而不是 DENY——由于 rm 不在默认的 allowed list 中,_check_commands 会产生 POLICY002 (COMMAND_REVIEW, MEDIUM, NEEDS_HUMAN_REVIEW)。而且 _recursive_rm 返回 None,所以没有 FILE001。_SHELL_META_RE 不会匹配 FOO=bar rm -rf /(没有 shell 元字符)。所以最终决策为 NEEDS_HUMAN_REVIEW。确认是从 DENY 降级。

但要注意:env-prefixed 模式与 timeout 模式(测试显示为 REVIEW)一致。因此,从 rm 不是段中第一个 token 这一点来看,两者可能都被归类为“可接受”。但 env-assignment FOO=bar 在语义上与 timeout 包装器不同——它在 shell 中是一个常见的模式,其中 rm 仍然是直接调用的命令。这比 timeout 的情况更值得关注。我将其报告为 Warning(与 -R 问题合并,因为根本原因相同:recursive-rm 检测不完整导致降级为 REVIEW)。

实际上 -Rf 和 env-prefix 是不同的根本原因。但两者都导致相同的影响(rm -rf DENY 降级为 REVIEW)。根据指导,如果影响/修复相似则合并,但根本原因不同。-Rf 的修复是将 r 改为不区分大小写;env-prefix 的修复是让 _recursive_rm 通过 _command_name 风格的跳过逻辑来查找 rm。我将把它们报告为单独的 Warning,因为修复方式不同。

让我完成最终报告。

发现的问题

🚨 Critical

  • trpc_agent_sdk/tools/safety/_bash_rules.py:161:递归删除检测漏掉 POSIX 大写 -R,导致 rm -Rf / 仅触发 review 而非 deny
    • _recursive_rm"r" in item[1:] 判断递归标志,区分大小写;但 rm -R 是 POSIX 标准的递归删除写法(GNU/BSD/macOS 均支持)。rm -Rf / 不会命中 FILE001(DENY),仅因 rm 不在允许命令列表中产生 POLICY002(NEEDS_HUMAN_REVIEW),把本应 DENY 的递归删除降级为人工审核。建议改为 "r" in item[1:].lower() 或显式匹配 R
    if any(item == "--recursive" or (not item.startswith("--") and "r" in item[1:]) for item in options):

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_bash_rules.py:155-163_recursive_rm 仅识别 rm 为片段首 token 的情况,环境变量前缀写法可绕过 DENY

    • FOO=bar rm -rf /shlex.splittokens[0]FOO=bar_recursive_rm 直接 continue,不会产生 FILE001;而 _check_commands 通过 _command_name 跳过 VAR=val 赋值后能识别出 rm,最终仅产生 POLICY002(NEEDS_HUMAN_REVIEW)。同段 timeout 10 rm -rf /test_scanner.py:586 已断言为 REVIEW)属同类降级,但 env 前缀是常见 shell 写法、rm 仍是直接命令,风险更高且无测试覆盖。建议让 _recursive_rm 复用 _command_name 的赋值跳过逻辑定位真正的 rm
  • examples/tool_safety_guard/mcp_server.py:66-85:示例用 shlex.split + create_subprocess_exec 执行,与 tool-review/mcp-review 场景声明的管道语义不一致

    • real_agent.pymcp-review 场景要求执行 echo mcp-review | cat,但 execute_commandshlex.split(command) 得到 ['echo', 'mcp-review', '|', 'cat'] 后以非 shell 方式 create_subprocess_exec|/cat 会作为 echo 的字面参数而非管道,演示结果与“管道触发 review”的预期不符。建议示例明确说明该入口不经过 shell,或改用与扫描器一致的语义,避免误导复用者。

总结

整体实现结构清晰、审计 fail-closed 与脱敏处理到位;存在一个明确的 Critical 问题——递归删除检测未覆盖大写 -R,导致 rm -Rf / 从 DENY 降级为 REVIEW,建议在合入前修复。

测试建议

  • 补充 rm -Rf /rm --recursive /rm -r 大小写变体的扫描测试,断言命中 FILE001 且 decision == DENY
  • 补充 FOO=bar rm -rf / 等环境变量前缀写法的测试,确认递归删除仍被 DENY(或在策略中显式记录该降级为预期)。

if not tokens or os.path.basename(tokens[0]).lower() != "rm":
continue
options = [item for item in tokens[1:] if item.startswith("-")]
if any(item == "--recursive" or (not item.startswith("--") and "r" in item[1:]) for item in options):

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

递归删除检测漏掉 POSIX 大写 -R

_recursive_rm 用 "r" in item[1:] 判断递归标志且区分大小写,但 rm -R 是 POSIX 标准递归删除写法(GNU/BSD/macOS 均支持)。rm -Rf / 不会命中 FILE001(DENY),仅因 rm 不在允许命令列表产生 POLICY002(NEEDS_HUMAN_REVIEW),把本应 DENY 的递归删除降级为人工审核。建议改为 "r" in item[1:].lower() 或显式匹配 R。

@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

没有 TODO。现在让我再次通过追踪带有 -o 空值的 _ssh_remaps 来验证 while 1 的主张,确保不存在实际的 bug,并确保我对 Python while 1 的发现是准确的。

Python while 1:ast.parse("while 1:\n pass")node.testast.Constant,其 value=1 (int)。代码检查 node.test.value is True1 is True → 为 False。所以 RES001 不会被触发。已确认。并且 while 1 没有其他 RES 规则。所以 Python 中的 while 1: 在静态检查中是 ALLOW(允许)的,仅受执行器超时限制。这是一个实际的扫描漏洞。

让我来写最终的审查报告。

发现的问题

🚨 Critical

无。

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_python_rules.py:315-318while 1: 等非 True 常量无限循环未被检测
    • visit_While 仅在 node.testConstantvalue is True 时触发 RES001,而 while 1:while 1==1: 等等价死循环不会被标记,与 Bash 侧 while true/: 的检测不对称。安全扫描器对此类无界循环静默放行,可能让使用者误判已覆盖。虽然 SafetyGuardedCodeExecutorasyncio.wait_for 会在策略超时后终止执行,风险被限制在超时阈值内,但建议放宽判断条件:当 node.test 为常量且布尔求值为真(如 isinstance(node.test, ast.Constant) and node.test.value)时即触发,并补一条针对 while 1: 的测试。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_audit.py:30-31_PATH_LOCKS 为模块级普通字典,_shared_path_lock 只增不删,随审计路径变化长期累积会造成内存增长。常规部署路径固定、影响有限,但可改用 WeakValueDictionary 或在 sink 析构时清理对应锁,避免长生命周期进程中的泄漏。

总结

整体安全模块设计为 fail-closed(扫描失败、适配失败、审计失败均降级为 review/deny),扫描规则与测试覆盖较为完整,未发现阻塞级别的安全或正确性缺陷。唯一值得关注的是 Python 侧 while 1: 类无限循环未被静态规则覆盖(与 Bash 侧不对称),建议补齐。

测试建议

  • 补充 Python 无限循环的等价变体用例:while 1:while 1 == 1:while not 0:,验证是否触发 RES001,以闭合与 while True: 之间的检测缺口。

@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

I've read the full diff (5237 lines) and verified key downstream contracts (BaseCodeExecutor fields, create_code_execution_result, FilterResult/BaseFilter.run short-circuit behavior, BashTool timeout arg). Here is my review.

发现的问题

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_cli.py:90-91(经 scripts/tool_safety_check.py 入口):run_cli 调用 emit_report(JsonlAuditSink(args.audit), ...),但 emit_report 在审计写入失败时会抛 SafetyAuditError(继承 RuntimeError),而 main 只捕获 (ValueError, OSError)。结果是审计写盘失败(如路径不可写、磁盘满)会以未捕获异常崩溃并打印完整堆栈,而非返回 README 约定的 1=CLI/config error 结构化 JSON 退出。建议在 run_cli/main 中显式捕获 SafetyAuditError,经 sanitizer 清洗后返回 EXIT_ERROR{"error": ...}

  • tests/tools/safety/test_quality_constraints.py:38SAFETY_PACKAGE = Path("trpc_agent_sdk/tools/safety") 为相对路径,且校验逻辑为 assert not failures——当工作目录不是仓库根(或路径不存在)时,glob("*.py") 返回空列表,failures 恒为空,质量约束(文件/函数行数、参数数上限)会静默通过,等于这条约束实际未被执行。建议先断言扫描到的 .py 文件数大于 0,或在缺失时直接 fail,避免空集保护。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_integration.py:3856-3857adapt_tool_request 仅在 name == "bash" 时设置 execution_home/execution_root,而 workspace_exec/skill_run/skill_exec(同属 _BASH_TOOL_NAMES)保持 execution_home=None,导致同样的 ~ 路径在 Bash 工具能解析、在 skill/workspace 工具一律落到 FILE003 人工复核,行为不一致且偏保守。建议对同一类命令执行工具统一解析 execution_home,或在策略侧显式声明该差异以便维护。

总结

整体设计稳健(fail-closed 审计、脱敏优先于截断、超时与输出限制闭环),未发现阻塞性安全漏洞或核心逻辑错误。两处 Warning 分别是 CLI 审计失败路径的异常未收敛、以及质量约束测试在非根目录下静默失效,建议合入前修复。

测试建议

  • _cli.py 补充一条用例:传入不可写的 --audit 路径,断言退出码为 EXIT_ERROR(1) 且 stdout 为结构化 {"error": ...}、不含异常堆栈。
  • test_quality_constraints.py 增加断言:扫描到的 SAFETY_PACKAGE/*.py 文件数 > 0,确保约束确实落到真实源文件上。

@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

没有 conftestEXAMPLE_DIR 是相对路径,依赖于根目录的 cwd。这是一个真实的鲁棒性缺口。让我来写这份审查报告。

发现的问题

⚠️ Warning

  • tests/tools/safety/test_cli_and_acceptance.py:22EXAMPLE_DIR 使用相对路径,依赖 pytest 在仓库根目录执行
    • EXAMPLE_DIR = Path("examples/tool_safety_guard") 是相对路径,验收和 CLI 测试能否找到样本与 manifest 完全取决于运行时工作目录。仓库已有测试(tests/code_executors/local/test_unsafe_local_code_executor.py:134 通过 os.chdir(tmpdir) 改变全局 CWD),若该测试异常退出或测试顺序变化,相对路径会失效并导致 manifest.yaml/样本文件读不到。建议改为基于文件位置解析,如 Path(__file__).resolve().parents[3] / "examples/tool_safety_guard",使其与运行目录解耦。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_integration.py:232-233allow 分支直接用 request.effective_timeout_seconds(float)覆写 req[timeout_arg_name],对 timeout/timeout_sec 声明为 int 的 Skill 工具(如 SkillRunInput.timeout: int)会注入浮点值。Pydantic v2 通常能 coerce 整数浮点,但缺少针对真实 BashTool/skill_run 的端到端注入测试(现有 test_allow_invokes_handler_and_injects_timeout 仅用 SimpleNamespace)。建议补一个对真实工具参数校验路径的注入断言,避免静默类型不匹配。

总结

整体实现质量高、测试覆盖充分,未发现必须修复的阻塞性问题。唯一值得改进的是 tests/tools/safety/test_cli_and_acceptance.pyEXAMPLE_DIR 依赖运行时 CWD 的脆弱性,建议改为基于源文件位置解析。

测试建议

  • 暂无额外测试建议;现有覆盖已较完整。如采纳上述 Suggestion,可补一条验证 ToolSafetyFilter 向真实 BashTool/skill_run 参数 schema 注入 effective_timeout_seconds 后能通过其输入校验的集成断言。

@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

确认:os.remove/os.unlink/os.rmdir 存在于 _DIRECT_FILE_CALLS 中,但没有删除规则。测试中仅使用了 shutil.rmtree。通过 os.remove/os.unlink/os.rmdir 进行的 Python 文件删除未被标记为删除(仅在路径明确为禁止/系统路径时才会标记)。这是一个真正的覆盖缺口。

我现在已经有了明确的审查结果。让我再检查一遍,确认 _DIRECT_FILE_CALLS 条目是否触发了 _is_write_call 来检测删除。对于 os.rmdir_is_write_call 返回 False(不是写入)。因此,在普通路径上的 os.rmdir("/tmp/important") 会被放行。没错。这是一个真正的 Bug。

让我最后确认一下:_DELETE_CALLS 仅包含 shutil.rmtree。如果存在 shutil.rmtree 的别名(例如 import shutil as files; files.rmtree(...))会怎样?测试 test_recursive_delete_denied_python_alias 涵盖了 import shutil as files; files.rmtree(...)。访问者通过 _symbolic_value_aliasesfiles 别名为 shutil。让我验证一下 _name 是否解析了它。import shutil as filesvisit_Import 设置 self._aliases["files"] = "shutil"。然后 files.rmtree_name → "shutil.rmtree" → 在 _DELETE_CALLS 中。很好,别名已处理。所以 shutil.rmtree 无论别名如何都会被捕获,但 os.remove 及其相关方法从未被作为删除操作捕获。

现在让我汇总最终的审查结果。让我写下来。

其实,让我重新考虑一下 os.remove 的缺口是否是有意为之。README 指出 FILE001 “删除递归删除或将其限制在批准的工作区内”。规则 ID FILE001 用于递归删除。_DELETE_CALLS = shutil.rmtreeos.rmdir/os.remove 是单文件/空目录删除 —— 可以说是没那么危险。但 os.remove 仍然可以删除重要文件。而 bash 的 rm(非递归 rm file)——让我检查一下:_recursive_rm 仅标记带有 -rrm。所以 bash 的 rm important_file(非递归)→ 不会被 _recursive_rm 标记。但是 _check_commandsrm 不在 allowed_commands 中 → COMMAND_REVIEW。所以 bash 的 rm file → review(因为 rm 不在白名单中)。但是在 Python 中,os.remove → 没有命令白名单(仅文件访问检查)→ 在普通路径上放行。所以 Python 的 os.remove("/tmp/important") 被放行了。这是一个真正的不一致性。

然而,鉴于此防护明确侧重于递归/破坏性删除(FILE001 规范显示为“递归删除”),单文件删除可能超出了范围。但 os.rmdir 可以删除目录。而且示例 danger_delete.py 仅显示 shutil.rmtree。我认为这是一个值得注意的真正缺口,但将其定为 Warning(警告)而非 Critical(严重),因为单文件删除可以说破坏性比递归删除小,而且路径检查仍然可以捕获系统/禁止路径。

让我最终确定我的审查结果:

  • Warning(警告): Python 删除操作 os.remove/os.unlink/os.rmdir 没有被标记为删除操作;在非禁止路径上被放行,这与 shutil.rmtree 和 bash 的 rm -rf 形成了不一致(它们会被拒绝/审查)。
  • Warning(警告): os.open 的写入检测将动态/变量标志视为非写入,导致系统路径写入绕过(而 open() 正确地将未知模式视为写入)。

让我合并或保留它们。它们是不同的规则和修复方法,所以我将分开列出。实际上两者都是 _python_rules.py_scan_file_access/_is_write_call 的缺口。但修复方法不同(添加删除规则 vs 保守处理 os.open 标志)。我会分开列出。

让我验证一下建议的数量 —— 最多 2 条。我将专注于此。

让我最后检查一下 _estimated_size"x" * 1000000 的处理 —— text_node 是字符串字面量,count_node 是常量 1000000 → 大小 = 1*1000000 → 如果 > large_write_bytes → 审查。很好。但对于 "x" * variable → 没有 count_node → 0 → 放行。对于安全工具来说,这是可以接受的(无法静态确定)。

现在让我写下审查报告。我已经完成了对 diff 的全面审查,并对照仓库上下文验证了被调用方(BashTool、skill 工具参数名、create_code_execution_resultBaseTool.name 等)。下面是结论。

发现的问题

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_python_rules.py:4424_DELETE_CALLS)/ _python_rules.py:4839_scan_file_access):Python 文件删除检测仅覆盖 shutil.rmtreeos.remove/os.unlink/os.rmdir 不触发删除规则

    • 这三个调用虽在 _DIRECT_FILE_CALLS 中,但只走 _scan_file_access 的 forbidden/system 路径检查;当目标是普通路径(如 /tmp/important、工作区外非系统路径)时既不触发 FILE001 也不触发 FILE002,会被放行。这与 shutil.rmtree(DENY)和 Bash rm -rf(DENY)形成不一致的删除检测缺口,可绕过破坏性文件操作防护。建议将 os.remove/os.unlink/os.rmdir(可考虑 os.rename/os.replace)纳入删除/写删除规则,或在 _is_write_call 中将其视为写删除做保守标记。
  • trpc_agent_sdk/tools/safety/_python_rules.py:4860_is_write_callos.open 分支):os.open 写入判定对动态 flag 失败闭合不一致

    • open(...) 在 mode 为变量时保守判定为写(mode is None or ... 为 True),但 os.open(path, flags)flags 是变量/非字面量时 flag_text 为空,_is_write_call 返回 False,导致 os.open("/etc/x", flags) 写系统路径不触发 FILE001。建议当 os.open 的 flag 无法静态判定时保守视为写,与 open() 的处理保持一致。
  • examples/tool_safety_guard/mcp_server.py:266:超时分支 process.kill() 后再次 await process.communicate(),未在 kill() 与回收之间设置上限

    • communicate() 在子进程已被 kill 后通常很快返回,但若子进程忽略 SIGKILL(如处于不可中断系统调用,理论上少见)或管道已被部分消费,二次 communicate() 可能阻塞;建议在 kill 后用 asyncio.wait_for(process.wait(), timeout=...) 兜底并 best-effort 关闭管道,避免 demo 执行器在异常路径下挂起。

💡 Suggestion

  • examples/tool_safety_guard/README.md:75:文档称超时返回 CodeExecutionResult(is_timed_out=True),但 create_code_execution_resulttrpc_agent_sdk/code_executors/_types.py:303)实际不把 is_timed_out 存入结果对象,仅映射为 outcome=DEADLINE_EXCEEDED 和 stderr 文本。建议修正措辞为“返回 outcome=DEADLINE_EXCEEDED 且 output 含超时提示”,避免接入方按 is_timed_out 字段取值落空。

总结

整体实现稳健:fail-closed 的扫描/审计/超时链路、安全文件打开(O_NOFOLLOW+samestat+0600)、证据脱敏、并发锁、超时注入与输出截断均有对应测试覆盖,未发现安全凭证泄露或命令注入类 Critical 问题。存在两处 Python 静态删除/写入检测的不一致缺口(os.remove/os.unlink/os.rmdir 与动态 flag 的 os.open),建议修复但不构成必须阻塞。

测试建议

  • 补充 os.remove/os.unlink/os.rmdir 删除普通路径文件、以及 os.open("/etc/x", flags)flags 为变量)写系统路径的用例,验证二者不再被放行。
  • mcp_server.py 超时路径补充子进程未及时退出的兜底回收测试(当前仅覆盖正常超时返回)。

@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

我已经完成了彻底的审查。让我来写最终的审查结论。

发现的问题

🚨 Critical

  • examples/tool_safety_guard/mcp_server.py:283-292_scan_request_context 附近行):mcp_server.execute_commandScriptScanRequest 未设置 env_keys/argv/background/tty,但真正的问题是该 server 在 GUARD.scan 判定为 ALLOW 后直接用 shlex.split(command) + create_subprocess_exec 执行——扫描器对 rm -rf safety-demo-trash 这类命令会判定 deny,但策略 tool_safety_policy.yaml 允许了 bash/cat/curl/echo/pytest/python,且 _check_commands 只检查命令名是否在白名单。对于 echo mcp-allow(allow 场景)会通过并真实执行子进程;该执行器在 README 中明确标注“包含真实本地执行器”,但示例缺少沙箱。建议在示例中明确这是演示用途且默认不应在生产运行(README 已有声明),实际执行风险由示例性质承担,等级下调为下面 Warning 处理。

(经核对,mcp_server.py 的核心执行路径与扫描阻断逻辑一致,未发现可绕过扫描的真实执行漏洞;上述更接近示例固有风险,不计为 Critical。)

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_integration.py:240-246ToolSafetyFilter._before):在 report.decision == ALLOW 后才注入 req[request.timeout_arg_name],但 adapt_tool_request_timeout 对非有限值(NaN/Inf)走的是 requested 原样返回分支,requested_timeout_seconds 会是 nanscan_limitsrequested is None or 0 < requested <= max 判定不会对 NaN 触发 review(NaN 比较全为 False,走到 if requested <= 0: return [], False 也是 False,于是继续构造 evidence 走 POLICY001)。这条路径测试已覆盖 NaN 走 effective=30,但 _before 注入 value = request.effective_timeout_seconds(30.0)会覆盖原 nan,逻辑上是对的;风险点在于 _timeoutrequested(NaN)也返回了,scan_limitsrequested 非 None 且 0 < requested <= max 为 False、requested <= 0 也为 False,会落到 evidence = f"requested timeout {requested}s exceeds ..." 误把 NaN 判为超限并生成 POLICY001 review,与“NaN 应回退到策略上限并放行”的测试意图相悖。建议在 scan_limits 中对非有限值单独处理(视为无效并回退/阻断),或在 _timeout 阶段把 NaN 规整为 None

  • trpc_agent_sdk/tools/safety/_audit.py:131-140CompositeAuditSink.emit):降级路径里把 execution_blocked 强制置为 True、并把 decision 风格化,但 event.model_copy(update={...}) 只更新了 decision/risk_level/execution_blocked,未更新 redacted;若原事件 redacted=False 而降级实际发生了改写,审计中不会标记。这是审计保真度问题而非安全漏洞,建议在降级事件上置 redacted=True 以反映事件被改写。

  • trpc_agent_sdk/tools/safety/_audit.py:208-215emit_report):_shared_sink_lock(sink)WeakKeyDictionary 以 sink 实例为键。对每次调用都新建的 sink(如 LoggingAuditSink()CompositeAuditSink(...))不同实例拿到不同锁,无法在多线程并发 emit_report 时对“同一逻辑 sink”串行化;只有长期复用的同一实例(如 JsonlAuditSink,且其内部另有 path lock)才被串行。自定义 sink 若每次新建实例则降级为无锁并发。建议以 sink 的稳定标识(如路径/类名+构造参数)或要求 sink 自带可重入锁,避免对外部自定义 sink 给出虚假的串行化保证。

  • trpc_agent_sdk/tools/safety/_sanitizer.py:88-99truncate_output 列表分支):循环中对每个字符串元素用 max(remaining, 0) 截断并把 remaining 减去已用字节,但 truncate_textmax_bytes <= 0 时返回 ("", True),此时 remaining -= 0 不再下降,后续元素也都被截为 "",总字节确实受控;不过 was_truncated 在列表分支未被设置(仅 dict 分支设置 truncated 标记),导致列表输出被截断时调用方无法从结果中得知发生过截断(与 dict 分支行为不一致)。建议列表分支也返回截断标记或对列表结果补 truncated 指示。

  • examples/tool_safety_guard/mcp_server.py:108-111:成功执行返回时只截断 stdout/stderrMAX_OUTPUT_CHARS,但未设置 truncated 标记,且未通过 truncate_output 限制总字节(与 ToolSafetyFilter_after 限制逻辑不一致);同时 mcp_server 不走 Filter 的 _after,Agent 拿到的输出可能超过策略 max_output_bytes。建议在 MCP server 返回前统一用 truncate_output(response, GUARD.policy.max_output_bytes) 限制,保持与 Filter 路径一致的输出边界。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_scanner.py:171-176_scan_request_context):把 cwd/env_keys/argv" ".join 拼成单段文本再交给 scan_paths 的正则提取,多空格/特殊字符会导致路径 token 边界不稳定(例如 cwd 含空格时被切成多 token)。维护性上建议对 cwdargv 各自单独调用 scan_paths,避免拼接导致的边界歧义。

总结

整体是一个结构清晰、测试覆盖较全的静态安全扫描模块,扫描规则、审计 fail-closed、脱敏与输出截断设计基本正确,未发现可绕过扫描执行恶意代码的 Critical 漏洞。主要风险集中在 NaN 超时被误判为 review、自定义审计 sink 的并发串行化保证不完整、以及 MCP 示例输出截断与 Filter 路径不一致等 Warning 级问题,建议合入前修复上述 Warning。

测试建议

  • 补充 scan_limitsrequested_timeout_seconds = NaN/Inf 的断言(当前测试只断言 effective_timeout_seconds,未断言 decision/rule_id,无法暴露误判为 POLICY001 的问题)。
  • 补充并发测试:每次 emit_report 传入新构造的同类型 sink 实例,验证是否仍能串行化(当前 test_emit_report_serializes_custom_sink_calls 复用同一实例,无法暴露 WeakKeyDictionary 对新实例不共享锁的问题)。

Normalize non-finite timeouts, serialize custom audit sinks, and expose a public output limiter so integrations avoid private APIs.
@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

testpaths = ["tests"],因此 examples/ 并不直接包含在测试路径中,但 test_cli_and_acceptance.py 导入了 examples.tool_safety_guard.mcp_server 并对 execute_command 进行了测试。因此,mcp_server.py 确实位于测试运行路径中。双重 communicate 问题仅在挂起进程的真实执行中才会显现;而测试中使用了 mock,所以 CI 测试可以通过。但根据准则:“对测试代码、示例代码、演示代码也按真实执行风险评估”。mcp_server 是一个在真实调用时可能崩溃的示例。

让我更严谨地验证 asyncio 的双重 communicate 行为。实际上,在现代 asyncio(Python 3.10+)中,在因超时取消第一次 communicate() 后调用第二次 process.communicate() 是一个已知的模式,通常能正常工作——第一次调用被取消,管道依然存在,第二次 communicate() 会读取剩余数据并等待进程结束。已取消的 communicate 任务会被妥善清理。因此,这实际上可能不是崩溃,而是一种惯用法。最初的 communicate() 调用是一个被 wait_for 包装的协程;取消操作会清理读取器。许多实现都使用这种模式。我不再过度断定这一点,但会在“稳定性”维度下将其作为一项“警告”提出,并如实说明其实际影响。

让我再检查一下:超时后的 _limit_result 只会截断 output,但 CodeExecutionResult 中的 stderr 可能包含敏感信息?stderr 并未被清理。此外,在 filter 的 _after 阶段,truncate_output 应用于完整的响应字典,如果证据中包含密钥,则可能产生泄露。但输出的大小限制(而非清理)是既定目标——清理是在证据/审计层面进行的。问题在于,工具输出可能包含原始密钥,而 filter 在不进行清理的情况下,仅根据大小进行截断。这是设计使然(filter 的文档说明“限制输出大小”,而非清理)。好的。

现在,一个真正的问题:在 SafetyGuardedCodeExecutor._limit_result 中,只有 output 被截断了,而 stderr 没有。但 create_code_execution_result 会将 stderr 合并到 output 中,所以 stderr 确实被包含在 output 里了。没问题。

让我再检查一下 _integration.py 中的 set_safety_span_attributes —— 它在 except 之后被调用,没问题。

让我重新考虑是否存在一个真正的严重问题。让我重新审视 _timeout:当 requested 为非有限值时,返回 (None, float(policy.max_timeout_seconds), arg_name)。因此 requested_timeout_seconds=Noneeffective=policy max。在 scan_limits 中,requested is None → 返回 [], False。所以非有限超时不会产生审查结论。但是 test_adapt_non_finite_timeout_falls_back_to_policy_limit 测试预期是 effective=30requested=None —— 这是适配器层面的,没问题。但 test_non_finite_scan_timeout_is_reviewed_as_invalid(guard) 扫描测试使用 timeout=float("nan") 来调用 _request,并直接将 requested_timeout_seconds=nan 传入。然后 scan_limitsrequested=nanrequested is None 为 False,0 < nan <= max 为 False,not isfinite(nan) → 为 True → POLICY001 审查。很好。所以 nan 在扫描层被处理了,但适配器层将其转换为 None。一致。

等等——这里确实存在不一致。通过 filter 的 adapt_tool_requestnan 超时 → requested=None(通过非有限值处理),因此不会产生审查,它会正常执行。但是 CLI/直接扫描路径会进行审查。这是预期的设计:适配器限制了非有限值。这没问题。

现在,关于 ToolSafetyFilter._before_before 注入超时的一个真正问题:if request and request.applicable and request.timeout_arg_name:。它注入了 req[request.timeout_arg_name] = value。对于 skill_run 工具,timeout_arg_name="timeout"。但 skill_run 是否真正使用 timeout 作为参数是另一个问题。该 filter 是通用的。没问题。

我已经有足够的内容来撰写审查报告了。让我整理一下发现的问题。让我做最后一次检查,看看 _bash_rules 中的 _SHELL_META_RE_check_commands 是否会导致合法的安全命令产生误报,但这属于策略权衡,而非 bug。

让我核实最后一个真正的问题:_scan_request_context 使用 scan_paths 扫描 request.cwd 以及 env_keysargv 上下文。对于 argv=["~/.ssh/id_rsa"]scan_paths("~/.ssh/id_rsa", request, policy)。如果未设置 execution_home(在适配器中对于 bash 工具,execution_home 是通过 Path.home() 设置的),对于 bash 工具它是被设置的。path_forbidden("~/.ssh/id_rsa", ...)execution_home 已设置 → 解析 ~ → 检查 /root/.ssh 是否匹配 forbidden_paths 中的 ~/.sshforbidden_paths=["~/.ssh"]execution_home 已设置 → candidates 包含 /root/.sshforms 包含 /root/.ssh/id_rsaform.startswith("/root/.ssh" + "/") → 为 True。很好。

现在让我写出审查报告。让我重新考虑最强有力的发现,并确保我没有捏造问题。让我核实一下 _open_secure_file 符号链接竞态问题:它打开时带有 O_NOFOLLOW(如果可用),然后验证 lstat==fstatO_NOFOLLOW 已经会拒绝符号链接。重复检查。没问题。os.fchmod 确保了即使是已存在的文件也是 0600 权限。不错。

让我检查一个真实但细微的问题:CompositeAuditSink.emit 在回退前会吞掉主 sink 抛出的任何异常(不仅仅是 OSError),通过 except Exception: pass。因此,来自主 sink 的编程错误会被静默降级。这是预期的(弹性)。没问题。

最后让我核实一下 test_quality_constraints 中的 MAX_FILE_LINES=1000 是否会被任何文件违反——_python_rules.py 有 558 行,_bash_rules.py 420 行,都在限制内。函数参数:_limit_result(result, max_output_bytes) 为 2 个参数。adapt_tool_request(tool, args, policy) 为 3 个。均在 4 以内。没问题。

现在,关于双重 communicate,我需要精确一点。让我再次检查 mcp_server 的代码并如实描述。我将其定性为“警告”——在真实进程超时时存在崩溃风险。我也指出了 mcp_server 仅将 stdout/stderr 截断为 MAX_OUTPUT_CHARS(4096 字符),然后调用 GUARD.limit_output(response)(按 max_output_bytes=1MB 进行限制)。因此,最终的输出限制是基于 max_output_bytes 的。4096 字符的预限制是额外的。不是 bug。

让我再次思考 truncate_output 对列表的处理是否存在真实 bug。考虑 max_bytes=20,list=["x"*5, "y"*20](来自 test_guard_limits_output_with_policy_budgetmax_output_bytes=20)。string_bytes = 5+20=25 > 20marker_bytes=12reserve_marker = 20>=12 and 25>20 = Trueremaining = 20-12 = 8。第一项 "x"*5:truncate_text("xxxxx", 8) → 未改变,remaining=8-5=3。第二项 "y"*20:truncate_text("y"*20, max(3,0)=3) → "yyy",changed=True,remaining=3-3=0truncated=Truereserve_marker=True → 追加 "[TRUNCATED]"。结果 = ["xxxxx","yyy","[TRUNCATED]"]。总和 = 5+3+12=20 <= 20。测试预期 result[-1]=="[TRUNCATED]"sum <= 20。通过。很好,这里没有 bug。

但我发现了一个真实的细微问题:在列表截断中,truncated 仅通过 truncate_text 返回的 changed 来设置。如果 string_bytes <= max_bytes(无需截断),则没有问题。但如果某个项是非字符串,string_bytes 会排除它,但非字符串项保持不变——没问题。

现在确实存在一个 bug:reserve_marker = max_bytes >= marker_bytes and string_bytes > max_bytes。如果 string_bytes > max_bytes 但实际上没有任何项被截断(例如,项的总和超过 max_bytes,但各项都在限制内,只是累加溢出)——那么循环可能会在中间项超出 remaining 时将其截断,并将 changed 设为 True。没问题,逻辑是一致的。

我现在对我的发现很有把握。让我定稿。让我再重新考虑一下示例 tool_safety_audit.jsonl 中的 timestamp "2026-01-01" 是否重要——它是一个生成的示例产物,README 中说可以重新生成。不是 bug。

让我核实一下 README 中记录的退出码 "1=CLI/config error" 是否与 _exit_code 匹配——_exit_code 只处理 DENY/REVIEW/ALLOW;错误通过 EXIT_ERROR=1 返回。一致。

让我撰写最终的审查报告。让我重新考虑双重 communicate 的严重性——我需要保持诚实。由于测试在 mock 上通过,且真实世界的模式在 Python 3.10 中通常有效,我将其表述为“警告”,并注明需要验证。

实际上,让我重新考虑一个更具体、更站得住脚的问题。在 mcp_server.execute_command 中,如果执行成功(未被阻止),它会返回 GUARD.limit_output(response)。但 GUARD 是使用位于示例目录的 POLICY_PATH 构建的。tool_safety_policy.yaml 中的 max_output_bytes 为 1048576。limit_output 应用于一个包含 return_codestdoutstderr 的字典。truncate_output 只会截断 stdout/stderr/output,并在截断时设置 truncated=True。没问题。

但是等等——真正的问题:MAX_OUTPUT_CHARS = 4096 用于 stdout/stderr,但随后 GUARD.limit_output(response) 按 1MB 进行截断。所以双重限制没问题。但如果 return_codeNone(超时路径返回单独的字典),则不会调用 limit_output。一致。

现在让我定稿。让我重新考虑是否有任何严重问题。让我再检查一遍 _scan_payload:对于 PYTHON,它调用 scan_python 但不调用 scan_paths!对于 BASH,它调用 scan_paths(payload.content,...) 然后 scan_bash。对于 Python,字面路径检测是通过 AST 访问者 (_scan_file_access) 进行的,而不是通过 scan_paths 的正则表达式。所以 Python 的 forbidden-path-in-comment 或非 AST 上下文不会被 scan_paths 捕获,但 AST 访问者处理字符串字面量。没问题。

但注意:对于 Python,_scan_request_context 仍然对 cwdenv_keysargv 调用 scan_paths。因此 argv 路径在所有语言中都会被检查。很好。

我现在准备好写审查报告了。让我最终确定列表:

  1. 警告mcp_server.pyasyncio.TimeoutError 时对子进程重复调用 process.communicate();在真实(非 mock)子进程上,这种模式不可靠,可能会引发异常/导致进程泄漏,因为处理程序没有第二次 communicate 的防护。测试中使用了 mock,因此该风险未被发现。

实际上,让我降低对该问题的断言程度并诚实表述。让我再找找是否还有更站得住脚的问题。

让我重新检查 _dynamic_network_command:对于 curl https://api.example.com(允许的),_network_command_text 仅包含 curl 片段,scan_urls 检查主机 api.example.com → 允许。并且 _dynamic_network_command 返回 None,因为存在 https?://。很好。对于 env curl "$URL",片段 → _unwrap_tokens 剥离 envcurl 是命令。_dynamic_network_command:在片段中 re.search("https?://") → 无匹配 → 返回片段 → NET002 审查。很好。

现在,关于 _network_option_findingscurl -sKconfig https://api.example.com 的一个真实担忧:_option_matches("-K", "config")?token 是 "-sKconfig"。option="-K"len==2option.startswith("-") 为 True,not token.startswith("--") 为 True → option[1] in token[1:] → "K" in "sKconfig" → 为 True。所以匹配到 remap → NET003 拒绝。很好,测试通过。

我现在有信心了。让我定稿。让我重新考虑 real_agent.py_mcp_toolset 是否传递了 filters=[safety_filter]——并且 bash.add_one_filter(safety_filter) 添加了同一个 filter 实例,然后 skill toolset 也添加了同一个实例。一个 filter 实例在多个工具间复用。BaseFilter_handle_co 中有可变状态吗?不,它是无状态的。没问题。

让我写下审查报告。我已经审查了完整的 diff 以及相关的仓库上下文(filter runner、BaseCodeExecutorCodeExecutionResultMCPToolset)。以下是我的审查结论。

发现的问题

⚠️ Warning

  • examples/tool_safety_guard/mcp_server.py:91-101:超时回收路径对同一个 subprocess 重复调用 process.communicate()

    • 外层 await asyncio.wait_for(process.communicate(), ...) 超时后会取消该协程,紧接着 process.kill() 后再次 process.communicate()。在真实进程上,被取消的第一次 communicate() 可能已部分占用管道/transport,第二次调用在某些 asyncio 版本下会抛 RuntimeError 而非 TimeoutError,从而逃出外层 except asyncio.TimeoutError 导致 handler 崩溃且子进程残留。测试 test_mcp_timeout_reap_is_bounded_HungProcess mock 掩盖了该路径,真实执行风险未被覆盖。
    • 建议改用 process.wait() 配合 process.kill() 做回收,或对第二次 communicate() 捕获 Exception,并保证进程一定被 reap:
      except asyncio.TimeoutError:
          process.kill()
          try:
              await asyncio.wait_for(process.wait(), timeout=PROCESS_REAP_TIMEOUT_SECONDS)
          except (asyncio.TimeoutError, ProcessLookupError):
              pass
  • examples/tool_safety_guard/mcp_server.py:103-113:超时分支返回的响应未经 limit_output 处理

    • 正常路径返回前会经 GUARD.limit_output(response) 截断,但超时分支直接返回 {"stdout": "", "stderr": "...", "timed_out": True},绕过了统一的输出限制。虽然此处字段较短,但破坏了“所有返回都受 max_output_bytes 约束”的不变式;建议超时分支也走 GUARD.limit_output(response),保持与正常路径一致。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_bash_rules.py:317-329_check_commands 仅排除 {do,done,then,fi},对 for/while/if/case/else/elif 等 shell 关键字会逐段产生 POLICY002 review。这是有意保守,但这些关键字 review 与真正的“未授权命令”混在一起会稀释信号;可考虑把 shell 保留关键字一并排除,减少噪音。

总结

整体实现质量高,安全边界(fail-closed、审计失败即阻断、脱敏前置截断、路径/网络/重映射规则覆盖全面)设计严密,测试覆盖了绝大多数风险路径。存在 2 个 Warning 集中在示例 MCP server 的超时回收路径上,因测试以 mock 绕过而未暴露真实执行风险;无 Critical 阻塞问题。

测试建议

  • 建议为 mcp_server.execute_command 超时路径补充一个真实子进程(如 sleep 命令)的端到端测试,验证第二次进程回收不会抛出非 TimeoutError 异常且进程被 kill。
  • 建议补充一条断言:超时返回的响应同样受 max_output_bytes 限制(或显式记录该分支不受限的设计取舍)。

Reap timed-out subprocesses without reusing canceled pipe readers, enforce output limits on timeout responses, and reduce shell keyword review noise.
@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

CodeExecutionResult 并非定义在已 checkout 的源码中,而是通过 types 包导入的。它是被重新导出的。无论如何,_limit_result 仅截断了 output 字段(该字段聚合了 stdout 和 stderr)。由于 create_code_execution_result 会将 stderr 合并到 output 中,截断 output 基本足够了。这并非明确的 Bug。

我现在已经进行了彻底的审查。该代码质量很高,且采用了故障安全(fail-safe)设计。让我确认最后几点——特别是 _FORK_BOMB_RE 和递归删除检测是否存在明显的绕过路径。该工具是静态的“尽力而为”的扫描器;绕过是固有的,并非本次审查的重点。

让我重新考虑一个真正的顾虑:adapt_tool_requestenv_keys 泄露。它仅存储键(而非值)——已通过测试验证。很好。

让我再重新考虑一下 _before 超时注入,这在测试中得到确认:args["timeout"] == 10(policy max 为 10)。这是故障安全设计(限制超时)。没问题。

根据我的审查,我将撰写审查意见。我能辩护的最具体的问题是:

  • 警告 (Warning)_integration.py:231-236 — 当工具本身未提供超时(originalNone)时,过滤器会注入 effective_timeout_seconds(浮点数)。如果工具的参数校验器要求整数类型,这可能会破坏需要整数类型的下游工具。但仅当原始值为 int 时才会执行 int 分支,否则为浮点数。这是一个小问题,考虑到测试明确期望注入浮点数,这可能不成立。

让我重新考虑第 234 行:if isinstance(original, int) and not isinstance(original, bool): value = int(value)。所以如果 original 存在且为 int,则注入 int;如果 originalNone、float 或缺失,则注入 float。对于原本需要 int 超时的工具,如果用户省略了它,注入 float 可能会导致校验错误。但策略最大值是一个 int,effective_timeout_seconds = min(requested, float(policy.max_timeout_seconds)) —— 总是浮点数。嗯。这是一个合理的兼容性警告,但比较微妙。

老实说,这段代码经过了精心设计和测试。让我重新考虑是否有任何真正的问题,而不是强求。最合理的具体问题是:

  1. _limit_result 仅截断 CodeExecutionResult.output,如果代理未填充 output,则单独的 stderr/stdout 字段将不被受限 —— 但根据 create_code_execution_resultoutput 是聚合字段。作为警告是可以辩护的,但比较薄弱。

鉴于进行了深入的审查且没有发现明确的 Bug,我应该报告“未发现明显阻塞问题”,并仅在有真正维护价值的情况下提出最多 1-2 条建议。让我确认在放弃之前没有漏掉真正的正确性/安全性 Bug。

让我仔细重新检查一个安全问题:JsonlAuditSink.emit_open_secure_file 中使用了 os.O_NOFOLLOW 并检查了 samestat 以防止符号链接。这很稳健。但是 self._path.parent.mkdir(parents=True, exist_ok=True) 可能会创建目录 —— 如果 path 的父目录是 /tmp 等受控路径,那是没问题的。目录创建并未强制使用安全模式,但审计路径是由部署人员配置的。可接受。

让我检查最后一件事 —— 如果满足两个条件,CompositeAuditSink.emit 是否会抛出 SafetyAuditDegradedError,以及 emit_report 是否能正确传播它:是的,第 202-203 行会重新抛出。并且 _before 捕获 SafetyAuditErrorSafetyAuditDegradedError 的父类)→ 阻塞。故障安全。很好。

我已经足够得出结论了。代码是稳健的。我将报告没有阻塞性问题,并提出 1-2 条可选建议。

让我快速验证一个真正的顾虑:_shared_path_lock 在每次 emit 时都使用 str(path.resolve())。如果路径在不同进程间有所不同(例如带有 pid 的 tmpdir),则不会有跨进程锁定 —— 但每个进程内部是线程安全的,这是声明的作用域(“单进程”)。没问题。

让我重新考虑 error_report 是否设置了 summary="needs_human_review: safety scan failed."decision=NEEDS_HUMAN_REVIEW —— 意味着扫描错误会阻止执行(需要审核 = 在自动模式下不允许)。故障安全。很好。

最终结论:没有明确的关键/警告级 Bug。我将记录下这些微小的可选建议。

发现的问题

经逐文件审查 pr.diff 中的全部新增代码(trpc_agent_sdk/tools/safety/*scripts/tool_safety_check.py、示例与测试),未发现明确的逻辑错误、安全漏洞或会导致核心功能失败的阻塞性问题。

关键路径的 fail-safe 设计已落实到位,且可由现有测试验证:

  • 扫描/适配异常经 error_report 转为 NEEDS_HUMAN_REVIEW 并阻断执行(_integration.py:214-215295-307)。
  • 审计失败一律拒绝执行,不存在 fail-open(_integration.py:216-225297-304test_filter.pytest_audit_failure_stops_handler / test_primary_audit_degradation_stops_handler 覆盖)。
  • 输出脱敏先于截断(_sanitizer.py:63-76),审计仅落库脱敏后的 tool_name/rule_ids_audit.py:176-192),env 仅保留 key(_integration.py:121test_adapters.py 验证不泄露值)。
  • 审计文件使用 O_NOFOLLOW + samestat 防符号链接、0o600 权限(_audit.py:147-160)。
  • 超时在 filter 中被 clamp 到 policy 上限并回写(_integration.py:231-236,测试验证回写为 policy max)。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_sanitizer.py:92-111truncate_output 列表分支在 max_bytes < marker_bytes(即预算小于 11)且字符串总量超预算时,会截断但不追加 [TRUNCATED] 标记(reserve_marker 为 False),存在静默截断。若希望所有截断都可观测,可在该分支也追加标记或显式记录。
  • trpc_agent_sdk/tools/safety/_integration.py:231-236:超时回写仅当原始值为 int 时才转 int,否则(包括原始值缺失/None)回写 float。若某些受保护工具的 schema 将 timeout 限定为整数,用户省略 timeout 时注入 float 可能触发下游校验报错;可考虑按目标工具期望类型统一转换。

总结

整体风险低,未发现必须修复的 Critical 或 Warning 级问题;安全扫描器在异常、审计失败、超时、脱敏等关键路径均按 fail-safe 设计并配有测试覆盖。两条 Suggestion 仅为可观测性与类型兼容性的可选改进。

测试建议

暂无额外测试建议。现有测试已覆盖 deny/review/allow 决策、审计失败阻断、超时 clamp 与回写、并发审计写入、脱敏优先截断等高风险路径;如后续采纳上述 Suggestion,可补充「预算小于 marker_bytes 的列表截断仍可观测」与「省略 timeout 时回写类型符合目标工具 schema」两条用例。

Keep list truncation observable under tiny budgets and match integer timeout schemas when defaults are injected.
@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

发现的问题

🚨 Critical

未发现 Critical 问题。所有安全关键路径(DENY 拦截、审计失败 fail-closed、超时注入与裁剪)均有对应测试覆盖,且在异常/降级路径上正确阻断执行。

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_sanitizer.py:90-91:字符串输出截断缺少截断标记

    • truncate_outputlist 会追加 [TRUNCATED]/[T],对 dict 会写入 truncated: True,但对 str 分支只调用 truncate_text 后直接返回,被截断时无任何标识。字符串是 Bash/Tool 常见返回形态,静默截断会让 Agent 误判输出完整。建议字符串分支同样返回带标记或显式 truncated 信号,与 list/dict 行为对齐。
  • trpc_agent_sdk/tools/safety/_bash_rules.py:338-340(配合 _command_name 144-170):allowed_commands 策略可被 shell 关键字前缀绕过

    • _check_commands 仅取 segment 首个 token 作为命令名,遇到 do/then/else_SHELL_KEYWORDS 时整段跳过,导致关键字后的非白名单命令不触发 POLICY002。例如多行 for x in y\ndo whoami\ndone 因无 ; 也不触发 _SHELL_META_RE,最终 ALLOW,而裸 whoami 会被 review。危险命令(rm/网络/安装)仍由专项规则兜底,但信息收集类等仅由白名单约束的命令会被绕过。建议在 _command_name/_check_commands 中跳过前导关键字后再判断实际命令。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_audit.py:82JsonlAuditSink.emit 通过 mkdir(parents=True, exist_ok=True) 创建父目录但未收敛权限(默认 umask),在共享临时目录下父目录可能世界可读;审计文件本身已 0o600,影响有限,可考虑对新建父目录设 0o700 以闭合权限边界。

总结

整体实现稳健,安全关键路径与异常降级均按 fail-closed 处理并有充分测试。存在两处可验证的检测/输出一致性弱点(字符串截断无标记、关键字前缀绕过白名单),均非立即失败但建议修复;无必须修复的 Critical 问题。

测试建议

  • 补充 truncate_output 字符串分支的截断可见性测试(断言截断后可被感知)。
  • 补充多行 do/then 后接非白名单命令(如 whoami)应触发 POLICY002 的用例,固化关键字后命令的扫描预期。

Restrict newly created audit directories and continue scanning commands after shell control keywords.
- add visible truncation markers for string and list outputs
- cover zero-budget truncation and os.open fail-closed behavior
- harden JsonlAuditSink directory permissions and keep tests portable
@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

发现的问题

🚨 Critical

无。

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_python_rules.py:43-57_static_truthyast.literal_eval 求值 while 条件,可被恶意脚本用来对扫描器本身做资源耗尽攻击。
    • literal_eval 支持 Mult,对 while [0] * 100000000:while "x" * 100000000: 这类字面量会真正物化出 1 亿元素列表 / 100MB 字符串再判断真值,导致扫描器进程内存飙升甚至被 OOM-kill;该模块的威胁模型正是扫描不可信的 Agent 生成脚本。虽然上游 except Exception 会把 MemoryError 转成 needs_human_review,但内存峰值已发生。建议改为结构性判断真值(非空容器/非零数字即 truthy),或在物化前对字面量规模设限,避免真正构造对象:
      # 例如对 BinOp Mult 仅判断左右是否为“非空字面量”,不执行 left * right

💡 Suggestion

  • tests/tools/safety/test_cli_and_acceptance.py:11import examples.tool_safety_guard.mcp_server 隐式依赖仓库根目录位于 sys.path(即从仓库根运行 pytest)。
    • 仓库中其他导入示例代码的测试(如 tests/evaluation/test_optimize_quickstart_example.py)都会显式 sys.path.insert,而这里没有;若从其他 cwd 运行该测试文件,模块收集阶段即失败。建议在 [tool.pytest.ini_options]pythonpath = ["."],或在测试内显式插入仓库根,使导入不依赖 cwd。

总结

整体实现严谨:审计 sink 的 fail-closed、symlink/权限加固、超时注入与类型保留、secret 脱敏及异常链清理均有对应测试覆盖,未发现阻塞级缺陷。主要风险是扫描器对 while 条件中大型字面量的 literal_eval 求值可被用于 DoS,建议修复后再合入。

测试建议

  • 补充一个针对 while [0] * N: / while "x" * N:(N 为大字面量)的扫描用例,断言扫描在限定时间内返回 needs_human_reviewdeny 且不触发大内存分配,以固化 _static_truthy 的资源边界。

- replace literal_eval truthiness with structural checks
- deny large repeated literal while conditions without allocation
- make example acceptance import independent of cwd
@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

BashTool 返回 stdoutstderrformatted_outputtruncate_output 截断了 stdout/stderr,但 formatted_output(包含完整的 stdout+stderr 以及工作目录/返回代码)未被处理。已确认。

现在让我检查 _function_tool.py 是否返回了 result —— 不在 _OUTPUT_KEYS 中。但该示例并未使用安全过滤器包装 function tools。属于范围之外。

我已有足够的信息来撰写审查报告。让我整理一下。

发现的问题

🚨 Critical

(无)

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_sanitizer.py:124-139_OUTPUT_KEYStruncate_output 的 dict 分支):输出截断白名单只覆盖 stdout/stderr/output,遗漏了 BashTool 实际返回的 formatted_output 字段。
    • BashTool._run_async_impl 返回的 dict 中 formatted_output 把完整 stdout+stderr 拼在一起(trpc_agent_sdk/tools/file_tools/_bash_tool.py:212),而 ToolSafetyFilter._afterfinalize_responsetruncate_output 只裁剪 stdout/stderrformatted_output 原样透传给 Agent,使 _after 阶段“限制返回输出大小”的目标被绕过,大输出仍可回到上下文。建议把 formatted_output 纳入 _OUTPUT_KEYS,或在 truncate_output 中对未识别的 string 字段也按剩余预算裁剪。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_integration.py:255-261SafetyGuardedCodeExecutor.execute_code 超时分支):asyncio.wait_for 超时只取消协程,UnsafeLocalCodeExecutor 派生的子进程不会被回收,可能继续占用资源直到自然结束。README 已声明该限制,但代码层未对 delegate 子进程做 kill;若后续有可观测的 delegate 句柄,建议在超时分支补充进程级清理,长期可维护性更好。

总结

整体安全设计稳健(fail-closed、审计失败即阻断、脱敏先行、POSIX 文件加固与 symlink 拒绝均有覆盖)。未发现必须修复的 Critical 问题;主要可操作问题是 truncate_output 的输出字段白名单遗漏 formatted_output,导致 BashTool 的输出大小限制被部分绕过,建议修复。

测试建议

  • 补充一条 truncate_output 测试:输入包含 formatted_output 字段且内容超过 max_output_bytes,断言该字段也被裁剪且 truncated=True,覆盖当前 _OUTPUT_KEYS 未命中的字段路径。

Prioritize formatted_output so BashTool responses cannot return oversized agent-facing output after stdout/stderr are clipped.
@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

I have completed a thorough review of the diff. Below are my findings.

发现的问题

🚨 Critical

未发现 Critical 级别问题。

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_sanitizer.py:123-136truncate_output 对 dict 只截断 _OUTPUT_KEYSformatted_output/stdout/stderr/output),其他键原样放行。

    • ToolSafetyFilter._after 声称"限制返回给 Agent 的 output 大小",但任意 Tool 返回 {"data": "x"*10MB}{"html": ...} 等非约定字段时不会被截断,输出上限形同虚设,可能把超大/敏感内容透传给模型。建议对 dict 内所有 string 值计入同一预算,或显式文档化"仅约束已知输出字段"为安全边界的一部分。
  • trpc_agent_sdk/tools/safety/_python_rules.py:4824-4836_static_truthinessBinOp(Mult) 当两侧均为数值字面量时返回 None,导致 while 1 * 2: / while 2 * 3: 这类确定恒真的循环不被 visit_While 标记为 RES001,可绕过资源阻断规则。

    • 建议当两侧均为有限数值字面量时直接计算乘积真值(1*2 非零即真),归入 RES001 DENY,与 while 1: / while [0]*N: 的处理保持一致。
  • trpc_agent_sdk/tools/safety/_audit.py:3216-3232JsonlAuditSink.emitparent.exists() 循环外用 mkdir(parents=True, exist_ok=True) 创建目录,再对"缺失父目录"逐个 chmod(0o700);但若中间某层已存在但权限为 0o755,missing_parents 不会包含它,既有目录权限不会被收紧,审计目录可能保留过宽权限。

    • 建议对所有最终审计路径的祖先目录显式校验/收紧权限,而非仅对新增目录,避免权限边界被既有目录削弱。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_audit.py:3306-3314_shared_sink_lock 对非 JsonlAuditSink(如 CompositeAuditSink、自定义 sink)统一回退到全局 _FALLBACK_SINK_LOCK,会使互不相关的审计 sink 在高并发下被串行化。可按 sink 实例(id 弱引用)分配独立锁,减少不必要的序列化。

总结

整体设计稳健、失败闭合(fail-closed)处理到位、测试覆盖较全面,未发现必须修复的阻塞性问题;主要风险集中在输出截断字段覆盖不全、数值乘法恒真循环检测漏网以及审计目录权限收紧不彻底三处,建议按上述修复方向补齐。

测试建议

  • 补充 truncate_output 对含非 _OUTPUT_KEYS 大字符串字段的 dict 不再无限制透传的用例(如 {"data": "x"*10_000_000} 应触发截断或 truncated 标记)。
  • 补充 while 1 * 2: / while 2 * 3: 等数值乘法恒真循环应被判为 RES001 DENY 的用例。

- truncate all string fields in tool output dicts
- treat numeric multiplication in loop conditions as static truthy
- avoid chmodding cwd for plain relative audit paths
@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

Bash 环境需要批准,且无法运行 Python。我已经从 _types.py 中获取了足够的信息:CodeExecutionResult 拥有 output 字段(通过 create_code_execution_result 设置)和 outcome_limit_resultresult.output 进行截断并使用 model_copy 进行更新——这看起来是正确的,因为 output 是真实字段。

现在我有足够的上下文了。让我基于已知信息完成审查。我将跳过 Python 执行验证,仅依赖静态阅读。

发现的问题

🚨 Critical

  • trpc_agent_sdk/tools/safety/_integration.py:280-285:审计失败时调用 self.guard.error_report(error) 构造阻断报告,但 error_report 内部 make_finding(..., error, ...) 会用 SafetySanitizer.sanitize(error) 处理异常对象——str(error) 可能包含原始异常消息(如 sink 抛出的 "audit unavailable" 或底层 OSError 文本),随后作为 evidence 进入报告并通过 ToolSafetyViolation(report)report.summary/findings.evidence 暴露给调用方。虽然 _audit.pyemit_report 已尽力丢弃异常链,但此路径把未脱敏的异常文本塞进了结构化报告。建议对 error_report 的 evidence 同样走 sanitizer 并避免回显原始异常消息(注释也声称"expose the same structured review result",但实际 evidence 仍含原始 error 文本)。

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_integration.py:230-238_beforereport.decision == ALLOWrequest.applicablerequest.timeout_arg_name 存在时才注入超时。但当 adapt_tool_request 抛异常(被 except Exception 捕获生成 error_report,decision 为 NEEDS_HUMAN_REVIEW)时 request 仍为 None,后续 if request and ... 会跳过注入——逻辑正确;但 _timeout_valueBash 系工具在 original is None 时强制 int(value),而 workspace_exectimeout_sec 字段语义在策略中可能要求浮点。若上游 Tool 期望 timeout_sec 为 float,强制 int 截断可能与原契约不一致,建议按 timeout_arg_name 区分 timeout/timeout_sec 的类型保留策略。

  • trpc_agent_sdk/tools/safety/_sanitizer.py:65-74_NAMED_SECRET_RE 的正则字面量以 , ) 结尾(多了一个逗号),虽可正常编译(逗号被 re.compile 的第二参数位置忽略),但该尾随逗号会让阅读者误以为是问题,且若未来有人移除 re.IGNORECASE|re.DOTALL 之外的其他标志位排列,可能误把逗号当作 flag。建议清理尾随逗号以避免歧义。

  • examples/tool_safety_guard/mcp_server.py:99-104:超时分支中 process.kill() 后仅 wait_for(process.wait(), PROCESS_REAP_TIMEOUT_SECONDS);若子进程忽略 SIGKILL(如内核态等待)或产生孙子进程,reap 超时后直接返回而未确保进程树回收,且 process.communicate() 在 kill 后未再次调用以排空管道,可能留下僵尸进程或未关闭的 pipe。建议在 reap 失败时记录告警,并考虑 process.wait() 的最终状态检查。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_audit.py:97-108JsonlAuditSink.emitmissing_parents 收集逻辑先沿 parent 向上遍历到已存在目录,再 mkdir(parents=True, exist_ok=True) 再逐个 chmod,存在重复遍历;可简化为 mkdir 后只对新增层级 chmod,减少可读性负担。

总结

整体安全设计较完整(脱敏、审计 fail-closed、输出截断、symlink 防护),无明确阻塞性漏洞;Critical 一条为审计失败路径下异常文本可能未经脱敏进入结构化报告,建议修复。Warning 涉及超时类型契约、正则尾随逗号歧义和 MCP 子进程回收健壮性。

测试建议

  • 补充审计 sink 抛出含敏感词异常时 ToolSafetyViolation.report 的 evidence/summary 不泄露原始异常文本的断言测试。
  • 补充 MCP 超时后子进程及管道被正确回收(无僵尸进程、pipe 已关闭)的测试。

delegate: BaseCodeExecutor
guard: Any
audit_sink: Any

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

审计失败路径异常文本未脱敏进入结构化报告

error_report 构造阻断报告时,make_finding 对异常对象调用 sanitize(error),但 str(error) 可能含原始异常消息(如 sink 抛出的 OSError 文本),作为 evidence 暴露给调用方。建议对 evidence 同样走 sanitizer 并避免回显原始异常消息。

- avoid leaking audit exception text in safety reports
- preserve timeout_sec float injection for workspace tools
- drain MCP subprocess output after timeout
- simplify audit directory permission handling
Ensure newly created audit parent directories are tightened to 0700 again while keeping plain relative paths from chmodding cwd.
@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

发现的问题

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_python_rules.py:45-57while 循环条件静态求值会实例化超大字面量容器

    • visit_While_static_truthy_static_truthiness 在判断可静态求值前先调用 _static_value(node),而 _static_valueTuple/List/Set/Dict 字面量会完整构造对象。扫描器输入是不可信的 Agent 生成代码,一个 while {0:1, …百万键…}: 形式的脚本会强制分配并构造百万级字典,造成内存/延迟放大(DoS)。_static_truthiness 已对 BinOp Mult 做了免实例化短路,但裸容器字面量没有同等保护。建议对容器节点先按元素数量或 ast 体积做上限短路,超限即返回 _UNKNOWN_VALUE
  • trpc_agent_sdk/tools/safety/_integration.py:108-112workspace_exectimeout_sec 注入 float 会被目标 Tool 截断

    • _timeout_valuetimeout_arg_name == "timeout_sec" 一律返回 float,但被注入对象 workspace_exec._ExecInput.timeout_sec 声明为 inttrpc_agent_sdk/skills/tools/_workspace_exec.py:151),model_validate 会把 30.0 截断为 30。现有测试 test_workspace_exec_injects_float_timeout_sec_when_omitted 用 mock handler 只断言了注入类型,未覆盖真实 Tool 的截断行为。影响有限(仅丢失亚秒精度),建议按目标 schema 注入 int,或补一条针对真实 _ExecInput 的截断断言。
  • examples/tool_safety_guard/mcp_server.py:88-112:超时回收路径的二次 communicate() 可能对已关闭管道抛错

    • asyncio.wait_for 取消首个 communicate() 后,对同一进程再次调用 communicate();在真实管道已关闭/重入场景下可能抛 RuntimeError/ValueError,已用 except 兜底并设 reap_timed_out=True,行为正确。但该路径会真实执行(real_agent.py 以 stdio 子进程启动本服务),且 create_subprocess_exec 使用 cwd=WORK_DIR 直接执行通过 shlex.split(command) 得到的 argv,建议显式确认 argv[0] 走 PATH 解析时不受 MCP_ENV_KEYS 透传的 PATH 影响,避免示例被复用时 argv 解析面被放大。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_integration.py:159-184adapt_code_execution_input 未传递 cwd/env/请求超时上下文,导致 CodeExecutor 路径下 scan_paths 的 cwd 比对和 scan_limits 的超限检查无法生效;若后续要统一扫描语义,可在该 adapter 中补齐上下文字段。

总结

整体安全设计扎实(fail-closed、审计失败即阻断、密钥脱敏先于截断、重复 YAML 键拒绝、未知执行入口保守扫描),未发现明确的权限绕过或命令注入类 Critical 问题。存在一处静态求值的资源放大隐患(建议修复)和两处低影响的行为/测试覆盖缺口。

测试建议

  • 补充一条针对 while 测试节点为超大字面量容器(如百万元素 dict/list)的扫描用例,断言扫描时长与内存受限;
  • 补充一条 workspace_exec 真实 _ExecInput.model_validate 对注入 float timeout_sec 截断行为的断言,以覆盖当前仅 mock 的路径。

@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

CodeExecutionResult_types.py 中是从 trpc_agent_sdk.types 导入的,但在 types/ 中找不到该定义。这可能是包级别的导入问题。我不在这上面花时间了 —— 这与 diff 无关。

我已经有足够的上下文来编写审查报告。让我整理一下我的发现:

可验证的问题:

  1. _static_truthy 未能捕获 while not not 1: 这种形式的无限循环(双重否定绕过)——这是一个可验证的静态分析漏洞。将其标记为警告,因为静态扫描被记录为非沙箱机制。
  2. SafetyGuardedCodeExecutor._limit_result 仅截断了 output 字段,如果代理返回大型的 stdout/stderr 或其他字符串字段,这些字段将不会被截断——但实际上 create_code_execution_result 只填充 output。风险较低。跳过。
  3. 审计父目录 chmod 0o700 的副作用——已记录/已测试。跳过。

让我重新考虑一下是否真的存在关键问题。让我再检查一件事——real_agent.py / mcp_server.py 中的示例是否在生产代码路径中运行(CI 测试会导入 mcp_server)。测试 test_cli_and_acceptance.py 在模块导入时执行了 importlib.import_module("examples.tool_safety_guard.mcp_server"),这会触发 GUARD = ToolScriptSafetyGuard.from_policy(POLICY_PATH)。这会读取策略文件。如果策略文件缺失,导入就会失败——但它是与示例一起提交的。没问题。

让我再检查一下:在 mcp_server.py 中,effective_timeout = min(requested_timeout or float(GUARD.policy.max_timeout_seconds), float(GUARD.policy.max_timeout_seconds))。如果 requested_timeout0.0(假值),0.0 or 300 → 300。但 0.0 在上游已被归零为 None。没问题。如果 requested_timeout 是像 0.5 这样的合法浮点数,0.5 or 300 → 0.5,min(0.5, 300) → 0.5。没问题。

现在看一个真正值得关注的问题:子进程超时使用了 request.effective_timeout_seconds,但 request 是通过 requested_timeout_seconds=requested_timeout, effective_timeout_seconds=effective_timeout 构建的。所以超时时间 = effective_timeout = min(requested, 300)。很好。但 reap(回收)超时逻辑在第二个 communicate 中使用了 PROCESS_REAP_TIMEOUT_SECONDS,这是模块级的,并且在测试中被 monkeypatched 了。没问题。

让我确认一下没有明显的 bug,并写审查报告。再快速检查一下 scan_limitstimeout_sec 非整数检查与 _timeout 返回 int(value) 的交互——这已经在测试中覆盖了。没问题。

现在让我检查 examples/tool_safety_guard/tool_safety_audit.jsonltool_safety_report.json 是否作为“示例工件”提交。README 说它们可以重新生成。但它们包含硬编码的 timestamp: 2026-01-01。这些只是示例文件,不是契约。低风险。跳过。

让我再看一个:mcp_server.py 使用了 shlex.split(command, posix=True) 然后是 create_subprocess_exec(*argv)。如果模型发送的 commandecho mcp-allow,argv = ['echo', 'mcp-allow']cwd=WORK_DIR 会执行。安全过滤器已批准。但请注意:安全过滤器是通过对 command 进行 bash 扫描来批准的,而 bash 扫描允许 echo。执行避免了 shell。这实际上比过滤器批准的更安全。没问题。

我认为没有关键问题了。让我把双重否定问题写成警告,或许再加一条关于 sanitizer 的建议。其实,让我再核实一下双重否定是否真的可被利用且足够常见:while not not True:。是的,这是有记录的 Python 惯用法变体。值得标记。

让我再核实一下 _static_truthywhile 1 == 1: 的处理——比较路径:left=1, right=1,两者都是已知值,Eq → operator.eq(1,1)=True → bool → True。已拒绝。很好。测试证实了这一点。

现在让我再考虑一下 while not 0: 是否会被捕获。_static_truthy(UnaryOp(Not, Constant(0)))truthiness = _static_truthiness(Constant(0)) = bool(0) = False(非 None)。→ return False if truthiness is None else not truthinessnot False = True。已拒绝。很好,测试证实了这一点。

所以只有双重否定(以及更深的嵌套)会漏掉。这是一个真实的、尽管范围有限的绕过。我会将其标记为警告。

让我再核实一个可验证的 bug:在 CompositeAuditSink.emit 中,当主节点成功时,它返回。当主节点失败时,它会创建降级事件并进行回退。如果回退失败,则引发 SafetyAuditError。如果回退成功,则引发 SafetyAuditDegradedError。然后在 emit_report 中,它用锁包装。注意:emit_report 捕获 SafetyAuditDegradedError 并重新引发它,捕获通用异常并引发 SafetyAuditError。但在 CompositeAuditSink.emit 中,回退 LoggingAuditSink.emit 可能引发 SafetyAuditError(“回退审计失败”)。然后 CompositeAuditSink.emit 捕获它并引发 SafetyAuditError("all tool safety audit sinks failed")。测试 test_composite_primary_success_and_logging_failure 预期 LoggingAuditSink().emit(event) 在 logger.warning 引发 RuntimeError 时引发 SafetyAuditError, match="fallback audit failed"。是一致的。

我已经可以下结论了。让我写审查报告。鉴于可验证的问题有限,双重否定是唯一真正值得注意的发现。让我看看还有没有更具体的。

其实,让我重新检查 mcp_server.py 的超时 reap:process.kill() 之后,它调用了 process.communicate()。但如果 kill() 引发 ProcessLookupError,那是被捕获的。但 kill 之后没有 process.wait() —— communicate 也能处理回收(reaping)。没问题。

让我最终定稿。我会保持简洁。

再次检查:示例中的 tool_safety_audit.jsonltimestamp:"2026-01-01T00:00:00+00:00" —— 只是一个示例。没关系。

最终答案。我会将双重否定标记为警告,并添加一条关于测试预期与 create_code_execution_result 的 stderr 处理之间相互作用的小建议。实际上,后者不是问题。

让我重新考虑严重程度:双重否定绕过——安全工具漏掉无限循环是功能上的漏洞,但由于文档说明静态分析不是沙箱,所以是警告。我就定这个。

再重新考虑一下:是否有任何关键问题?_together 超时注入将 int 注入到 req["timeout"] 中。BashTool 执行 args.get("timeout", 300) 并使用 asyncio.wait_for(..., timeout=timeout)。注入了 int(300)。没问题。但对于 workspace_exec,_ExecInput.model_validate 期望 timeout_sec 是 int。注入了 int。测试 test_workspace_exec_timeout_sec_matches_real_model_validate 通过。很好。

没有关键问题。写审查报告。我已经完整审阅了 pr.diff(约 6000 行,新增 Tool Script Safety Guard 模块及其测试/示例)以及相关的被调用方代码(BashToolBaseCodeExecutorcreate_code_execution_result)。下面是基于 diff 可定位、可验证的结论。

发现的问题

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_python_rules.py:4964-4991_static_truthy 无法识别多重 not 的恒真循环条件,导致 RES001 漏报
    • while not not 1: / while not not True: 这类双重否定条件,_static_truthy 进入 UnaryOp(Not) 分支后调用 _static_truthiness(operand),而 _static_truthiness 未处理 UnaryOp,返回 None,于是 return False if truthiness is None else not truthiness 直接返回 False,判定为“非恒真”,RES001 不触发,最终 decision=ALLOW,一个真实的死循环被放行。这是安全扫描器的功能性漏报,建议在 _static_truthiness 中递归处理 UnaryOp(Not)(对内层再求一次 truthiness 并取反,内层不可判定时返回 None 而非当作假),或对 UnaryOp(Not)not not 折叠后再判断。注意这与 while not 0:(单层)能正确拒绝形成对比,是回归性缺口。

💡 Suggestion

  • examples/tool_safety_guard/mcp_server.py:223-226effective_timeout = min(requested_timeout or float(...max...), float(...max...)) 使用 or 折叠 0/空值
    • 虽然上游已把 <=0 归零为 Noneor 写法在此能工作,但语义上依赖隐式假值判断,可读性弱;显式写成 requested_timeout if requested_timeout is not None else float(...) 会更清晰,且能避免未来若放宽 0 值处理时引入的歧义。不影响当前正确性。

总结

整体是一个结构清晰、测试覆盖充分的安全扫描模块,核心阻断/审计/超时/输出截断链路逻辑正确,未发现安全凭证泄露、命令注入或核心功能失败的 Critical 问题。唯一值得修复的是 _static_truthy 对多重 not 的恒真循环存在可验证的漏报(Warning 级)。

测试建议

  • tests/tools/safety/test_scanner.pytest_python_truthy_constant_loops_are_denied 参数化用例中补充 "while not not 1:\n pass""while not not True:\n pass",断言 decision == DENY 且命中 RES001,以锁定上述漏报修复。

@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

我现在已经完全理解了。让我来撰写最终的评审意见。

发现的问题

🚨 Critical

未发现 Critical 级别问题。

⚠️ Warning

  • examples/tool_safety_guard/mcp_server.py:280-286:超时回收路径在 process.kill() 后再次调用 process.communicate(),但首次 communicate() 抛出 TimeoutError 时 transport 状态已被打断,二次调用在某些 asyncio 实现上可能抛出未被捕获的异常类型(如 RuntimeError: pipe closed 之外的其他错误),导致整个 tool 调用向 MCP 客户端抛原始异常而非结构化错误。建议将二次回收的 except 收宽为 except Exception 或确保外层有兜底,保证超时分支始终返回结构化响应。

    ...
    except (asyncio.TimeoutError, ProcessLookupError, RuntimeError, ValueError):
        reap_timed_out = True
    ...
  • trpc_agent_sdk/tools/safety/_audit.py:3380-3387JsonlAuditSink.emitself._lock(路径锁)之外执行父目录探测、mkdirchmod。多线程并发首次写入同一尚不存在的审计目录时,parent.exists() 循环与 mkdir(parents=True, exist_ok=True) 之间无保护;虽然 exist_ok=True 可避免冲突,但对 missing_parentschmod(0o700) 会与另一线程的目录创建交错,存在竞态下目录权限短暂放宽或 chmod 失败被 OSError 吞为 SafetyAuditError 的可能。建议把目录准备整体移入 self._lock 内,或在创建后统一收紧权限。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_integration.py:4404-4407_timeout_valueoriginal is None 且工具名属于 _BASH_TOOL_NAMES 时返回 int(value),但 effective_timeout_seconds 来自 float(...);当策略 max_timeout_seconds 非整数时 int(value) 会截断而非四舍五入,可能使注入的 timeout 比策略上限略短。若意图是整数秒,建议显式 round 或在策略校验中限定整数,避免隐式截断语义。

总结

整体风险较低:新增的 safety 模块逻辑自洽、fail-safe(审计失败即阻断执行、扫描异常返回 review),静态规则与脱敏设计稳妥,测试覆盖了并发、审计降级、超时注入、扫描异常等关键路径。存在两处稳定性相关的边界问题(MCP 超时回收异常处理、审计目录创建的并发竞态),建议修复但不构成核心功能阻塞。

测试建议

  • 暂无额外测试建议。现有测试已覆盖主要风险路径;如要加固,可补一条 mcp_server 超时回收二次 communicate() 抛非预期异常时仍返回结构化响应的用例。

@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

合成构件,无敏感信息。很好。

现在让我重新考虑 _audit.pyJsonlAuditSink.emit 可能存在的真正问题——即 chmod 在 Windows 上会失败。在非 POSIX 系统上,第 90 行的 self._path.parent.chmod(0o700) 不会报错(chmod 在 Windows 上是有效的空操作),但测试是受平台限制的。在非 POSIX 系统上 os.fchmod 被限制在 os.name == "posix" 内。这没问题。

现在让我再核实一下关于审计 mkdir/chmod 的一个真正问题:如果现有的父目录归其他用户所有且不可写,chmod 会引发 OSErrorSafetyAuditError。故障安全(Fails safe)。可以接受。

让我重新考虑 _open_secure_file 中可能存在的真正 Bug:os.fdopen(descriptor, "a", ...)。使用 O_APPEND 模式,"a" 模式是一致的。很好。文件以 0o600 权限创建。很好。

现在让我检查最后一个真正值得关注的问题:_shared_path_lock 是否通过 weakref.WeakValueDictionary 正确地在 sink 之间共享锁。使用相同的路径 → 相同的锁。已测试。很好。但在使用 str(path.resolve()) 作为键时存在一个细微的问题:如果路径不存在,resolve() 仍然会返回绝对路径。很好。没有 Bug。

我已经完成了全面的审查。代码质量很高,采用了故障安全(fails-closed)设计,且测试充分。我真正想要提出的问题:

让我再次确认问题 A —— 仔细检查 mcp_server 返回的输出是否确实经过了清理。process.communicate() 的 stdout 被解码,并截断为 MAX_OUTPUT_CHARS,然后执行 GUARD.limit_output(response),这会调用 truncate_output(仅字节截断,无清理)。因此,子进程 stdout 中的机密信息会未经修改地流向 MCP 客户端(Agent)。这是一个演示用的 MCP 服务器,但它确实可以运行,且该安全防护框架明确旨在防止机密信息泄露到外部接收器。这是一个值得注意的真正漏洞,但仅限于示例代码,所以定为警告级别。

让我再核实一下 ToolSafetyFilter._after / finalize_response 是否也仅执行截断,而不进行清理 —— 确实如此(第 254-256 行,limit_outputtruncate_output)。因此,生产环境过滤器路径也不会对工具输出中的机密信息进行清理。但是,安全防护旨在防止写入机密的脚本执行,而不是对任意工具输出进行清理。过滤器执行扫描(预执行),如果输出包含机密,则扫描阶段应该在执行前捕获到 SECRET001。因此,对于通过批准的脚本产生的输出,假设其是安全的。这种设计是一致的。所以 mcp_server 的情况也是类似的 —— 它只在 ALLOW(允许)扫描后执行。因此,只有当扫描器未检测到机密时,输出才可能包含机密信息。该漏洞是真实存在的,但范围有限。我最多将其标记为低优先级的建议。

鉴于指令要求不要编造低价值的建议,且代码确实稳健,我将提供一个聚焦的审查。让我重新考虑一下 timeout 运行器是否漏掉了 rm,这是否应该被标记 —— 测试 test_process_runner_cannot_hide_nested_command 断言它仅返回 NHR(而非 DENY),明确接受了 timeout 10 rm -rf / 仅作为审查而非拒绝处理。这是一个真正的检测弱点(rm -rf / 被允许执行,仅标记为人工审查,而不是被拒绝)。但设计意图是 NHR。我可以将其标记为警告,因为 rm -rf / 在等待人工审查期间可能会执行。然而,NHR 意味着 execution_blocked=True(过滤器会阻断)。查看 _before 第 238 行:decision != ALLOWrsp.is_continue = False(处理程序未运行)。因此,NHR 同样会阻断执行。所以 timeout 10 rm -rf / 会被阻断(而非执行)。是安全的。所以这并不是一个真正的漏洞 —— NHR 也会阻断。很好,不是问题。

所以甚至问题 A 的漏洞也仅限于扫描器未检测到的机密(罕见)。我将如实记录:未发现关键/阻塞性问题,并提供真正值得注意的少量观察结果。

让我再仔细核实一个我想要确定的真正候选问题:_static_rules 中的 nohup 检查 —— 第 381 行:any(_tokens(segment.strip())[:1] == ["nohup"] ...)。这会检查任何片段是否以 nohup 开头。但是 _unwrap_tokens 会剥离 command/exec/nohup 包装器,因此 _command_name 会看到 nohup 后面的命令。nohup 检查使用原始的 _tokens(而非解包后的),因此 nohup sleep 2h → 第一个 token 为 nohup → PROC003。此外,_runner_command 检查 _PROCESS_RUNNERS(包含 timeout),但不包含 nohup。并且嵌套的 sleep 通过 _command_name(会解包 nohup)被 _sleep_values 扫描 → 检测到 sleep。测试 test_nohup_requires_review_and_scans_wrapped_sleep 断言同时存在 PROC003 和 RES002。很好。

我现在对我的审查结果充满信心。代码很稳健。让我撰写最终的审查意见,并附上真正的观察结果。

让我最后核实一个潜在的实际 Bug:_python_rules 中的 _static_truthy 处理 while 1 == 1_static_truthy(Compare(1==1))_static_truthiness 在 Compare 上返回 None(未处理)。然后 _static_truthy 第 54 行:Compare 包含 1 个操作符,1 个比较器。left=_static_value(Constant(1))=1,right=1。两者均非 UNKNOWN。ast.Eqoperator.eq(1,1)=True → 返回 bool(True)=True。所以 while 1 == 1: → 静态为真 → DENY。测试通过 ✓。

让我核实一下 while [0] * 100000000 是否返回 False(空 → 非真)。_static_truthiness(BinOp Mult):left=_static_value(List([0]))=[0],right=_static_value(Constant(100000000))=100000000(int)。left 是列表,right 是 int(非 bool) → bool(left) and right != 0 = True and True = True。等等,测试 test_python_truthy_constant_loops_are_denied 包含 "while [0] * 100000000",并断言为 DENY。而 _static_truthiness 对于 [] * 100000000 返回 False。对于 [0](非空),结果为 True → DENY。很好。测试 test_static_truthiness_handles_mult_without_materializing 断言 [] * 100000000 为 False。一致 ✓。

一切都已验证且稳健。代码库通过了彻底的审查。我将提供一份重点突出的审查意见,包含真正有价值的观察结果。

基于我对整个 PR 的彻底审查,代码质量很高,采用了故障安全(fails-closed)的设计,并且经过了充分测试。以下是我的审查意见:

发现的问题

⚠️ Warning

  • examples/tool_safety_guard/mcp_server.py:115-120:MCP 执行结果未做密钥脱敏
    • 该示例 MCP server 在命令通过扫描后,将 process.communicate() 的 stdout/stderr 原样(仅字符截断 + limit_output 字节截断)返回给调用方。truncate_output 只做长度限制不做脱敏,若被允许执行的命令输出中包含密钥(扫描器未识别到写入密钥的场景),会原样回传给 agent。建议对返回体增加 SafetySanitizer.sanitize 处理,与 SECRET001 的防护意图保持一致。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_audit.py:172-180:非 JsonlAuditSink 的 sink 共用全局 _FALLBACK_SINK_LOCK
    • 所有非 JsonlAuditSink 实例(含多个独立的 CompositeAuditSink)共用同一把全局 RLock,会把彼此无关的审计写入串行化,高并发审计时可能成为吞吐瓶颈。这是测试中明确确认的行为(test_independent_audit_sinks_share_fallback_lock),仅在审计量很大时才需考虑按 sink 实例化锁。

总结

整体风险低,未发现必须修复的阻塞问题。安全防护链路(扫描失败、审计失败、降级)均按 fail-closed 处理,超时注入的类型/边界处理与 _ExecInput/BashTool 上下游契约一致,私有密钥/凭证脱敏覆盖较全。仅示例 MCP server 的输出未做密钥脱敏一点值得在示例层面补齐。

测试建议

暂无额外测试建议。现有测试已覆盖 fail-closed、超时类型保持、脱敏边界、并发审计不交错、嵌套 payload 深度等关键路径;若采纳上述脱敏建议,可补一条“命令输出含密钥时返回体被脱敏”的用例。

@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

我已经确认了关键细节。让我再确认一下 _before 是否没有重置 rsp.rsp,以及在 ALLOW 路径下,req 的变更是否会持久化以供处理器(handler)使用。确实如此(req 是共享字典)。

现在,最后一个需要验证的真正关注点:关于 _before 的错误处理。如果 adapt_tool_request 抛出异常(例如,env 不是字典但 isinstance(env, dict) 为 False → env_keys=[],没问题)。如果 args.get("cwd") 不是字符串 → str(args.get("cwd") or "")。没问题。_timeout 可能会抛出异常吗?float(raw)isinstance(raw, (int,float)) 保护。没问题。所以 adapt 很少抛出异常,但 error_report 处理了这种情况。很好。

我现在非常确定。让我写下审查意见。我会将 chmod 的那个问题列为 Warning,并可能把 mcp_server 环境继承的问题作为示例改进建议(Suggestion)。让我重新考虑一下 chmod 是属于 Critical 还是 Warning:它仅在审计父目录由其他用户拥有时才会导致阻塞,这是一种特定的部署情况,且失败关闭(fail-closed)是安全的。Warning 是合适的等级。

让我再重新考虑一下是否还有我遗漏的真正 Critical 问题——带有攻击者控制路径的 JsonlAuditSink 路径穿越问题?路径来自策略/配置(JsonlAuditSink(path)),而不是攻击者控制。而且 _open_secure_file 会拒绝软链接。所以不存在路径穿越问题。

再看一个:create_audit_event 使用 _AUDIT_SANITIZER.sanitize(tool_name)tool_name 进行脱敏——tool_name 来自 getattr(tool, "name", ...),由工具定义,并非攻击者控制。没问题。

我准备好撰写审查意见了。再重新考虑一下 _deduplicate 的排序以及去重是否会丢失不同的发现结果——不会。还有 truncate_outputdict 分支即使在没有发生截断的情况下,如果非输出字符串键被更改……实际上 was_truncated 仅在 changed 时设置。如果字符串键没有溢出预算,truncate_text 返回 (item, False),所以 was_truncated 保持为 False。很好,没有误报的 truncated

现在让我写下最终的审查意见。基于我对 pr.diff 的审查(新增的 trpc_agent_sdk/tools/safety/ 模块、示例、脚本以及全面的测试套件),以下是我的发现。

发现的问题

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_audit.py:90JsonlAuditSink.emit 在每次写入时对已存在的审计父目录执行 chmod(0o700)。当审计路径配置在进程无权修改的预存目录下(例如系统管理的 /var/log/...),chmod 会抛出 OSError,被捕获后转换为 SafetyAuditError,从而触发 fail-closed,导致所有受保护工具执行被阻断。建议仅对本次新建的 missing_parents 做 chmod,跳过已存在的父目录(或仅在创建时设置一次),避免对预存目录的反复权限操作。

    ...
    self._path.parent.mkdir(parents=True, exist_ok=True)
    if self._path.parent != Path("."):
        self._path.parent.chmod(0o700)   # 对已存在且非自身拥有的目录会失败
    for directory in missing_parents:
        directory.chmod(0o700)
    ...
  • examples/tool_safety_guard/mcp_server.py:279execute_command 在扫描 ALLOW 后用 create_subprocess_exec(*argv, cwd=WORK_DIR, ...) 执行,但未显式传入 env,子进程会继承父进程完整环境(含 TRPC_AGENT_API_KEY 等密钥)。作为可被复用的示例服务端,建议显式构建白名单环境或过滤敏感变量后传入 env=,避免被守护工具意外将密钥泄露给子进程。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_scanner.py:198_deduplicate(rule_id, evidence) 作为去重键,而 evidence 已经过 sanitizer 脱敏/截断;不同原始证据若脱敏后相同会被合并。当前不影响安全判定,但会丢失同规则不同位置的区分。若需保留多位置信息,可额外加入源码偏移或 source 字段参与去重。

总结

整体实现稳健,安全语义(deny/review/allow)、fail-closed(适配异常、审计失败即阻断)、输出截断与脱敏、超时注入均有充分测试覆盖,未发现阻塞性的正确性或安全漏洞。主要风险为审计 sink 对预存父目录的反复 chmod 可能在特定部署下意外阻断执行,建议修复。

测试建议

暂无额外测试建议。现有测试已覆盖 deny/review/allow、审计失败 fail-closed、并发写入、超时回收与输出脱敏等关键路径;如采纳上述 chmod 修复,可补一条“审计父目录已存在且不可写时仍能正常审计或按预期安全失败”的用例。

@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

发现的问题

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_audit.py:91-93:审计落盘只 flushfsync,崩溃可丢审计

    • JsonlAuditSink.emit 写入后仅调用 stream.flush(),只保证 Python 缓冲区刷到内核,未 os.fsync() 落到磁盘。模块明确将审计作为“无持久记录即不执行”的安全前提(见 _integration.py:311 注释),进程/系统崩溃时审计行可能丢失,导致安全决策无追溯。建议在 flush 后对底层 fd 执行 os.fsync(stream.fileno())
    ...
    stream.write(line)
    stream.flush()
    ...
  • trpc_agent_sdk/tools/safety/_audit.py:82-90:审计路径锁为进程内 weakref 字典,跨进程写同一 JSONL 可交错

    • _PATH_LOCKS 是进程内 WeakValueDictionaryJsonlAuditSink 类注释也仅声明“Single-process”。但示例与 CLI(scripts/tool_safety_check.pyreal_agent.py 多进程 MCP 子进程)可能在同一审计文件上并发追加;POSIX append 对超过 pipe buffer 的单次写入不保证原子,会产生半行交错、破坏 JSONL 可解析性。建议在文档/__init__ 公开 API 注释中明确“单进程”约束,或追加时按行 O_APPEND 单次 write 并限制单行大小。

总结

整体安全性设计较扎实(fail-closed、脱敏、符号链接/权限加固、超时注入类型保持均有覆盖)。未发现必须修复的 Critical 问题;主要风险集中在审计持久性与多进程并发语义上,建议至少补齐 fsync 与单进程约束说明。

测试建议

  • 建议补一个测试:审计写入后模拟崩溃(os.fsync 缺失场景下断言文件内容完整可解析),以固化耐久性预期。
  • 建议补一个并发/跨进程写同一 JSONL 的测试,验证当前是否仅适用于单进程,避免被误用为多进程审计。

@helloopenworld

Copy link
Copy Markdown
Contributor

AI Code Review

基于我对完整 diff 的审查,我的审查结论如下。

发现的问题

⚠️ Warning

  • trpc_agent_sdk/tools/safety/_audit.py:3468-3470:审计目录创建存在权限收敛窗口

    • JsonlAuditSink.emit 先用 self._path.parent.mkdir(parents=True, exist_ok=True) 创建父目录(受进程 umask 影响,常见为 0o755,其他用户可读/可穿越),再在循环中对新创建的父目录 chmod(0o700)。在创建与 chmod 之间存在时间窗口,期间审计目录对外可见,与 README 声称的"父目录强制 0o700"保证不完全一致。建议改为按层级 os.mkdir(path, 0o700) 逐级创建并 chmod,或创建前临时 os.umask(0o077),避免窗口期。
      self._path.parent.mkdir(parents=True, exist_ok=True)
      for directory in missing_parents:
          directory.chmod(0o700)
  • tests/tools/safety/test_scanner.py:2738-2782:绑定跟踪/资源规则用例仅断言 report.findings 非空

    • test_python_visitor_tracks_bindings_and_resource_shapes 覆盖了别名导入、for 重绑定、augassign、lambda、comprehension 等高风险绑定失效路径,但最终只 assert report.findings,未校验预期命中的具体 rule_id(如 RES002/PROC001/FILE003)。若绑定逻辑出现回归导致部分危险构造漏报,只要残留任一 finding 该测试仍会通过,无法捕获真实漏报。建议改为断言预期 rule_id 集合是当前 findings 的子集。

💡 Suggestion

  • trpc_agent_sdk/tools/safety/_audit.py:3553-3561:非 JsonlAuditSink 的自定义 sink 全部共享同一全局 _FALLBACK_SINK_LOCK(RLock),会把所有自定义审计 sink 的写入串行化。出于安全串行是有意为之,但在高并发审计场景下可能成为吞吐瓶颈;如未来需要并行自定义 sink,可在文档中标注该约束或提供按实例加锁的扩展点。

总结

整体实现质量较高:执行前静态扫描、审计 fail-closed、symlink 拒绝、O_NOFOLLOW+fchmod、子进程 argv 化与 env 过滤、密钥脱敏先行再截断等关键安全路径均有对应测试覆盖。未发现 Critical 级别阻塞问题;存在 1 处审计目录权限收敛窗口和 1 处弱断言测试,建议在合入前修复以落实 README 声称的安全保证。

测试建议

  • 补充一个用例:在多级不存在父目录场景下,验证目录创建后立即(创建路径内)即为 0o700,或显式验证 umask 非默认时权限仍满足 0o700
  • test_python_visitor_tracks_bindings_and_resource_shapes 改为断言预期 rule_id 子集(如 {"RES002","PROC001","FILE003"})以防止绑定回归漏报。

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

构建 Tool 执行脚本安全扫描、Filter 拦截与监控机制

2 participants