两大 AI Agent 平台源码分析:代码执行沙箱机制对比(Dify vs Coze)

两大 AI Agent 平台源码分析:代码执行沙箱机制对比(Dify vs Coze)

随着 AI Agent 平台的日益成熟,越来越多的产品提供了“代码执行”能力,使得 Agent 能以更强的自主性完成任务。然而,执行用户自定义代码也带来了安全隐患,这就必须引入“代码沙箱”机制来隔离风险。

本篇我们来深入剖析两大主流平台 —— DifyCoze 的源码实现,从架构设计、安全机制、性能代价等多个角度分析它们的沙箱实现差异。

⚠️ 两个版本:原文(纯手敲)和AI整理过的原文

Dify:

Dify的设计架构里代码沙箱是放在一个单独的docker容器里执行的,这和coze有很大的区别。

这是dify代码沙箱的项目:https://github.com/langgenius/dify-sandbox


Dify主服务如何调用的

Dify中代码执行作为一个内置工具来使用,在源码的/api/core/tools/builtin_tool/providers/code/tools/simple_code.py#L30路径里,
调用了CodeExecutor.execute_code方法,我们查看execute_code方法的代码/api/core/helper/code_executor/code_executor.py#L60,
方法调用了dify-sandbox服务的Api接口/v1/sandbox/run,到此dify主服务如何调用执行代码沙箱完毕,我们继续看dify-sandbox源码。


dify-sandbox

首先从路由开始/internal/controller/router.go#L36,路由的执行函数为RunSandboxController,然后我们看执行函数中的service.RunPython3Code
RunPython3Code中就是整个代码执行的核心了,主要在这段代码:

1
2
3
4
runner := python.PythonRunner{}
stdout, stderr, done, err := runner.Run(
code, timeout, nil, preload, options,
)

runner.Run方法中调用了InitializeEnvironment方法,这个方法做了几件事:

  • 使用uuid给代码文件命名

  • 给执行代码模版文件prescript.py里的uid、gid、enable_network、preload、code赋值,源码中使用embed将prescript.py文件内容作为了sandbox_fs变量的值

    1
    2
    //go:embed prescript.py
    var sandbox_fs []byte
  • 加密代码并base64后写入LIB_PATH路径中,最终方法返回代码文件的路径和b64后的加密key

然后我们回到Run方法中,接着就是拼接cmd执行命令了,

1
2
3
4
5
6
7
// create a new process
cmd := exec.Command(
configuration.PythonPath,
untrusted_code_path,
LIB_PATH,
key,
)

本质上就是 python /var/sandbox/sandbox-python/tmp/$uuid.py /var/sandbox/sandbox-python b64-key

我们再看下prescript.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import ctypes
import os
import sys
import traceback
# setup sys.excepthook
def excepthook(type, value, tb):
sys.stderr.write("".join(traceback.format_exception(type, value, tb)))
sys.stderr.flush()
sys.exit(-1)

sys.excepthook = excepthook

lib = ctypes.CDLL("./python.so")
lib.DifySeccomp.argtypes = [ctypes.c_uint32, ctypes.c_uint32, ctypes.c_bool]
lib.DifySeccomp.restype = None

# get running path
running_path = sys.argv[1]
if not running_path:
exit(-1)

# get decrypt key
key = sys.argv[2]
if not key:
exit(-1)

from base64 import b64decode
key = b64decode(key)

os.chdir(running_path)

{{preload}}

lib.DifySeccomp({{uid}}, {{gid}}, {{enable_network}})

code = b64decode("{{code}}")

def decrypt(code, key):
key_len = len(key)
code_len = len(code)
code = bytearray(code)
for i in range(code_len):
code[i] = code[i] ^ key[i % key_len]
return bytes(code)

code = decrypt(code, key)
exec(code)

文件中主要是Seccomp比较特殊, 这也是这个项目的重点,调用文件在/internal/core/lib/python/add_seccomp.go

seccomp 是Linux 内核提供的一种安全机制,全称是Secure Computing Mode。它允许您限制进程可以使用的系统调用,从而增强系统的安全性。通过seccomp,您可以限制进程能进行的系统调用,减少系统的暴露面,防止潜在的安全风险。

