隔离 Python 执行器

函数库里的 Python 代码由 Java 侧调度执行：鉴权、存储、参数转换、并发控制和超时都由 Java 负责，Python 只负责运行原版函数代码本身。每次调试或执行都是一次独立的执行单元，结束后不留容器、不留文件。

这样做换来的是几条很实在的好处：

• 用户代码永远不以 Java 进程的身份运行。 执行环境不可用时一律拒绝执行，代码里不存在"降级到 Java 进程里跑"的分支。
• 宿主机密钥进不去。 模型密钥、数据库口令都不在传给执行单元的环境变量里。
• 执行环境本身没有状态。 一次性容器加内存临时目录，跑完即销毁，不需要为每个函数维护常驻运行时。
• 参数转换和必填校验在 Java 侧完成。 Python 函数按声明的类型接收参数即可，不用自己处理字符串到数字的转换。

执行环境怎么接（Docker 本机/远程，或宿主沙箱）见隔离 Python 执行环境配置；对外接口见函数库与调用接口。

一、交换协议

Java 与执行单元之间只用标准输入输出交换 JSON，不通过命令行传用户代码或参数，也不落脚本文件。

请求（Java 写入子进程 stdin）：

{
  "code": "def main(a, b):\n    return a + b",
  "params": { "a": 2, "b": 40 },
  "entrypoint": "main",
  "mode": "execute"
}

Java 侧组装这段请求的代码很薄，长度校验和并发闸门都在同一个方法里完成：

import java.nio.charset.StandardCharsets;
import java.util.concurrent.Semaphore;
import com.alibaba.fastjson2.JSONObject;

public class IsolatedPythonExecutor {
  private static final Semaphore SLOTS = new Semaphore(2);

  public Object execute(String code, JSONObject params, String entrypoint, boolean lint) {
    if (code == null || code.isBlank() || code.getBytes(StandardCharsets.UTF_8).length > 65536) {
      throw new IllegalArgumentException("Python 代码不能为空且不能超过 64 KiB");
    }
    JSONObject request = JSONObject.of("code", code, "params", params == null ? new JSONObject() : params,
        "entrypoint", entrypoint, "mode", lint ? "lint" : "execute");
    byte[] input = request.toJSONString().getBytes(StandardCharsets.UTF_8);
    if (input.length > 131072) {
      throw new IllegalArgumentException("函数输入超过 128 KiB");
    }
    if (!SLOTS.tryAcquire()) {
      throw new IllegalStateException("Python 执行器繁忙，请稍后重试");
    }
    // 之后启动一次性执行单元、把 input 写进它的 stdin、按秒等待，再用 JSON 解码 stdout
  }
}

响应（执行单元写入 stdout）：

{ "ok": true, "data": 42 }
{ "ok": false, "error": "division by zero" }

ok=false 时 Java 直接把 error 作为失败原因返回；用户函数里的 print 会被重定向到一个受限缓冲（8 KiB），不会污染这段协议输出。

执行单元这一侧同样很短：读一行请求、按 mode 分流、把返回值序列化回一行 JSON。

import ast
import contextlib
import io
import json
import sys


class LimitedOutput(io.StringIO):
    def write(self, value):
        room = max(0, 8192 - self.tell())
        super().write(value[:room])
        return len(value)


def run(request):
    code = request.get("code", "")
    tree = ast.parse(code, filename="function.py")
    if request.get("mode") == "lint":
        return []
    functions = [node.name for node in tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))]
    if not functions:
        raise ValueError("代码需要至少一个顶层函数")
    entry = request.get("entrypoint") or functions[-1]
    namespace = {"__name__": "maxkb_function"}
    output = LimitedOutput()
    with contextlib.redirect_stdout(output), contextlib.redirect_stderr(output):
        exec(compile(tree, "function.py", "exec"), namespace, namespace)
        return namespace[entry](**request.get("params", {}))


