沙箱改用系统自带的隔离原语,移除 Docker 依赖

跑一段不可信代码不需要一整套容器运行时。macOS 用 Seatbelt
(sandbox-exec,系统自带),Linux 用 bubblewrap;冷启动从 ~250ms
降到 ~25ms,且不再需要装任何东西。

守住三件事:默认断网、家目录不可读、只有工作区可写。策略由内核
强制且对子进程继承——沙箱内 subprocess 出来的 cat 读家目录同样是
Operation not permitted。解释器改用 sys.base_prefix 下那个干净的
Python,沙箱代码因此 import 不到后端的 FastAPI、SQLAlchemy 和凭据
处理代码。

如实标注能力边界,不给虚假的安全感:
- macOS 不强制 RLIMIT_AS,内存限不住(实测设 256MB 仍可分配 900MB),
  health 接口申报 enforced 字段,界面把限不住的项划掉
- Seatbelt 是访问控制而非虚拟化:进程表共享,以当前用户身份运行
- bubblewrap 侧多隔离 PID/IPC/挂载视图,内存限额真实生效

顺带修两个 bug:
- LocalSandbox 设 RLIMIT_NPROC=64 会误伤——macOS 上它按 uid 全局
  计数,用户已有上千进程时沙箱内任何 fork 都失败,连 bash 都起不来
- 模型所属 provider 被停用时,原先静默换一家继续用原模型名,必然
  404 且报错指不到原因;改为直接说清是哪个 provider 被停用了

写 SBPL 的坑记在注释里:(deny default) 会让 Python 直接 SIGABRT,
改用「默认允许 + 精确拒绝」。

