主题
原文：https://www.mstx.cn/harness/423.html · 创建：2026-09-05 · 更新：2026-09-05 · 阅读：1105
一个让人后怕的假设
想象这样一个场景：你的Agent有一个execute_sql工具，可以查询数据库。某一天，用户问了一个看似无害的问题："帮我看看最近一周的订单总数。"
Agent决定调用execute_sql工具，生成了以下SQL：
SELECT COUNT(*) FROM orders WHERE created_at > '2026-08-28'这个查询没问题。但同一天，另一个用户问："帮我把所有超过30天未支付的订单取消掉。"
Agent理解了用户的意图，生成了：
DELETE FROM orders WHERE status = 'pending' AND created_at < '2026-07-28'如果这条SQL真的被执行了，会发生什么？ 所有符合条件的未支付订单被删除，而这些订单本应进入催款流程而非直接删除。
问题不在于Agent"理解了"用户的意图——用户确实想取消这些订单。问题在于：Agent的判断可能出错，而错误的后果不可逆。
这就是安全沙箱存在的意义：在工具执行和外部世界之间建立一道可控的屏障。 Agent的决策可能有误，但沙箱确保错误的影响范围是有限的、可恢复的、经过授权的。
安全沙箱的四个层次
工具的安全沙箱不是一个单一的机制，而是多层防御的组合。每一层解决不同类型的安全问题：
┌─────────────────────────────────────────────────┐
│ 第1层：权限声明（Permission Declaration） │
│ · 工具声明自己的危险级别 │
│ · 决定执行前是否需要确认 │
├─────────────────────────────────────────────────┤
│ 第2层：参数校验（Parameter Validation） │
│ · 在执行前验证参数是否合法 │
│ · 拒绝明显危险或异常的输入 │
├─────────────────────────────────────────────────┤
│ 第3层：资源限制（Resource Limitation） │
│ · 限制执行时间、内存、网络访问 │
│ · 防止工具失控或消耗过多资源 │
├─────────────────────────────────────────────────┤
│ 第4层：审计记录（Audit Logging） │
│ · 记录每次工具调用的完整信息 │
│ · 支持事后追溯和问题定位 │
└─────────────────────────────────────────────────┘第1层：权限声明与执行前确认
在4.1节中，你已经为工具定义了category和permission。现在把它们转化为实际的安全策略。
python
from enum import Enum
from dataclasses import dataclass
from typing import Any, Callable, Dict, Optional, List
class ToolCategory(Enum):
READ = "read"
WRITE = "write"
EXECUTE = "execute"
COMMUNICATE = "communicate"
DANGEROUS = "dangerous"
class ToolPermission(Enum):
AUTO = "auto"
USER_CONFIRM = "user_confirm"
ADMIN_ONLY = "admin_only"
class PermissionDeniedError(Exception):
"""权限被拒绝"""
def __init__(self, tool_name: str, reason: str,
required_permission: ToolPermission,
current_permission: ToolPermission):
self.tool_name = tool_name
self.reason = reason
self.required_permission = required_permission
self.current_permission = current_permission
super().__init__(
f"工具 '{tool_name}' 执行被拒绝。"
f"原因：{reason}。"
f"需要权限：{required_permission.value}，"
f"当前权限：{current_permission.value}"
)权限确认的触发条件
并非所有工具都需要在执行前询问用户。确认的触发条件应该是基于风险而非固定规则的：
python
class PermissionManager:
"""管理工具执行的权限"""
def __init__(self):
# 需要确认的权限级别
self.confirmation_required = {
ToolPermission.USER_CONFIRM,
ToolPermission.ADMIN_ONLY,
}
# 需要确认的分类
self.confirmation_categories = {
ToolCategory.WRITE,
ToolCategory.DANGEROUS,
ToolCategory.COMMUNICATE,
}
# 需要确认的特定操作模式
self.dangerous_patterns = [
"delete", "drop", "truncate", "remove", "purge",
"execute", "shell", "command", "script",
"send", "publish", "broadcast",
"transfer", "withdraw", "payment",
]
def should_require_confirmation(self, tool_def, arguments: Dict[str, Any]) -> bool:
"""判断当前工具调用是否需要用户确认"""
# 规则1：权限级别要求确认
if tool_def.permission in self.confirmation_required:
return True
# 规则2：分类要求确认
if tool_def.category in self.confirmation_categories:
return True
# 规则3：参数中出现危险模式
args_str = str(arguments).lower()
for pattern in self.dangerous_patterns:
if pattern in args_str:
return True
# 规则4：工具自身标记需要确认
if tool_def.metadata.get("always_confirm", False):
return True
return False
def check_permission(self, tool_def, arguments: Dict[str, Any],
user_confirmed: bool = False,
user_role: str = "user") -> Optional[PermissionDeniedError]:
"""检查权限，返回错误（如果拒绝）"""
# ADMIN_ONLY权限检查
if tool_def.permission == ToolPermission.ADMIN_ONLY:
if user_role != "admin":
return PermissionDeniedError(
tool_name=tool_def.name,
reason="此工具仅管理员可用",
required_permission=ToolPermission.ADMIN_ONLY,
current_permission=ToolPermission.USER_CONFIRM,
)
# 需要确认但用户未确认
if self.should_require_confirmation(tool_def, arguments):
if not user_confirmed:
return PermissionDeniedError(
tool_name=tool_def.name,
reason=f"工具 '{tool_def.name}' 属于{tool_def.category.value}类操作，"
f"执行前需要用户确认",
required_permission=ToolPermission.USER_CONFIRM,
current_permission=ToolPermission.AUTO if not user_confirmed else ToolPermission.USER_CONFIRM,
)
return None这个PermissionManager的设计有几个关键点：
第一：确认的触发是基于多条件的。 不仅仅看权限级别，还看工具分类和参数内容。一个READ类的工具，如果参数中出现了"delete"字样，也会触发确认——因为这可能是Prompt注入的尝试。
第二：拒绝和确认是分开的。 ADMIN_ONLY权限的拒绝是硬性的（用户无法通过确认来绕过），而USER_CONFIRM的确认是软性的（用户确认后可以执行）。这两种机制解决不同的问题：前者解决"谁有权执行"，后者解决"是否确认要执行"。
第三：参数内容也参与安全判断。 这很重要——LLM可能被注入攻击，在看似无害的工具调用中填充危险参数。检查参数内容是第二道防线。
第2层：参数校验与净化
权限确认在工具执行之前发生，但确认只能防止"已经意识到的风险"。还有一些风险是不可预见的——用户的输入可能包含异常字符、异常格式、或者意图不明确但后果严重的内容。
参数校验的目标是：在参数被传入工具函数之前，验证它们是否合理。
python
class ParameterValidator:
"""参数校验器"""
def __init__(self):
self.validators = {}
def register_validator(self, tool_name: str, param_name: str, validator_func: Callable):
"""为一个工具的参数注册校验器"""
key = f"{tool_name}.{param_name}"
self.validators[key] = validator_func
def validate(self, tool_name: str, arguments: Dict[str, Any]) -> List[str]:
"""验证参数，返回错误列表（空列表表示通过）"""
errors = []
# 通用校验
for key, value in arguments.items():
# 检查是否为空
if value is None or value == "":
errors.append(f"参数 '{key}' 不能为空")
continue
# 检查字符串长度
if isinstance(value, str) and len(value) > 10000:
errors.append(f"参数 '{key}' 长度（{len(value)}）超过最大允许长度（10000）")
# 检查嵌套结构（防止深度嵌套攻击）
if isinstance(value, (dict, list)):
depth = self._get_depth(value)
if depth > 5:
errors.append(f"参数 '{key}' 嵌套深度（{depth}）超过限制（5）")
# 工具特定的校验
for key, validator in self.validators.items():
validator_tool, validator_param = key.split(".", 1)
if validator_tool == tool_name and validator_param in arguments:
try:
result = validator(arguments[validator_param])
if isinstance(result, str):
errors.append(result)
elif result is False:
errors.append(f"参数 '{validator_param}' 校验失败")
except Exception as e:
errors.append(f"参数 '{validator_param}' 校验异常：{e}")
return errors
def _get_depth(self, obj) -> int:
"""递归计算嵌套深度"""
if not isinstance(obj, (dict, list)):
return 0
if isinstance(obj, dict):
if not obj:
return 1
return 1 + max(self._get_depth(v) for v in obj.values())
if isinstance(obj, list):
if not obj:
return 1
return 1 + max(self._get_depth(item) for item in obj)
return 0使用示例——为SQL执行工具注册参数校验：
python
import re
validator = ParameterValidator()
def validate_sql_param(sql: str):
"""校验SQL参数"""
# 拒绝多语句（防止SQL注入）
if ";" in sql and not sql.strip().endswith(";"):
return "SQL包含多条语句，仅允许单条语句执行"
# 拒绝危险的SQL模式
dangerous_patterns = [
(r"DROP\s+TABLE", "不允许DROP TABLE操作"),
(r"DROP\s+DATABASE", "不允许DROP DATABASE操作"),
(r"TRUNCATE", "不允许TRUNCATE操作"),
(r"ALTER\s+TABLE", "不允许ALTER TABLE操作"),
(r"GRANT\s+", "不允许GRANT操作"),
(r"REVOKE\s+", "不允许REVOKE操作"),
(r"INTO\s+OUTFILE", "不允许导出文件操作"),
(r"LOAD_FILE", "不允许加载文件操作"),
]
sql_upper = sql.upper()
for pattern, error_msg in dangerous_patterns:
if re.search(pattern, sql_upper):
return error_msg
# 必须包含LIMIT（防止全表扫描）
if "SELECT" in sql_upper and "LIMIT" not in sql_upper:
return "SELECT查询必须包含LIMIT子句"
return None # 校验通过
validator.register_validator("execute_sql", "query", validate_sql_param)这个SQL校验器展示了一个重要的设计原则：白名单比黑名单更安全。 与其列举所有不允许的操作（黑名单），不如定义允许的操作模式（白名单）。上面的校验器实际上在向白名单方向靠拢——它要求SELECT必须带LIMIT，拒绝所有结构性变更操作。
第3层：资源限制
即使参数合法，工具执行仍然可能失控：网络请求挂起、数据库查询消耗大量内存、无限循环。资源限制确保工具的执行在物理上也是受控的。
python
import signal
import resource
import threading
import time
from contextlib import contextmanager
from typing import Any, Dict, Optional
class ResourceLimiter:
"""资源限制器"""
def __init__(self,
max_execution_time: int = 10,
max_memory_mb: int = 512,
max_output_size: int = 10000):
self.max_execution_time = max_execution_time
self.max_memory_mb = max_memory_mb
self.max_output_size = max_output_size
@contextmanager
def time_limit(self, seconds: int):
"""时间限制上下文管理器"""
def handler(signum, frame):
raise TimeoutError(f"执行超过{seconds}秒")
old_handler = signal.signal(signal.SIGALRM, handler)
signal.alarm(seconds)
try:
yield
finally:
signal.alarm(0)
signal.signal(signal.SIGALRM, old_handler)
@contextmanager
def memory_limit(self, mb: int):
"""内存限制上下文管理器（仅Unix）"""
try:
soft, hard = resource.getrlimit(resource.RLIMIT_AS)
limit = mb * 1024 * 1024
resource.setrlimit(resource.RLIMIT_AS, (limit, hard))
yield
except (ValueError, resource.error) as e:
# 系统不支持内存限制时，仅记录而不失败
print(f"警告：无法设置内存限制：{e}")
yield
finally:
try:
resource.setrlimit(resource.RLIMIT_AS, (soft, hard))
except:
pass
def limit_output(self, result: str) -> str:
"""限制输出大小"""
if len(result) > self.max_output_size:
truncated = result[:self.max_output_size]
return truncated + f"\n...[输出已截断：原{len(result)}字符，显示前{self.max_output_size}字符]"
return result将资源限制集成到工具执行中：
python
class ToolRegistry:
def __init__(self,
default_timeout: int = 10,
default_max_memory: int = 512,
default_max_output: int = 10000):
self._tools = {}
self.default_timeout = default_timeout
self.default_max_memory = default_max_memory
self.default_max_output = default_max_output
self.limiter = ResourceLimiter(
max_execution_time=default_timeout,
max_memory_mb=default_max_memory,
max_output_size=default_max_output,
)
self.permission_manager = PermissionManager()
self.validator = ParameterValidator()
def execute(self, name: str, arguments: Dict[str, Any],
user_confirmed: bool = False,
user_role: str = "user") -> ToolResult:
"""执行工具，带完整的安全检查"""
import time
start = time.time()
if name not in self._tools:
return ToolResult(
success=False, content="",
error=f"未知工具：{name}",
tool_name=name
)
tool_def = self._tools[name]
# 第1层：权限检查
permission_error = self.permission_manager.check_permission(
tool_def, arguments, user_confirmed, user_role
)
if permission_error:
return ToolResult(
success=False, content="",
error=str(permission_error),
tool_name=name
)
# 第2层：参数校验
validation_errors = self.validator.validate(name, arguments)
if validation_errors:
return ToolResult(
success=False, content="",
error=f"参数校验失败：{'; '.join(validation_errors)}",
tool_name=name
)
# 第3层：资源限制下的执行
timeout = tool_def.timeout_seconds or self.default_timeout
try:
with self.limiter.time_limit(timeout):
result = tool_def.func(**arguments)
# 限制输出大小
result_str = str(result)
result_str = self.limiter.limit_output(result_str)
duration = int((time.time() - start) * 1000)
return ToolResult(
success=True,
content=result_str,
tool_name=name,
duration_ms=duration
)
except TimeoutError as e:
duration = int((time.time() - start) * 1000)
return ToolResult(
success=False, content="",
error=f"工具执行超时：{e}",
tool_name=name,
duration_ms=duration
)
except Exception as e:
duration = int((time.time() - start) * 1000)
return ToolResult(
success=False, content="",
error=f"工具执行异常：{type(e).__name__}: {e}",
tool_name=name,
duration_ms=duration
)第4层：审计记录
即使有前面三层防线，工具执行仍然可能出错。当错误发生时，你需要知道发生了什么、谁做的、什么时候、参数是什么。这就是审计记录的价值。
python
import json
import uuid
from dataclasses import dataclass, field, asdict
from datetime import datetime
@dataclass
class AuditRecord:
"""一次工具调用的审计记录"""
audit_id: str
timestamp: str
tool_name: str
category: str
permission_required: str
user_confirmed: bool
user_role: str
arguments: Dict[str, Any]
success: bool
error: Optional[str] = None
duration_ms: int = 0
result_preview: str = ""
trace_id: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
def to_json(self) -> str:
return json.dumps(self.to_dict(), ensure_ascii=False, indent=2)
class AuditLogger:
"""审计日志记录器"""
def __init__(self, log_file: Optional[str] = None):
self.log_file = log_file
self.records: List[AuditRecord] = []
def record(self, tool_name: str, category: str,
permission_required: str, user_confirmed: bool,
user_role: str, arguments: Dict[str, Any],
success: bool, error: Optional[str],
duration_ms: int, result_preview: str,
trace_id: Optional[str] = None) -> AuditRecord:
"""记录一次工具调用"""
record = AuditRecord(
audit_id=str(uuid.uuid4())[:12],
timestamp=datetime.now().isoformat(),
tool_name=tool_name,
category=category,
permission_required=permission_required,
user_confirmed=user_confirmed,
user_role=user_role,
arguments=arguments,
success=success,
error=error,
duration_ms=duration_ms,
result_preview=result_preview[:200],
trace_id=trace_id,
)
self.records.append(record)
# 写入文件（如果有配置）
if self.log_file:
with open(self.log_file, 'a', encoding='utf-8') as f:
f.write(record.to_json() + "\n")
return record
def get_records_by_tool(self, tool_name: str) -> List[AuditRecord]:
"""获取特定工具的所有审计记录"""
return [r for r in self.records if r.tool_name == tool_name]
def get_denied_records(self) -> List[AuditRecord]:
"""获取所有被拒绝的记录"""
return [r for r in self.records if not r.success]
def get_records_by_time_range(self, start: datetime, end: datetime) -> List[AuditRecord]:
"""获取时间范围内的记录"""
return [
r for r in self.records
if start <= datetime.fromisoformat(r.timestamp) <= end
]审计记录与普通日志的区别：
- 普通日志：记录"系统发生了什么"，用于调试
- 审计记录：记录"谁做了什么操作"，用于合规和追溯
审计记录的关键特征：
- 不可变：一旦记录，不应被修改
- 完整：包含判断"这个操作是否合理"所需的全部信息
- 可查询：支持按工具、时间、结果等维度检索
- 持久化：写入文件或数据库，不能只存在内存中
将安全沙箱集成到Agent循环中
现在将这些安全机制集成到Runtime的执行流程中：
python
class AgentRuntime:
def __init__(self, config, tool_registry, memory,
output_schema=None, tracer=None, logger=None,
user_role: str = "user",
confirm_callback: Optional[Callable] = None):
# ... 之前的初始化 ...
self.user_role = user_role
self.confirm_callback = confirm_callback # 用户确认回调
def _execute_with_safety(self, call: ToolCallRequest):
"""带安全沙箱的工具执行"""
tool_def = self.tools._tools.get(call.name)
if not tool_def:
return ToolResult(
success=False, content="",
error=f"未知工具：{call.name}",
tool_name=call.name
)
# 检查是否需要确认
user_confirmed = False
if self.tools.permission_manager.should_require_confirmation(tool_def, call.arguments):
if self.confirm_callback:
# 调用确认回调，让用户确认
user_confirmed = self.confirm_callback(
tool_name=call.name,
arguments=call.arguments,
category=tool_def.category.value,
)
if not user_confirmed:
return ToolResult(
success=False, content="",
error=f"用户取消了工具 '{call.name}' 的执行",
tool_name=call.name
)
else:
# 没有确认回调，直接拒绝
return ToolResult(
success=False, content="",
error=f"工具 '{call.name}' 需要用户确认，但当前没有确认机制",
tool_name=call.name
)
# 执行（带完整安全检查）
return self.tools.execute(
name=call.name,
arguments=call.arguments,
user_confirmed=user_confirmed,
user_role=self.user_role,
)确认回调的设计：在实际应用中，用户确认是一个异步交互过程。Agent暂停执行，等待用户响应。在简单的CLI场景中，可以用input()实现：
python
def cli_confirm_callback(tool_name: str, arguments: Dict[str, Any],
category: str) -> bool:
"""CLI场景下的确认回调"""
print(f"\n⚠️ Agent请求执行{tool_name}工具")
print(f" 分类：{category}")
print(f" 参数：{json.dumps(arguments, ensure_ascii=False, indent=2)}")
response = input(" 是否允许执行？(y/n): ").strip().lower()
return response in ('y', 'yes', '是')一个完整的执行流程示例
将以上各层整合，看一个完整的流程：
用户："帮我取消订单ORD12345678"
Agent循环：
1. LLM判断：需要调用cancel_order工具
2. Runtime准备执行cancel_order(order_id="ORD12345678")
3. 安全沙箱检查开始：
第1层-权限检查：
- cancel_order属于WRITE类
- 权限级别为USER_CONFIRM
- should_require_confirmation返回True
- 触发用户确认流程
用户看到：
"⚠️ Agent请求执行cancel_order工具
分类：write
参数：{"order_id": "ORD12345678"}
是否允许执行？(y/n)"
用户输入：n（发现订单号不对）
结果：工具执行被拒绝
4. 错误信息返回给LLM："用户取消了工具 'cancel_order' 的执行"
5. LLM看到错误后重新决策：
- 可能询问用户正确的订单号
- 可能告知用户取消操作已被阻止这个流程展示了安全沙箱的核心价值：它在Agent的自主性和系统的安全性之间建立了一个可控的平衡点。Agent可以自由地提出执行请求，但实际执行需要经过安全检查。
本节小结
安全沙箱是Agent与外部世界之间的受控边界。本节实现了四层防御：
- 权限声明与确认：决定"谁可以执行什么、什么需要确认"
- 参数校验与净化：在执行前验证参数的合法性和安全性
- 资源限制：确保工具执行在时间和资源上受控
- 审计记录：记录每次调用的完整信息，支持事后追溯
核心设计原则：
- 默认拒绝：不确定是否安全时，拒绝执行
- 最小权限：工具只获得完成任务所需的最小权限
- 深度防御：多层检查，不依赖单层防线
- 可追溯：每次执行都有完整的审计记录
在下一节中，我们将讨论工具错误处理与重试策略——当工具执行失败时，如何优雅地恢复。