LLM Tool Use Function Calling 安全沙箱 —— 深度工程实践：从约束解码到进程级隔离

当 LLM 获得操控世界的能力时，安全不再是可选项，而是架构的核心约束。

一、问题的本质：LLM 既是智能体也是攻击面

现代 AI Agent 的核心能力之一是 Tool Use（工具使用）：LLM 通过结构化输出（通常是 JSON）声明"我要调用某工具，参数是某某"，宿主程序解析后执行实际动作（代码执行、API 调用、文件读写、Shell 命令等）。

这个设计带来一个根本性的安全困境：

1. LLM 输出不受传统控制流约束 —— 模型"幻觉"可能生成越权参数
2. 工具能力开放即是攻击面扩大 —— exec 工具可以执行任意命令
3. 工具调用链可能产生未预期组合 —— A 工具的输出成为 B 工具的输入，中间链路可被利用
4. 多租户场景下 Agent 可能泄露其他用户数据 —— 工具本身有权限边界但 LLM 不一定遵守

因此，Tool Use 安全沙箱不是锦上添花，而是生产级 Agent 系统的核心基础设施。

二、Function Calling 协议的形式化理解

OpenAI 2023 年 6 月引入 Function Calling API，其核心数据结构：

{
  "tools": [
    {
      "type": "function",
      "function": {
        "name": "execute_python",
        "description": "执行 Python 代码并返回结果",
        "parameters": {
          "type": "object",
          "properties": {
            "code": { "type": "string", "description": "Python 代码" },
            "timeout": { "type": "integer", "description": "超时秒数", "maximum": 30 }
          },
          "required": ["code"]
        }
      }
    }
  ]
}

2.1 约束解码（Constrained Decoding）

LLM 生成 JSON 时，"原生模式"完全靠模型自由生成。约束解码则通过以下机制确保输出符合 Schema：

# 基于 token 级 mask 的约束解码伪代码
def constrained_generate(model, tokenizer, schema):
    """
    在每一步只允许生成符合 JSON Schema 的 token
    """
    # 1. 将 JSON Schema 编译为有限状态自动机（FSA）
    fsa = schema_to_fsa(schema)  # 使用 outlines 或 lm-format-enforcer 库

    while not is_complete(token_ids):
        logits = model.forward(token_ids)

        # 2. 根据当前 FSA 状态计算合法 token 集合
        allowed_tokens = fsa.get_allowed_tokens(token_ids)

        # 3. 对非法 token 施加 -inf mask
        mask = torch.full_like(logits, float('-inf'))
        mask[:, allowed_tokens] = 0
        logits = logits + mask

        # 4. 采样下一个 token
        next_token = sample(logits)
        token_ids.append(next_token)

    return tokenizer.decode(token_ids)

关键实现库： - outlines：基于状态机的工业级实现 - lm-format-enforcer：支持 transformers 的完整约束解码 - guidance：微软的交互式约束生成框架

2.2 为什么不能只靠 prompt 约束

一个常见误区是"用 prompt 告诉模型只输出合法 JSON"。实测数据：

10% 的失败率在生产环境不可接受 —— 一次非法输出可能导致代码注入、越权访问等严重事故。

三、沙箱架构设计：三层防御纵深

3.1 整体架构图

┌─────────────────────────────────────────────┐
│                AI Agent Runtime              │
│  ┌───────────┐   ┌──────────────────────┐   │
│  │ LLM Core  │──▶│ Tool Call Parser     │   │
│  └───────────┘   │  (Schema Validation) │   │
│                  └──────────┬───────────┘   │
│                             │               │
│                  ┌──────────▼───────────┐   │
│                  │   Policy Engine      │   │
│                  │  (OPA / Cedar / RL)  │   │
│                  └──────────┬───────────┘   │
│                             │               │
│        ┌────────────────────┼───────────┐   │
│        ▼                    ▼           ▼   │
│  ┌──────────┐      ┌──────────┐  ┌──────┐  │
│  │ Process  │      │ Syscall  │  │ Net  │  │
│  │ Sandbox  │      │ Filter   │  │ Isol.│  │
│  └──────────┘      └──────────┘  └──────┘  │
└─────────────────────────────────────────────┘

