如何安全运行别人上传的Python代码？

前阵子晚上十一点多，我在公司楼下便利店买咖啡，运维给我打电话，说线上有台机器 CPU 打满了，top 一看一个 python3 进程 400% 占着不放，一查是个“给客户开放的脚本运行功能”，客户自己上传了个 Python 脚本跑报表，结果一行 while True: pass 写嗨了，顺带还顺手扫了下整个磁盘…当场我就决定：以后别人上传的 Python 代码，绝对不能直接跑在主进程里，要给它关小黑屋。

所以今天就跟你聊聊，怎么相对安全地运行别人上传的 Python 代码，别整成“线上挖矿场”。

在同一台机器、同一个解释器里，把别人的代码“完全安全”地跑起来，基本做不到。

你能做的是：多加几层保险，让它就算闹事，也只是在一个很小的范围里闹。

我按「从最危险到相对安全」这么聊，你自己对号入座就行。

第一个坑：别在当前进程里直接 exec / eval

很多人最开始都这么写，你看看是不是眼熟：

# 非常危险的示例，别这么干！
defrun_user_code_very_dangerous(code: str):
# 想着“我给你隔离个命名空间就安全了”
    user_globals = {}
    user_locals = {}
    exec(code, user_globals, user_locals)

看着好像给了个干净的 globals，其实屁用不大，别人代码照样可以：

import os
os.system("rm -rf /")          # 直接干系统
open("/etc/passwd").read()     # 看敏感文件
import socket, psutil, subprocess  # 扫网、扫进程、乱杀

因为你跟它在同一个进程，内存空间是共享的，C 扩展也共享，它想怎么玩就怎么玩，什么“禁用内置函数”“删掉 import”这些小伎俩，稍微懂点底层的人分分钟绕过。

所以可以直接记一句：

只要是 exec / eval 这种在当前进程跑用户代码的方案，就不要往“安全”两个字上靠。

你可以在开发机上玩玩，但上生产，别想。

换个思路：把别人的代码关进“小黑屋”

安全的核心思路就一句话：隔离。

就跟你不可能把陌生人领回家，让他自己在你卧室里翻箱倒柜一样。正确姿势是：给他一个只放了他需要东西的房间，门口还站个保安，时间到了拉闸。

对应到技术上，大概几层：

1. 换进程：用户代码在独立进程里跑，挂了就挂，不拖主进程。
2. 限资源：限制 CPU 时间、内存、打开文件数、磁盘占用。
3. 限文件：让它在一个临时目录里转，不给它访问系统关键目录。
4. 限网络：多数场景直接断网，或者只给白名单。
5. 可销毁：跑完一把火烧掉——容器删掉、临时目录删掉。

我先给你一个“最小能用”的 Python 跑脚本示例，你能在 Linux 环境跑通的那种。

来个最小可用版的“脚本小黑屋”

下面这个例子做的事情是：

• 把用户上传的代码写到一个临时目录 main.py
• 用 subprocess 起一个子进程来跑
• 用 resource 限制 CPU 时间、内存
• 加一个超时时间，跑太久就杀掉

import os
import tempfile
import subprocess
import textwrap
import resource  # 只在类 Unix 系统可用
from dataclasses import dataclass
@dataclass
classRunResult:
    returncode: int
    stdout: str
    stderr: str
    timeout: bool
def_limit_resources():
"""
    在子进程里调用，限制资源。
    注意：只在 Linux / macOS 这类 Unix 上生效。
    """
# 限制 CPU 时间，单位秒，比如 2 秒
    resource.setrlimit(resource.RLIMIT_CPU, (2, 2))
# 限制内存，单位字节，比如 256MB
    mem_bytes = 256 * 1024 * 1024
    resource.setrlimit(resource.RLIMIT_AS, (mem_bytes, mem_bytes))
# 限制可以创建的文件大小，单位字节，比如 10MB
    file_bytes = 10 * 1024 * 1024
    resource.setrlimit(resource.RLIMIT_FSIZE, (file_bytes, file_bytes))
defrun_user_code_sandboxed(code: str, input_data: str = "", timeout: int = 3) -> RunResult:
"""
    在临时目录 + 受限子进程里执行用户代码。
    这只是一个“轻量沙箱”，不是绝对安全。
    """
# 简单规整一下缩进，防止用户贴进来的代码缩进乱掉
    code = textwrap.dedent(code)
with tempfile.TemporaryDirectory(prefix="user_code_") as tmpdir:
        script_path = os.path.join(tmpdir, "main.py")
