7.5 权限与安全
在上一节(7.4)中,我们讨论了如何为 Agent 选择合适的工具——从工具描述质量、参数设计到匹配策略,核心关注的是"选对工具"。然而,选对工具只是第一步:一个被正确选中的工具如果缺乏安全管控,同样可能造成灾难性后果。试想,Agent 正确地选择了 delete_file 工具,但如果调用它的用户本不该拥有删除权限,或者传入的路径参数被恶意构造为 ../../../etc/passwd,后果将不堪设想。
本节将从权限与安全的视角,回答一个核心问题:工具被选中后,如何安全地执行它? 我们将以"门禁系统"为核心类比,系统讲解 Agent 工具调用中的权限控制、沙箱隔离、人工确认机制和参数安全校验。
7.5.1 Agent 安全风险全景
Agent 的工具调用能力是一把双刃剑——它让模型从"只能说"变为"能做事",但也带来了前所未有的安全挑战。传统软件的安全边界相对清晰:用户通过 UI 触发操作,每一层都有明确的输入校验和权限检查。而 Agent 的工具调用往往由 LLM 自主决策发起,输入来源是自然语言,攻击面急剧扩大。
想象一栋大楼的门禁系统。传统软件就像一栋只有一个入口的大楼,保安在门口逐一检查身份证件即可。而 Agent 系统更像一栋有上百个入口的大楼,每个入口连接着不同的房间(工具),且"访客"是一个能理解自然语言、可以自主决策的智能体——它可能被"说服"去打开任何一扇门。
下表列出了 Agent 工具调用面临的主要安全风险:
| 风险类型 | 示例 | 严重程度 |
|---|---|---|
| 提示注入 | 用户输入中嵌入恶意指令,诱导 Agent 调用危险工具 | 🔴 严重 |
| 权限越界 | 工具访问了调用者不应访问的资源 | 🔴 严重 |
| 数据泄露 | 工具返回了敏感信息(密钥、用户隐私等) | 🟡 中等 |
| 参数篡改 | 恶意构造的参数导致意外行为(如路径遍历) | 🟡 中等 |
| 拒绝服务 | 工具调用耗尽系统资源(CPU、内存、磁盘) | 🟡 中等 |
| 供应链攻击 | 第三方工具包中包含恶意代码 | 🔴 严重 |
表 7-5-1:Agent 工具调用安全风险矩阵
这些风险并非停留在理论层面。以下是一些真实世界中的案例缩影:
- 提示注入导致数据删除:某客服 Agent 因用户输入中嵌入的提示注入指令,被诱导调用了
delete_all_orders工具,导致订单数据被批量删除。 - 恶意 SQL 执行:某代码助手 Agent 执行了用户提供的"示例 SQL",其中包含
DROP TABLE语句,导致数据库表被删除。 - 路径遍历读取敏感文件:某文件管理 Agent 因未对路径参数做校验,被构造的
../../../etc/passwd参数利用,读取了系统敏感文件。 - 供应链植入后门:某 Agent 集成的第三方 NPM 包被注入恶意代码,在工具执行过程中窃取了环境变量中的 API 密钥。
这些案例的共同点是:安全问题的根源往往不在于工具本身,而在于工具调用链条中缺少足够的管控层级。 正如一栋大楼不能只靠一道门禁,Agent 的安全也需要多层级、纵深防御的体系。
7.5.2 权限模型设计——Agent 的"门禁系统"
门禁系统类比
要理解 Agent 权限控制的设计思路,最贴切的类比就是现实世界中的门禁系统。一栋现代化办公大楼的门禁系统通常包含以下几个层次:
身份认证(Authentication)——"你是谁?":访客在大楼入口刷卡或刷脸,系统验证其身份。这对应 Agent 中的用户认证层(OAuth / JWT / API Key)。
角色授权(Authorization)——"你能进哪些区域?":即使通过了身份认证,访客也只能进入与其角色匹配的区域。普通员工不能进机房,访客不能进办公区。这对应 Agent 中的 RBAC(基于角色的访问控制)。
操作约束(Parameter Validation)——"你能在这里做什么?":即使进入了某个房间,也不是所有操作都被允许。会议室可以开会但不能做饭,工具间可以使用工具但不能带走设备。这对应 Agent 中的参数校验层。
执行监控(Execution Control)——"你的行为是否在可控范围内?":大楼内有监控摄像头和保安巡逻,限制每个区域的人数和停留时间。这对应 Agent 中的速率限制、超时控制和资源配额。
特殊审批(Human-in-the-Loop)——"这个操作需要主管签字":某些特殊操作(如进入金库、修改消防系统)需要额外的人工确认。这对应 Agent 中的高风险操作审批机制。
这五层防护构成了纵深防御体系,任何单层被突破都不会导致整体安全失效。
分层权限架构
一个健壮的 Agent 权限系统应遵循最小权限原则(Principle of Least Privilege):每个角色只拥有完成其任务所需的最低权限,不多也不少。这就像门禁系统中,访客卡只能开大堂的门,员工卡能开办公区的门,而机房卡只有运维人员才有。
┌─────────────────────────────────────────────────────────┐
│ 分层权限架构 │
├─────────────────────────────────────────────────────────┤
│ │
│ Layer 1: 用户认证 (Authentication) │
│ ├── 身份验证 (OAuth / JWT / API Key) │
│ └── 会话管理 │
│ —— 门禁类比:大楼入口刷卡验证身份 │
│ │
│ Layer 2: 角色授权 (Authorization - RBAC) │
│ ├── 角色定义 (admin / editor / viewer) │
│ ├── 权限分配 (read / write / execute / delete) │
│ └── 工具级别权限控制 │
│ —— 门禁类比:不同卡片对应不同区域通行权 │
│ │
│ Layer 3: 参数校验 (Parameter Validation) │
│ ├── 类型检查 │
│ ├── 范围限制 (数值大小 / 字符串长度) │
│ └── 白名单过滤 (只允许安全值) │
│ —— 门禁类比:进入区域后对具体行为的约束 │
│ │
│ Layer 4: 执行控制 (Execution Control) │
│ ├── 速率限制 (Rate Limiting) │
│ ├── 超时控制 │
│ └── 资源配额 (memory / CPU / disk) │
│ —— 门禁类比:监控系统对停留时间和行为的限制 │
│ │
└─────────────────────────────────────────────────────────┘图 7-5-1:四层安全架构与门禁系统类比
RBAC 权限控制实现
下面用 Python 实现一个完整的 RBAC 权限控制系统。代码中每一行都配有详细注释,帮助理解每个设计决策背后的考量。
from enum import Enum, Flag, auto # auto 自动生成唯一标志位值
from typing import Dict, Optional, Any # 类型注解,提高代码可读性
from dataclasses import dataclass # dataclass 简化数据类定义
import functools # 用于 wraps 保留原函数元信息
# ========== 权限定义 ==========
class Permission(Flag):
"""权限标志位——使用 Flag 而非普通 Enum,支持位运算组合权限"""
NONE = 0 # 无权限,用于初始化和比较基线
READ = auto() # 读取权限,如读取文件内容
WRITE = auto() # 写入权限,如创建或修改文件
EXECUTE = auto() # 执行权限,如运行代码或 SQL
DELETE = auto() # 删除权限,如删除文件或记录
ADMIN = auto() # 管理权限,如修改权限配置本身
# 组合权限——利用 Flag 的位运算特性
READ_WRITE = READ | WRITE # 读写组合
FULL = READ | WRITE | EXECUTE | DELETE | ADMIN # 全部权限
class Role(Enum):
"""预定义角色——对应门禁系统中的卡片类型"""
VIEWER = "viewer" # 只读角色,类似访客卡
EDITOR = "editor" # 读写角色,类似普通员工卡
OPERATOR = "operator" # 可执行操作,类似操作员卡
ADMIN = "admin" # 全部权限,类似管理员卡
# 角色-权限映射表——定义每种角色拥有哪些权限
# 这是 RBAC 的核心:权限不直接分配给用户,而是分配给角色
ROLE_PERMISSIONS = {
Role.VIEWER: Permission.READ, # 访客:只读
Role.EDITOR: Permission.READ | Permission.WRITE, # 编辑:读写
Role.OPERATOR: Permission.READ | Permission.WRITE | Permission.EXECUTE,
Role.ADMIN: Permission.FULL, # 管理员:全部权限
}
@dataclass
class ToolPermission:
"""工具级别的权限配置——为每个工具定义独立的访问规则"""
tool_name: str # 工具名称,如 "read_file"
required_permission: Permission # 调用此工具所需的最低权限
max_calls_per_minute: int = 60 # 每分钟最大调用次数(防滥用)
max_calls_per_session: int = 1000 # 单次会话最大调用次数
max_execution_time: float = 30.0 # 单次执行最大时间(秒),防卡死
requires_approval: bool = False # 是否需要人工审批(高风险工具设为 True)
class PermissionManager:
"""权限管理器——门禁系统的核心控制中心"""
def __init__(self):
# 存储每个工具的权限配置,key 为工具名
self._tool_permissions: Dict[str, ToolPermission] = {}
# 存储每个用户的角色,key 为用户 ID
self._user_roles: Dict[str, Role] = {}
# 调用计数器:{user_id: {tool_name: count}}
# 用于速率限制,记录每个用户对每个工具的调用次数
self._call_counters: Dict[str, Dict[str, int]] = {}
def register_tool_permission(self, perm: ToolPermission):
"""注册工具的权限要求——相当于在门禁系统中录入新门禁点"""
self._tool_permissions[perm.tool_name] = perm
def set_user_role(self, user_id: str, role: Role):
"""设置用户角色——相当于为用户发放特定类型的门禁卡"""
self._user_roles[user_id] = role
def check_permission(self, user_id: str, tool_name: str) -> bool:
"""检查用户是否有权限调用工具——门禁系统的刷卡验证"""
# 第一步:获取用户角色,默认为 VIEWER(访客)
role = self._user_roles.get(user_id, Role.VIEWER)
# 根据角色获取用户拥有的权限集合
user_perm = ROLE_PERMISSIONS[role]
# 第二步:获取工具所需的权限
tool_perm = self._tool_permissions.get(tool_name)
if not tool_perm:
return False # 未注册的工具默认拒绝——安全优先原则
# 第三步:位运算检查——用户权限是否包含工具所需权限
# 使用 &(按位与)检查,结果等于所需权限则说明权限满足
# 例如:用户有 READ|WRITE,工具需要 READ
# (READ|WRITE) & READ == READ → True
return (user_perm & tool_perm.required_permission) == tool_perm.required_permission
def check_rate_limit(self, user_id: str, tool_name: str) -> bool:
"""检查速率限制——防止短时间内大量调用导致系统过载"""
# 初始化用户计数器(如果尚未存在)
if user_id not in self._call_counters:
self._call_counters[user_id] = {}
# 获取当前调用次数
count = self._call_counters[user_id].get(tool_name, 0)
# 获取该工具的速率限制配置
limit = self._tool_permissions[tool_name].max_calls_per_minute
# 当前次数未达上限时返回 True(允许调用)
return count < limit
def record_call(self, user_id: str, tool_name: str):
"""记录一次工具调用——更新计数器,用于速率限制统计"""
self._call_counters.setdefault(user_id, {}) # 确保用户计数器存在
self._call_counters[user_id][tool_name] = \
self._call_counters[user_id].get(tool_name, 0) + 1 # 计数 +1
# ========== 权限检查装饰器 ==========
def require_permission(tool_name: str):
"""
权限检查装饰器——将权限逻辑与业务逻辑解耦
使用方式:在工具函数上方加上 @require_permission("tool_name")
"""
def decorator(func):
@functools.wraps(func) # 保留原函数的名称和文档字符串
async def wrapper(user_id: str, *args, **kwargs):
# 第一步:权限检查——用户是否有权调用此工具
if not perm_manager.check_permission(user_id, tool_name):
raise PermissionError(f"用户 {user_id} 无权限调用 {tool_name}")
# 第二步:速率限制检查——是否调用过于频繁
if not perm_manager.check_rate_limit(user_id, tool_name):
raise PermissionError(f"工具 {tool_name} 调用频率超限")
# 第三步:记录本次调用(更新计数器)
perm_manager.record_call(user_id, tool_name)
# 第四步:权限检查通过,执行实际工具函数
return await func(*args, **kwargs)
return wrapper
return decorator
# ========== 配置示例 ==========
perm_manager = PermissionManager()
# 注册工具权限——相当于在门禁系统中配置每个门的通行要求
# read_file 工具:只需读取权限,速率限制较宽松(100 次/分钟)
perm_manager.register_tool_permission(ToolPermission(
tool_name="read_file",
required_permission=Permission.READ,
max_calls_per_minute=100
))
# delete_file 工具:需要删除权限,速率限制严格(10 次/分钟),且需要人工审批
perm_manager.register_tool_permission(ToolPermission(
tool_name="delete_file",
required_permission=Permission.DELETE,
max_calls_per_minute=10,
requires_approval=True # 高风险操作,需要主管签字
))
# execute_sql 工具:需要执行权限,执行时间限制为 10 秒
perm_manager.register_tool_permission(ToolPermission(
tool_name="execute_sql",
required_permission=Permission.EXECUTE,
max_calls_per_minute=30,
max_execution_time=10.0 # SQL 执行应快速完成
))
# 设置用户角色——相当于发放不同级别的门禁卡
perm_manager.set_user_role("user_001", Role.VIEWER) # 访客卡
perm_manager.set_user_role("user_002", Role.EDITOR) # 员工卡
perm_manager.set_user_role("admin_001", Role.ADMIN) # 管理员卡
# 验证权限检查
print(f"user_001 读文件: {perm_manager.check_permission('user_001', 'read_file')}")
# 输出: True —— 访客拥有读取权限
print(f"user_001 删文件: {perm_manager.check_permission('user_001', 'delete_file')}")
# 输出: False —— 访客无删除权限
print(f"admin_001 删文件: {perm_manager.check_permission('admin_001', 'delete_file')}")
# 输出: True —— 管理员拥有全部权限上述代码的核心设计要点:
- 使用
Flag而非普通Enum:Flag支持位运算,可以用|组合权限,用&检查权限,非常适合权限系统的"包含"语义。 - 权限不直接绑定用户:通过角色间接分配权限,当人员变动时只需调整角色,不需要逐个修改权限——这正是 RBAC 的精髓。
- 未注册工具默认拒绝:
check_permission中对未注册工具返回False,遵循"默认拒绝"的安全原则。 - 装饰器解耦:
require_permission装饰器将权限检查逻辑与业务逻辑分离,工具开发者只需关注功能实现。
7.5.3 沙箱隔离——为危险操作建一道"防爆墙"
为什么需要沙箱
回到门禁系统的类比:权限控制解决了"谁能进哪个门"的问题,但即便一个人有合法权限进入了某个房间,他在房间里的行为也需要被约束。如果一个操作员有权限执行代码,但他执行的代码包含无限循环或恶意命令怎么办?
沙箱(Sandbox)就是为了解决这个问题而存在的。它就像化工实验室里的"防爆墙"——即使内部发生爆炸,也不会波及外部。在 Agent 系统中,沙箱通过隔离执行环境,确保即使工具代码出现异常或恶意行为,也不会影响宿主系统。
沙箱隔离的核心目标包括:
- 网络隔离:限制网络访问,防止数据外泄或拉取恶意载荷
- 文件系统隔离:限制可访问的路径,防止读写敏感文件
- 资源限制:控制 CPU、内存、磁盘使用,防止资源耗尽
- 执行超时:防止死循环或长时间阻塞
- 权限降级:以最低权限用户运行,减少攻击面
沙箱架构设计
┌─────────────────────────────────────────────────────────┐
│ 沙箱隔离架构 │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────┐ │
│ │ 宿主机 (Host) │ │
│ │ ┌───────────────────────────────────────────┐ │ │
│ │ │ 沙箱容器 (Container) │ │ │
│ │ │ ┌─────────────────────────────────────┐ │ │ │
│ │ │ │ 受限执行环境 │ │ │ │
│ │ │ │ • 网络: 仅允许白名单域名 │ │ │ │
│ │ │ │ • 文件: 只读挂载,限制路径 │ │ │ │
│ │ │ │ • 内存: 上限 256MB │ │ │ │
│ │ │ │ • CPU: 上限 1 核 │ │ │ │
│ │ │ │ • 超时: 30 秒自动终止 │ │ │ │
│ │ │ └─────────────────────────────────────┘ │ │ │
│ │ └───────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘图 7-5-2:沙箱隔离架构
安全代码执行沙箱实现
以下代码实现了两种沙箱方案:基于 subprocess 的轻量级沙箱和基于 Docker 的完全隔离沙箱。
import subprocess # 子进程管理,用于隔离执行
import tempfile # 临时文件管理
import os # 操作系统接口,用于文件清理
import signal # 信号处理,用于超时控制
import resource # 资源限制,用于内存配额
from typing import Dict, Any, Optional
from dataclasses import dataclass
import json # JSON 解析,用于沙箱与外部通信
@dataclass
class SandboxConfig:
"""沙箱配置——定义沙箱的各项限制参数"""
max_memory_mb: int = 256 # 最大内存使用量(MB)
max_cpu_time: float = 30.0 # 最大 CPU 时间(秒)
max_output_size: int = 1024 * 1024 # 最大输出大小(1MB),防止输出爆炸
allowed_modules: list = None # Python 模块白名单(None 表示使用默认)
network_enabled: bool = False # 是否允许网络访问(默认禁用)
filesystem_readonly: bool = True # 文件系统是否只读(默认只读)
class SafeCodeExecutor:
"""安全代码执行沙箱——轻量级方案"""
def __init__(self, config: SandboxConfig = None):
# 如果未提供配置,使用默认配置
self.config = config or SandboxConfig()
def execute_python(self, code: str, timeout: float = 30.0) -> Dict[str, Any]:
"""在受限环境中执行 Python 代码"""
# ========== 第一步:代码静态检查 ==========
# 定义危险模式黑名单——这些模式可能导致系统安全问题
# 注意:黑名单方案并不完美,但可以作为第一道防线
dangerous_patterns = [
"import os", # os 模块可执行系统命令
"import subprocess", # subprocess 可启动任意进程
"import sys", # sys 可修改 Python 运行时
"import shutil", # shutil 可执行文件系统操作
"__import__", # 动态导入,可绕过静态检查
"eval(", # eval 可执行任意表达式
"exec(", # exec 可执行任意代码
"open(", # open 可读写文件
"compile(", # compile 可编译代码对象
"globals()", # globals 可访问全局命名空间
"locals()", # locals 可访问局部命名空间
]
# 逐一检查代码是否包含危险模式
for pattern in dangerous_patterns:
if pattern in code:
return {
"success": False,
"error": f"安全限制: 代码包含禁止的模式 '{pattern}'"
}
# ========== 第二步:创建安全包装代码 ==========
# 将用户代码包装在安全框架中,添加超时和内存限制
with tempfile.NamedTemporaryFile(
mode='w', suffix='.py', delete=False
) as f:
# 构造安全包装代码
safe_code = f"""
import sys
import signal
import json
# 超时处理——设置闹钟,超时后自动退出
def timeout_handler(signum, frame):
print(json.dumps({{"error": "执行超时"}}))
sys.exit(1)
# 注册 SIGALRM 信号处理函数
signal.signal(signal.SIGALRM, timeout_handler)
# 设置超时闹钟,超时后触发 timeout_handler
signal.alarm({int(timeout)})
# 限制内存使用——防止内存爆炸导致系统 OOM
try:
import resource
# 设置虚拟内存上限(RLIMIT_AS),超出则触发 MemoryError
resource.setrlimit(resource.RLIMIT_AS, ({self.config.max_memory_mb} * 1024 * 1024, -1))
except:
pass # 某些平台不支持 setrlimit,静默失败
# 执行用户代码,捕获所有异常
try:
{chr(10).join(' ' + line for line in code.split(chr(10)))}
result = None
# 输出 JSON 格式结果,便于外部解析
print(json.dumps({{"success": True, "result": str(result)}}))
except Exception as e:
print(json.dumps({{"success": False, "error": str(e)}}))
"""
f.write(safe_code) # 写入包装后的代码
temp_path = f.name # 记录临时文件路径
try:
# ========== 第三步:在子进程中执行 ==========
# 使用 subprocess 启动独立 Python 进程
# 即使子进程崩溃,也不会影响主进程
result = subprocess.run(
["python3", temp_path], # 执行临时文件
capture_output=True, # 捕获 stdout 和 stderr
timeout=timeout + 5, # 外层超时(比内层多 5 秒作为缓冲)
text=True, # 输出为文本而非字节
env={"PATH": "/usr/bin:/bin"} # 最小化环境变量,移除敏感信息
)
# 检查子进程退出码
if result.returncode != 0:
return {"success": False, "error": result.stderr.strip()}
# 检查输出大小——防止输出爆炸
if len(result.stdout) > self.config.max_output_size:
return {"success": False, "error": "输出超出大小限制"}
# 解析 JSON 输出
return json.loads(result.stdout.strip())
except subprocess.TimeoutExpired:
# 子进程超时被强制终止
return {"success": False, "error": "执行超时"}
finally:
# 无论成功或失败,都清理临时文件
os.unlink(temp_path)
def execute_shell(self, command: str, timeout: float = 10.0) -> Dict[str, Any]:
"""在受限子进程中执行 Shell 命令"""
# ========== 命令白名单检查 ==========
# 只允许执行预定义的安全命令
allowed_commands = ["ls", "cat", "echo", "head", "tail", "wc", "grep", "find"]
cmd_parts = command.strip().split()
# 检查命令是否在白名单中
if not cmd_parts or cmd_parts[0] not in allowed_commands:
return {
"success": False,
"error": f"不允许的命令: {cmd_parts[0] if cmd_parts else '空'}。"
f"允许: {', '.join(allowed_commands)}"
}
try:
# 执行命令,工作目录限制在 /tmp
result = subprocess.run(
command,
shell=True, # 通过 shell 执行
capture_output=True, # 捕获输出
timeout=timeout, # 超时限制
text=True, # 文本输出
cwd="/tmp", # 限制工作目录为 /tmp
env={"PATH": "/usr/bin:/bin", "HOME": "/tmp"} # 最小化环境变量
)
return {
"success": True,
"stdout": result.stdout[:10000], # 限制输出长度(10000 字符)
"stderr": result.stderr[:1000], # 限制错误输出长度
"returncode": result.returncode
}
except subprocess.TimeoutExpired:
return {"success": False, "error": "命令执行超时"}
# ========== 基于 Docker 的完全隔离沙箱(生产推荐) ==========
class DockerSandbox:
"""基于 Docker 的完全隔离沙箱——最强的隔离方案"""
def __init__(self, image: str = "python:3.11-slim"):
# 使用精简版 Python 镜像,减少攻击面
self.image = image
def execute(self, code: str, timeout: int = 30) -> Dict[str, Any]:
"""在 Docker 容器中执行代码——完全隔离的执行环境"""
# 构造 Docker 运行命令,每一项都是一个安全限制
docker_cmd = [
"docker", "run", "--rm", # 运行后自动删除容器
"--memory", "256m", # 内存限制:256MB
"--cpus", "1", # CPU 限制:1 核
"--network", "none", # 完全禁用网络——最关键的安全限制
"--read-only", # 文件系统只读——防止写入恶意文件
"--tmpfs", "/tmp:size=100m", # 仅 /tmp 可写,限制 100MB
"--timeout", str(timeout), # 容器超时自动终止
self.image, # 使用的 Docker 镜像
"python3", "-c", code # 在容器内执行 Python 代码
]
try:
result = subprocess.run(
docker_cmd,
capture_output=True,
timeout=timeout + 10, # 外层超时比容器超时多 10 秒
text=True
)
return {
"success": result.returncode == 0,
"stdout": result.stdout[:50000], # 限制输出大小
"stderr": result.stderr[:5000]
}
except subprocess.TimeoutExpired:
return {"success": False, "error": "Docker 执行超时"}
except FileNotFoundError:
return {"success": False, "error": "Docker 未安装或不可用"}两种沙箱方案的对比:
| 维度 | subprocess 沙箱 | Docker 沙箱 |
|---|---|---|
| 隔离强度 | 中等(进程级) | 高(容器级) |
| 启动开销 | 低(毫秒级) | 较高(秒级) |
| 网络隔离 | 依赖环境配置 | 原生支持 |
| 文件系统隔离 | 依赖路径检查 | 原生只读挂载 |
| 适用场景 | 开发测试、轻量任务 | 生产环境、不可信代码 |
表 7-5-2:两种沙箱方案对比
7.5.4 Human-in-the-Loop 确认机制——"主管签字"制度
设计原则
在门禁系统中,某些区域(如金库、数据中心核心机房)需要额外的授权流程——即使持卡人有基础通行权限,也需要主管现场签字或远程确认才能进入。Agent 系统中的 Human-in-the-Loop(HITL)机制与此异曲同工。
对于高风险操作(删除数据、发送邮件、执行支付、修改系统配置),在工具执行前引入人工确认,可以有效防止 Agent 被误导后的灾难性后果。这是纵深防御体系的"最后一道闸门"。
HITL 机制的核心设计原则是风险分级:不是所有操作都需要人工确认,否则系统将完全失去自动化价值。操作的风险等级决定是否需要人工介入以及需要何种级别的确认。
┌─────────────────────────────────────────────────────────┐
│ Human-in-the-Loop 工作流 │
├─────────────────────────────────────────────────────────┤
│ │
│ LLM 决定调用工具 │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ 风险评估 │ │
│ │ (风险等级?) │ │
│ └──────┬───────┘ │
│ │ │
│ ┌────┴────┐ │
│ ▼ ▼ │
│ 低风险 高风险 │
│ │ │ │
│ │ ▼ │
│ │ ┌──────────────┐ │
│ │ │ 等待人工审批 │ │
│ │ │ (确认/拒绝) │ │
│ │ └──────┬───────┘ │
│ │ │ │
│ │ ┌────┴────┐ │
│ │ ▼ ▼ │
│ │ 确认 拒绝 │
│ │ │ │ │
│ └─────┴─────────┘ │
│ │ │ │
│ ▼ ▼ │
│ 执行工具 取消操作 │
│ │
└─────────────────────────────────────────────────────────┘图 7-5-3:Human-in-the-Loop 决策流程
确认机制实现
from enum import Enum # 枚举类型,用于定义风险等级
from typing import Callable, Optional
import asyncio # 异步 IO,支持等待外部审批回调
class RiskLevel(Enum):
"""风险等级定义——对应门禁系统中的不同审批要求"""
LOW = "low" # 低风险:自动执行,无需确认(如读取文件)
MEDIUM = "medium" # 中风险:记录日志后自动执行(如修改配置)
HIGH = "high" # 高风险:需要人工确认(如删除单条记录)
CRITICAL = "critical" # 严重风险:需要多重确认(如批量删除)
class ApprovalRequest:
"""审批请求——一次高风险操作的审批单"""
def __init__(self, tool_name: str, arguments: dict,
risk_level: RiskLevel, reason: str):
self.tool_name = tool_name # 被请求执行的工具名
self.arguments = arguments # 工具参数
self.risk_level = risk_level # 风险等级
self.reason = reason # 风险评估原因说明
self.status = "pending" # 审批状态:pending / approved / rejected
self.approver = None # 审批人(确认后填充)
class ApprovalManager:
"""审批管理器——门禁系统中的审批中心"""
def __init__(self, approval_callback: Optional[Callable] = None):
# 待处理审批请求字典
self._pending_approvals: Dict[str, ApprovalRequest] = {}
# 审批回调函数(实际系统中可能是 WebSocket 通知或消息队列)
self._approval_callback = approval_callback
# 风险等级与处理策略的映射
self._risk_thresholds = {
RiskLevel.LOW: "auto", # 自动执行
RiskLevel.MEDIUM: "log", # 记录日志后自动执行
RiskLevel.HIGH: "approve", # 需要人工审批
RiskLevel.CRITICAL: "multi_approve", # 需要多重审批
}
def assess_risk(self, tool_name: str, arguments: dict) -> tuple[RiskLevel, str]:
"""评估操作风险等级——自动分析工具调用是否包含高危特征"""
# 定义高风险关键词——这些关键词出现在参数中时提升风险等级
destructive_keywords = ["delete", "drop", "truncate", "remove", "erase"]
sensitive_keywords = ["password", "token", "secret", "key", "credit"]
financial_keywords = ["pay", "charge", "refund", "transfer", "order"]
# 将参数转换为小写字符串进行关键词匹配
arg_str = str(arguments).lower()
# 检查破坏性操作——如果是"全部删除"类操作,升级为 CRITICAL
if any(kw in arg_str for kw in destructive_keywords):
# 检查是否包含"全部"类关键词——大规模操作风险更高
if any(kw in arg_str for kw in ["all", "everything", "cascade"]):
return RiskLevel.CRITICAL, "大规模删除操作"
return RiskLevel.HIGH, "包含删除操作"
# 检查金融操作——涉及资金流动的操作属于高风险
if any(kw in arg_str for kw in financial_keywords):
return RiskLevel.HIGH, "包含金融操作"
# 检查敏感数据操作——涉及密码、密钥等属于中风险
if any(kw in arg_str for kw in sensitive_keywords):
return RiskLevel.MEDIUM, "涉及敏感数据"
# 常规操作——低风险
return RiskLevel.LOW, "常规操作"
async def request_approval(self, tool_name: str, arguments: dict) -> bool:
"""请求审批——根据风险等级决定是否需要人工确认"""
# 第一步:评估风险等级
risk, reason = self.assess_risk(tool_name, arguments)
# 第二步:低风险和中风险自动通过
if risk in [RiskLevel.LOW, RiskLevel.MEDIUM]:
if risk == RiskLevel.MEDIUM:
# 中风险操作记录日志,但不阻塞执行
print(f"⚠️ [自动执行-日志记录] {tool_name}: {reason}")
return True
# 第三步:高风险操作需要人工确认
request = ApprovalRequest(tool_name, arguments, risk, reason)
request_id = f"{tool_name}_{id(request)}"
self._pending_approvals[request_id] = request
# 打印审批提示信息
print(f"""
🔴 需要人工审批
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
操作: {tool_name}
风险等级: {risk.value.upper()}
原因: {reason}
参数: {arguments}
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
""")
# 第四步:调用审批回调(实际系统中通过 WebSocket 或消息队列通知审批人)
if self._approval_callback:
approved = await self._approval_callback(request)
# 更新审批状态
request.status = "approved" if approved else "rejected"
return approved
# 模拟审批通过(实际系统中应阻塞等待外部审批结果)
print("等待审批... (y/n): ", end="")
return True # 演示用途,实际应等待真实审批
def approve(self, request_id: str, approver: str):
"""批准请求——审批人确认操作"""
if request_id in self._pending_approvals:
self._pending_approvals[request_id].status = "approved"
self._pending_approvals[request_id].approver = approver
def reject(self, request_id: str, approver: str, reason: str = ""):
"""拒绝请求--审批人拒绝操作"""
if request_id in self._pending_approvals:
self._pending_approvals[request_id].status = "rejected"
self._pending_approvals[request_id].approver = approver集成到 Agent 工具调用
将权限控制、审批机制和参数校验组合起来,形成一个完整的安全 Agent。这就像将门禁系统中的身份验证、区域通行、行为约束和主管审批整合为一套完整流程。
class SecureAgent:
"""带安全控制的 Agent——门禁系统的完整实现"""
def __init__(self, llm, tools: dict, approval_manager: ApprovalManager):
self.llm = llm # LLM 模型实例
self.tools = tools # 工具字典: name -> callable
self.approval = approval_manager # 审批管理器
self.permission = PermissionManager() # 权限管理器
async def execute_tool(self, user_id: str, tool_name: str,
arguments: dict) -> dict:
"""安全执行工具——完整的五步安全流程"""
# 第一步:权限检查——用户是否有权调用此工具
if not self.permission.check_permission(user_id, tool_name):
return {"error": "权限不足", "tool": tool_name}
# 第二步:速率限制——防止频繁调用
if not self.permission.check_rate_limit(user_id, tool_name):
return {"error": "调用频率超限", "tool": tool_name}
# 第三步:风险评估与审批——高风险操作需要人工确认
approved = await self.approval.request_approval(tool_name, arguments)
if not approved:
return {"error": "操作被拒绝", "reason": "人工审批未通过"}
# 第四步:参数校验——清理和验证输入参数
validated_args = self._validate_arguments(tool_name, arguments)
# 第五步:执行工具——记录调用并执行
self.permission.record_call(user_id, tool_name)
try:
result = await self.tools[tool_name](**validated_args)
return {"success": True, "result": result}
except Exception as e:
return {"success": False, "error": str(e)}
def _validate_arguments(self, tool_name: str, arguments: dict) -> dict:
"""参数校验与清理——防止参数注入和滥用"""
safe_args = {}
for key, value in arguments.items():
# 字符串长度限制——防止超长输入导致内存问题或缓冲区溢出
if isinstance(value, str) and len(value) > 10000:
raise ValueError(f"参数 {key} 过长")
# 数值范围限制——防止超大数值导致计算异常
if isinstance(value, (int, float)):
if abs(value) > 1e9:
raise ValueError(f"参数 {key} 数值过大")
safe_args[key] = value
return safe_args7.5.5 参数安全校验——"安检仪"的最后一道扫描
在前面的门禁类比中,参数校验对应的是"进入区域后对具体行为的约束"。更贴切的类比是机场安检仪:即使旅客持有合法登机牌(权限通过),行李中的物品(参数内容)仍然需要经过 X 光机扫描,确保没有违禁品。
Agent 工具的参数来自 LLM 的输出,而 LLM 的输出受到用户输入的间接影响。如果没有严格的参数校验,攻击者可以通过精心构造的提示,诱导 LLM 生成包含恶意内容的工具参数。
常见的参数攻击手段包括:
- 路径遍历:参数值为
../../../etc/passwd,试图访问非授权文件 - SQL 注入:参数值为
1; DROP TABLE users; --,试图执行恶意 SQL - 命令注入:参数值为
; rm -rf /,试图在 Shell 命令中注入恶意操作 - 超长输入:发送超长字符串,试图耗尽内存或触发缓冲区溢出
以下是一个完整的参数校验框架实现:
import re # 正则表达式,用于模式匹配
from typing import Any, Dict, Optional
from dataclasses import dataclass
from enum import Enum
class ParamType(Enum):
"""参数类型定义——每种类型对应不同的校验规则"""
STRING = "string" # 普通字符串
INTEGER = "integer" # 整数
FLOAT = "number" # 浮点数
BOOLEAN = "boolean" # 布尔值
EMAIL = "email" # 邮箱地址
URL = "url" # URL 地址
FILEPATH = "filepath" # 文件路径
SQL_IDENTIFIER = "sql_identifier" # SQL 标识符(表名、列名等)
@dataclass
class ParamConstraint:
"""参数约束——定义每个参数的校验规则"""
type: ParamType # 参数类型
min_length: Optional[int] = None # 字符串最小长度
max_length: Optional[int] = None # 字符串最大长度
min_value: Optional[float] = None # 数值最小值
max_value: Optional[float] = None # 数值最大值
pattern: Optional[str] = None # 自定义正则表达式
allowed_values: Optional[list] = None # 白名单(只允许特定值)
sanitize: bool = True # 是否自动清理危险字符
class ParameterValidator:
"""参数安全校验器——门禁系统中的安检仪"""
# 安全正则模式——每种参数类型的合法格式
PATTERNS = {
ParamType.EMAIL: r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$',
ParamType.URL: r'^https?://[a-zA-Z0-9.-]+(?:\.[a-zA-Z]{2,})+(?:/[\w\-./?%&=]*)?$',
ParamType.FILEPATH: r'^[\w\-./]+$', # 不允许 .., ~, $ 等
ParamType.SQL_IDENTIFIER: r'^[a-zA-Z_][a-zA-Z0-9_]*$', # 只允许字母、数字、下划线
}
# 危险字符——根据参数类型清理不同类别的危险字符
DANGEROUS_CHARS = {
ParamType.STRING: ['\x00', '\r', '\n'], # 控制字符
ParamType.FILEPATH: ['..', '~', '$', '`', '|', ';'], # 路径遍历和命令注入
ParamType.SQL_IDENTIFIER: [';', '--', '/*', '*/', 'DROP', 'DELETE'], # SQL 注入
}
def validate(self, value: Any,
constraint: ParamConstraint) -> tuple[bool, str, Any]:
"""校验并清理参数值——安检仪的核心扫描逻辑"""
# ========== 第一步:类型检查 ==========
# 定义每种类型的检查函数
type_checks = {
ParamType.STRING: lambda v: isinstance(v, str),
ParamType.INTEGER: lambda v: isinstance(v, int) and not isinstance(v, bool),
ParamType.FLOAT: lambda v: isinstance(v, (int, float)) and not isinstance(v, bool),
ParamType.BOOLEAN: lambda v: isinstance(v, bool),
}
# 执行类型检查
if constraint.type in type_checks:
if not type_checks[constraint.type](value):
return False, f"类型错误: 期望 {constraint.type.value}", None
# ========== 第二步:危险字符清理 ==========
if constraint.sanitize and isinstance(value, str):
value = self._sanitize(value, constraint.type)
# ========== 第三步:长度检查 ==========
if constraint.type == ParamType.STRING:
if constraint.min_length and len(value) < constraint.min_length:
return False, f"长度不足: 最小 {constraint.min_length}", None
if constraint.max_length and len(value) > constraint.max_length:
return False, f"长度超限: 最大 {constraint.max_length}", None
# ========== 第四步:数值范围检查 ==========
if isinstance(value, (int, float)):
if constraint.min_value is not None and value < constraint.min_value:
return False, f"值过小: 最小 {constraint.min_value}", None
if constraint.max_value is not None and value > constraint.max_value:
return False, f"值过大: 最大 {constraint.max_value}", None
# ========== 第五步:正则匹配 ==========
# 自定义正则优先
if constraint.pattern:
if not re.match(constraint.pattern, str(value)):
return False, f"格式不匹配: {constraint.pattern}", None
# 检查预定义类型模式
if constraint.type in self.PATTERNS:
if not re.match(self.PATTERNS[constraint.type], str(value)):
return False, f"不符合 {constraint.type.value} 格式", None
# ========== 第六步:白名单检查 ==========
if constraint.allowed_values and value not in constraint.allowed_values:
return False, f"不在允许值列表中: {constraint.allowed_values}", None
# 所有检查通过
return True, "OK", value
def _sanitize(self, value: str, param_type: ParamType) -> str:
"""清理危险字符——移除特定类型参数中的危险字符"""
dangerous = self.DANGEROUS_CHARS.get(param_type, [])
for char in dangerous:
value = value.replace(char, '') # 逐个替换危险字符
return value.strip() # 去除首尾空白
# ========== 测试参数校验 ==========
validator = ParameterValidator()
# 定义测试用例——包含正常和攻击场景
test_cases = [
# 正常邮箱
("hello@example.com", ParamConstraint(type=ParamType.EMAIL)),
# 非法邮箱格式
("not-an-email", ParamConstraint(type=ParamType.EMAIL)),
# 路径遍历攻击——尝试访问 /etc/passwd
("../../../etc/passwd", ParamConstraint(type=ParamType.FILEPATH)),
# 数值超出范围
(150, ParamConstraint(type=ParamType.INTEGER, min_value=0, max_value=100)),
# SQL 注入攻击
("DROP TABLE users", ParamConstraint(type=ParamType.SQL_IDENTIFIER)),
]
# 执行测试
for value, constraint in test_cases:
ok, msg, cleaned = validator.validate(value, constraint)
status = "✅" if ok else "❌"
print(f"{status} {value} -> {msg} | 清理后: {cleaned}")运行上述测试,输出如下:
✅ hello@example.com -> OK | 清理后: hello@example.com
❌ not-an-email -> 不符合 email 格式 | 清理后: None
❌ ../../../etc/passwd -> 不符合 filepath 格式 | 清理后: None
❌ 150 -> 值过大: 最大 100 | 清理后: None
❌ DROP TABLE users -> 不符合 sql_identifier 格式 | 清理后: None可以看到,路径遍历攻击和 SQL 注入攻击都在参数校验阶段被成功拦截。
7.5.6 审计日志——门禁系统的"监控录像"
门禁系统的最后一个关键组件是监控录像——即使所有防护层都被绕过或出现遗漏,事后仍需要通过录像追溯发生了什么、由谁发起、造成了什么影响。在 Agent 系统中,这就是审计日志。
审计日志的设计要点:
- 完整性:记录所有工具调用请求,无论成功还是失败
- 不可篡改:日志应存储在只可追加(append-only)的存储中
- 可追溯:包含足够的信息用于事后分析(时间、用户、工具、参数、结果)
- 实时告警:对异常行为(如短时间内大量调用高风险工具)实时告警
import hashlib # 哈希算法,用于生成请求唯一 ID
from datetime import datetime # 时间戳
class AuditLogger:
"""审计日志记录器——门禁系统的监控录像"""
def __init__(self):
self._logs = [] # 日志存储(生产环境应使用持久化存储)
def _generate_request_id(self, user_id: str, tool_name: str,
arguments: dict) -> str:
"""生成请求唯一 ID——便于跨系统追踪"""
raw = f"{user_id}:{tool_name}:{str(arguments)}"
return hashlib.md5(raw.encode()).hexdigest()[:8]
def log(self, user_id: str, tool_name: str, arguments: dict,
action: str, detail: str):
"""记录一条审计日志"""
request_id = self._generate_request_id(user_id, tool_name, arguments)
entry = {
"timestamp": datetime.now().isoformat(), # ISO 格式时间戳
"request_id": request_id, # 请求唯一标识
"user_id": user_id, # 操作用户
"tool_name": tool_name, # 调用的工具
"arguments": arguments, # 工具参数
"action": action, # 动作类型
"detail": detail, # 详细信息
}
self._logs.append(entry)
print(f"[{entry['timestamp']}] {action}: {detail}")
def get_logs(self, user_id: str = None,
tool_name: str = None) -> list:
"""查询日志——支持按用户和工具过滤"""
results = self._logs
if user_id:
results = [l for l in results if l["user_id"] == user_id]
if tool_name:
results = [l for l in results if l["tool_name"] == tool_name]
return results
# ========== 完整的安全工具执行框架 ==========
# 集成权限、审批、沙箱和审计日志
class SecureToolExecutor:
"""安全工具执行框架——门禁系统的完整实现"""
def __init__(self):
self.permission_manager = PermissionManager() # 权限管理
self.approval_manager = ApprovalManager() # 审批管理
self.sandbox = SafeCodeExecutor() # 代码执行沙箱
self.audit_log = AuditLogger() # 审计日志
def configure_tool(self, name: str, permission: Permission,
risk_triggers: list = None):
"""配置工具安全策略——为每个工具设定权限要求"""
self.permission_manager.register_tool_permission(ToolPermission(
tool_name=name,
required_permission=permission,
requires_approval=(permission in [Permission.DELETE, Permission.ADMIN])
))
async def execute(self, user_id: str, tool_name: str,
arguments: dict) -> dict:
"""安全执行入口——所有工具调用都经过此方法"""
# 生成请求 ID,用于全链路追踪
request_id = self.audit_log._generate_request_id(
user_id, tool_name, arguments
)
try:
# Step 1: 权限检查——用户是否有权调用此工具
if not self.permission_manager.check_permission(user_id, tool_name):
self.audit_log.log(user_id, tool_name, arguments, "DENIED", "权限不足")
return {"error": "权限不足"}
# Step 2: 审批——高风险操作需要人工确认
approved = await self.approval_manager.request_approval(
tool_name, arguments
)
if not approved:
self.audit_log.log(user_id, tool_name, arguments, "REJECTED", "审批未通过")
return {"error": "操作被拒绝"}
# Step 3: 执行——带超时保护
self.audit_log.log(user_id, tool_name, arguments, "EXECUTING",
f"执行 {tool_name}")
result = await asyncio.wait_for(
self._execute_tool(tool_name, arguments),
timeout=30.0
)
# 记录成功日志
self.audit_log.log(user_id, tool_name, arguments, "SUCCESS",
str(result)[:100])
return {"success": True, "result": result}
except asyncio.TimeoutError:
# 超时被记录为 TIMEOUT
self.audit_log.log(user_id, tool_name, arguments, "TIMEOUT", "执行超时")
return {"error": "执行超时"}
except Exception as e:
# 其他异常被记录为 ERROR
self.audit_log.log(user_id, tool_name, arguments, "ERROR", str(e))
return {"error": str(e)}
async def _execute_tool(self, tool_name: str, args: dict) -> Any:
"""实际工具执行——根据工具类型选择执行方式"""
# 代码执行类工具使用沙箱隔离
if tool_name == "execute_code":
return self.sandbox.execute_python(args.get("code", ""))
# 其他工具直接执行
return {"tool": tool_name, "args": args, "status": "ok"}
# ========== 端到端测试 ==========
async def test_security():
executor = SecureToolExecutor()
# 配置工具安全策略
executor.configure_tool("read_file", Permission.READ)
executor.configure_tool("delete_file", Permission.DELETE)
executor.configure_tool("execute_code", Permission.EXECUTE)
# 设置用户角色
executor.permission_manager.set_user_role("alice", Role.EDITOR)
executor.permission_manager.set_user_role("bob", Role.VIEWER)
# 测试1: 正常读取(应通过)
print("\n=== 测试1: 正常读取 ===")
result = await executor.execute("alice", "read_file", {"path": "/tmp/test.txt"})
print(f"结果: {result}")
# 测试2: 权限不足(应拒绝)
print("\n=== 测试2: 权限不足 ===")
result = await executor.execute("bob", "delete_file", {"path": "/tmp/test.txt"})
print(f"结果: {result}")
# 测试3: 高风险操作(需审批)
print("\n=== 测试3: 高风险操作 ===")
result = await executor.execute("alice", "delete_file", {"path": "/tmp/test.txt"})
print(f"结果: {result}")
asyncio.run(test_security())7.5.7 常见误区
在实际项目中,Agent 安全控制常存在以下误区:
误区一:"LLM 本身会注意安全,不需要额外控制"
这是最常见的误解。LLM 确实经过安全对齐训练,但安全对齐是"尽力而为"的,不是"保证"。在提示注入攻击下,LLM 可能被诱导执行本不该执行的操作。安全控制必须在代码层面而非 LLM 层面实现——代码层的检查是确定性的,LLM 的判断是概率性的。
误区二:"只做黑名单过滤就够了"
黑名单(如禁止 import os)只能拦截已知的危险模式,无法应对未知的攻击方式。例如,攻击者可以用 __builtins__.__dict__['os'] 绕过对 import os 的检查。正确的做法是白名单优先、黑名单兜底:默认拒绝所有,只允许已知安全的操作。
误区三:"有了权限控制就不需要沙箱"
权限控制解决的是"谁能做什么"的问题,沙箱解决的是"做了之后的影响范围"问题。即使用户有合法权限执行代码,执行代码时的副作用(如内存泄漏、死循环、文件读写)仍需要沙箱来限制。两者是互补关系,不能互相替代。
误区四:"人工审批会严重降低系统效率"
这种担忧基于一个错误假设——所有操作都需要审批。实际上,风险分级机制确保只有高风险操作才需要人工确认。在典型系统中,90% 以上的操作是低风险的(读取、查询),只有少数破坏性或金融操作才需要审批。合理设计的审批机制对整体效率影响极小。
误区五:"审计日志只是为了合规检查"
审计日志不仅是合规要求,更是安全运营的核心工具。通过分析审计日志,可以发现异常行为模式(如某用户突然开始大量调用删除工具)、追踪安全事件的根因、度量安全策略的效果。没有审计日志的安全系统就像没有监控录像的大楼——出了事无法追溯。
误区六:"沙箱只需在开发环境使用"
生产环境恰恰是最需要沙箱的地方。开发环境中的代码通常是受控的,而生产环境中 Agent 可能处理来自真实用户的不受控输入。沙箱的启动开销(秒级)相对于安全风险而言是值得付出的代价。
7.5.8 本节小结
本节以"门禁系统"为核心类比,系统讲解了 Agent 工具调用中的权限与安全控制。核心内容回顾如下:
| 安全层级 | 门禁类比 | 核心机制 | 关键原则 |
|---|---|---|---|
| 用户认证 | 入口刷卡 | OAuth / JWT / API Key | 确认"你是谁" |
| 角色授权 (RBAC) | 区域通行权 | 角色-权限映射,位运算检查 | 最小权限原则 |
| 参数校验 | 安检仪扫描 | 类型检查、正则匹配、白名单 | 白名单优先,黑名单兜底 |
| 沙箱隔离 | 防爆墙 | 进程隔离、资源限制、网络禁用 | 纵深防御,不信任任何代码 |
| 人工审批 | 主管签字 | 风险分级、HITL 确认 | 只对高风险操作介入 |
| 审计日志 | 监控录像 | 全链路记录、可追溯 | 记录一切,事后可查 |
表 7-5-3:安全体系总览
关键要点总结:
纵深防御是核心思想:没有任何单一安全措施是充分的。权限控制、参数校验、沙箱隔离、人工审批和审计日志层层叠加,形成纵深防御体系。
最小权限原则:每个角色只拥有完成其任务所需的最低权限。宁可因权限不足导致功能受限,也不可因权限过大导致安全风险。
默认拒绝:未注册的工具默认拒绝,未通过校验的参数默认拒绝。这是安全系统设计的基本原则——宁可误杀,不可放过。
风险分级:安全控制不是非黑即白的。通过风险分级,在安全性和自动化效率之间取得平衡——低风险自动通过,高风险人工确认。
沙箱是执行不可信代码的底线:当代码来源不可控时(如用户提供的代码、LLM 生成的代码),必须在沙箱中执行。Docker 容器隔离是生产环境推荐方案。
审计日志是最后一道防线:即使所有防护都失效,审计日志仍能提供事后追溯能力。没有审计日志的系统无法回答"发生了什么"的问题。
在下一节(7.6 结果解析)中,我们将讨论工具执行后的下一步——如何解析和处理工具返回的结果。Agent 拿到工具执行结果后,需要判断执行是否成功、提取关键信息、决定后续行动(是否需要调用其他工具、是否可以直接回答用户)。结果解析的质量直接影响 Agent 的整体表现,是工具调用闭环中的关键一环。