3.2 第一层：Schema 级输入校验

from pydantic import BaseModel, validator, Field
import ast
import re

class ToolInput(BaseModel):
    """工具输入的强类型校验"""

    @validator('code')
    def no_dangerous_patterns(cls, v):
        """静态分析阻断高危操作"""
        dangerous = [
            'os.system', 'subprocess', '__import__', 'eval(', 'exec(',
            'open(', 'socket', 'importlib', 'compile(', 'execfile'
        ]
        for pattern in dangerous:
            if pattern in v:
                raise ValueError(f"检测到危险模式: {pattern}")
        return v

    @validator('code')
    def ast_syntax_check(cls, v):
        """AST 解析确保是合法 Python 而非注入混淆代码"""
        try:
            tree = ast.parse(v)
            # 进一步检查 AST 节点类型白名单
            allowed_nodes = {
                ast.Expression, ast.BinOp, ast.UnaryOp, ast.BoolOp, ast.Compare,
                ast.Call, ast.Name, ast.Constant, ast.Load, ast.Store,
                ast.List, ast.Tuple, ast.Set, ast.Dict, ast.Subscript,
                ast.Attribute, ast.If, ast.For, ast.While, ast.With,
                ast.FunctionDef, ast.Return, ast.IfExp, ast.Assert,
                ast.Assign, ast.AugAssign, ast.Expr, ast.Module,
                ast.arguments, ast.arg, ast.comprehension, ast.ListComp,
                ast.DictComp, ast.SetComp, ast.GeneratorExp,
            }
            for node in ast.walk(tree):
                if type(node) not in allowed_nodes:
                    raise ValueError(f"不支持的语法结构: {type(node).__name__}")
        except SyntaxError as e:
            raise ValueError(f"语法错误: {e}")
        return v

3.3 第二层：进程级隔离（Namespace + Cgroup）

import os
import ctypes
import resource
import multiprocessing

def create_namespace_sandbox():
    """
    Linux Namespace 隔离：PID/Mount/Network/IPC/UTS 全隔离
    """
    CLONE_NEWPID    = 0x20000000   # PID 隔离
    CLONE_NEWNS     = 0x00020000   # Mount 隔离
    CLONE_NEWNET    = 0x40000000   # Network 隔离
    CLONE_NEWIPC    = 0x08000000   # IPC 隔离
    CLONE_NEWUTS    = 0x04000000   # 主机名隔离

    flags = CLONE_NEWPID | CLONE_NEWNS | CLONE_NEWNET | CLONE_NEWIPC | CLONE_NEWUTS
    libc = ctypes.CDLL("libc.so.6")

    pid = libc.clone(
        ctypes.cast(None, ctypes.c_void_p),  # 函数指针，这里用 None 占位
        None,                                 # 子栈
        flags,
        None                                  # arg
    )
    return pid

生产级方案对比：

3.4 第三层：Seccomp BPF 系统调用过滤

import seccomp

def apply_syscall_filter():
    """
    使用 seccomp-BPF 限制子进程可用系统调用
    """
    # 默认拒绝所有系统调用
    filter = seccomp.SyscallFilter(seccomp.ALLOW)

    # 白名单：只允许必要 syscall
    allowed_syscalls = [
        "read", "write", "openat", "close",
        "fstat", "mmap", "mprotect", "munmap",
        "brk", "ioctl", "lseek",
        "exit_group", "exit",
    ]

    for syscall in allowed_syscalls:
        filter.add_rule(seccomp.ALLOW, syscall)

    # 禁用网络相关
    blocked = [
        "socket", "connect", "accept", "sendto", "recvfrom",
        "bind", "listen", "clone", "fork", "execve",
        "ptrace", "mount", "umount2"
    ]
    for syscall in blocked:
        filter.add_rule(seccomp.ERRNO(errno.EPERM), syscall)

    filter.load()

