如何在Django/DRF中实现类LeetCode的用户代码测试用例运行功能

核心安全前提（必做，否则服务器存在被入侵风险） #

• 绝对禁止在Django主进程、服务器宿主环境直接执行用户提交的代码，必须做多层环境隔离，否则恶意代码可直接窃取服务器密钥、删除数据、入侵内网
• 隔离方案优先级：Docker容器 > 系统级沙箱（nsjail、firejail） > 受限子进程，中小流量场景用Docker足够，维护成本最低
• 所有代码执行必须加硬性限制：CPU运行时长上限、内存占用上限、系统调用白名单、全禁网络访问、文件读写仅开放独立临时目录

第一步：基础数据模型设计 #

先定义核心表结构，对应题目、测试用例、用户提交记录三类核心数据：

# problems/models.py
from django.db import models
from django.contrib.auth import get_user_model

User = get_user_model()

class Problem(models.Model):
    """编程题目表"""
    title = models.CharField(max_length=200)
    description = models.TextField()
    # 支持的编程语言，如python3、java、cpp
    support_langs = models.JSONField(default=list)
    # 预置代码骨架，用户补全核心逻辑即可
    code_template = models.JSONField(default=dict)
    time_limit = models.IntegerField(default=1000, help_text="单用例最大运行时长，单位ms")
    memory_limit = models.IntegerField(default=128*1024, help_text="单用例最大内存占用，单位KB")
    created_at = models.DateTimeField(auto_now_add=True)

class TestCase(models.Model):
    """预置测试用例表"""
    problem = models.ForeignKey(Problem, on_delete=models.CASCADE, related_name="test_cases")
    # 入参按函数参数顺序存储，如add(1,2)的入参为[1,2]
    input_params = models.JSONField()
    expected_output = models.JSONField()
    # 公开用例提交后可展示错误详情，隐藏用例仅返回对错
    is_public = models.BooleanField(default=False)
    weight = models.IntegerField(default=1, help_text="用例权重，用于计算得分")

class Submission(models.Model):
    """用户提交记录表"""
    STATUS_CHOICES = (
        ("pending", "等待判题"),
        ("running", "判题中"),
        ("passed", "全部通过"),
        ("failed", "未通过"),
        ("compile_error", "编译错误"),
        ("runtime_error", "运行错误"),
        ("time_limit_exceeded", "超时"),
        ("memory_limit_exceeded", "内存超限"),
    )
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    problem = models.ForeignKey(Problem, on_delete=models.CASCADE)
    code = models.TextField()
    lang = models.CharField(max_length=20)
    status = models.CharField(max_length=30, choices=STATUS_CHOICES, default="pending")
    # 存储判题详情：错误用例、实际输出、运行时长、内存占用等
    judge_detail = models.JSONField(default=dict)
    total_run_time = models.IntegerField(default=0, help_text="总运行时长，单位ms")
    created_at = models.DateTimeField(auto_now_add=True)

后续按DRF常规流程编写对应序列化器、提交接口即可，注意接口收到用户提交的代码后，先存为pending状态的提交记录，不要同步执行判题逻辑阻塞请求。

# submissions/serializers.py
from rest_framework import serializers
from .models import Submission

class SubmissionCreateSerializer(serializers.ModelSerializer):
    class Meta:
        model = Submission
        fields = ["id", "problem", "code", "lang"]
        read_only_fields = ["id", "status", "judge_detail"]

    def validate(self, attrs):
        problem = attrs["problem"]
        if attrs["lang"] not in problem.support_langs:
            raise serializers.ValidationError("当前题目不支持该编程语言")
        return attrs

第二步：异步判题队列搭建 #

• 用Celery+Redis/RabbitMQ搭建异步任务队列，用户提交代码后，接口只负责把提交ID丢入任务队列，立刻返回提交ID给前端
• 前端可通过轮询、Websocket两种方式获取后续判题状态和结果，避免长时间等待请求超时
• 队列侧加并发数限制，控制同一时间运行的判题任务数量，避免服务器资源被占满

第三步：核心判题逻辑实现 #

