平时开发后台服务或构建在线平台时，你可能常遇到这样的需求：同事发来一个 xxx.py 文件，让你“在服务里跑一下”；或者你需要搭建一个在线练习、脚本执行平台，允许用户直接上传并运行 Python 代码。
这件事如果没有足够的安全意识，很可能导致服务器被“玩坏”。想象一下，你把服务器的键盘和鼠标交给一个陌生人，对方能做什么，别人上传的代码基本就能做什么。
典型的运行风险主要包括以下几类：
-
破坏数据
最直接的方式：删库、删文件、改配置。例如，仅仅一行代码就足够造成严重破坏：
import shutil
shutil.rmtree("/var/www")
-
窃取数据
读取配置文件、环境变量，甚至扫描整个磁盘寻找敏感信息：
from pathlib import Path
for p in Path("/").rglob("*.env"):
print(p, p.read_text()[:200])
你以为只是帮忙运行一个算法题，对方可能顺手就把你所有的密钥扫描了一遍。
-
滥用计算资源
包括死循环打满 CPU，或疯狂占用内存直至 OOM（内存溢出）：
# CPU 打满版
x = 0
while True:
x += 1
或者：
# 内存炸到 OOM
big = []
while True:
big.append("x" * 10_000_000)
-
滥用网络资源
利用你的服务器进行挖矿、恶意刷接口或端口扫描等行为：
import requests
while True:
requests.get("https://target.example.com")
因此，一个基本的安全结论是：绝对不要在你的主进程或主环境中直接 exec 或 eval 他人的代码。 构建安全防线，至少需要两层：语言层约束 与 系统层隔离。
第一道防线：语言层粗筛（AST 静态分析）
即便你准备了容器或虚拟机作为最终屏障，在裸跑代码前进行一轮简单的静态检查也能拦截明显的危险操作，降低风险。
Python 标准库中的 ast 模块可以将源代码解析为抽象语法树（AST），我们可以利用它做一些基础检查：
- 禁止导入高危模块，如
os, sys, subprocess, socket 等。
- 禁止访问双下划线属性，如
__dict__, __class__, __subclasses__ 等，这些常用于突破沙箱。
- 限制某些语句类型，例如不允许无条件的
while True 循环（虽然不能100%防止死循环）。
下面是一个简化的“安全检查器”示例（请注意，这种方法仅用于降低风险，无法保证绝对安全）：
import ast
FORBIDDEN_MODULES = {
"os", "sys", "subprocess", "socket", "shutil",
"pathlib", "inspect", "builtins"
}
FORBIDDEN_ATTR_PREFIX = "__"
class UnsafeCodeError(Exception):
pass
class SafeVisitor(ast.NodeVisitor):
def visit_Import(self, node):
for alias in node.names:
if alias.name.split('.')[0] in FORBIDDEN_MODULES:
raise UnsafeCodeError(f"禁止导入模块: {alias.name}")
self.generic_visit(node)
def visit_ImportFrom(self, node):
if node.module and node.module.split('.')[0] in FORBIDDEN_MODULES:
raise UnsafeCodeError(f"禁止 from {node.module} import ...")
self.generic_visit(node)
def visit_Attribute(self, node):
# 限制 obj.__xxx__
if isinstance(node.attr, str) and node.attr.startswith(FORBIDDEN_ATTR_PREFIX):
raise UnsafeCodeError(f"禁止访问属性: {node.attr}")
self.generic_visit(node)
def visit_While(self, node):
# 简单粗暴：禁止无条件 while True
if isinstance(node.test, ast.Constant) and node.test.value is True:
raise UnsafeCodeError("禁止 while True 死循环")
self.generic_visit(node)
def check_code_safe(source: str):
try:
tree = ast.parse(source)
except SyntaxError as e:
raise UnsafeCodeError(f"语法错误: {e}") from e
SafeVisitor().visit(tree)
这种基于 ast 的静态分析检查容易被高水平攻击者绕过，但对于拦截普通的“误操作”脚本或简单的恶意代码仍然有一定作用。真正的安全不能只依赖它。
第二道防线：使用子进程与受控环境
即使不打算使用容器，也至少应该做到以下几点：
- 启动一个独立的子进程来运行用户代码。
- 在子进程中严格限制资源（CPU、内存、进程数）。
- 主进程只负责传入代码、收集结果，并设置超时强制终止子进程。
以下是一个在 Linux 环境下最基础但比直接执行安全得多的实现方案：
import subprocess
import tempfile
import textwrap
import os
import resource
import sys
from pathlib import Path
def _limit_resources():
# 限制 CPU 时间：3 秒
resource.setrlimit(resource.RLIMIT_CPU, (3, 3))
# 限制内存：256MB
mem_bytes = 256 * 1024 * 1024
resource.setrlimit(resource.RLIMIT_AS, (mem_bytes, mem_bytes))
# 限制打开文件数
resource.setrlimit(resource.RLIMIT_NOFILE, (16, 16))
def run_user_code_basic(source: str, input_data: str = "", timeout=5):
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir_path = Path(tmpdir)
code_file = tmpdir_path / "main.py"
code_file.write_text(source, encoding="utf-8")
cmd = [sys.executable, str(code_file)]
proc = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd=tmpdir,
text=True,
preexec_fn=_limit_resources, # 在子进程启动前调用资源限制函数
)
try:
stdout, stderr = proc.communicate(input_data, timeout=timeout)
except subprocess.TimeoutExpired:
proc.kill()
return {"ok": False, "error": "运行超时", "stdout": "", "stderr": ""}
return {
"ok": proc.returncode == 0,
"stdout": stdout[:8000], # 防止输出太多
"stderr": stderr[:8000],
"returncode": proc.returncode,
}
这里的 _limit_resources() 函数使用了 Linux 的 resource 模块进行资源限制（Windows 不支持此模块，可退而求其次仅依赖 timeout 或借助外部工具）。注意：这个版本的代码仍然可以访问宿主机的文件和网络，对于生产环境而言只能算是一个半成品。
第三道防线：置于容器中运行
对于多数线上环境，使用 Docker 或 Kubernetes 等容器技术，让他人代码在独立容器中运行是更常见且可靠的方案。许多在线判题系统（OJ）、交互式笔记本（Notebook）都采用这种模式。
基本思路如下：
- 准备一个基础镜像，例如
python:3.10-slim，并在其中安装好允许使用的库。
- 将用户代码写入一个临时目录。
- 使用
docker run 启动一个一次性容器，并施加严格限制：
- 禁用网络：
--network=none
- 限制 CPU/内存：
--cpus=1 --memory=256m
- 文件系统只读：
--read-only，仅挂载一个临时目录供其写入
- 在容器内使用非 root 用户运行
以下为思路示意代码（简化版）：
import subprocess
import tempfile
from pathlib import Path
import textwrap
import uuid
def run_in_docker(source: str, input_data: str = "", timeout=5):
image = "python:3.10-slim" # 使用自制的沙箱镜像更佳
with tempfile.TemporaryDirectory() as tmpdir:
tmpdir_path = Path(tmpdir)
code_file = tmpdir_path / "main.py"
input_file = tmpdir_path / "input.txt"
code_file.write_text(source, encoding="utf-8")
input_file.write_text(input_data, encoding="utf-8")
container_name = f"user-code-{uuid.uuid4().hex[:8]}"
cmd = [
"docker", "run",
"--rm",
"--name", container_name,
"--network=none",
"--cpus=1",
"--memory=256m",
"--pids-limit=64",
"--read-only",
"-v", f"{tmpdir}:/app:ro", # 以只读方式挂载宿主机目录
"-w", "/app",
image,
"bash", "-lc",
"python main.py < input.txt" # 在容器内执行
]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout,
)
except subprocess.TimeoutExpired:
# 超时后强制清理容器
subprocess.run(["docker", "rm", "-f", container_name],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL)
return {"ok": False, "error": "容器运行超时", "stdout": "", "stderr": ""}
return {
"ok": result.returncode == 0,
"stdout": result.stdout[:8000],
"stderr": result.stderr[:8000],
"returncode": result.returncode,
}
在实际项目中，你通常还会：
- 制作一个专用的
sandbox-python 镜像，移除无关工具，仅包含允许的第三方库。
- 在 Dockerfile 中使用非 root 用户（例如
RUN useradd -m sandbox && USER sandbox）。
- 配置额外的安全策略，如
seccomp 配置文件、apparmor 策略，或使用 --cap-drop=ALL 删除所有内核权限。
即便如此，做到上述水平，其安全性也已远超“在 Web 服务中直接 exec ”的做法。这种容器化隔离与资源控制是构建可靠沙箱的基石。
其他容易忽略的安全细节
-
限制输出大小
用户可能使用 print("x" * 10**9) 打爆你的日志或内存，务必对输出进行截断：
MAX_OUTPUT = 8000
stdout = result.stdout
stderr = result.stderr
if len(stdout) > MAX_OUTPUT:
stdout = stdout[:MAX_OUTPUT] + "\n...输出过长，已截断..."
-
避免泄露环境变量中的机密
在容器内，用户代码依然可以执行 import os; print(os.environ) 来获取所有环境变量。因此，运行用户代码的容器环境变量应保持清洁，切勿注入数据库密码等敏感信息。
-
采用“白名单”而非“黑名单”
对于算法题等场景，用户通常只需要 math、itertools、collections 等少数标准库。一个更有效的做法是提供“模板代码”，用户仅在预定义的函数内编写逻辑，而非任意编写顶层脚本，从而施加更强的约束。
-
不要依赖纯 Python 沙盒的“绝对安全”
Python 历史上曾有过受限执行模式，但因太容易被绕过而被移除。因此，生产环境的安全必须依赖系统层的隔离（容器、虚拟机或 nsjail 等工具）。
整合：一个简单的安全运行服务流程
可以设计一个高层函数，其大致流程如下：
- 接收用户提交的代码。
- 使用 AST 进行第一层静态安全检查 (
check_code_safe)。
- 检查通过后，将代码提交至 Docker 沙箱中执行。
- 如有输入数据，一并传入容器。
- 获取输出或错误信息，截断后返回给用户。
简化版的组合伪代码如下：
def run_user_python(source: str, input_data: str = ""):
# 第 1 道：静态检查
try:
check_code_safe(source)
except UnsafeCodeError as e:
return {
"ok": False,
"stage": "static_check",
"error": str(e),
}
# 第 2 道：容器沙盒
result = run_in_docker(source, input_data=input_data, timeout=5)
# 简单清洗输出
result["stdout"] = (result.get("stdout") or "")[:8000]
result["stderr"] = (result.get("stderr") or "")[:8000]
return result
在此基础上，可以进行进一步的工程化优化，例如引入任务队列防止瞬时高并发打垮机器、增加监控和审计日志、限制单用户调用频率等，这些思路与构建高并发、可扩展的后端服务一脉相承。
总结
“安全运行他人上传的代码”这一需求，其核心本质是将“不可信代码”隔离在一个你可控的“盒子”里。
你可以将这个盒子设计得非常精巧（容器 + seccomp + cgroup + AST + 代码模板），也可以从简易版本起步（子进程 + 资源限制 + 超时控制）。但切忌为了省事而在主进程中直接执行，那无异于将服务器的 root 权限拱手相让。
希望本文提供的思路和代码示例，能帮助你在 云栈社区 或其他技术平台上构建更安全的代码执行服务。安全无小事，层层设防是关键。
发表于 7 天前
|
查看: 13|
回复: 0