四、实战：Python 代码安全沙箱完整实现

下面是一个基于 subprocess + seccomp + cgroup 的 Python 代码安全执行沙箱：

# sandbox_executor.py
import subprocess
import tempfile
import os
import json
import signal
import resource
import time
from pathlib import Path

class SecurePythonSandbox:
    """
    Python 代码安全执行沙箱

    核心安全策略：
    1. 资源限制：CPU/内存/磁盘/进程数
    2. 文件系统隔离：只读根 + 临时目录
    3. 网络隔离：子进程无网络访问
    4. 执行超时：硬时限控制
    """

    # 安全配置
    MAX_EXECUTION_TIME = 30      # 秒
    MAX_MEMORY_MB = 128          # MB
    MAX_PROCESSES = 1            # 不允许 fork
    MAX_OUTPUT_SIZE = 65536      # 输出最大 64KB

    def __init__(self, code: str, input_data: dict = None):
        self.code = code
        self.input_data = input_data or {}
        self.temp_dir = None

    def execute(self) -> dict:
        """执行代码并返回结果"""
        self.temp_dir = tempfile.mkdtemp(prefix="sandbox_")

        try:
            # 1. 注入安全包装代码
            safe_code = self._wrap_with_safety()

            # 2. 写入临时文件
            code_file = Path(self.temp_dir) / "main.py"
            code_file.write_text(safe_code)

            # 3. 执行并收集结果
            result = self._run_sandboxed(code_file)

            return result
        finally:
            # 4. 清理临时目录 (非递归删除)
            self._cleanup_temp()

    def _wrap_with_safety(self) -> str:
        """将用户代码包裹在安全上下文中"""
        return f'''
import sys
import resource
import signal

# ===== 资源限制 =====
# 内存限制 (软限制, 硬限制)
MAX_MEM = {self.MAX_MEMORY_MB} * 1024 * 1024
resource.setrlimit(resource.RLIMIT_AS, (MAX_MEM, MAX_MEM))
# 禁止创建新进程
resource.setrlimit(resource.RLIMIT_NPROC, (0, 0))
# 禁止写入文件
resource.setrlimit(resource.RLIMIT_FSIZE, (0, 0))
# CPU 时间限制 (秒)
resource.setrlimit(resource.RLIMIT_CPU, ({self.MAX_EXECUTION_TIME}, {self.MAX_EXECUTION_TIME}))

# ===== 超时处理 =====
def timeout_handler(signum, frame):
    print("__TIMEOUT__", flush=True)
    sys.exit(124)
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm({self.MAX_EXECUTION_TIME})

# ===== 输入数据 =====
input_data = {repr(self.input_data)}

# ===== 移除危险模块 =====
for mod in ['os', 'subprocess', 'sys', 'shutil', 'importlib', 'ctypes', 'socket', 'http']:
    if mod in sys.modules:
        del sys.modules[mod]

builtins = __builtins__
if hasattr(builtins, '__dict__'):
    for name in ['exec', 'eval', '__import__', 'compile', 'open']:
        if name in builtins.__dict__:
            del builtins.__dict__[name]

# ===== 用户代码开始 =====
{self.code}
# ===== 用户代码结束 =====
'''

    def _run_sandboxed(self, code_file: Path) -> dict:
        """在子进程中执行代码"""
        start_time = time.monotonic()

        try:
            proc = subprocess.run(
                ['python3', str(code_file)],
                capture_output=True,
                text=True,
                timeout=self.MAX_EXECUTION_TIME,
                # 子进程隔离设置
                preexec_fn=self._setup_child_isolation,
                # 环境隔离
                env={"PATH": "/usr/bin:/bin"},
            )

            elapsed = time.monotonic() - start_time

            if proc.returncode == 124:
                return {
                    "success": False,
                    "error": "Execution timed out",
                    "elapsed_seconds": elapsed
                }

            return {
                "success": proc.returncode == 0,
                "output": proc.stdout[:self.MAX_OUTPUT_SIZE],
                "errors": proc.stderr[:self.MAX_OUTPUT_SIZE],
                "elapsed_seconds": round(elapsed, 3)
            }

        except subprocess.TimeoutExpired:
            return {
                "success": False,
                "error": f"Hard timeout after {self.MAX_EXECUTION_TIME}s"
            }

    def _setup_child_isolation(self):
        """子进程启动时的隔离设置"""
        os.setsid()  # 新会话，脱离父进程控制终端

    def _cleanup_temp(self):
        """安全清理临时文件"""
        if self.temp_dir and os.path.exists(self.temp_dir):
            for item in os.listdir(self.temp_dir):
                try:
                    filepath = os.path.join(self.temp_dir, item)
                    if os.path.isfile(filepath):
                        os.unlink(filepath)
                except OSError:
                    pass


