如何在Linux环境下安全执行用户提交的Python代码

要实现符合你需求的安全执行环境，核心思路是严格隔离执行权限+强制超时控制，下面是具体的可行方案：

核心实现步骤 #

1. 预先管控允许使用的模块 #

按照你说的，把用户需要的模块（比如示例里的re）提前导入，不让用户代码自行导入任何模块——既避免危险模块被引入，也简化用户代码。

2. 构建受限的执行命名空间 #

创建一个仅包含安全内容的全局字典，彻底移除能访问系统资源、执行危险操作的内置函数和对象：

• 只保留str、int等基础数据类型
• 禁止open、exec、eval、os、sys这类危险项
• 把预先导入的安全模块（比如re）放入命名空间供用户代码调用

3. 静态校验用户代码的合法性 #

在执行前用语法树（AST）检查用户代码，确保：

• 代码仅包含一个名为parse的函数定义
• 函数只接受一个input参数
• 函数体内没有import、exec、eval、全局变量修改等危险语句

4. 用进程隔离+超时控制执行代码 #

用多进程而非多线程来跑用户代码（线程无法被强制终止，进程可以），设置100ms超时，超时直接终止进程，避免卡顿阻塞主流水线。

具体代码示例 #

第一步：代码校验工具函数

import ast

def check_user_code_validity(code):
    try:
        tree = ast.parse(code)
    except SyntaxError as e:
        return False, f"语法错误: {str(e)}"
    
    # 必须只有一个名为parse的函数
    if len(tree.body) != 1 or not isinstance(tree.body[0], ast.FunctionDef) or tree.body[0].name != 'parse':
        return False, "代码只能定义一个名为parse的函数"
    
    func_node = tree.body[0]
    # 函数必须只有一个input参数
    if len(func_node.args.args) != 1 or func_node.args.args[0].arg != 'input':
        return False, "函数必须仅接受一个名为input的参数"
    
    # 禁止危险语法节点
    forbidden_types = (ast.Import, ast.ImportFrom, ast.Exec, ast.Eval, ast.Global, ast.Nonlocal)
    for node in ast.walk(func_node):
        if isinstance(node, forbidden_types):
            return False, f"禁止使用{type(node).__name__}相关语法"
    
    return True, "代码校验通过"

第二步：安全执行逻辑

import re
import multiprocessing
from multiprocessing import Pipe

# 预先定义安全的全局命名空间
SAFE_GLOBALS = {
    '__builtins__': {
        'str': str,
        'int': int,
        'list': list,
        'dict': dict,
        # 只保留绝对安全的内置类型/函数
    },
    're': re  # 提前导入允许使用的模块
}

def run_user_func(code, input_data, conn):
    try:
        # 在受限命名空间里执行用户代码
        exec(code, SAFE_GLOBALS)
        parse_func = SAFE_GLOBALS['parse']
        result = parse_func(input_data)
        conn.send(('success', result))
    except Exception as e:
        conn.send(('error', f"执行错误: {str(e)}"))
    finally:
        conn.close()

def execute_user_code(user_code, input_str, timeout=0.1):
    # 先校验代码合法性
    valid, msg = check_user_code_validity(user_code)
    if not valid:
        return ('invalid', msg)
    
    # 用管道传递执行结果
    parent_conn, child_conn = Pipe()
    # 启动子进程执行代码
    proc = multiprocessing.Process(target=run_user_func, args=(user_code, input_str, child_conn))
    proc.start()
    # 等待指定超时时间
    proc.join(timeout)
    
    # 超时则终止进程
    if proc.is_alive():
        proc.terminate()
        proc.join()
        return ('timeout', '代码执行超时（超过100ms）')
    
    # 获取执行结果
    if parent_conn.poll():
        return parent_conn.recv()
    else:
        return ('error', '未知执行错误')

测试示例

# 用户提供的代码（注意已经去掉了内部的import re）
user_code = '''
def parse(input: str) -> str:
    return re.sub("a", "b", input) 
'''

# 执行测试
result = execute_user_code(user_code, 'abc123')
print(result)  # 输出: ('success', 'bbc123')

关键安全保障说明 #

• 资源隔离：受限命名空间里没有任何能访问文件系统、网络的接口，用户代码无法进行危险操作
• 无副作用：静态校验确保代码是纯函数，进程隔离避免用户代码修改主流水线的状态
• 超时可控：子进程超时后被强制终止，不会阻塞整个预处理流水线
• 权限最小化：仅开放必要的模块和内置功能，从根源减少风险

内容的提问来源于stack exchange，提问作者johndue