feat: add tool script safety guard

Author: Spock12138

examples/tool_safety_guard/README.md

@@ -0,0 +1,47 @@
+# Tool Safety Guard
+
+This example shows how to add a static safety scan before a tool or code executor runs script-like input.
+
+## Tool Filter
+
+```python
+from trpc_agent_sdk.tools import FunctionTool
+from trpc_agent_sdk.tools.safety import ToolSafetyFilter
+
+
+def run_command(command: str):
+    return {"ran": command}
+
+
+tool = FunctionTool(
+    run_command,
+    filters=[
+        ToolSafetyFilter(policy_path="examples/tool_safety_guard/tool_safety_policy.yaml"),
+    ],
+)
+```
+
+If you prefer `filters_name=["tool_safety_guard"]`, import `trpc_agent_sdk.tools.safety` first so the registered filter is available.
+
+## Code Executor Wrapper
+
+```python
+from trpc_agent_sdk.code_executors.local import UnsafeLocalCodeExecutor
+from trpc_agent_sdk.tools.safety import SafetyGuardedCodeExecutor
+
+
+code_executor = SafetyGuardedCodeExecutor(
+    delegate=UnsafeLocalCodeExecutor(),
+    policy_path="examples/tool_safety_guard/tool_safety_policy.yaml",
+)
+```
+
+## What It Checks
+
+The guard scans Python and Bash-like inputs for dangerous file operations, secret file access, non-whitelisted network egress, subprocess and shell patterns, dependency installation, resource abuse, and sensitive output.
+
+Reports include `decision`, `risk_level`, `rule_id`, `evidence`, and `recommendation`. Audit events are written as JSONL and the filter sets current OpenTelemetry span attributes such as `tool.safety.decision` and `tool.safety.rule_ids`.
+
+## Limits
+
+This guard is a pre-execution static scan. It can have false positives, false negatives, and bypasses through obfuscation or dynamic code construction. It does not replace sandboxing, container isolation, OS permissions, network controls, timeouts, or resource limits. Use it as an early policy and observability layer before a properly isolated runtime.

examples/tool_safety_guard/tool_safety_audit.jsonl

@@ -0,0 +1 @@
+{"timestamp":"2026-06-30T00:00:00+00:00","tool_name":"run_command","decision":"deny","risk_level":"critical","rule_ids":["FILE_RECURSIVE_DELETE"],"elapsed_ms":0.42,"redacted":false,"blocked":true,"language":"bash","cwd":"","metadata":{}}

examples/tool_safety_guard/tool_safety_policy.yaml

@@ -0,0 +1,27 @@
+mode: standard
+fail_closed: false
+block_on_review: true
+allowed_domains:
+  - api.example.com
+  - "*.trusted.internal"
+allowed_commands:
+  - curl
+  - wget
+  - echo
+  - cat
+  - grep
+  - python
+  - python3
+  - pip
+denied_paths:
+  - "~/.ssh"
+  - ".env"
+  - "/etc"
+  - "/var/secrets"
+max_timeout_seconds: 300
+max_output_bytes: 10000
+audit_log_path: tool_safety_audit.jsonl
+rules:
+  DEPENDENCY_INSTALL:
+    enabled: true
+    decision: needs_human_review

examples/tool_safety_guard/tool_safety_report.json

@@ -0,0 +1,24 @@
+{
+  "decision": "deny",
+  "risk_level": "critical",
+  "findings": [
+    {
+      "rule_id": "FILE_RECURSIVE_DELETE",
+      "risk_type": "file_operation",
+      "risk_level": "critical",
+      "decision": "deny",
+      "message": "Dangerous recursive delete detected.",
+      "evidence": "rm -rf /tmp/build-output",
+      "recommendation": "Avoid recursive deletion or restrict it to an explicitly approved workspace path.",
+      "line": null,
+      "column": null
+    }
+  ],
+  "elapsed_ms": 0.42,
+  "redacted": false,
+  "blocked": true,
+  "language": "bash",
+  "tool_name": "run_command",
+  "scanner_version": "1",
+  "error": null
+}

tests/tools/safety/samples.yaml