# 写入用户脚本
with open(script_path, "w", encoding="utf-8") as f:
            f.write(code)
try:
            proc = subprocess.run(
                ["python3", script_path],
                input=input_data.encode("utf-8"),
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                cwd=tmpdir,          # 工作目录限制在临时目录
                timeout=timeout,     # 超时杀进程
                preexec_fn=_limit_resources  # 设置资源限制（Unix）
            )
return RunResult(
                returncode=proc.returncode,
                stdout=proc.stdout.decode("utf-8", errors="replace"),
                stderr=proc.stderr.decode("utf-8", errors="replace"),
                timeout=False
            )
except subprocess.TimeoutExpired as e:
# 超时的话，返回部分输出
            stdout = e.stdout.decode("utf-8", errors="replace") if e.stdout else""
            stderr = e.stderr.decode("utf-8", errors="replace") if e.stderr else""
return RunResult(
                returncode=-1,
                stdout=stdout,
                stderr=stderr or"Execution timed out",
                timeout=True
            )

这个东西其实已经比“直接 exec”强太多了：

• 进程独立，挂了不影响主进程
• CPU、内存、文件大小都有限制
• 在一个空目录里跑，访问不到你项目里的源码文件

你在本地可以试一下：

if __name__ == "__main__":
    user_code = """
    import time
    print("start")
    time.sleep(5)
    print("end")
    """
    result = run_user_code_sandboxed(user_code, timeout=2)
    print("timeout:", result.timeout)
    print("stdout:", result.stdout)
    print("stderr:", result.stderr)

按理说会超时，stdout 只会有个 start。

不过，这还远远不够。

再加一层：把危险模块先拦一拦（但不要迷信这一招）

很多时候，业务上你其实只需要用户写一些“算发代码”（比如只用 math、random、简单 for 循环），完全不需要让他接触 os、subprocess、socket 这些东西。

可以做个非常粗暴的静态检查，当辅助用：

import ast
DANGEROUS_MODULES = {
"os", "sys", "subprocess", "socket", "shutil",
"pathlib", "inspect", "importlib", "ctypes"
}
defbasic_code_check(code: str) -> None:
"""
    非严格的静态检查：
    - 禁止导入部分危险模块
    - 发现有问题就抛异常
    """
    tree = ast.parse(code)
for node in ast.walk(tree):
# 处理 `import os`
if isinstance(node, ast.Import):
for alias in node.names:
if alias.name.split(".")[0] in DANGEROUS_MODULES:
raise ValueError(f"禁止导入模块: {alias.name}")
# 处理 `from os import path`
if isinstance(node, ast.ImportFrom):
if node.module and node.module.split(".")[0] in DANGEROUS_MODULES:
raise ValueError(f"禁止 from {node.module} import ...")
# 也可以禁止 exec/eval 之类
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
if node.func.id in {"eval", "exec", "__import__"}:
raise ValueError(f"禁止调用函数: {node.func.id}")

然后在刚才的 run_user_code_sandboxed 之前加一行：

defrun_user_code_safe_entry(code: str, input_data: str = "", timeout: int = 3) -> RunResult:
    basic_code_check(code)           # 有问题直接抛异常
return run_user_code_sandboxed(code, input_data, timeout)

注意，我一直在强调这个只是辅助。

安全这块有个铁律：只靠黑名单是守不住的。

但在很多“内部脚本平台”“数据分析平台”这种半信任场景下，配合资源限制，用起来还是很香的——起码能挡住大多数“误操作”。

上强度：把用户代码扔进 Docker 里

如果你的场景稍微严肃一点，比如：

• 面向外部用户的在线 Python 运行
• 在线判题系统（OJ）
• Saas 里的自动化脚本功能

那我强烈建议：用容器/虚拟机隔离一层。

最常见就 Docker，思路大概是：

1. 做一个专门的“执行镜像”：

• 只装 Python 和你允许用户用的库
• 不放任何业务代码、配置、密钥

• 用 --network none 或者只开白名单网络
• 用 --cpus、--memory 限制资源
• 用挂载卷把“工作目录”映射进去

伪代码长这样：

import subprocess
import tempfile
import os
from dataclasses import dataclass
@dataclass
classDockerRunResult:
    returncode: int
    stdout: str
    stderr: str
    timeout: bool
defrun_user_code_in_docker(code: str, image: str = "python:3.11-slim") -> DockerRunResult:
with tempfile.TemporaryDirectory(prefix="user_code_") as tmpdir:
        script_path = os.path.join(tmpdir, "main.py")