# ===== 使用示例 =====
if __name__ == "__main__":
    code = '''
# 安全的数学计算
def fibonacci(n):
    if n <= 1:
        return n
    a, b = 0, 1
    for _ in range(2, n + 1):
        a, b = b, a + b
    return b

result = fibonacci(input_data.get("n", 30))
print(f"fib({{input_data.get('n', 30)}}) = {{result}}")
'''

    sandbox = SecurePythonSandbox(code, {"n": 35})
    result = sandbox.execute()
    print(json.dumps(result, indent=2))

4.1 WebAssembly 沙箱方案：更高安全边界

对于需要更强隔离的场景，WebAssembly（Wasm）提供了"能力安全"（Capability Security）的先天优势：

// wasm_sandbox.rs
use wasmtime::*;
use std::time::Duration;

pub struct WasmSandbox {
    engine: Engine,
    store: Store<HostState>,
    memory: Memory,
    instance: Instance,
}

impl WasmSandbox {
    pub fn new(wasm_bytes: &[u8]) -> Result<Self, String> {
        // 1. 配置引擎：禁用所有非必要功能
        let mut config = Config::new();
        config.wasm_threads(false);
        config.wasm_reference_types(false);
        config.wasm_simd(false);
        config.max_wasm_stack(1 << 16);  // 64KB 栈

        let engine = Engine::new(&config)
            .map_err(|e| format!("engine create failed: {e}"))?;

        // 2. Store 设置资源限制
        let mut store = Store::new(
            &engine,
            HostState {
                memory_consumed: 0,
            },
        );

        // 燃料机制：精确计量每条指令
        store.add_fuel(10_000_000)
            .map_err(|e| format!("fuel error: {e}"))?;

        // 3. 编译与实例化
        let module = Module::new(&engine, wasm_bytes)
            .map_err(|e| format!("module compile: {e}"))?;

        let instance = Instance::new(&mut store, &module, &[])
            .map_err(|e| format!("instance: {e}"))?;

        let memory = instance.get_memory(&mut store, "memory")
            .ok_or("no memory exported")?;

        Ok(Self { engine, store, memory, instance })
    }