add_seccomp.go文件中InitSeccomp方法主要以下几点:

  1. 创建一个受限的执行环境(沙盒)
  2. 限制 Python 代码可以执行的系统调用
  3. 设置适当的用户权限
  4. 防止特权提升

文件系统隔离

  • 使用chroot 创建了一个隔离的文件系统环境,进程只能访问当前目录及其子目录
    1
    2
    syscall.Chroot(".")  // 将进程限制在当前目录
    syscall.Chdir("/") // 在chroot后切换到根目录

权限控制

  • 确保进程及其子进程不能获取更多权限
    1
    lib.SetNoNewPrivs()  // 防止进程提升权限

系统调用过滤

  • 使用 seccomp 机制限制进程可以执行的系统调用
  • 有两个列表:
    • allowed_syscalls: 完全允许的系统调用
    • allowed_not_kill_syscalls: 允许但不杀死进程的系统调用(即使失败也继续)
      1
      lib.Seccomp(allowed_syscalls, allowed_not_kill_syscalls)

用户权限设置

  • 将进程权限降为指定的非特权用户
    1
    2
    syscall.Setuid(uid)  // 设置用户ID
    syscall.Setgid(gid) // 设置组ID
    总结一下就是dify sandbox使用了seccomp和文件隔离来保证不受信代码的运行。

Coze:

Coze目前开源的代码执行就相对没那么复杂,并没有单独作为一个服务来管理,而是直接在主服务中进行,分了两种策略:
官方线上版本通过分析使用的是sandbox策略的(Deno+pyodide)

1
2
3
4
5
# Workflow Code Runner Configuration
# Supported code runner types: sandbox / local
# Default using local
# - sandbox: execute python code in a sandboxed env with deno + pyodide
# - local: using venv, no env isolation

核心执行代码在以下目录:

https://github.com/coze-dev/coze-studio/tree/main/backend/infra/impl/coderunner

  • direct目录下为runner types为local时的策略
  • sandbox目录下为runner types为sandbox时的策略
  • script目录下为两种策略的执行代码模版文件

direct里就一个runner.go,代码也很明确,直接将节点用户的代码传入进来后,和var pythonCode拼接了一下,然后就是命令行的调用执行,没有任何的安全措施。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
// /backend/infra/impl/coderunner/direct/runner.go

var pythonCode = `
import asyncio
import json
import sys

class Args:
def __init__(self, params):
self.params = params

class Output(dict):
pass

%s

try:
result = asyncio.run(main( Args(json.loads(sys.argv[1]))))
print(json.dumps(result))
except Exception as e:
print(f"{type(e).__name__}: {str(e)}", file=sys.stderr)
sys.exit(1)

`
func (r *runner) pythonCmdRun(_ context.Context, code string, params map[string]any) (map[string]any, error) {
bs, _ := sonic.Marshal(params)
cmd := exec.Command(goutil.GetPython3Path(), "-c", fmt.Sprintf(pythonCode, code), string(bs)) // ignore_security_alert RCE
stdout := new(bytes.Buffer)
stderr := new(bytes.Buffer)
cmd.Stdout = stdout
cmd.Stderr = stderr
err := cmd.Run()

script里的python_script.py貌似目前在Coze项目里没有使用到,但是也是一直运行不受信代码的方案,使用了RestrictedPython项目。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
try:
from RestrictedPython import safe_builtins, limited_builtins, utility_builtins
except ModuleNotFoundError:
print("RestrictedPython module required, please run pip install RestrictedPython",file=sys.stderr)
sys.exit(1)

custom_builtins = safe_builtins.copy()

custom_builtins['__import__'] = __import__
custom_builtins['asyncio'] = asyncio
custom_builtins['json'] = json
custom_builtins['time'] = time
custom_builtins['random'] = random

restricted_globals = {
'__builtins__': custom_builtins,
'_utility_builtins': utility_builtins,
'_limited_builtins': limited_builtins,
'__name__': '__main__',
'dict': dict,
'list': list,
'print': print,
'set': set,

}

class Args:
def __init__(self, params):
self.params = params


DefaultCode = """
class Args:
def __init__(self, params):
self.params = params
class Output(dict):
pass
"""


async def run_main(app_code, params):
try:
complete_code = DefaultCode + app_code
locals_dict = {"args": Args(params=params)}
exec(complete_code, restricted_globals, locals_dict) # ignore_security_alert
main_func = locals_dict['main']
ret = await main_func(locals_dict['args'])
except Exception as e:
print(f"{type(e).__name__}: {str(e)}", file=sys.stderr)
sys.exit(1)
return ret


sandbox 使用的Deno在run时安装jsr:@langchain/pyodide-sandbox@0.0.4来做的沙箱,具体代码在/backend/infra/impl/coderunner/script/sandbox.py,
这里比较特殊的就是sandbox/runner.go里的Run方法使用了两个管道来和python进程数据交互。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
// /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")
}
b, err := json.Marshal(req{
Config: runner.config,
Code: request.Code,
Params: request.Params,
})
if err != nil {
return nil, err
}
pr, pw, err := os.Pipe()
if err != nil {
return nil, err
}
r, w, err := os.Pipe()
if err != nil {
return nil, err
}
if _, err = pw.Write(b); err != nil {
return nil, err
}
if err = pw.Close(); err != nil {
return nil, err
}
cmd := exec.Command(runner.pyPath, runner.scriptPath)
cmd.ExtraFiles = []*os.File{w, pr}
if err = cmd.Start(); err != nil {
return nil, err
}
if err = w.Close(); err != nil {
return nil, err
}
result := &resp{}
d := json.NewDecoder(r)
...省略部分代码
1
2
cmd := exec.Command(runner.pyPath, runner.scriptPath)
cmd.ExtraFiles = []*os.File{w, pr}