7 个模板全部回归通过。
This commit is contained in:
ThearchyHelios
2026-09-13 18:04:36 +08:00
parent 964eeaa829
commit dd57c12016
12 changed files with 589 additions and 375 deletions
+2 -2
View File
@@ -6,10 +6,10 @@
# AGENTLAB_WEB_PORT=5273
# ---- 沙箱 ----
# auto(有 Docker 用 Docker,否则降级本地子进程)| docker | local | off
# auto(macOS→seatbelt,Linux→bubblewrap,都没有则 local)| seatbelt | bubblewrap | local | off
# AGENTLAB_SANDBOX_BACKEND=auto
# AGENTLAB_SANDBOX_IMAGE=python:3.12-slim
# AGENTLAB_SANDBOX_TIMEOUT=30
# 内存限额仅在 Linux 生效,macOS 不强制 RLIMIT_AS
# AGENTLAB_SANDBOX_MEMORY_MB=512
# AGENTLAB_SANDBOX_NETWORK=false
+25 -10
View File
@@ -20,7 +20,7 @@
## 快速开始
需要 Python 3.12+(`uv` 会自动装)、Node 20+、pnpm;Docker 可选但强烈建议(代码沙箱靠它做隔离)。
需要 Python 3.12+(`uv` 会自动装)、Node 20+、pnpm。**不需要 Docker** —— 代码沙箱用操作系统自带的隔离原语。
```bash
./scripts/dev.sh
@@ -44,7 +44,7 @@
| **可视化编排** | 15 种节点拖拽连线;运行时节点实时高亮、边上数据流动画、卡片里直接看到流式 token |
| **多模型接入** | Anthropic / OpenAI / 任意 OpenAI 兼容服务(DeepSeek、Kimi、通义、智谱、硅基流动、Ollama…)+ Mock |
| **工具链** | 15 个内置工具 + 自定义工具(HTTP 模板 / 沙箱 Python)+ MCP server 接入 |
| **代码沙箱** | Docker 隔离:断网、只读根文件系统、非 root、内存/CPU/进程数限额、超时强杀 |
| **代码沙箱** | macOS 用 Seatbelt、Linux 用 bubblewrap:断网、家目录不可读、只有工作区可写、超时强杀,冷启动 ~25ms |
| **人工介入** | 图暂停并落盘,人在界面上批准/驳回/改稿,从断点继续——进程重启也不丢 |
| **记忆与知识库** | 跨运行的长期记忆(自动去重)+ 文档知识库(向量 + BM25 混合检索,可调配比) |
| **Skill 管理** | 把「怎么做事」抽成可复用的方法论,挂到节点上注入 system prompt |
@@ -85,7 +85,7 @@ backend/ FastAPI + LangGraph
expressions.py 模板插值 + AST 白名单表达式求值
nodes/ 各类节点的执行器
app/providers/ 多模型接入与成本估算
app/sandbox/ Docker / 本地子进程双后端
app/sandbox/ Seatbelt / bubblewrap / 裸子进程三后端
app/tools/ 工具注册表、内置工具、MCP、自定义工具
app/memory/ 记忆、知识库、混合检索
app/api/ REST + WebSocket
@@ -111,11 +111,24 @@ LangGraph 的 `@task` 里,结果进 checkpoint,重放时直接取缓存。
`RunEvent` 落库并广播。画布高亮、token 流、工具卡片、时间线全部由同一份事件驱动,
所以刷新页面或中途接入都能拿到完整过程(WebSocket 先补历史再接实时)。
**沙箱的隔离姿态。** 容器 `read_only` + tmpfs 工作区 + `cap_drop ALL` +
`no-new-privileges` + 非 root + 默认断网 + 内存/CPU/pids 限额 + 超时整容器销毁。
注意 Docker 的 `put_archive` 在只读容器上会被拒,所以代码是用 exec + base64 管道写进去的
——为的是不牺牲只读根文件系统。Docker 不可用时降级到子进程 + rlimit,
**那条路径没有内核级隔离**,界面上会明确标注。
**沙箱不用容器。** 跑一段不可信代码不需要一整套容器运行时——操作系统自己就有
内核级的强制访问控制。macOS 用 Seatbelt(`sandbox-exec`,系统自带),Linux 用
bubblewrap(几百 KB 的二进制)。冷启动 ~25ms,对比容器方案的 ~250ms,且不用装任何东西。
守住的是三件事:**默认断网**、**家目录不可读**、**只有工作区可写**。策略由内核强制且
对子进程继承——沙箱里 `subprocess` 出来的 `cat` 读家目录一样是 `Operation not permitted`。
解释器刻意用 `sys.base_prefix` 下那个干净的 Python,而不是后端自己的 venv,
所以沙箱代码 import 不到 FastAPI、SQLAlchemy 和凭据处理相关的任何东西。
**和容器比,少了什么要说清楚**:Seatbelt 做的是访问控制而非虚拟化——进程表是共享的
(能看到宿主机进程列表),代码以当前用户身份运行(不像容器会降到 nobody),
而且 **macOS 不强制 `RLIMIT_AS`,内存用量限不住**(实测设 256MB 仍能分配 900MB)。
CPU 时间、单文件大小、墙钟超时这三项有效。bubblewrap 那边多隔离了 PID/IPC/挂载视图,
内存限额也真实生效。要更强的隔离(内存配额、完整虚拟化),该上的是 gVisor 或 microVM,
而不是回到容器。
写 SBPL 策略有个坑:`(deny default)` 会让 Python 直接 SIGABRT 起不来——要枚举的路径和
syscall 太多,漏一个就崩。所以用的是「默认允许 + 精确拒绝」,只把那三件事收紧。
**检索为什么是混合的。** 默认 embedding 是本地特征哈希,不联网不下模型,零配置可用,
但语义泛化弱;BM25 补上精确关键词匹配这一半。两者各自归一化后加权,权重在界面上可调,
@@ -127,8 +140,10 @@ LangGraph 的 `@task` 里,结果进 checkpoint,重放时直接取缓存。
## 常见问题
**沙箱不可用?** 设置页「运行环境」会显示实际后端和原因。Docker 没启动时会自动降级到
本地子进程,功能可用但没有隔离。首次执行会拉 `python:3.12-slim` 镜像,需要等一会儿。
**沙箱用的是哪个后端?** 设置页「运行环境」会列出所有候选及其可用性,并标出当前选中的那个。
macOS 上应该是 `seatbelt`;Linux 上装了 `bwrap`(`apt install bubblewrap`)就是 `bubblewrap`,
没装则降级到 `local`——**那条路径没有任何访问控制**,界面上会明确标红。
可以用 `AGENTLAB_SANDBOX_BACKEND` 强制指定。
**模型返回空内容?** Claude 4.6 之后的模型默认开着 thinking,`max_tokens` 给小了会出现
「思考完就没额度写正文」。节点里把 max_tokens 调大,或把思考模式设为关闭。时间线里会有提示。
+1 -2
View File
@@ -31,8 +31,7 @@ class Settings(BaseSettings):
secret_key: str = ""
# --- 沙箱 ---
sandbox_backend: str = "auto" # auto | docker | local | off
sandbox_image: str = "python:3.12-slim"
sandbox_backend: str = "auto" # auto | seatbelt | bubblewrap | local | off
sandbox_timeout: int = 30
sandbox_memory_mb: int = 512
sandbox_cpus: float = 1.0
+12
View File
@@ -61,6 +61,18 @@ async def resolve_provider(
for p in enabled:
if p.default_model == model:
return p
# 模型认得出来,但它所属的 provider 被停用了。这时候不能默默换一家 ——
# 拿着 A 家的模型名去调 B 家,必然 404,而且报错完全指不到真正的原因。
for p in providers:
if not p.enabled and (
any((m or {}).get("id") == model for m in (p.models or []))
or p.default_model == model
):
raise ProviderNotConfigured(
f"模型 {model!r} 属于 provider {p.name!r},但它已被停用。"
f"请在设置里启用它,或给节点换一个模型。"
)
# 认不出来的模型名就放行:用户可能手填了一个目录里还没有的新模型
real = [p for p in enabled if p.kind != "mock"]
return (real or enabled or [None])[0]
+168
View File
@@ -0,0 +1,168 @@
from __future__ import annotations
import asyncio
import os
import resource
import shutil
import sys
import time
from pathlib import Path
from app.core.config import settings
from app.sandbox.base import ExecResult, Sandbox, SandboxLimits, truncate
_LANG = {
"python": ("main.py", ["python3", "-I", "-u", "/work/main.py"]),
"bash": ("main.sh", ["/bin/bash", "/work/main.sh"]),
"sh": ("main.sh", ["/bin/sh", "/work/main.sh"]),
"node": ("main.js", ["node", "/work/main.js"]),
"javascript": ("main.js", ["node", "/work/main.js"]),
}
# 只读挂进沙箱的系统目录,存在才挂
_RO_BINDS = ["/usr", "/lib", "/lib64", "/bin", "/sbin", "/etc/alternatives", "/etc/ssl"]
class BubblewrapSandbox(Sandbox):
"""Linux 上的对等方案:bubblewrap(Flatpak 的沙箱内核)。
一个几百 KB 的二进制,直接用 namespace + seccomp 建隔离环境,
不需要 daemon、不需要 root、不需要镜像。相比 Seatbelt 它还多隔离了
PID、IPC 和挂载视图,隔离强度更接近容器。
"""
name = "bubblewrap"
def __init__(self) -> None:
self._root = settings.workspace_dir / "bwrap"
self._bwrap = shutil.which("bwrap")
@staticmethod
def available() -> bool:
return sys.platform.startswith("linux") and shutil.which("bwrap") is not None
async def health(self) -> dict[str, object]:
ok = self.available()
return {
"backend": self.name,
"available": ok,
"isolated": ok,
"isolation": "Linux namespace(mount/pid/ipc/uts/net)+ 只读根",
"limits_note": "内存与 CPU 由 rlimit 限制;如需 cgroups 级配额请配合 systemd-run",
"enforced": {"timeout": True, "cpu": True, "memory": True, "network": True, "fsize": True},
"root": str(self._root),
**({} if ok else {"error": "未找到 bwrap,可执行 apt install bubblewrap 安装"}),
}
def _session_dir(self, session_id: str) -> Path:
safe = "".join(c for c in session_id if c.isalnum() or c in "-_")[:64] or "default"
path = self._root / safe
path.mkdir(parents=True, exist_ok=True)
return path
@staticmethod
def _preexec(limits: SandboxLimits): # pragma: no cover - 子进程里执行
def _apply() -> None:
resource.setrlimit(resource.RLIMIT_CPU, (limits.timeout, limits.timeout + 2))
# Linux 上 RLIMIT_AS 是真正生效的(macOS 不是)
nbytes = limits.memory_mb * 1024 * 1024
resource.setrlimit(resource.RLIMIT_AS, (nbytes, nbytes))
resource.setrlimit(resource.RLIMIT_FSIZE, (256 * 1024 * 1024,) * 2)
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
os.setsid()
return _apply
def _argv(self, workdir: Path, limits: SandboxLimits, cmd: list[str]) -> list[str]:
argv = [
self._bwrap or "bwrap",
"--unshare-all", # mount/pid/ipc/uts/cgroup/net 全部隔离
*(["--share-net"] if limits.network else []),
"--die-with-parent", # 父进程没了,沙箱跟着走
"--new-session", # 摘掉控制终端,防 TIOCSTI 注入
"--proc", "/proc",
"--dev", "/dev",
"--tmpfs", "/tmp",
"--bind", str(workdir), "/work",
"--chdir", "/work",
"--setenv", "HOME", "/work",
"--setenv", "TMPDIR", "/tmp",
"--setenv", "PATH", "/usr/bin:/bin",
"--setenv", "PYTHONDONTWRITEBYTECODE", "1",
]
for path in _RO_BINDS:
if Path(path).exists():
argv += ["--ro-bind", path, path]
return argv + ["--", *cmd]
async def run(
self,
code: str,
*,
language: str = "python",
limits: SandboxLimits | None = None,
session_id: str | None = None,
env: dict[str, str] | None = None,
files: dict[str, str] | None = None,
) -> ExecResult:
limits = limits or SandboxLimits()
language = (language or "python").lower()
if language not in _LANG:
return ExecResult(ok=False, backend=self.name, error=f"不支持的语言:{language}")
if not self.available():
return ExecResult(ok=False, backend=self.name, error="bwrap 不可用")
filename, cmd = _LANG[language]
workdir = self._session_dir(session_id or f"tmp-{int(time.time() * 1000)}")
(workdir / filename).write_text(code)
for rel, content in (files or {}).items():
target = (workdir / rel).resolve()
if not str(target).startswith(str(workdir.resolve())):
return ExecResult(ok=False, backend=self.name, error=f"非法文件路径:{rel}")
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(content)
argv = self._argv(workdir, limits, cmd)
for key, value in (env or {}).items():
argv[1:1] = ["--setenv", key, value]
started = time.perf_counter()
try:
proc = await asyncio.create_subprocess_exec(
*argv,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
preexec_fn=self._preexec(limits),
)
out, err = await asyncio.wait_for(proc.communicate(), timeout=limits.timeout)
except asyncio.TimeoutError:
try:
os.killpg(os.getpgid(proc.pid), 9)
except (ProcessLookupError, PermissionError):
proc.kill()
await proc.wait()
return ExecResult(
ok=False, backend=self.name, timed_out=True, exit_code=124,
duration_ms=int((time.perf_counter() - started) * 1000),
error=f"执行超过 {limits.timeout}s 被终止",
)
except Exception as e: # noqa: BLE001
return ExecResult(ok=False, backend=self.name, error=f"{type(e).__name__}: {e}")
stdout, t1 = truncate(out.decode(errors="replace"), limits.max_output)
stderr, t2 = truncate(err.decode(errors="replace"), limits.max_output)
code_ = proc.returncode or 0
return ExecResult(
ok=code_ == 0,
exit_code=code_,
stdout=stdout,
stderr=stderr,
truncated=t1 or t2,
timed_out=code_ in (-24, 152),
backend=self.name,
duration_ms=int((time.perf_counter() - started) * 1000),
files_written=[p.name for p in workdir.iterdir() if p.is_file() and p.name != filename][:50],
)
async def cleanup(self, session_id: str) -> None:
await asyncio.to_thread(shutil.rmtree, self._session_dir(session_id), True)
-314
View File
@@ -1,314 +0,0 @@
from __future__ import annotations
import asyncio
import base64
import shlex
import time
import uuid
from typing import Any
from app.core.config import settings
from app.sandbox.base import ExecResult, Sandbox, SandboxLimits, truncate
_WORKDIR = "/workspace"
_LANG_CMD = {
"python": ["python", "-u", "/workspace/.main.py"],
"bash": ["bash", "/workspace/.main.sh"],
"sh": ["sh", "/workspace/.main.sh"],
"node": ["node", "/workspace/.main.js"],
"javascript": ["node", "/workspace/.main.js"],
}
_LANG_FILE = {
"python": ".main.py",
"bash": ".main.sh",
"sh": ".main.sh",
"node": ".main.js",
"javascript": ".main.js",
}
# 单条 shell 命令里塞的 base64 分块大小。留足余量避开 ARG_MAX。
_CHUNK = 48 * 1024
_MAX_FILE_BYTES = 4 * 1024 * 1024
def _write_commands(files: dict[str, str]) -> list[list[str]]:
"""生成往容器里写文件的 shell 命令。
这里不用 docker 的 put_archive:那个 API 会被 `read_only` 的容器直接拒掉
(即便目标是可写的 tmpfs 挂载点)。改成 exec + base64 管道写入,
就能同时保住只读根文件系统和往 /workspace 落文件这两件事。
"""
commands: list[list[str]] = []
for path, content in files.items():
rel = path.lstrip("/")
if ".." in rel.split("/"):
raise ValueError(f"非法文件路径:{path}")
full = f"{_WORKDIR}/{rel}"
data = content.encode()
if len(data) > _MAX_FILE_BYTES:
raise ValueError(f"{path} 超过 {_MAX_FILE_BYTES // 1024 // 1024}MB 上限")
blob = base64.b64encode(data).decode()
quoted = shlex.quote(full)
commands.append(["sh", "-c", f"mkdir -p $(dirname {quoted})"])
if not blob:
commands.append(["sh", "-c", f": > {quoted}"])
continue
for i in range(0, len(blob), _CHUNK):
chunk = blob[i : i + _CHUNK]
redirect = ">" if i == 0 else ">>"
commands.append(["sh", "-c", f"printf %s {chunk} {redirect} {quoted}.b64"])
commands.append(["sh", "-c", f"base64 -d < {quoted}.b64 > {quoted} && rm -f {quoted}.b64"])
return commands
class DockerSandbox(Sandbox):
"""容器隔离的代码执行。
安全姿态:丢掉所有 capability、禁 setuid 提权、只读根文件系统 + 可写 tmpfs、
默认断网、非 root 用户、限制内存 / CPU / 进程数。
同一个 session_id 复用容器,这样多次执行之间文件是连续的。
"""
name = "docker"
def __init__(self, image: str | None = None) -> None:
self.image = image or settings.sandbox_image
self._client: Any = None
self._sessions: dict[str, str] = {} # session_id -> container id
self._lock = asyncio.Lock()
self._image_ready = False
# ---------------- 基础设施 ----------------
def _get_client(self) -> Any:
if self._client is None:
import docker
self._client = docker.from_env()
return self._client
async def health(self) -> dict[str, Any]:
try:
info = await asyncio.to_thread(lambda: self._get_client().version())
has_image = await asyncio.to_thread(self._has_image)
return {
"backend": self.name,
"available": True,
"docker_version": info.get("Version"),
"image": self.image,
"image_pulled": has_image,
"sessions": len(self._sessions),
}
except Exception as e: # noqa: BLE001
return {"backend": self.name, "available": False, "error": str(e)}
def _has_image(self) -> bool:
import docker.errors
try:
self._get_client().images.get(self.image)
return True
except docker.errors.ImageNotFound:
return False
except Exception: # noqa: BLE001
return False
async def ensure_image(self) -> None:
"""首次执行时拉镜像。拉取可能要几十秒,所以单独暴露出来让 API 可以预热。"""
if self._image_ready:
return
if await asyncio.to_thread(self._has_image):
self._image_ready = True
return
await asyncio.to_thread(lambda: self._get_client().images.pull(self.image))
self._image_ready = True
def _create_container(self, limits: SandboxLimits) -> Any:
client = self._get_client()
return client.containers.run(
self.image,
command=["sleep", str(max(limits.timeout * 20, 600))],
detach=True,
working_dir=_WORKDIR,
network_disabled=not limits.network,
mem_limit=f"{limits.memory_mb}m",
memswap_limit=f"{limits.memory_mb}m", # 禁止用 swap 绕过内存限制
nano_cpus=int(limits.cpus * 1e9),
pids_limit=128,
cap_drop=["ALL"],
security_opt=["no-new-privileges"],
read_only=True,
# mode=1777:tmpfs 默认属 root、755,非 root 的 nobody 会写不进去
tmpfs={
_WORKDIR: "rw,exec,size=128m,mode=1777",
"/tmp": "rw,exec,size=64m,mode=1777",
},
user="nobody",
environment={"HOME": "/tmp", "PYTHONDONTWRITEBYTECODE": "1"},
labels={"agentlab": "sandbox"},
auto_remove=False,
)
async def _get_container(self, session_id: str | None, limits: SandboxLimits) -> Any:
client = self._get_client()
async with self._lock:
if session_id and session_id in self._sessions:
try:
container = await asyncio.to_thread(
client.containers.get, self._sessions[session_id]
)
if container.status == "running":
return container
except Exception: # noqa: BLE001 - 容器没了就重建
self._sessions.pop(session_id, None)
container = await asyncio.to_thread(self._create_container, limits)
if session_id:
self._sessions[session_id] = container.id
return container
# ---------------- 执行 ----------------
async def run(
self,
code: str,
*,
language: str = "python",
limits: SandboxLimits | None = None,
session_id: str | None = None,
env: dict[str, str] | None = None,
files: dict[str, str] | None = None,
) -> ExecResult:
limits = limits or SandboxLimits()
language = (language or "python").lower()
if language not in _LANG_CMD:
return ExecResult(
ok=False, backend=self.name, error=f"不支持的语言:{language}"
)
started = time.perf_counter()
ephemeral = session_id is None
sid = session_id or f"tmp-{uuid.uuid4().hex[:8]}"
container = None
try:
await self.ensure_image()
container = await self._get_container(sid, limits)
payload = dict(files or {})
payload[_LANG_FILE[language]] = code
def _stage() -> tuple[int, bytes]:
for cmd in _write_commands(payload):
res = container.exec_run(cmd, workdir=_WORKDIR, user="nobody")
if res.exit_code != 0:
return res.exit_code, res.output or b""
return 0, b""
stage_code, stage_out = await asyncio.to_thread(_stage)
if stage_code != 0:
return ExecResult(
ok=False,
backend=self.name,
exit_code=stage_code,
error=f"写入代码失败:{stage_out.decode(errors='replace')[:500]}",
duration_ms=int((time.perf_counter() - started) * 1000),
)
cmd = _LANG_CMD[language]
exec_env = {"HOME": "/tmp", **(env or {})}
def _exec() -> tuple[int, bytes, bytes]:
res = container.exec_run(
cmd,
workdir=_WORKDIR,
environment=exec_env,
demux=True,
user="nobody",
)
out, err = res.output if isinstance(res.output, tuple) else (res.output, b"")
return res.exit_code, out or b"", err or b""
try:
exit_code, out, err = await asyncio.wait_for(
asyncio.to_thread(_exec), timeout=limits.timeout
)
except asyncio.TimeoutError:
# 超时就把容器整个干掉,防止里面还有进程在烧 CPU
await self._kill(sid)
return ExecResult(
ok=False,
backend=self.name,
timed_out=True,
exit_code=124,
duration_ms=int((time.perf_counter() - started) * 1000),
error=f"执行超过 {limits.timeout}s 被终止",
)
stdout, t1 = truncate(out.decode(errors="replace"), limits.max_output)
stderr, t2 = truncate(err.decode(errors="replace"), limits.max_output)
return ExecResult(
ok=exit_code == 0,
exit_code=exit_code,
stdout=stdout,
stderr=stderr,
truncated=t1 or t2,
backend=self.name,
duration_ms=int((time.perf_counter() - started) * 1000),
)
except Exception as e: # noqa: BLE001
return ExecResult(
ok=False,
backend=self.name,
error=f"{type(e).__name__}: {e}",
duration_ms=int((time.perf_counter() - started) * 1000),
)
finally:
if ephemeral:
await self._kill(sid)
async def _kill(self, session_id: str) -> None:
cid = self._sessions.pop(session_id, None)
if not cid:
return
client = self._get_client()
def _remove() -> None:
try:
client.containers.get(cid).remove(force=True)
except Exception: # noqa: BLE001
pass
await asyncio.to_thread(_remove)
async def cleanup(self, session_id: str) -> None:
await self._kill(session_id)
async def close(self) -> None:
for sid in list(self._sessions):
await self._kill(sid)
async def reap_orphans(self) -> int:
"""清掉上次进程崩溃留下的沙箱容器。"""
def _reap() -> int:
client = self._get_client()
containers = client.containers.list(
all=True, filters={"label": "agentlab=sandbox"}
)
live = set(self._sessions.values())
count = 0
for c in containers:
if c.id in live:
continue
try:
c.remove(force=True)
count += 1
except Exception: # noqa: BLE001
pass
return count
try:
return await asyncio.to_thread(_reap)
except Exception: # noqa: BLE001
return 0
+10 -6
View File
@@ -21,11 +21,11 @@ _LANG = {
class LocalSandbox(Sandbox):
"""子进程执行,Docker 不可用时的降级方案。
"""裸子进程执行,所有隔离后端都不可用时的最后兜底。
有 rlimit(CPU / 内存 / 文件大小 / 进程数)和独立工作目录,但**没有内核级隔离**:
代码仍然跑在宿主机上,能读到你的文件系统。只适合本地自己玩,
别拿它跑不信任的代码 —— 需要真隔离就把 Docker 起起来。
只有 rlimit 和一个独立工作目录,**没有任何访问控制**:代码能读你的整个
文件系统、能联网。正常情况下不会走到这里 —— macOS 有 Seatbelt、
Linux 装个 bubblewrap 即可。真要用它跑不信任的代码,别。
"""
name = "local"
@@ -39,7 +39,9 @@ class LocalSandbox(Sandbox):
"backend": self.name,
"available": True,
"isolated": False,
"warning": "本地子进程执行,没有内核级隔离,不要跑不信任的代码",
"warning": "裸子进程执行,无访问控制,不要跑不信任的代码",
"enforced": {"timeout": True, "cpu": True,
"memory": not sys.platform == "darwin", "network": False, "fsize": True},
"root": str(self._root),
}
@@ -53,6 +55,7 @@ class LocalSandbox(Sandbox):
def _preexec(limits: SandboxLimits): # pragma: no cover - 只在子进程里跑
def _apply() -> None:
resource.setrlimit(resource.RLIMIT_CPU, (limits.timeout, limits.timeout + 2))
# 注意 macOS 不强制执行 RLIMIT_AS,这条在 Darwin 上是无效的
nbytes = limits.memory_mb * 1024 * 1024
for res_name in ("RLIMIT_AS", "RLIMIT_DATA"):
res_id = getattr(resource, res_name, None)
@@ -62,8 +65,9 @@ class LocalSandbox(Sandbox):
except (ValueError, OSError):
pass
resource.setrlimit(resource.RLIMIT_FSIZE, (64 * 1024 * 1024,) * 2)
resource.setrlimit(resource.RLIMIT_NPROC, (64, 64))
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
# 不设 RLIMIT_NPROC:macOS 上它按 uid 全局计数,设小了会让沙箱里
# 任何 fork 都失败(bash 直接起不来),而不是只限制沙箱自己。
os.setsid() # 独立进程组,超时时能整组 kill
return _apply
+53 -21
View File
@@ -1,12 +1,14 @@
from __future__ import annotations
import asyncio
import sys
from typing import Any
from app.core.config import settings
from app.sandbox.base import ExecResult, Sandbox, SandboxLimits
from app.sandbox.docker_sandbox import DockerSandbox
from app.sandbox.bubblewrap_sandbox import BubblewrapSandbox
from app.sandbox.local_sandbox import LocalSandbox
from app.sandbox.seatbelt_sandbox import SeatbeltSandbox
class DisabledSandbox(Sandbox):
@@ -20,11 +22,24 @@ class DisabledSandbox(Sandbox):
)
async def health(self) -> dict[str, Any]:
return {"backend": self.name, "available": False, "reason": "已关闭"}
return {"backend": self.name, "available": False, "isolated": False, "reason": "已关闭"}
# 按隔离强度从高到低排,auto 模式挑第一个能用的
_BACKENDS: list[tuple[str, type[Sandbox]]] = [
("bubblewrap", BubblewrapSandbox),
("seatbelt", SeatbeltSandbox),
("local", LocalSandbox),
]
class SandboxManager:
"""按配置挑一个可用后端。auto = 有 Docker 用 Docker,没有就降级到本地子进程。"""
"""挑选并持有代码执行后端。
用操作系统自带的隔离原语,不依赖容器运行时:macOS 走 Seatbelt
(`sandbox-exec`,系统自带),Linux 走 bubblewrap。两者都是进程级启动,
没有虚拟机开销,冷启动在几十毫秒量级。
"""
def __init__(self) -> None:
self._backend: Sandbox | None = None
@@ -37,28 +52,33 @@ class SandboxManager:
async with self._lock:
if self._backend is not None:
return self._backend
self._backend = await self._resolve()
self._backend = self._resolve()
return self._backend
async def _resolve(self) -> Sandbox:
def _resolve(self) -> Sandbox:
choice = (settings.sandbox_backend or "auto").lower()
if choice == "off":
self._resolved_from = "配置关闭"
return DisabledSandbox()
if choice == "local":
self._resolved_from = "配置指定"
return LocalSandbox()
if choice == "docker":
self._resolved_from = "配置指定"
return DockerSandbox()
docker = DockerSandbox()
health = await docker.health()
if health.get("available"):
self._resolved_from = "自动探测到 Docker"
await docker.reap_orphans()
return docker
self._resolved_from = f"Docker 不可用({health.get('error', '未知原因')}),降级到本地"
if choice == "off":
self._resolved_from = "配置为关闭"
return DisabledSandbox()
named = dict(_BACKENDS)
if choice in named:
impl = named[choice]
backend = impl()
available = getattr(impl, "available", lambda: True)()
self._resolved_from = (
f"配置指定 {choice}" if available
else f"配置指定 {choice},但当前环境不可用"
)
return backend
for name, impl in _BACKENDS:
if getattr(impl, "available", lambda: True)():
self._resolved_from = f"自动选择 {name}({_why(name)})"
return impl()
self._resolved_from = "没有可用的隔离后端,降级到本地子进程"
return LocalSandbox()
async def run(
@@ -87,6 +107,10 @@ class SandboxManager:
info["selected_because"] = self._resolved_from
info["configured"] = settings.sandbox_backend
info["defaults"] = SandboxLimits().model_dump()
info["candidates"] = [
{"name": name, "available": getattr(impl, "available", lambda: True)()}
for name, impl in _BACKENDS
]
return info
async def cleanup(self, session_id: str) -> None:
@@ -98,4 +122,12 @@ class SandboxManager:
await self._backend.close()
def _why(name: str) -> str:
return {
"seatbelt": "macOS 自带 sandbox-exec",
"bubblewrap": "检测到 bwrap",
"local": "无可用隔离原语",
}.get(name, "")
sandbox_manager = SandboxManager()
+283
View File
@@ -0,0 +1,283 @@
from __future__ import annotations
import asyncio
import os
import resource
import shutil
import sys
import time
from pathlib import Path
from app.core.config import settings
from app.sandbox.base import ExecResult, Sandbox, SandboxLimits, truncate
SANDBOX_EXEC = "/usr/bin/sandbox-exec"
def _real(path: str | Path) -> str:
"""Seatbelt 按真实路径匹配,符号链接必须先解开。
macOS 上 /tmp 其实是 /private/tmp,不解析的话策略会整条失效。
"""
return str(Path(path).resolve())
def sandbox_python() -> str:
"""沙箱里用的 Python 解释器。
刻意用 `sys.base_prefix` 下的那个,而不是后端自己的 venv:venv 会把
FastAPI、SQLAlchemy、provider 的凭据处理代码等一整套依赖挂进 sys.path,
沙箱代码不该看得见它们。base_prefix 是一个干净的标准库环境。
"""
base = Path(sys.base_prefix)
major, minor = sys.version_info.major, sys.version_info.minor
for candidate in (
base / "bin" / f"python{major}.{minor}",
base / "bin" / f"python{major}",
base / "bin" / "python",
):
if candidate.exists():
return _real(candidate)
return _real(sys.executable)
_LANG = {
"python": ("main.py", lambda: [sandbox_python(), "-I", "-u", "main.py"]),
"bash": ("main.sh", lambda: ["/bin/bash", "main.sh"]),
"sh": ("main.sh", lambda: ["/bin/sh", "main.sh"]),
"node": ("main.js", lambda: [_real(shutil.which("node") or "node"), "main.js"]),
"javascript": ("main.js", lambda: [_real(shutil.which("node") or "node"), "main.js"]),
}
def _interpreter_roots(language: str) -> list[str]:
"""解释器自身需要能读到的目录。
只放行解释器的安装根(只读),不放行 sys.path —— 后者会把后端的
site-packages 一起开出去。uv 把 Python 装在 ~/.local/share/uv/python/ 下,
正好落在要禁读的家目录里,所以这条放行是必需的。
"""
roots: set[str] = {
"/usr", "/System", "/Library", "/bin", "/sbin", "/opt",
"/private/var/select", "/private/var/db",
}
if language == "python":
roots.add(_real(sys.base_prefix))
elif language in ("node", "javascript"):
node = shutil.which("node")
if node:
roots.add(str(Path(_real(node)).parent.parent))
return sorted(roots)
def build_policy(workdir: Path, *, language: str, network: bool) -> str:
"""生成 Seatbelt 策略(SBPL)。
用「默认允许 + 精确拒绝」而不是 `(deny default)`:后者要把解释器启动所需的
每一次 syscall 和路径都枚举出来,稍有遗漏进程就直接 SIGABRT,实测 Python
根本起不来。这里换个思路 —— 默认放行,然后把真正要守的三件事收紧:
写盘只限工作区、家目录不可读、默认断网。
规则从上往下匹配,后面的覆盖前面的,所以 allow 必须写在对应 deny 之后。
"""
work = _real(workdir)
home = _real(Path.home())
lines = [
"(version 1)",
"(allow default)",
"",
";; ---- 网络 ----",
]
if network:
lines.append("(allow network*)")
else:
lines.append("(deny network*)")
lines += [
"",
";; ---- 写:只有工作区 ----",
"(deny file-write*)",
f'(allow file-write* (subpath "{work}"))',
'(allow file-write-data (literal "/dev/null") (literal "/dev/stdout")'
' (literal "/dev/stderr") (literal "/dev/dtracehelper"))',
'(allow file-write* (subpath "/private/var/folders"))', # 临时目录,Python 要用
"",
";; ---- 读:家目录挡住,放行工作区和解释器 ----",
f'(deny file-read* (subpath "{home}"))',
f'(allow file-read* (subpath "{work}"))',
]
for root in _interpreter_roots(language):
lines.append(f'(allow file-read* (subpath "{root}"))')
lines += [
"",
";; ---- 其他 ----",
"(allow process-exec process-fork)", # 子进程同样受这份策略约束
"(allow sysctl-read)",
"(allow mach-lookup)",
"(deny file-read* (subpath \"/Users/Shared\"))",
]
return "\n".join(lines) + "\n"
class SeatbeltSandbox(Sandbox):
"""基于 macOS 原生 Seatbelt(sandbox-exec)的沙箱。
内核级的强制访问控制,不需要虚拟机,启动开销约等于起一个进程。
策略对子进程是继承的,所以沙箱里再 exec 出来的东西一样跑不掉。
和容器方案的差别要说清楚:这里做的是**访问控制**而不是虚拟化 ——
文件系统和网络守得住,但进程表是共享的(能看到宿主机进程列表),
而且 macOS 不认 RLIMIT_AS,**内存用量限不住**。CPU 时间、文件大小、
墙钟超时这几项由 rlimit 和超时杀进程组来兜。
"""
name = "seatbelt"
def __init__(self) -> None:
self._root = settings.workspace_dir / "seatbelt"
self._policies = settings.data_dir / "sandbox-policies"
@staticmethod
def available() -> bool:
return sys.platform == "darwin" and Path(SANDBOX_EXEC).exists()
async def health(self) -> dict[str, object]:
ok = self.available()
return {
"backend": self.name,
"available": ok,
"isolated": ok,
"isolation": "macOS Seatbelt:文件系统 + 网络(内核强制,子进程继承)",
"limits_note": "macOS 不支持 RLIMIT_AS,内存用量无法限制;CPU 时间与墙钟超时有效",
# 如实申报:界面按这个把限不住的项划掉,避免给人虚假的安全感
"enforced": {"timeout": True, "cpu": True, "memory": False, "network": True, "fsize": True},
"root": str(self._root),
**({} if ok else {"error": "当前系统不是 macOS 或缺少 /usr/bin/sandbox-exec"}),
}
def _session_dir(self, session_id: str) -> Path:
safe = "".join(c for c in session_id if c.isalnum() or c in "-_")[:64] or "default"
path = self._root / safe
path.mkdir(parents=True, exist_ok=True)
return path
@staticmethod
def _preexec(limits: SandboxLimits): # pragma: no cover - 只在子进程里执行
def _apply() -> None:
# CPU 时间:能挡住纯计算的死循环(sleep 不计入,墙钟超时另外兜)
resource.setrlimit(resource.RLIMIT_CPU, (limits.timeout, limits.timeout + 2))
# 单文件大小,防止把磁盘写满
resource.setrlimit(resource.RLIMIT_FSIZE, (256 * 1024 * 1024,) * 2)
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
# 注意:不设 RLIMIT_NPROC —— macOS 上它按 uid 全局计数,
# 设成小值会让沙箱里任何 fork 都失败(连 bash 都起不来)。
# 也不设 RLIMIT_AS —— Darwin 不强制执行,设了是自欺欺人。
os.setsid() # 独立进程组,超时能整组干掉
return _apply
async def run(
self,
code: str,
*,
language: str = "python",
limits: SandboxLimits | None = None,
session_id: str | None = None,
env: dict[str, str] | None = None,
files: dict[str, str] | None = None,
) -> ExecResult:
limits = limits or SandboxLimits()
language = (language or "python").lower()
if language not in _LANG:
return ExecResult(ok=False, backend=self.name, error=f"不支持的语言:{language}")
if not self.available():
return ExecResult(ok=False, backend=self.name, error="sandbox-exec 不可用")
filename, argv_of = _LANG[language]
workdir = self._session_dir(session_id or f"tmp-{int(time.time() * 1000)}")
(workdir / filename).write_text(code)
for rel, content in (files or {}).items():
target = (workdir / rel).resolve()
if not str(target).startswith(str(workdir.resolve())):
return ExecResult(ok=False, backend=self.name, error=f"非法文件路径:{rel}")
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(content)
self._policies.mkdir(parents=True, exist_ok=True)
policy_file = self._policies / f"{workdir.name}-{language}.sb"
policy_file.write_text(build_policy(workdir, language=language, network=limits.network))
argv = [SANDBOX_EXEC, "-f", str(policy_file), *argv_of()]
# HOME 指向工作区:既挡住家目录,又让解释器的缓存有地方落
child_env = {
"PATH": "/usr/bin:/bin:/usr/sbin:/sbin:/opt/homebrew/bin",
"HOME": str(workdir),
"TMPDIR": str(workdir),
"LANG": "en_US.UTF-8",
"PYTHONDONTWRITEBYTECODE": "1",
"PYTHONUNBUFFERED": "1",
**(env or {}),
}
started = time.perf_counter()
try:
proc = await asyncio.create_subprocess_exec(
*argv,
cwd=str(workdir),
env=child_env,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
preexec_fn=self._preexec(limits),
)
except Exception as e: # noqa: BLE001
return ExecResult(ok=False, backend=self.name, error=f"{type(e).__name__}: {e}")
try:
out, err = await asyncio.wait_for(proc.communicate(), timeout=limits.timeout)
except asyncio.TimeoutError:
self._kill(proc)
await proc.wait()
return ExecResult(
ok=False,
backend=self.name,
timed_out=True,
exit_code=124,
duration_ms=int((time.perf_counter() - started) * 1000),
error=f"执行超过 {limits.timeout}s 被终止",
)
stdout, t1 = truncate(out.decode(errors="replace"), limits.max_output)
stderr, t2 = truncate(err.decode(errors="replace"), limits.max_output)
code_ = proc.returncode or 0
# CPU 时间用尽是 SIGXCPU(24),归到超时里更好理解
timed_out = code_ in (-24, 152)
written = [
p.name for p in workdir.iterdir() if p.is_file() and p.name != filename
][:50]
return ExecResult(
ok=code_ == 0,
exit_code=code_,
stdout=stdout,
stderr=stderr if not timed_out else (stderr or "CPU 时间超限,已终止"),
truncated=t1 or t2,
timed_out=timed_out,
backend=self.name,
duration_ms=int((time.perf_counter() - started) * 1000),
files_written=written,
)
@staticmethod
def _kill(proc: asyncio.subprocess.Process) -> None:
try:
os.killpg(os.getpgid(proc.pid), 9)
except (ProcessLookupError, PermissionError):
proc.kill()
async def cleanup(self, session_id: str) -> None:
path = self._session_dir(session_id)
await asyncio.to_thread(shutil.rmtree, path, True)
for stale in self._policies.glob(f"{path.name}-*.sb"):
stale.unlink(missing_ok=True)
-1
View File
@@ -20,7 +20,6 @@ dependencies = [
"langchain-openai",
"langchain-mcp-adapters",
"mcp",
"docker>=7.1",
"numpy>=2.0",
"jsonschema>=4.23",
"beautifulsoup4>=4.12",
-16
View File
@@ -16,7 +16,6 @@ dependencies = [
{ name = "aiosqlite" },
{ name = "beautifulsoup4" },
{ name = "cryptography" },
{ name = "docker" },
{ name = "fastapi" },
{ name = "greenlet" },
{ name = "httpx" },
@@ -41,7 +40,6 @@ requires-dist = [
{ name = "aiosqlite", specifier = ">=0.20" },
{ name = "beautifulsoup4", specifier = ">=4.12" },
{ name = "cryptography", specifier = ">=43.0" },
{ name = "docker", specifier = ">=7.1" },
{ name = "fastapi", specifier = ">=0.115" },
{ name = "greenlet", specifier = ">=3.1" },
{ name = "httpx", specifier = ">=0.28" },
@@ -434,20 +432,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/12/b3/231ffd4ab1fc9d679809f356cebee130ac7daa00d6d6f3206dd4fd137e9e/distro-1.9.0-py3-none-any.whl", hash = "sha256:7bffd925d65168f85027d8da9af6bddab658135b840670a223589bc0c8ef02b2", size = 20277, upload-time = "2023-12-24T09:54:30.421Z" },
]
[[package]]
name = "docker"
version = "7.2.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pywin32", marker = "sys_platform == 'win32'" },
{ name = "requests" },
{ name = "urllib3" },
]
sdist = { url = "https://files.pythonhosted.org/packages/88/7f/731ff914b0255d3d065f45fd4e626d4b8c95dbcbaada049f337a6ac16410/docker-7.2.0.tar.gz", hash = "sha256:cebb93773d334f778e023a7ee352a8d6e13ab1bd3b863a4d4a59dec897df43ac", size = 118731, upload-time = "2026-07-09T14:53:46.39Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/75/23/529140fe1aab80fc6992f93a706deec709140a6397439139a054e1515c45/docker-7.2.0-py3-none-any.whl", hash = "sha256:a3f45fdeb9165e2d25d9a1d02ddf3bc70fb572cf5ebbf9b58558c22caf29b71f", size = 148775, upload-time = "2026-07-09T14:53:45.224Z" },
]
[[package]]
name = "docstring-parser"
version = "0.18.0"
+35 -3
View File
@@ -409,8 +409,22 @@ function SystemTab() {
</span>
} />
<Row label="选择原因" value={sandbox.selected_because} />
{sandbox.image && <Row label="镜像" value={`${sandbox.image}${sandbox.image_pulled ? '' : '(未拉取,首次执行会自动拉)'}`} />}
{sandbox.docker_version && <Row label="Docker" value={sandbox.docker_version} />}
{sandbox.isolation && <Row label="隔离方式" value={sandbox.isolation} />}
{sandbox.limits_note && <Row label="限额说明" value={sandbox.limits_note} />}
{!!sandbox.candidates?.length && (
<Row label="可用后端" value={
<span className="flex flex-wrap gap-1.5">
{sandbox.candidates.map((c: any) => (
<span key={c.name} className="chip"
style={c.name === sandbox.backend
? { borderColor: 'var(--ok)', color: 'var(--ok)' }
: c.available ? undefined : { opacity: 0.45 }}>
{c.name}{c.available ? '' : ' · 不可用'}
</span>
))}
</span>
} />
)}
{sandbox.warning && (
<div className="mt-2 rounded border px-2 py-1.5 text-[10.5px]"
style={{ borderColor: 'var(--warn)', color: 'var(--warn)' }}>
@@ -419,7 +433,12 @@ function SystemTab() {
)}
{sandbox.defaults && (
<Row label="默认限额" value={
`${sandbox.defaults.timeout}s · ${sandbox.defaults.memory_mb}MB · ${sandbox.defaults.cpus} CPU · ${sandbox.defaults.network ? '联网' : '断网'}`
<span className="flex flex-wrap items-center gap-2">
<Limit on={sandbox.enforced?.timeout !== false} text={`${sandbox.defaults.timeout}s 超时`} />
<Limit on={sandbox.enforced?.memory !== false} text={`${sandbox.defaults.memory_mb}MB 内存`} />
<Limit on={sandbox.enforced?.cpu !== false} text={`${sandbox.defaults.cpus} CPU`} />
<Limit on={sandbox.enforced?.network !== false} text={sandbox.defaults.network ? '联网' : '断网'} />
</span>
} />
)}
</div>
@@ -439,6 +458,19 @@ function SystemTab() {
)
}
/** 限额项:当前后端管不住的,划掉并注明,别给人虚假的安全感。 */
function Limit({ on, text }: { on: boolean; text: string }) {
return on ? (
<span className="chip" style={{ color: 'var(--ok)', borderColor: 'color-mix(in srgb, var(--ok) 40%, transparent)' }}>
{text}
</span>
) : (
<span className="chip line-through opacity-60" title="当前后端不支持这项限额">
{text} · 不生效
</span>
)
}
function Row({ label, value }: { label: string; value: any }) {
return (
<div className="flex gap-3">