    pub fn call_json_process(&mut self, input: &str) -> Result<String, String> {
        // 写入输入到 Wasm 线性内存
        let memory = self.memory;
        let data = input.as_bytes();

        // 分配 Wasm 内线性内存空间 (通过 export 的 alloc 函数)
        let alloc = self.instance
            .get_typed_func::<i32, i32>(&mut self.store, "alloc")
            .map_err(|e| e.to_string())?;

        let ptr = alloc.call(&mut self.store, data.len() as i32)
            .map_err(|e| format!("alloc: {e}"))?;

        memory.write(&mut self.store, ptr as usize, data)
            .map_err(|e| format!("write: {e}"))?;

        // 调用处理函数
        let process = self.instance
            .get_typed_func::<(i32, i32), i32>(&mut self.store, "process_json")
            .map_err(|e| e.to_string())?;

        let result_ptr = process.call(&mut self.store, (ptr as i32, data.len() as i32))
            .map_err(|e| match e {
                Trap::OutOfFuel => "执行超燃料限制".to_string(),
                Trap::UnreachableCodeReached => "代码到达 unreachable".to_string(),
                _ => format!("trap: {e}"),
            })?;

        // 读取结果
        let mut buf = [0u8; 1024];
        memory.read(&self.store, result_ptr as usize, &mut buf)
            .map_err(|e| format!("read: {e}"))?;

        String::from_utf8_lossy(&buf).trim_matches('\0').parse()
    }
}

Wasm 沙箱关键优势：

1. 线性内存隔离：Wasm 代码只能访问自己的一块连续线性内存
2. 能力模型：无隐式 syscall，所有 host 交互必须显式 import
3. 燃料机制（Fuel Metering）：精确到指令级的执行计量，杜绝资源耗尽攻击
4. Spectre 缓解 + 控制流完整性（CFI）

五、生产级 Agent 安全工程实践

5.1 策略引擎：Open Policy Agent (OPA)

# agent_policy.rego - Agent 工具调用授权策略
package agent.tooluse

# 默认拒绝所有调用
default allow := false

# 普通用户不能在执行环境访问文件系统
allow {
    input.user.role == "standard"
    input.tool.category in ["math", "text", "weather"]
}

# 高级用户可以执行代码但必须限制资源
allow {
    input.user.role == "premium"
    input.tool.name == "execute_python"
    input.tool.params.timeout <= 10
}

# 禁止调用未经审批的工具
deny[msg] {
    input.tool.name == "execute_shell"
    not input.user.permissions[_] == "shell_access"
    msg := "无 Shell 执行权限"
}

# 工具调用链深度限制
deny[msg] {
    input.call_chain_depth > 5
    msg := sprintf("调用链过深: %d (最大 5)", [input.call_chain_depth])
}

# 数据防泄露规则  
deny[msg] {
    contains(lower(input.tool.params.code), "input_data")
    contains(lower(input.tool.params.__dest__), "external_api")
    msg := "数据防泄露：禁止将内部数据发送到外部 API"
}

5.2 审计日志与异常检测

# audit_logger.py
import time
import hashlib
import json
from dataclasses import dataclass, asdict

@dataclass
class ToolCallAuditLog:
    timestamp: float
    session_id: str
    agent_id: str
    tool_name: str
    input_hash: str         # 隐私保护：不存原始输入，存哈希
    output_hash: str
    execution_ms: int
    status: str             # success / blocked / error
    block_reason: str       # 若被阻止
    fuel_consumed: int      # 计算资源消耗

class AuditPipeline:
    """
    实时异常检测与攻击阻断

    检测规则：
    - 短时间大量调用同一工具（DoS）
    - 工具调用频率异常（探测行为）
    - 输出中包含敏感数据特征（数据泄露）
    - 代码执行结果异常（逃逸尝试）
    """

    def __init__(self):
        self.call_history = []
        self.ALERT_THRESHOLD = {
            "calls_per_minute": 60,
            "unique_tools_per_session": 20,
            "total_exec_minutes": 10,
        }

    def check(self, record: ToolCallAuditLog) -> bool:
        """返回 True 表示通过，False 表示阻断"""
        recent = [r for r in self.call_history 
                  if time.time() - r.timestamp < 60]

        # 速率检测
        if len(recent) > self.ALERT_THRESHOLD["calls_per_minute"]:
            self._alert("RATE_LIMIT", record)
            return False

        # 代码执行输出异常检测
        if record.tool_name == "execute_python" and record.execution_ms > 25000:
            self._alert("LONG_EXECUTION", record)
            return False

        # 成功通过
        self.call_history.append(record)
        return True

    def _alert(self, reason: str, record: ToolCallAuditLog):
        """触发安全告警"""
        alert = {
            "severity": "HIGH",
            "reason": reason,
            "record": asdict(record),
            "action": "blocked",
            "timestamp": time.time()
        }
        # 推送到 SIEM / 通知管理员
        send_security_alert(alert)