这里将文件描述符(fd3 fd4)传递给python子进程。

为什么是 3 和 4?
这与 UNIX/Linux 下文件描述符有关:

  • 标准输入 stdin:fd=0
  • 标准输出 stdout:fd=1
  • 标准错误 stderr:fd=2
  • 文件描述符从 3 开始就是额外自定义的文件/管道。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# /backend/infra/impl/coderunner/script/sandbox.py
...省略部分代码

if __name__ == "__main__":
w = os.fdopen(3, "wb", )
r = os.fdopen(4, "rb", )

try:
req = json.load(r)
user_code, params, config = req["code"], req["params"], req["config"] or {}
sandbox = Sandbox(**config)

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)
result = json.dumps(dataclasses.asdict(resp), ensure_ascii=False)
w.write(str.encode(result))
w.flush()
w.close()
except Exception as e:
print("sandbox exec error", e)
w.write(str.encode(json.dumps({"sandbox_error": str(e)})))
w.flush()
w.close()

这里 os.fdopen(3, “wb”) 和 os.fdopen(4, “rb”) 的意思是:

w:打开文件描述符 3 作为二进制写入流(写给父进程/调用者)。
r:打开文件描述符 4 作为二进制读取流(从父进程/调用者读数据)。
后面通过 r 读取请求,通过 w 写回结果。

Go 端描述 子进程中 fd Python 中对应的变量
w (Go写→Python读) fd=3 w = os.fdopen(3, "wb")
pr (Python写→Go读) fd=4 r = os.fdopen(4, "rb")

所以Coze这里每执行一个代码节点,就会产生一个python和deno进程。 高并发下会?


相关沙箱项目:


⚠️ 注意:以下是AI整理过的 选择性阅读📖

⚠️ 注意:以下是AI整理过的 选择性阅读📖

一、Dify 的代码执行沙箱设计

独立服务架构:沙箱容器化执行

Dify 采用了 服务解耦 + 容器化 的方式来实现代码执行沙箱功能。沙箱运行逻辑被独立为一个专门的服务:
👉 https://github.com/langgenius/dify-sandbox

在 Dify 主服务中,代码执行被封装为内置工具,具体调用路径如下:

  • 调用入口:/api/core/tools/builtin_tool/providers/code/tools/simple_code.py
  • 核心方法:CodeExecutor.execute_code()
  • 实际调用:访问 dify-sandbox/v1/sandbox/run 接口

这一设计具备高安全性和可扩展性,尤其适合多用户环境。


运行流程详解