def main():
    try:
        request = json.loads(sys.stdin.buffer.read(131073))
        response = json.dumps({"ok": True, "data": run(request)}, ensure_ascii=False, allow_nan=False)
    except BaseException as error:
        response = json.dumps({"ok": False, "error": str(error)[:1000]}, ensure_ascii=False)
    sys.stdout.write(response)

命名空间里的 __name__ 固定是 maxkb_function，所以源码里 if __name__ == "__main__": 分支不会被执行，写在那里的代码不会运行。

lint 模式

mode=lint 只做 ast.parse：语法正确时返回空数组，语法错误时把行列信息包成诊断数组返回，不会执行函数体。

{ "line": 1, "column": 9, "endLine": 1, "endColumn": 11, "message": "invalid syntax", "type": "error" }

字段与前端编辑器的诊断格式一一对应，所以语法错误能直接画成波浪线并弹出提示。它只覆盖语法层面，"这行代码运行起来对不对"要靠在调试抽屉里真的跑一次。

二、两种 runner 的命令形状

docker runner

docker [--host <引擎地址>] run --rm --pull=never --name maxkb-python-<uuid> \
  --network=none --read-only --cap-drop=ALL --security-opt=no-new-privileges --pids-limit=32 \
  --memory=256m --memory-swap=256m --cpus=1 --user=65534:65534 \
  --tmpfs=/tmp:rw,noexec,nosuid,size=16m --workdir=/tmp --log-driver=none -i \
  python:3.12-slim python -I -B -c <base64 引导脚本>

容器名带一次执行一个的随机后缀（maxkb-python-<uuid>），不挂载宿主机目录，也不挂载 Docker socket。

host runner

/usr/sbin/runuser -u sandbox -- python3 -I -B -c <base64 引导脚本>

子进程的工作目录是沙箱目录，环境变量只保留白名单（见第六节）。为什么默认用 runuser 而不是 su，见环境配置里的实测对比。

三、为什么用 base64 引导脚本，而不是直接传源码

两个独立的原因，缺一不可：

1. Windows 会破坏参数里的双引号。 直接把 runner.py 源码作为 python -c 的参数传，CreateProcess 会把它解析坏：容器能起来，但 Python 报 SyntaxError: unterminated string literal。改成不含空格、不含双引号的 base64 引导脚本（exec(__import__('base64').b64decode('...').decode())）后各平台行为一致。
2. 不落脚本文件，就没有属主问题。 官方 MaxKB 的做法是把代码写成一个 .py 文件再 su -c "exec(open(file).read())"，这需要把文件 chown 给沙箱用户——只有以 root 运行的部署才做得到。走 stdin 既避开了这个问题，也不会留下残留文件。

引导脚本只包含 base64 字符和单引号，所以即使经由 shell 传递也不会被改写；单元测试固定了这一点（断言引导脚本里不含 $、反引号、反斜杠和双引号，并且能解码回 runner.py 原文）。

四、资源限制与并发

docker runner 的这几项由容器参数强制；host runner 只有超时和输入输出限制来自 Java 侧，内存、CPU、进程数没有上限，这一点在环境配置的边界对比表里同样列出。

超时触发后杀的是提权进程，用户代码是它的子进程，因此会连同整棵进程树一起强制终止，避免在宿主上留下沙箱进程。

五、失败语义：一律拒绝，不降级

执行器在任何一环不可用时都返回明确错误，不会把用户代码改到 Java 进程里跑：

用户函数自身抛出的异常不算执行器故障，会原样作为失败原因返回（例如 division by zero），便于在前端直接展示。

六、执行单元能看到什么