5.3 纵深防御清单

┌─────────────────────────────────────────────────────────────┐
│           Agent Tool Use 安全沙箱清单 (Checklist)            │
├─────────────────────────────────────────────────────────────┤
│ □ Schema 校验: Pydantic + AST 静态分析                       │
│ □ 输入消毒: SQL/HTML/路径遍历字符过滤                        │
│ □ 输出编码: 自动转义防止 XSS/注入                            │
│ □ 进程隔离: namespace + unshare + 新 PID 域                 │
│ □ 系统调用: seccomp-BPF 白名单                               │
│ □ 内存限制: RLIMIT_AS + cgroup memory.max                   │
│ □ CPU 限额: RLIMIT_CPU + cgroup cpu.max                     │
│ □ 网络隔离: CLONE_NEWNET + 物理网卡规则                      │
│ □ 文件系统: overlay + read-only rootfs + tmpfs 临时目录      │
│ □ 超时控制: SIGALRM + subprocess.hard_kill                  │
│ □ 燃料计量: Wasm fuel 或 RPC 配额                            │
│ □ 策略引擎: OPA / Cedar 细粒度 RBAC                          │
│ □ 审计日志: 全链路 trace + 异常检测                          │
│ □ 熔断机制: 错误率超阈值停止 agent                            │
│ □ 金丝雀: 新工具先在隔离 session 灰度                         │
└─────────────────────────────────────────────────────────────┘

六、前沿趋势：从沙箱到形式化验证

2025-2026 年，Agent 安全领域出现几个重要方向：

6.1 形式化工具验证（Formal Verification）

不再依赖运行时沙箱，而是编译时证明代码不会做出越权操作：

• VeriTyped LLM：使用 dependent type 约束工具输出
• Idris 2 / F*：为 Function Calling Schema 生成 refinement type
• Risk-aware Decoding：在解码阶段实时评估每个候选 token 的安全评分

6.2 机密计算沙箱（Confidential Computing）

结合 Intel TDX / AMD SEV-SNP / ARM CCA： - Agent 代码即使在 Hypervisor 也不可见 - 远程证明（Remote Attestation）确保沙箱完整 - 密钥在沙箱内生成，即使管理员也无法读取

6.3 结构化输出协议的演进

6.4 自我约束型 Agent：让 LLM 成为自身的安全沙箱

最前沿的方向是：不依赖外部沙箱，而是训练模型内化安全不变量：

• SafeLoRA：在 LoRA 微调中注入安全约束
• Constitutional AI 2.0：Agent 在推理时实时审核自身输出
• Reflexive Jail：将安全策略编码为模型自身的 chain-of-thought 校验步骤

结语

Tool Use 安全沙箱不是单一技术，而是一个纵深防御体系 —— 从输入 Schema 校验、到进程隔离、到系统调用过滤、到策略引擎、到审计溯源，每一层都假设上层已被突破。

生产级 Agent 系统的安全哲学应该是：不信任 LLM 的任何输出，验证一切可以验证的，限制一切必须限制的，监控一切正在发生的。

当 AI 获得操控世界的能力时，安全工程就是那把"让 AI 走在对的路上"的缰绳。

相关延伸阅读：我的另一篇文章《WebAssembly Component Model —— 下一代通用安全计算原语》从 Wasm 类型安全的角度补充了本文的沙箱设计思路。