s06
沙箱系统
能力层安全隔离执行
SandboxProvider 抽象 + 虚拟路径映射本地/Docker/K8s 三种模式统一接口,虚拟路径让 Agent 无感切换环境
安全隔离执行
SandboxProvider 抽象 + 虚拟路径映射本地/Docker/K8s 三种模式统一接口,虚拟路径让 Agent 无感切换环境
Agent 会执行用户给出的任意代码(Python 脚本、Shell 命令等)。如果直接在宿主机运行,恶意或有缺陷的代码可能删除文件、窃取密钥、耗尽资源。沙箱系统通过容器隔离 + 虚拟路径映射,让 Agent 在安全边界内执行代码,同时保持统一的文件访问接口——切换执行环境时 Agent 代码零修改。
| 虚拟路径 (Agent 视角) | Local 宿主机 | Docker 容器 | K8s Pod |
|---|---|---|---|
| /workspace/output.csv | /tmp/sandbox/output.csv | /app/output.csv | /mnt/vol/output.csv |
| /workspace/script.py | /tmp/sandbox/script.py | /app/script.py | /mnt/vol/script.py |
Sandbox 抽象基类 — 定义 Agent 可用的全部文件系统操作,所有执行模式必须实现这些方法
class Sandbox(ABC):
@abstractmethod
def execute_command(self, command: str) -> str: ...
@abstractmethod
def read_file(self, path: str) -> str: ...
@abstractmethod
def write_file(self, path: str, content: str,
append: bool = False) -> None: ...
@abstractmethod
def list_dir(self, path: str, max_depth=2) -> list[str]: ...SandboxProvider 抽象 — 通过 resolve_class() 动态加载,config.yaml 一行切换执行模式
class SandboxProvider(ABC):
@abstractmethod
def acquire(self, thread_id: str | None = None) -> str: ...
@abstractmethod
def get(self, sandbox_id: str) -> Sandbox | None: ...
@abstractmethod
def release(self, sandbox_id: str) -> None: ...
# 动态加载: config.yaml 中 sandbox.use 指定类路径
def get_sandbox_provider(**kwargs) -> SandboxProvider:
config = get_app_config()
cls = resolve_class(config.sandbox.use, SandboxProvider)
return cls(**kwargs)虚拟路径映射 — Agent 只看到 /mnt/user-data/,宿主机路径对 Agent 完全透明
VIRTUAL_PATH_PREFIX = "/mnt/user-data"
class Paths:
def sandbox_work_dir(self, thread_id: str) -> Path:
# Host: {base}/threads/{tid}/user-data/workspace/
# Sandbox: /mnt/user-data/workspace/
return self.thread_dir(thread_id) / "user-data" / "workspace"
def resolve_virtual_path(self, thread_id: str,
virtual_path: str) -> Path:
"""将沙箱虚拟路径解析为宿主机真实路径,
同时检测路径穿越攻击"""
actual = (base / relative).resolve()
actual.relative_to(base) # 防止 ../../ 穿越
return actualBackend 策略模式 — provisioner_url 存在则用远程 K8s,否则走本地 Docker
def _create_backend(self) -> SandboxBackend:
provisioner_url = self._config.get("provisioner_url")
if provisioner_url:
return RemoteSandboxBackend(
provisioner_url=provisioner_url)
return LocalContainerBackend(
image=self._config["image"],
base_port=self._config["port"],
container_prefix=self._config["container_prefix"],
config_mounts=self._config["mounts"],
environment=self._config["environment"])