• docker runner：容器内 uid 固定为 65534，根文件系统只读，/tmp 是 16 MiB 的内存文件系统，没有网络；传给 docker 客户端的环境变量走白名单（PATH、Windows 上的 SYSTEMROOT/WINDIR/TEMP/TMP/USERPROFILE，以及 DOCKER_HOST/DOCKER_CONTEXT/DOCKER_TLS_VERIFY/DOCKER_CERT_PATH 等 DOCKER_* 项），应用密钥不会进入容器。SYSTEMROOT 这类只为了让 Windows 上的 docker 客户端能正常启动，容器内看不到它们。
• host runner：子进程以沙箱用户身份运行，工作目录与 HOME/TMPDIR 都指向沙箱目录，环境变量只保留显式清单 PATH、LANG、LC_ALL、LC_CTYPE、TZ、TERM（不是 LC_* 通配），外加固定的 USER/LOGNAME/SHELL。

两者都不把模型密钥和数据库凭据传给用户代码；区别在于 host runner 的沙箱用户能读到宿主上"其他人可读"的文件，因此部署时要收紧 my.txt、secrets.txt 的权限。

七、前端怎么用起来

编辑器和调试抽屉是这条链路的两个入口，二者都走执行器，但用途不同。

编辑器实时语法检查。 Python 代码编辑器在内容变化后调用语法检查接口，把返回的行列诊断画成波浪线，鼠标悬停显示 message：

import { linter, type Diagnostic } from '@codemirror/lint'
import FunctionApi from '@/api/function-lib'

function getRangeFromLineAndColumn(state: any, line: number, column: number, end_column?: number) {
  const l = state.doc.line(line)
  const from = l.from + column
  const to_end_column = l.from + end_column
  return {
    from: from > l.to ? l.to : from,
    to: end_column && to_end_column < l.to ? to_end_column : l.to
  }
}

const regexpLinter = linter(async (view) => {
  const diagnostics: Diagnostic[] = []
  await FunctionApi.pylint(view.state.doc.toString()).then((ok) => {
    ok.data.forEach((element: any) => {
      const range = getRangeFromLineAndColumn(
        view.state,
        element.line,
        element.column,
        element.endColumn
      )
      diagnostics.push({
        from: range.from,
        to: range.to,
        severity: element.type,
        message: element.message
      })
    })
  })
  return diagnostics
})

请求失败时不会打断编辑，只是不显示诊断；语法检查只覆盖语法层面。

调试抽屉。 它按函数的输入参数定义生成表单（必填项带星号），把界面上填的值和函数保存的初始参数一起交给调试接口，再直接展示返回值或错误信息。函数自己抛的异常会原样出现在"输出"里，例如 division by zero。

保存过的函数还可以被工作流节点等 Java 侧调用方复用，走的是同一个 execute 入口；前端目前接的是列表、增删改查、调试和语法检查这几条路径。

八、验证

mvn '-Dtest=IsolatedPythonExecutorTest' '-Dsurefire.failIfNoSpecifiedTests=false' '-Dmaven.javadoc.skip=true' '-Dgpg.skip=true' test

测试覆盖：容器命令不含宿主挂载且带全部限制参数、引导脚本的 shell 安全性、两种 runner 的命令形状、引擎地址作为全局参数出现在子命令之前、host runner 的七条拒绝路径（平台、沙箱用户缺失、沙箱用户是 root、Java 进程非 root、沙箱目录缺失、沙箱目录所有人可写、找不到提权命令）、环境变量不泄漏父进程变量、无引擎时拒绝执行、超时后终止并清理、响应超过 262144 字节被拒绝。

真实隔离效果用环境配置里的脚本验证，容器内的四项检查（uid 是否为 65534、根文件系统只读、外部网络不可达、宿主密钥不可见）实测结果：

{ "uid": 65534, "secret_absent": true, "read_only": true, "network_blocked": true }

接口层面的执行验证见函数库与调用接口。

相关章节

• 隔离 Python 执行环境配置：本机 WSL2 + Docker Engine 安装（含手动步骤）、Docker over TCP、宿主沙箱与验证脚本。
• 函数库与调用接口：函数存储、调试、执行与参数转换。