大家好，我是十三！欢迎来到十三Tech。
低代码平台里的"代码块"节点是个经典难题：用户在前端写一段代码，服务端怎么把它安全地执行掉再把结果送回去？这道题没有银弹——任何能跑用户代码的服务都面临 RCE、文件越权、资源耗尽这一连串风险。字节开源的 Coze Studio 给了一个相当系统的工程答案，正好拿来拆一遍。
一、核心实现：两种模式的权衡
Coze 没有走"一刀切"的路子，而是设计了两套截然不同的执行引擎。这背后的原则是场景分离：生产环境里安全是第一要务，开发测试环境里便捷性更重要。同一个 Runner 接口、两种实现，按部署场景挑用哪条路径。
无论走哪条路径，请求在到达 Runner 之前都要先穿过上层统一的模块黑白名单——这是第一道独立的安全防线，我们放到后面讲。
1. 生产环境的安全方案：Sandbox Runner
生产方案的核心是纵深防御：把执行链条切成四层，每层只承担一个明确的安全职责。整条链路长这样：Go 主进程 → Python 调度脚本 → Deno 运行时 → Pyodide (Wasm)。
第一层 · Go 调度层：Go 不直接碰用户代码，而是通过 os/exec 启动一个独立的 Python 子进程，用 os.Pipe 建双向管道做进程间通信。这是入口，也是边界——Go 进程崩了不会污染宿主。
// file: coze/coze-studio/backend/infra/impl/coderunner/sandbox/runner.go
func (runner *runner) Run(ctx context.Context, request *coderunner.RunRequest) (*coderunner.RunResponse, error) {
if request.Language == coderunner.JavaScript {
return nil, fmt.Errorf("js not supported yet") // JS被明确禁用
}
// 1. 将所有请求信息序列化为 JSON
b, err := json.Marshal(req{
Config: runner.config,
Code: request.Code,
Params: request.Params,
})
if err != nil {
return nil, err
}
// 2. 创建用于 "Go -> Python" 和 "Python -> Go" 的双向管道
pr, pw, err := os.Pipe() // pw (Go写) -> pr (Python读)
if err != nil {
return nil, err
}
r, w, err := os.Pipe() // w (Python写) -> r (Go读)
if err != nil {
return nil, err
}
// 3. 将数据写入管道并立即关闭写端
if _, err = pw.Write(b); err != nil {
return nil, err
}
if err = pw.Close(); err != nil {
return nil, err
}
// 4. 启动 Python 脚本子进程
cmd := exec.Command(runner.pyPath, runner.scriptPath)
// 5. 【关键】通过 ExtraFiles 将管道的文件描述符传递给子进程
cmd.ExtraFiles = []*os.File{w, pr}
if err = cmd.Start(); err != nil {
return nil, err
}
if err = w.Close(); err != nil { // 关闭Go端的写管道
return nil, err
}
// 6. 从输出管道中解码 Python 脚本返回的 JSON 结果
result := &resp{}
d := json.NewDecoder(r)
d.UseNumber() // 确保数字精度
if err = d.Decode(result); err != nil {
return nil, err
}
if err = cmd.Wait(); err != nil { // 等待子进程结束，回收资源
return nil, err
}
// 7. 检查执行状态
if result.Status != "success" {
return nil, fmt.Errorf("exec failed, stdout=%s, stderr=%s, sandbox_err=%s",
result.Stdout, result.Stderr, result.SandboxError)
}
return &coderunner.RunResponse{Result: result.Result}, nil
}
第二、三、四层 · Python 调度与 Deno 执行：Python 脚本是整个沙箱的桥接层。它从 fd 3/4 读出 Go 传来的请求，组装一条 deno run 命令，把权限 flag、V8 内存限制、Pyodide JSR 包一并拼好，然后用 subprocess 把活儿派给 Deno。
# file: coze/coze-studio/backend/infra/impl/coderunner/script/sandbox.py
class Sandbox:
def __init__(self, *, allow_env=False, allow_read=False, allow_write=False,
allow_net=False, allow_run=False, allow_ffi=False,
node_modules_dir="auto", **kwargs) -> None:
# 1. 根据 Go 传来的配置，生成 Deno 的权限参数列表
self.permissions = []
perm_defs = [
("--allow-env", allow_env, None),
("--allow-read", allow_read, ["node_modules"]),
("--allow-write", allow_write, ["node_modules"]),
("--allow-net", allow_net, None),
("--allow-run", allow_run, None),
("--allow-ffi", allow_ffi, None),
]
for flag, value, defaults in perm_defs:
perm = self._build_permission_flag(flag, value=value)
if perm is None and defaults is not None:
perm = f"{flag}={','.join(defaults)}"
if perm:
self.permissions.append(perm)
self.permissions.append(f"--node-modules-dir={node_modules_dir}")
def execute(self, code, *, timeout_seconds=None, memory_limit_mb=100, **kwargs) -> Output:
# 2. 组装 deno run 命令
cmd = ["deno", "run"]
cmd.extend(self.permissions)
# 3. 设置 V8 引擎参数（内存限制等）
v8_flags = ["--experimental-wasm-stack-switching"]
if memory_limit_mb is not None and memory_limit_mb > 0:
v8_flags.append(f"--max-old-space-size={memory_limit_mb}")
cmd.append(f"--v8-flags={','.join(v8_flags)}")
# 4. 运行一个基于 Pyodide (Python in Wasm) 的 JSR 包
cmd.append("jsr:@langchain/pyodide-sandbox@0.0.4")
cmd.extend(["--code", code])
# 5. 通过子进程运行 Deno，并捕获输出
try:
process = subprocess.run(cmd, capture_output=True, text=False,
timeout=timeout_seconds, check=False)
stdout = process.stdout.decode("utf-8", errors="replace")
if stdout:
full_result = json.loads(stdout)
result = full_result.get("result", None)
status = "success" if full_result.get("success", False) else "error"
else:
stderr = process.stderr.decode("utf-8", errors="replace")
status = "error"
except subprocess.TimeoutExpired:
status = "error"
stderr = f"Execution timed out after {timeout_seconds} seconds"
return Output(status=status, result=result, ...)
if __name__ == "__main__":
# 从 Go 进程传递来的文件描述符中读取输入
w = os.fdopen(3, "wb") # 写给 Go
r = os.fdopen(4, "rb") # 从 Go 读
try:
req = json.load(r)
user_code, params, config = req["code"], req["params"], req["config"] or {}
sandbox = Sandbox(**config)
# 包装用户代码为完整的 Python 脚本
if params is not None:
code = prefix + f'args={json.dumps(params)}\n' + user_code + suffix
else:
code = prefix + user_code + suffix
resp = sandbox.execute(code, **config)
# 将最终结果写回给 Go 进程
result = json.dumps(dataclasses.asdict(resp), ensure_ascii=False)
w.write(str.encode(result))
w.flush()
w.close()
except Exception as e:
# 异常情况下也要回传错误信息
w.write(str.encode(json.dumps({"sandbox_error": str(e)})))
w.flush()
w.close()
走到这里，用户的 Python 代码最终是在 Deno 严格限制权限、并跑在 WebAssembly 虚拟机内的 Pyodide 解释器里执行。四层嵌套不是为了花哨，而是把"用户代码能造成的伤害"逐层削掉——Go 隔离进程、Deno 卡住权限、Wasm 拦住系统调用。
2. 开发环境的便捷方案：Direct Runner
direct/runner.go 是另一个极端：直接通过 exec.Command 调系统装的 Python 解释器跑代码，没有沙箱、没有权限隔离。源码里一句 // ignore_security_alert RCE 把它的定位说得很清楚——只用于本地或受信任的测试场景。生产环境绝不能走这条路径。
3. 统一的安全前置校验
无论选哪个 Runner，代码在到达执行器之前，都要先经过上层业务逻辑里的一道前置校验——主要针对 Python 的 import 语句。这一步独立于沙箱实现，是 Runner 之外的第一道闸。
// file: coze/coze-studio/backend/domain/workflow/internal/nodes/code/code.go
// 完整的模块黑名单
var pythonBuiltinBlacklist = map[string]struct{}{
"curses": {}, "dbm": {}, "multiprocessing": {}, "threading": {},
"socket": {}, "pty": {}, "tty": {}, "fcntl": {}, "grp": {},
"pwd": {}, "resource": {}, "syslog": {}, "termios": {}, // ... 更多模块
}
// 第三方库白名单（非常有限）
var pythonThirdPartyWhitelist = map[string]struct{}{
"requests_async": {},
"numpy": {},
}
func validatePythonImports(code string) error {
imports := parsePythonImports(code) // 解析代码中的所有 import 语句
importErrors := make([]string, 0)
var blacklistedModules []string
var nonWhitelistedModules []string
for _, imp := range imports {
if _, ok := pythonBuiltinModules[imp]; ok {
// 检查是否在内置模块的黑名单中
if _, blacklisted := pythonBuiltinBlacklist[imp]; blacklisted {
blacklistedModules = append(blacklistedModules, imp)
}
} else {
// 检查是否在第三方库的白名单中
if _, whitelisted := pythonThirdPartyWhitelist[imp]; !whitelisted {
nonWhitelistedModules = append(nonWhitelistedModules, imp)
}
}
}
if len(blacklistedModules) > 0 {
moduleNames := fmt.Sprintf("'%s'", strings.Join(blacklistedModules, "', '"))
importErrors = append(importErrors, fmt.Sprintf(
"ModuleNotFoundError: The module(s) %s are removed from the Python standard library for security reasons",
moduleNames))
}
if len(nonWhitelistedModules) > 0 {
moduleNames := fmt.Sprintf("'%s'", strings.Join(nonWhitelistedModules, "', '"))
importErrors = append(importErrors, fmt.Sprintf(
"ModuleNotFoundError: No module named %s", moduleNames))
}
if len(importErrors) > 0 {
return errors.New(strings.Join(importErrors, ","))
}
return nil
}
逻辑很直白：内置模块走黑名单（socket、multiprocessing、pty 这类有系统副作用的直接拒掉），第三方库走白名单（默认只放 numpy、requests_async 这种"无害"的）。这套规则和 Coze 官方文档 的说明完全一致——它独立于 Runner 实现，是第一道、也是最廉价的一道防线。
二、执行流程
把 Sandbox Runner 的完整执行流程串起来，每一步的职责就一目了然了。
实线是调用方向、虚线是结果回程：用户请求触发后，Go 后端先做模块黑白名单校验；通过后 exec 启动 Python 子进程并用 os.Pipe 传代码；Python 组装 deno run 命令、设好权限 flag 和 V8 内存上限，把任务交给 Deno；用户代码在 Wasm 虚拟机内执行；结果沿调用链原路返回，最终由 Go 封装成响应送回用户。
三、回到最初的问题
回到文章开头那个问题——低代码平台的"代码块"功能到底怎么做到安全又稳定？Coze 给的答案可以归纳成三条原则，纵深防御 + 场景分离 + 源头拦截：
- 纵深防御靠分层：
Go → Python → Deno → Wasm四层嵌套，单层失守不会穿透到宿主，每一层都能独立审计。 - 场景分离靠接口：同一
Runner接口，生产走 Sandbox、开发走 Direct，把"零信任"和"便捷"放在两条不同路径上。 - 源头拦截靠前置校验：模块黑白名单在代码执行前就拆掉"炸弹引信"，比事后修复便宜得多。
这套组合拳的可贵之处不在任何单点，而在于它把"用户代码不可信"这个前提，铺到了每一层的设计里。
总结
Coze 的代码块节点设计为低代码平台的安全架构提供了一个完整范本：分层沙箱隔离 + 双执行器场景分离 + 模块前置校验。这些做法不只适用于代码执行场景——任何要"跑用户输入"的系统，都能从这套纵深防御里找到可以直接借鉴的工程模式。
Make Open Source Great Again!
关于十三 Tech
资深服务端研发工程师，AI 编程实践者。 专注分享真实的技术实践经验，相信 AI 是程序员的最佳搭档。 希望能和大家一起写出更优雅的代码！
联系方式：569893882@qq.com GitHub：@TriTechAI VX：TriTechAI（备注：十三 Tech）