with open(script_path, "w", encoding="utf-8") as f:
            f.write(code)
        container_name = f"user-code-{os.path.basename(tmpdir)}"
        cmd = [
"docker", "run", "--rm",
"--name", container_name,
"--network", "none",      # 禁止网络
"--cpus", "0.5",          # 限制 CPU
"--memory", "256m",       # 限制内存
"-v", f"{tmpdir}:/app",   # 只挂载工作目录
"-w", "/app",
            image,
"python", "main.py",
        ]
try:
            proc = subprocess.run(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                timeout=5,
            )
return DockerRunResult(
                returncode=proc.returncode,
                stdout=proc.stdout.decode("utf-8", errors="replace"),
                stderr=proc.stderr.decode("utf-8", errors="replace"),
                timeout=False
            )
except subprocess.TimeoutExpired as e:
# 超时了顺便尝试杀容器
            subprocess.run(["docker", "rm", "-f", container_name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
            stdout = e.stdout.decode("utf-8", errors="replace") if e.stdout else""
            stderr = e.stderr.decode("utf-8", errors="replace") if e.stderr else""
return DockerRunResult(
                returncode=-1,
                stdout=stdout,
                stderr=stderr or"Execution timed out",
                timeout=True
            )

这里面安全点就比刚才的“裸进程”多了很多：

• 就算用户代码跑崩了，也只影响容器
• 容器没网，扫不出去
• 容器里没你业务代码、没数据库密码
• 跑完容器删掉，临时文件一并清掉

当然，容器本身也不是绝对安全的，内核漏洞之类的咱就不展开了，日常业务足够用了。

别忘了：账号权限、文件、网络这些外围的事

这个也是很多人容易忽略的，你光在代码里搞沙箱不够，操作系统那一层也要配合一下。

我随口捋几个你可以对照一下：

• 专门建一个系统用户，比如 code_runner，所有用户代码进程都用这个账号跑，千万别用 root。
• 这个账号只给一个干净目录的读写权限，别让他去翻 /var/log、/home/xxx 那些奇怪地方。
• 所有“用户代码的输入”都当成不可信数据，不要直接丢给你的数据库、内部 HTTP 接口。
• 业务数据库账号也做权限拆分，给“执行脚本用的账号”一个只读库或者只读表。

这些东西写起来很无聊，但真出事的时候，全靠它们兜底。

再啰嗦两句流程上的事：别让上传脚本直接跑主库

很多同学一开始会写成这样：

1. 用户在页面上传一个 xxx.py
2. 后端接口收到文件之后，直接 run_user_code(...)
3. 共用一个主库连接，脚本里面可以跑 ORM、直接查业务库

这风险就有点大了，尤其是第三点。

稍微稳一点的玩法，是搞一个“任务队列式”的：

• web 服务只负责把脚本和参数丢进队列（比如 Redis / MQ）
• 后台有 worker 专门拿队列里的任务，起容器执行
• worker 用一个专门的“低权限数据库账号”
• 用户在前端轮询“任务结果”

流程大概是这样，小画一段伪代码意思一下：

# 伪代码：提交任务
defsubmit_task(code: str, params: dict) -> str:
    task_id = generate_task_id()
    save_task_meta(task_id, status="PENDING")
    push_to_queue({
"task_id": task_id,
"code": code,
"params": params,
    })
return task_id
# 伪代码：worker 消费任务
defworker_loop():
whileTrue:
        task = pop_from_queue(block=True)
        task_id = task["task_id"]
        code = task["code"]
        params = task["params"]
try:
            save_task_meta(task_id, status="RUNNING")
            result = run_user_code_in_docker(code)  # 这里用前面那个容器执行
            save_task_result(task_id, result)
except Exception as e:
            save_task_meta(task_id, status="FAILED", error=str(e))

这样有两个好处：

• 上传脚本那台机器压力很小，worker 爆了也不影响整个系统。
• 可以随时加 worker、缩 worker，弹性伸缩，很舒服。

最后再提两个很容易被忘掉的小点

这个是我在公司楼下抽烟的时候，边吐槽边跟我们组那个小李说的，你顺便听一耳朵：

1. 要有日志和审计

• 谁在什么时间运行了什么代码，大概输出了什么，最好都记一份
• 真出问题（比如被植入挖矿脚本），你至少知道是哪个用户干的

• 比如每个用户每天最多跑多少次脚本、每次最长跑多久、同时最多几个任务
• 不然别人不一定是恶意的，写个死循环不带资源限制，也能搞死你

行了，差不多就这些，安全这玩意儿永远是“多加几道锁”，不是说有一段银弹代码往那一贴世界就太平了。