Agent 安全沙箱：Code Interpreter 与数据验证

概念速查

为什么 Agent 需要代码执行能力

大语言模型有三大天生缺陷：复杂计算错误率超过 10%、超过 100 行的结构化数据无法全部放入上下文、多步推理容易逻辑断层。Code Interpreter 是填补这些缺陷的核心工具——Agent 生成代码，在隔离环境中执行并返回结果，实现数学计算、数据处理、图表生成等能力。

术语	定义
Agent	具备自主感知、决策、执行能力的 LLM 智能体，核心是调用工具完成复杂任务
Code Interpreter	Agent 的执行类工具，接收 Agent 生成的代码，在隔离环境中执行并返回结果
安全沙箱	Code Interpreter 的底座，一种高度隔离的运行环境，防止不可信代码影响宿主系统

安全风险一览表

风险类型	具体威胁	严重程度
文件系统	读取/删除宿主文件，写入恶意脚本	高
网络	向外发送数据、下载恶意负载、横向渗透	高
资源	无限循环耗尽 CPU、内存泄漏 OOM	中
进程	创建子进程、fork bomb 攻击	高
提权	容器逃逸、利用内核漏洞获取宿主权限	极高

根据风险评估模型 R = A × S × P（风险 = 攻击面 × 漏洞严重性 × 被利用概率），未加防护的代码执行功能的综合风险极高，必须在多层防御架构下运行。

各类风险的攻击路径与应对方案对比：

风险类型	攻击方式	利用难度	典型目标	首道防线	兜底防线
文件系统	读取 /etc/passwd、写入 cron 任务	低	凭据、密钥	AST 扫描 open()	只读文件系统
网络外传	HTTP POST 到 C2 服务器	低	数据窃取	AST 扫描 requests/urllib	network_mode="none"
资源耗尽	无限循环、大内存分配	低	拒绝服务	超时检测	cgroups 硬限制
命令执行	os.system('rm -rf /')	中	系统破坏	AST 扫描 os/subprocess	无特权 + 只读根文件系统
容器逃逸	挂载 /proc、利用内核漏洞	极高	宿主控制	cap_drop + seccomp	User Namespace 映射

每一类风险至少有两层防线兜底：一层在代码层拦截，一层在运行时隔离。单层失效不会导致全线崩溃。

安全防护层级速查

层级	防护手段	解决的问题
AST 扫描	Python ast 模块静态分析代码	在运行前拦截危险模块导入和危险函数调用
资源限制	cgroups CPU/内存硬限制	防止资源耗尽
Docker 沙箱	容器隔离 + Namespace	防止进程与文件系统逃逸
网络隔离	network_mode="none"	禁止数据外传
只读文件系统	read_only=True	防止写入恶意文件
无特权模式	security_opt=["no-new-privileges"]	防止提权
超时终止	container.wait(timeout=N)	防止死循环

底层原理

AST 静态扫描拦截危险代码

Python 的 ast 模块能将代码文本解析为抽象语法树（AST）。通过遍历 AST 节点，可以在不执行代码的前提下识别所有导入语句和函数调用，从而拦截危险操作。

import ast
from typing import List, Tuple

DANGEROUS_MODULES = {
    'os', 'subprocess', 'socket', 'requests', 'urllib',
    'paramiko', 'ftplib', 'ctypes', 'sys'
}

DANGEROUS_FUNCTIONS = {
    'system', 'popen', 'call', 'check_output', 'exec', 'eval',
    'compile', '__import__', 'open', 'fork', 'kill'
}