判题固定流程：拉取提交记录和对应题目的所有测试用例 -> 启动隔离沙箱 -> 拼接用户代码和对应语言的测试驱动脚本 -> 沙箱内逐用例运行 -> 收集运行结果、实际输出、报错、耗时、内存数据 -> 比对实际输出和预期输出 -> 汇总结果更新提交记录

Python3语言驱动脚本示例 #

驱动脚本负责加载用户代码、逐用例调用目标函数、捕获异常、统计运行资源：

# judge/drivers/python3.py
import sys
import json
import time
import tracemalloc
import importlib.util
from io import StringIO

def run_python_code(user_code: str, test_cases: list, time_limit: int, memory_limit: int):
    result = {
        "passed_count": 0,
        "total_count": len(test_cases),
        "case_results": [],
        "global_error": None
    }
    # 临时写入用户代码，避免exec上下文污染
    with open("/tmp/user_solution.py", "w", encoding="utf-8") as f:
        f.write(user_code)
    try:
        spec = importlib.util.spec_from_file_location("user_solution", "/tmp/user_solution.py")
        user_module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(user_module)
        # 题目要求实现的函数名可按题目标识配置，默认统一为solution
        solution_func = getattr(user_module, "solution")
    except Exception as e:
        result["global_error"] = f"编译/导入错误: {str(e)}"
        return result

    for case in test_cases:
        case_res = {
            "input": case["input_params"],
            "expected": case["expected_output"],
            "actual": None,
            "passed": False,
            "time_cost": 0,
            "memory_cost": 0,
            "error": None,
            "is_public": case["is_public"]
        }
        old_stdout = sys.stdout
        sys.stdout = StringIO()
        tracemalloc.start()
        start_time = time.time()
        try:
            actual = solution_func(*case["input_params"])
            run_time = int((time.time() - start_time)*1000)
            current, peak = tracemalloc.get_traced_memory()
            tracemalloc.stop()
            sys.stdout = old_stdout

            case_res["actual"] = actual
            case_res["time_cost"] = run_time
            case_res["memory_cost"] = int(peak/1024)

            if run_time > time_limit:
                case_res["error"] = "运行超时"
            elif case_res["memory_cost"] > memory_limit:
                case_res["error"] = "内存超限"
            elif actual == case["expected_output"]:
                case_res["passed"] = True
                result["passed_count"] += 1
        except Exception as e:
            tracemalloc.stop()
            sys.stdout = old_stdout
            case_res["error"] = f"运行错误: {str(e)}"
        result["case_results"].append(case_res)
    return result

Docker沙箱隔离实现 #

每个判题任务启动一个临时资源受限的Docker容器，挂载用户代码和驱动脚本运行，运行结束立刻销毁容器：

• 容器使用对应语言的官方精简镜像，比如python:3.11-slim
• 启动容器时加资源限制参数：--cpus=0.5 --memory=128m --network=none --read-only --tmpfs /tmp:rw,noexec,nosuid,size=64m，直接禁用网络、设置根文件系统只读、仅开放临时目录的受限读写权限
• 容器内用普通用户身份运行脚本，禁止用root权限，进一步降低容器逃逸风险
• 脚本运行结束后将判题结果以JSON格式输出到标准输出，宿主进程读取输出即可拿到结果，超时直接强制销毁容器

第四步：结果判定与返回 #

判题任务执行完成后更新提交记录状态：

• 存在编译/导入错误，状态标记为compile_error
• 单用例运行触发超时、内存超限、运行时异常，标记为对应错误状态
• 所有用例全部通过，状态标记为passed
• 存在用例输出不匹配，状态标记为failed
• 返回给前端的结果中，公开用例可展示输入、实际输出、错误详情，隐藏用例仅返回是否通过，避免用户暴力刷用例

常见坑点 #

• 输出比对要做严格类型校验，比如用户返回字符串"1"和预期数字1不能判定为通过
• 每个用例执行前要重新加载用户代码模块，避免全局变量污染导致后续用例运行结果异常
• 增加用户提交频率限制，防止恶意提交占满判题队列资源
• 多语言支持只需要对应编写各语言的驱动脚本即可，核心判题流程完全一致

内容的提问来源于stack exchange，提问作者Abdul Rehman Nawaz