@@ -0,0 +1,89 @@
+- name: safe_python
+  language: python
+  expected_decision: allow
+  expected_rule_ids: []
+  content: |
+    total = sum([1, 2, 3])
+    print(total)
+- name: dangerous_delete
+  language: bash
+  expected_decision: deny
+  expected_rule_ids:
+    - FILE_RECURSIVE_DELETE
+  content: |
+    rm -rf /tmp/build-output
+- name: read_secrets
+  language: python
+  expected_decision: deny
+  expected_rule_ids:
+    - FILE_SECRET_READ
+  content: |
+    with open(".env") as f:
+        print(f.read())
+- name: network_egress
+  language: python
+  expected_decision: deny
+  expected_rule_ids:
+    - NET_NON_WHITELIST_EGRESS
+  content: |
+    import requests
+    requests.get("https://evil.example/data")
+- name: whitelist_network
+  language: bash
+  expected_decision: allow
+  expected_rule_ids: []
+  content: |
+    curl https://api.example.com/status
+- name: subprocess_call
+  language: python
+  expected_decision: needs_human_review
+  expected_rule_ids:
+    - PROC_SUBPROCESS
+  content: |
+    import subprocess
+    subprocess.run(["ls", "-la"])
+- name: shell_injection
+  language: bash
+  expected_decision: needs_human_review
+  expected_rule_ids:
+    - SHELL_INJECTION
+  content: |
+    echo $(cat user_input.txt)
+- name: dependency_install
+  language: bash
+  expected_decision: needs_human_review
+  expected_rule_ids:
+    - DEPENDENCY_INSTALL
+  content: |
+    pip install suspicious-package
+- name: infinite_loop
+  language: python
+  expected_decision: needs_human_review
+  expected_rule_ids:
+    - RESOURCE_INFINITE_LOOP
+  content: |
+    while True:
+        pass
+- name: sensitive_output
+  language: python
+  expected_decision: deny
+  expected_rule_ids:
+    - SENSITIVE_OUTPUT
+  content: |
+    api_key = "sk-test-secret-value"
+    print(api_key)
+- name: bash_pipeline
+  language: bash
+  expected_decision: needs_human_review
+  expected_rule_ids:
+    - SHELL_PIPELINE
+  content: |
+    cat access.log | grep token
+- name: human_review_mixed
+  language: python
+  expected_decision: needs_human_review
+  expected_rule_ids:
+    - NET_CLIENT_USAGE
+  content: |
+    import socket
+    socket.create_connection((host, 443))