class CodeSecurityScanner(ast.NodeVisitor):
    def __init__(self):
        self.errors: List[Tuple[int, str]] = []

    def visit_Import(self, node):
        for alias in node.names:
            root_module = alias.name.split('.')[0]
            if root_module in DANGEROUS_MODULES:
                self.errors.append((node.lineno, f"危险模块: {alias.name}"))
        self.generic_visit(node)

    def visit_ImportFrom(self, node):
        if node.module:
            root_module = node.module.split('.')[0]
            if root_module in DANGEROUS_MODULES:
                self.errors.append((node.lineno, f"危险模块: {node.module}"))
        self.generic_visit(node)

    def visit_Call(self, node):
        if isinstance(node.func, ast.Name) and node.func.id in DANGEROUS_FUNCTIONS:
            self.errors.append((node.lineno, f"危险函数: {node.func.id}"))
        self.generic_visit(node)

def scan_python_code(code: str) -> Tuple[bool, List[str]]:
    try:
        tree = ast.parse(code)
    except SyntaxError as e:
        return False, [f"语法错误: 第{e.lineno}行: {e.msg}"]
    scanner = CodeSecurityScanner()
    scanner.visit(tree)
    if scanner.errors:
        return False, [f"第{lineno}行: {msg}" for lineno, msg in scanner.errors]
    return True, []

这种方法的优势是无侵入、零开销——扫描在代码到达沙箱之前完成。局限性在于无法检测动态导入（__import__('os')）或混淆后的代码，因此必须与其他防御层级配合。

AST 扫描的核心是 ast.NodeVisitor 的访问者模式。每个 visit_* 方法对应一类 AST 节点，在遍历语法树时自动触发：

• visit_Import / visit_ImportFrom：捕获 import os、from subprocess import call 等静态导入语句
• visit_Call：捕获所有函数调用，但需要区分 ast.Name（直接调用 exec(...)）和 ast.Attribute（方法调用 os.system(...)）

上面代码只覆盖了 ast.Name 类型的调用，而实际攻击更常通过 ast.Attribute 发起——os.system、subprocess.Popen、shutil.rmtree 都属于属性访问。增强版扫描器需在 visit_Call 中同时检查 node.func.attr in DANGEROUS_FUNCTIONS 且确定调用对象属于危险模块。

此外，ast.BinOp 可用于检测拼接混淆字符串，ast.ListComp / ast.DictComp 可用于标记过于复杂的单行表达式（潜在的 payload 隐藏手法），ast.Constant（Python 3.8+）可检查 Base64 编码字符串——每覆盖一种节点类型，就收窄一层攻击面。

Docker 沙箱的资源隔离机制

Docker 沙箱利用 Linux 内核的 6 大 Namespace 实现进程级别的隔离：

Namespace	隔离对象	作用
UTS	主机名	沙箱内无法识别宿主主机标识
IPC	进程间通信	无法通过信号量/共享内存与其他进程通信
PID	进程 ID	沙箱内 PID 独立，看不到宿主机进程
Mount	文件系统	只挂载必要的系统目录，宿主文件不可见
Network	网络栈	默认只保留 loopback 接口
User	用户权限	容器内 root 在宿主机映射为普通用户

再加上 cgroups 限制 CPU 和内存上限，seccomp 限制系统调用白名单，只读根文件系统禁止写入，构成完整的沙箱执行环境。

import docker
import uuid

client = docker.from_env()

def run_in_sandbox(code: str, timeout: int = 30, memory: str = "256m", cpu: float = 1.0):
    container = None
    name = f"sandbox-{uuid.uuid4().hex}"
    try:
        container = client.containers.run(
            image="python:3.11-slim",
            command=["python", "-c", code],
            name=name,
            detach=True,
            mem_limit=memory,
            nano_cpus=int(cpu * 1e9),
            network_mode="none",
            read_only=True,
            tmpfs={"/tmp": "size=128m"},
            user="nobody",
            security_opt=["no-new-privileges"],
            cap_drop=["ALL"],
        )
        result = container.wait(timeout=timeout)
        stdout = container.logs(stdout=True, stderr=False).decode("utf-8", errors="replace")
        stderr = container.logs(stdout=False, stderr=True).decode("utf-8", errors="replace")
        return {"ok": True, "stdout": stdout, "stderr": stderr, "exit": result["StatusCode"]}
    except TimeoutError:
        return {"ok": False, "error": f"超时 ({timeout}s)"}
    except docker.errors.APIError as e:
        return {"ok": False, "error": str(e)}
    finally:
        if container:
            try:
                container.remove(force=True)
            except Exception:
                pass

