Skip to content

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 权限控制的设计思路,最贴切的类比就是现实世界中的门禁系统。一栋现代化办公大楼的门禁系统通常包含以下几个层次:

  1. 身份认证(Authentication)——"你是谁?":访客在大楼入口刷卡或刷脸,系统验证其身份。这对应 Agent 中的用户认证层(OAuth / JWT / API Key)。

  2. 角色授权(Authorization)——"你能进哪些区域?":即使通过了身份认证,访客也只能进入与其角色匹配的区域。普通员工不能进机房,访客不能进办公区。这对应 Agent 中的 RBAC(基于角色的访问控制)。

  3. 操作约束(Parameter Validation)——"你能在这里做什么?":即使进入了某个房间,也不是所有操作都被允许。会议室可以开会但不能做饭,工具间可以使用工具但不能带走设备。这对应 Agent 中的参数校验层。

  4. 执行监控(Execution Control)——"你的行为是否在可控范围内?":大楼内有监控摄像头和保安巡逻,限制每个区域的人数和停留时间。这对应 Agent 中的速率限制、超时控制和资源配额。

  5. 特殊审批(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 权限控制系统。代码中每一行都配有详细注释,帮助理解每个设计决策背后的考量。

python
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 —— 管理员拥有全部权限

上述代码的核心设计要点:

  1. 使用 Flag 而非普通 EnumFlag 支持位运算,可以用 | 组合权限,用 & 检查权限,非常适合权限系统的"包含"语义。
  2. 权限不直接绑定用户:通过角色间接分配权限,当人员变动时只需调整角色,不需要逐个修改权限——这正是 RBAC 的精髓。
  3. 未注册工具默认拒绝check_permission 中对未注册工具返回 False,遵循"默认拒绝"的安全原则。
  4. 装饰器解耦require_permission 装饰器将权限检查逻辑与业务逻辑分离,工具开发者只需关注功能实现。

7.5.3 沙箱隔离——为危险操作建一道"防爆墙"

为什么需要沙箱

回到门禁系统的类比:权限控制解决了"谁能进哪个门"的问题,但即便一个人有合法权限进入了某个房间,他在房间里的行为也需要被约束。如果一个操作员有权限执行代码,但他执行的代码包含无限循环或恶意命令怎么办?

沙箱(Sandbox)就是为了解决这个问题而存在的。它就像化工实验室里的"防爆墙"——即使内部发生爆炸,也不会波及外部。在 Agent 系统中,沙箱通过隔离执行环境,确保即使工具代码出现异常或恶意行为,也不会影响宿主系统。

沙箱隔离的核心目标包括:

  • 网络隔离:限制网络访问,防止数据外泄或拉取恶意载荷
  • 文件系统隔离:限制可访问的路径,防止读写敏感文件
  • 资源限制:控制 CPU、内存、磁盘使用,防止资源耗尽
  • 执行超时:防止死循环或长时间阻塞
  • 权限降级:以最低权限用户运行,减少攻击面
沙箱架构设计
┌─────────────────────────────────────────────────────────┐
│                    沙箱隔离架构                           │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  ┌─────────────────────────────────────────────────┐   │
│  │                  宿主机 (Host)                    │   │
│  │  ┌───────────────────────────────────────────┐  │   │
│  │  │              沙箱容器 (Container)           │  │   │
│  │  │  ┌─────────────────────────────────────┐  │  │   │
│  │  │  │        受限执行环境                    │  │  │   │
│  │  │  │  • 网络: 仅允许白名单域名              │  │  │   │
│  │  │  │  • 文件: 只读挂载,限制路径            │  │  │   │
│  │  │  │  • 内存: 上限 256MB                   │  │  │   │
│  │  │  │  • CPU: 上限 1 核                     │  │  │   │
│  │  │  │  • 超时: 30 秒自动终止                 │  │  │   │
│  │  │  └─────────────────────────────────────┘  │  │   │
│  │  └───────────────────────────────────────────┘  │   │
│  └─────────────────────────────────────────────────┘   │
│                                                         │
└─────────────────────────────────────────────────────────┘

图 7-5-2:沙箱隔离架构

安全代码执行沙箱实现

以下代码实现了两种沙箱方案:基于 subprocess 的轻量级沙箱和基于 Docker 的完全隔离沙箱。

python
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 决策流程

确认机制实现
python
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。这就像将门禁系统中的身份验证、区域通行、行为约束和主管审批整合为一套完整流程。

python
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_args

7.5.5 参数安全校验——"安检仪"的最后一道扫描

在前面的门禁类比中,参数校验对应的是"进入区域后对具体行为的约束"。更贴切的类比是机场安检仪:即使旅客持有合法登机牌(权限通过),行李中的物品(参数内容)仍然需要经过 X 光机扫描,确保没有违禁品。

Agent 工具的参数来自 LLM 的输出,而 LLM 的输出受到用户输入的间接影响。如果没有严格的参数校验,攻击者可以通过精心构造的提示,诱导 LLM 生成包含恶意内容的工具参数。

常见的参数攻击手段包括:

  • 路径遍历:参数值为 ../../../etc/passwd,试图访问非授权文件
  • SQL 注入:参数值为 1; DROP TABLE users; --,试图执行恶意 SQL
  • 命令注入:参数值为 ; rm -rf /,试图在 Shell 命令中注入恶意操作
  • 超长输入:发送超长字符串,试图耗尽内存或触发缓冲区溢出

以下是一个完整的参数校验框架实现:

python
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)的存储中
  • 可追溯:包含足够的信息用于事后分析(时间、用户、工具、参数、结果)
  • 实时告警:对异常行为(如短时间内大量调用高风险工具)实时告警
python
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:安全体系总览

关键要点总结:

  1. 纵深防御是核心思想:没有任何单一安全措施是充分的。权限控制、参数校验、沙箱隔离、人工审批和审计日志层层叠加,形成纵深防御体系。

  2. 最小权限原则:每个角色只拥有完成其任务所需的最低权限。宁可因权限不足导致功能受限,也不可因权限过大导致安全风险。

  3. 默认拒绝:未注册的工具默认拒绝,未通过校验的参数默认拒绝。这是安全系统设计的基本原则——宁可误杀,不可放过。

  4. 风险分级:安全控制不是非黑即白的。通过风险分级,在安全性和自动化效率之间取得平衡——低风险自动通过,高风险人工确认。

  5. 沙箱是执行不可信代码的底线:当代码来源不可控时(如用户提供的代码、LLM 生成的代码),必须在沙箱中执行。Docker 容器隔离是生产环境推荐方案。

  6. 审计日志是最后一道防线:即使所有防护都失效,审计日志仍能提供事后追溯能力。没有审计日志的系统无法回答"发生了什么"的问题。

在下一节(7.6 结果解析)中,我们将讨论工具执行后的下一步——如何解析和处理工具返回的结果。Agent 拿到工具执行结果后,需要判断执行是否成功、提取关键信息、决定后续行动(是否需要调用其他工具、是否可以直接回答用户)。结果解析的质量直接影响 Agent 的整体表现,是工具调用闭环中的关键一环。