tests/tools/safety/test_filter_and_audit.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.
+"""Tests for tool safety filter, audit, and executor wrapper."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock
+
+from opentelemetry import trace
+
+from trpc_agent_sdk.code_executors import BaseCodeExecutor
+from trpc_agent_sdk.code_executors import CodeBlock
+from trpc_agent_sdk.code_executors import CodeExecutionInput
+from trpc_agent_sdk.code_executors import CodeExecutionResult
+from trpc_agent_sdk.code_executors import create_code_execution_result
+from trpc_agent_sdk.context import AgentContext
+from trpc_agent_sdk.context import InvocationContext
+from trpc_agent_sdk.tools import FunctionTool
+from trpc_agent_sdk.tools.safety import SafetyDecision
+from trpc_agent_sdk.tools.safety import SafetyGuardedCodeExecutor
+from trpc_agent_sdk.tools.safety import SafetyPolicy
+from trpc_agent_sdk.tools.safety import ToolSafetyFilter
+
+
+class DummyCodeExecutor(BaseCodeExecutor):
+    calls: int = 0
+
+    async def execute_code(
+        self,
+        invocation_context: InvocationContext,
+        code_execution_input: CodeExecutionInput,
+    ) -> CodeExecutionResult:
+        self.calls += 1
+        return create_code_execution_result(stdout="ok")
+
+
+def _mock_invocation_context() -> InvocationContext:
+    ctx = MagicMock(spec=InvocationContext)
+    ctx.agent = MagicMock()
+    ctx.agent.parallel_tool_calls = False
+    ctx.agent.before_tool_callback = None
+    ctx.agent.after_tool_callback = None
+    ctx.agent_context = AgentContext()
+    return ctx
+
+
+async def test_filter_blocks_dangerous_tool_and_writes_audit(tmp_path):
+    def run_command(command: str) -> dict:
+        return {"ran": command}
+
+    audit_path = tmp_path / "audit.jsonl"
+    policy = SafetyPolicy(audit_log_path=str(audit_path))
+    tool = FunctionTool(run_command, filters=[ToolSafetyFilter(policy=policy)])
+
+    result = await tool.run_async(tool_context=_mock_invocation_context(), args={"command": "rm -rf /tmp/out"})
+
+    assert result["error"] == "TOOL_SAFETY_BLOCKED"
+    report = result["safety_report"]
+    assert report["decision"] == SafetyDecision.DENY.value
+    assert "FILE_RECURSIVE_DELETE" in report["findings"][0]["rule_id"]
+
+    lines = audit_path.read_text(encoding="utf-8").splitlines()
+    assert len(lines) == 1
+    event = json.loads(lines[0])
+    assert event["tool_name"] == "run_command"
+    assert event["blocked"] is True
+    assert event["decision"] == SafetyDecision.DENY.value
+
+
+async def test_filter_allows_safe_tool(tmp_path):
+    def run_command(command: str) -> dict:
+        return {"ran": command}
+
+    policy = SafetyPolicy(audit_log_path=str(tmp_path / "audit.jsonl"))
+    tool = FunctionTool(run_command, filters=[ToolSafetyFilter(policy=policy)])
+
+    result = await tool.run_async(tool_context=_mock_invocation_context(), args={"command": "echo hello"})
+
+    assert result == {"ran": "echo hello"}
+
+
+def test_filter_writes_otel_span_attributes(monkeypatch, tmp_path):
+    span = MagicMock()
+    monkeypatch.setattr(trace, "get_current_span", lambda: span)
+    filter_instance = ToolSafetyFilter(policy=SafetyPolicy(audit_log_path=str(tmp_path / "audit.jsonl")))
+    tool = MagicMock()
+    tool.name = "exec"
+
+    async def run_filter():
+        from trpc_agent_sdk.abc import FilterResult
+        from trpc_agent_sdk.tools._context_var import reset_tool_var
+        from trpc_agent_sdk.tools._context_var import set_tool_var
+
+        token = set_tool_var(tool)
+        try:
+            rsp = FilterResult()
+            await filter_instance._before(AgentContext(), {"command": "rm -rf /tmp/out"}, rsp)
+            return rsp
+        finally:
+            reset_tool_var(token)
+
+    import asyncio
+
+    rsp = asyncio.run(run_filter())
+    assert rsp.is_continue is False
+    span.set_attribute.assert_any_call("tool.safety.decision", SafetyDecision.DENY.value)
+    span.set_attribute.assert_any_call("tool.safety.blocked", True)
+
+
+async def test_code_executor_wrapper_blocks_before_delegate(tmp_path):
+    delegate = DummyCodeExecutor()
+    executor = SafetyGuardedCodeExecutor(
+        delegate=delegate,
+        policy=SafetyPolicy(audit_log_path=str(tmp_path / "audit.jsonl")),
+    )
+
+    result = await executor.execute_code(
+        MagicMock(spec=InvocationContext),
+        CodeExecutionInput(code_blocks=[CodeBlock(language="python", code='open(".env").read()')]),
+    )
+
+    assert "TOOL_SAFETY_BLOCKED" in result.output
+    assert delegate.calls == 0
+
+
+async def test_code_executor_wrapper_delegates_safe_code(tmp_path):
+    delegate = DummyCodeExecutor()
+    executor = SafetyGuardedCodeExecutor(
+        delegate=delegate,
+        policy=SafetyPolicy(audit_log_path=str(tmp_path / "audit.jsonl")),
+    )
+    input_data = CodeExecutionInput(code_blocks=[CodeBlock(language="python", code="print('ok')")])
+
+    result = await executor.execute_code(MagicMock(spec=InvocationContext), input_data)
+
+    assert "ok" in result.output
+    assert delegate.calls == 1