关键参数说明：

• network_mode="none"：彻底切断网络，数据无法外泄
• read_only=True：根文件系统只读，恶意代码无法写入文件
• cap_drop=["ALL"]：移除所有 Linux Capability，无法执行提权操作
• user="nobody"：以最低权限用户运行
• tmpfs：唯一可写目录挂载为内存文件系统，容器销毁后数据自动清除

pandas 数据分析 Agent 实现流程

from langchain_core.tools import tool
import subprocess
import tempfile
import os

@tool
def python_repl(code: str):
    """在隔离环境执行 Python 代码"""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False, encoding='utf-8') as f:
        f.write(code)
        path = f.name
    try:
        result = subprocess.run(
            ['python', path],
            capture_output=True, text=True, timeout=10
        )
        if result.returncode == 0:
            return result.stdout
        return f"错误: {result.stderr}"
    except subprocess.TimeoutExpired:
        return "错误: 执行超时"
    finally:
        os.unlink(path)

Agent 的典型数据工作流：用户提问 → LLM 生成 pandas 代码 → 安全扫描 → 沙箱执行 → 结果回传给 LLM 解读 → 返回用户。执行出错时 LLM 自动修正代码并重试，形成反馈闭环。

上述 python_repl 使用了 tempfile + subprocess 的轻量级方案，适合快速原型。生产环境必须换成真正的 Docker 沙箱——区别在于进程隔离的完备性：subprocess 与宿主共享 PID、网络、文件系统，恶意代码可以 os.kill 其他进程或读取 /tmp 中的敏感文件。

完整的数据验证安全链路包含五个环节：

1. 输入验证：检查 CSV/Excel 文件格式、列名是否含 =CMD( 等注入特征、文件大小是否超限，拒绝畸形输入
2. 代码生成：LLM 基于数据概要（shape、dtype、describe）生成分析代码，原始数据不进入 prompt，减少泄漏面
3. 沙箱执行：数据副本复制进容器，原始文件不受影响；执行结果限制最大行数和最大输出长度
4. 输出验证：返回的 DataFrame 或图表 base64 需校验格式与大小，防止渲染恶意 HTML 或超大 payload 导致下游溢出
5. 审计日志：记录每次执行的代码 hash、输入数据 hash、执行耗时和退出码，支持事后追溯与异常检测

步骤 4 常被忽略——如果沙箱被用来生成格式异常的图表或超长字符串，未经校验的结果回传会造成二次风险。输出验证应作为安全链路的最后一环，与输入验证首尾呼应。

架构设计原则

最小权限原则。 能禁用的权限全部禁用。容器以 nobody 用户运行、丢弃所有 capability、只读根文件系统、网络完全切断。每一个开启的权限都是被攻击的可能入口。

多层防御原则。 任何单层安全措施都有可能被绕过。AST 扫描可以防住 90% 的常见攻击，但无法对抗混淆或动态执行。Docker 沙箱能兜住逃过扫描的恶意代码，但无法防止正确的代码执行错误的操作。至少需要静态扫描 → LLM 审核 → 沙箱隔离三层防线，每层解决不同维度的安全问题。

超时终止原则。 所有代码执行必须设定硬超时。Agent 生成的代码可能包含无意识无限循环，也可能是 Fork Bomb 攻击。超时时间取决于业务场景：数据分析类建议 30s，复杂计算建议 120s，极限不应超过 300s。超时后强制 kill 容器并销毁。

沙箱无状态原则。 每次执行新建沙箱，执行完毕立即强制销毁。不保留任何持久化存储。唯一可写目录是挂载在 tmpfs 的临时目录，随容器销毁自动清除。