dify-sandbox 源码看,核心逻辑大致如下:

  1. 路由 /v1/sandbox/run 对应控制器函数 RunSandboxController

  2. 执行函数调用 service.RunPython3Code,其内部通过:

    1
    2
    3
    4
    runner := python.PythonRunner{}
    stdout, stderr, done, err := runner.Run(
    code, timeout, nil, preload, options,
    )
  3. runner.Run() 中调用 InitializeEnvironment,完成如下准备:

    • 使用 uuid 命名代码文件
    • 使用 embed 模板(prescript.py)动态插入用户代码
    • 对代码加密并 Base64 编码,写入隔离目录
  4. 拼接命令调用:

    1
    cmd := exec.Command("python", code_path, lib_path, key)

    实质上就是运行一段带有沙箱逻辑的 Python 脚本。


prescript.py:安全核心逻辑

prescript.py 是沙箱的“引导器”,内容包括:

  • 设置 sys.excepthook 捕获异常
  • 使用 ctypes 调用编译好的 C 函数:DifySeccomp(...)
  • 解密用户代码后运行

其中 DifySeccomp 来源于 Go 文件 add_seccomp.go,封装了 Linux 系统调用控制逻辑。


Seccomp + chroot 的安全防护

Dify 的安全实现堪称严密:

  • ✅ 使用 seccomp 限制系统调用

  • ✅ 使用 chroot 文件系统隔离

  • ✅ 降低进程权限(Setuid, Setgid

  • ✅ 禁止权限提升(SetNoNewPrivs

其整体设计让人想起 Google 的 gVisor,是对“不可信代码执行”极为严肃的处理。


二、Coze 的代码执行机制

更加轻量化的实现策略

相较 Dify,Coze 的代码执行没有单独服务,而是内嵌在主服务中,策略支持两种模式:

1
# Supported code runner types: sandbox / local
  • local: 无隔离,直接使用本地 venv 环境

  • sandbox: 使用 Deno + Pyodide 做轻度沙箱隔离

代码路径参考:
👉 https://github.com/coze-dev/coze-studio/tree/main/backend/infra/impl/coderunner


Local Runner:无隔离执行

在 direct/runner.go 中实现了 local 模式:

1
cmd := exec.Command(goutil.GetPython3Path(), "-c", fmt.Sprintf(pythonCode, code), string(bs))
  • 用户代码拼接后直接通过 python -c 执行

  • 无权限控制、无文件隔离、无系统调用限制

  • 存在极大安全风险

这种方式适合内部信任环境,但不推荐对外开放使用。


Script Runner:RestrictedPython(暂未使用)

虽然 Coze 项目中包含了对 RestrictedPython 的集成代码(见 script/python_script.py),但目前尚未启用。

  • 提供有限内置函数

  • 只允许部分模块导入

  • 类似浏览器沙箱的逻辑

这部分潜力值得关注,未来可能用于更安全的轻量级沙箱策略。


Sandbox Runner:Deno + Pyodide + IPC 机制

sandbox/runner.go 中,Coze 采用了 Deno + Pyodide 的混合沙箱模型,并使用 Go 与 Python 间的双管道通信:

1
cmd.ExtraFiles = []*os.File{w, pr}

在Python子进程中:

1
2
w = os.fdopen(3, "wb")
r = os.fdopen(4, "rb")

实现了 Go → Python 参数传递 & Python → Go 结果回写 的双向数据流。

优点:

  • 使用 Pyodide 运行代码,避免直接触达系统资源

  • 通过 fd 通信,不暴露敏感环境变量

缺点:

  • 每次执行拉起 Python + Deno 两个进程

  • 并发性能存在瓶颈,需额外优化


三、总结与思考

项目 执行隔离 权限控制 性能成本 易用性 安全级别
Dify 容器+chroot Seccomp, Setuid 中等 中等 ✅ 高
Coze-local ❌ 低
Coze-sandbox Pyodide+Deno js 沙箱+fd隔离 ✅ 中
  • Dify 的沙箱实现适合需要强隔离、高安全的生产环境,尤其是对外提供 Agent 服务时。

  • Coze 的策略则更灵活,适合在轻量场景中快速运行,未来若完善 RestrictedPython 与 Deno 沙箱结合,将具备更强潜力。


🔗 推荐阅读与项目参考


两大 AI Agent 平台源码分析:代码执行沙箱机制对比(Dify vs Coze)
http://example.com/2025/07/28/sandbox/
作者
vvanglro
发布于
2025年7月28日
许可协议