Python 调用 Node.js vm2 沙箱实战教程

mkdir python-vm2
cd python-vm2

# 安装 vm2
npm init -y
npm install vm2

# Python 无需额外安装库（内置 subprocess）

const { VM } = require('vm2');
// 从命令行接收 Python 传过来的 JS 代码
const userCode = process.argv[2] || '';
try {
    // 创建安全沙箱
    const vm = new VM({
        timeout: 1000,
        allowAsync: false,
        sandbox: {
            // 你可以给沙箱开放安全变量
            PI: 3.14159
        }
    });
    // 执行代码
    const result = vm.run(userCode);
    // 返回结果给 Python
    console.log(JSON.stringify({ success: true, data: result }));
} catch (err) {
    // 错误返回
    console.log(JSON.stringify({ success: false, error: err.message }));
}

import subprocess
import json
import sys
def run_js_in_vm2(js_code: str):
    """
    Python 胶水函数：调用 Node.js + vm2 执行不安全 JS 代码
    """
    try:
        # 调用 node runner.js "代码"
        result = subprocess.run(
            [sys.executable.replace("python", "node"), "runner.js", js_code],
            capture_output=True,
            text=True,
            encoding="utf-8",
            timeout=3
        )
        # 解析返回结果
        output = result.stdout.strip()
        return json.loads(output)
    except Exception as e:
        return {"success": False, "error": str(e)}
# ====================== 测试 ======================
if __name__ == "__main__":
    # 测试 1：安全代码
    code1 = "1 + 2 * 3 + PI"
    print("执行结果：", run_js_in_vm2(code1))
    # 测试 2：沙箱禁止的危险代码（会被拦截）
    code2 = "process.exit()"
    print("危险代码执行：", run_js_in_vm2(code2))

python main.py

const { VM } = require('vm2');
const readline = require('readline');

// 创建沙箱
const vm = new VM({
    timeout: 1000,
    allowAsync: true,
    sandbox: { PI: 3.14159 }
});

// 读取 Python 传来的消息
const rl = readline.createInterface({
    input: process.stdin,
    output: process.stdout
});

console.log("VM2 RPC 服务已启动");

rl.on('line', (code) => {
    try {
        const res = vm.run(code);
        console.log(`[RESULT]${JSON.stringify(res)}`);
    } catch (err) {
        console.log(`[ERROR]${err.message}`);
    }
});

import subprocess
import threading
import json

class VM2Client:
    def __init__(self):
        # 启动 Node 服务
        self.proc = subprocess.Popen(
            ["node", "rpc_server.js"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            encoding="utf-8"
        )
        self.response = None

    def _read_output(self):
        while True:
            line = self.proc.stdout.readline()
            if not line:
                break
            self.response = line.strip()

    def run(self, code):
        # 启动监听线程
        threading.Thread(target=self._read_output, daemon=True).start()
        
        # 发送代码
        self.proc.stdin.write(code + "\n")
        self.proc.stdin.flush()
        
        # 等待返回
        while self.response is None:
            pass
        
        res = self.response
        self.response = None
        return res

# ====================== 测试 ======================
if __name__ == "__main__":
    client = VM2Client()

    print(client.run("100 + 200"))
    print(client.run("PI * 2"))
    print(client.run("process"))  # 被沙箱禁止

python rpc_client.py

const { NodeVM } = require('vm2');
const code = process.argv[2];

const vm = new NodeVM({
    timeout: 1500,
    eval: false,
    wasm: false,
    console: 'off',
    require: {
        builtin: [],
        external: false,
        deny: ['*']
    }
});

try {
    const result = vm.run(code);
    console.log(JSON.stringify({ ok: 1, data: result }));
} catch (e) {
    console.log(JSON.stringify({ ok: 0, err: e.message }));
}

import subprocess
import json
def js_sandbox(js_code: str):
    res = subprocess.run(
        ["node", "safe_runner.js", js_code],
        capture_output=True,
        text=True
    )
    return json.loads(res.stdout)
# 使用
print(js_sandbox("1 + 2 + 3"))