在前面六篇文章里，我们一步步完成了 mini Agent Runtime 的前两个大阶段：构建了可持久化恢复的两阶段状态机 AgentSessionV2，打通了从 mini vLLM 采样层 LogitsProcessor 到 AST 校验的三层结构化输出防御，并实现了高并发的 AsyncToolExecutor 与 DAGPlanAndSolveEngine。
到现在为止，我们的调度内核已经非常扎实了。但是，如果审视一下我们之前注册给 Agent 的那些工具——比如 list_python_files、count_file_lines、read_source_module——你会发现它们全是我们作为开发者预先写死的“专用小函数”。
这会带来一个根本性的矛盾：真实世界的任务是无穷无尽的，我们不可能为每一种文件格式转换、每一种日志正则统计、每一次代码编译或单元测试都提前写好一个专用的 Python 工具函数。
怎么让一个 Agent 真正具备通用问题求解能力（General Problem-Solving）？
答案大家都知道：给它赋予直接操作计算机底层的两个“元工具（Meta-Tools）”——bash_exec（执行任意 Shell 命令） 和 python_exec（执行任意生成的 Python 代码）。有了这两个元工具，当没有现成工具可用时，Agent 可以自己现场写一段脚本、自己调编译器、自己跑测试、自己看报错并改代码。
但作为系统工程师，当你准备在代码里写下 subprocess.run(cmd, shell=True) 并把它暴露给大模型的那一刻，你的后背一定会立刻发凉。
因为大模型并不是一个永远理智、永远懂系统安全的资深运维。在实际运行中，一旦你没有任何防护地裸跑 subprocess.run，你会立刻遭遇下面这 “五大沙箱惊魂现场”：
孤儿孙进程残留（Orphan Grandchild Processes）：当模型执行的命令是
bash -c "python3 train.py && sleep 1000"时，Python 的subprocess创建的是外层的/bin/bash子进程（PID 1001），而/bin/bash又 fork 出了真正的计算孙进程python3（PID 1002）。当执行超时触发proc.kill()时，标准库默认只杀了父进程 PID 1001！孙进程 PID 1002 会立刻变成挂载在init(PID 1) 下的孤儿进程，继续在后台疯狂吃满你的 CPU 和显存！输出流洪水打爆宿主机内存（Stdout/Stderr Flood OOM）：如果模型不小心写了一个
while True: print("debug line...")或者cat /dev/zero，子进程会在一秒钟内往标准输出管道（Pipe）里狂吐几百 MB 甚至几个 GB 的字节。如果你用proc.communicate()试图把全部输出读进内存，首先你的宿主机 Python 进程会直接 OOM 崩溃；其次，就算没 OOM，把几百万个字符作为Observation塞进AgentSession，下一轮请求发给mini vLLM时会瞬间撑爆 Context Window！路径穿越与目录逃逸（Path Traversal & Symlink Escape）：你本来规定 Agent 只能在
./workspace目录下读写文件，但模型在读写工具里传了一个"../../etc/passwd"，或者在工作区里先创建一个指向根目录的软链接ln -s / root_link，然后再去读./workspace/root_link/etc/shadow，轻松绕过简单的字符串前缀检查。宿主环境变量与密钥泄露（Environment Secret Leakage）：默认情况下，子进程会完整继承父进程的
os.environ。如果你在启动 Agent 服务的环境变量里配置了OPENAI_API_KEY、AWS_SECRET_ACCESS_KEY或数据库密码，模型只要跑一句env或print(os.environ)，所有核心机密就会原封不动地出现在日志和上下文轨迹里。CPU 死循环与 Fork 炸弹耗尽内核资源（Resource Exhaustion）：模型生成的一段递归代码如果没有终止条件，或者在脚本里无意间在循环里不断创建子进程/分配巨型数组，如果不从操作系统内核层面施加
rlimit（Resource Limits）紧箍咒，整台物理机都会陷入假死。
从这一篇（第 7 篇）开始，我们正式进入第三阶段：《执行环境、安全沙箱与生态协议》。
这一篇，我们就发挥 Linux 系统编程的看家本领，不依赖重型的虚拟机或外部云服务，直接用 Python 标准库配合 POSIX / Linux 内核原语（os.setsid、os.killpg、resource.setrlimit、os.path.realpath、异步流式环形缓冲区），从零手写一个轻量、强隔离、防 OOM、防逃逸的工业级本地安全执行沙箱 LocalSecureSandbox！
1. 这一篇要解决什么问题
我们先画出 LocalSecureSandbox 在接收到一个来自 AsyncToolExecutor 的代码或命令执行请求时，必须在操作系统层面建立的五道隔离防线：
ToolCall: bash_exec(cmd) / python_exec(code)
│
▼
┌───────────────────────────────────────────────────────────────────────┐
│ 防线 1：静态指令审计与路径规范化校验 (Path Canonicalization Guard) │
│ • 校验 realpath(target) 严格位于 realpath(workspace_root) 子树下 │
│ • 拦截穿透软链接 (Symlink Escape) 与高危命令黑名单模式 │
└───────┬───────────────────────────────────────────────────────────────┘
│
▼
┌───────────────────────────────────────────────────────────────────────┐
│ 防线 2：净室环境变量构造 (Clean-Room Environment Sanitization) │
│ • 剥离宿主机 os.environ 中的所有 API Key / Token / 敏感凭据 │
│ • 仅注入最小白名单环境 (PATH, HOME=workspace, LANG, PYTHONIOENCODING) │
└───────┬───────────────────────────────────────────────────────────────┘
│
▼ fork() + preexec_fn (在子进程 exec 之前执行内核级自缚)
┌───────────────────────────────────────────────────────────────────────┐
│ 防线 3：POSIX 进程组隔离与内核 rlimit 紧箍咒 │
│ • os.setsid() 创建独立 Session 与进程组 (PGID = PID) │
│ • resource.setrlimit: 限制地址空间内存 (RLIMIT_AS)、CPU 时间、 │
│ 最大子进程数 (RLIMIT_NPROC 防 fork 炸弹)、最大写入文件大小 (FSIZE) │
└───────┬───────────────────────────────────────────────────────────────┘
│
▼ 异步非阻塞管道读取 (asyncio.subprocess.PIPE)
┌───────────────────────────────────────────────────────────────────────┐
│ 防线 4：Head + Tail 双端有界流式截断器 (Bounded Stream Capturer) │
│ • 实时统计已产出字节数，仅保留头部 B_head 与尾部 B_tail (环形滑窗) │
│ • 一旦单次输出突破硬上限 B_kill，立即提前终止进程防止 I/O 洪泛 │
└───────┬───────────────────────────────────────────────────────────────┘
│
▼ 当触发超时或异常退出时
┌───────────────────────────────────────────────────────────────────────┐
│ 防线 5：进程组连坐斩杀 (Process Group Tree Termination) │
│ • os.killpg(pgid, signal.SIGKILL) 一键清空父进程及所有派生的孙进程！ │
└───────────────────────────────────────────────────────────────────────┘
这五道防线环环相扣：静态检查防越权访问，净室环境防凭据泄露，rlimit 防内核资源耗尽，双端截断防内存与 Context 爆炸，killpg 防后台僵尸进程残留。
2. 沙箱隔离的形式化安全不变量与流式截断数学模型
在写代码之前，我们把其中最核心的三个系统机制用数学与操作系统原语严格推导清楚。
2.1 规范化路径闭包不变量（Canonical Path Containment Invariant）
假设沙箱分配给当前会话的专属工作目录为 W。在文件系统中，由于存在相对路径符号（.、..）以及符号链接（Symbolic Links，即软链接 l ↦ target），一个表面上看起来在工作区内的路径字符串 p（例如 p = "./workspace/subdir/../../etc/passwd" 或指向外部的软链接 ./workspace/link_to_root/etc/passwd），其最终指向的物理 inode 可能完全脱离了 W。
设操作系统内核提供的路径规范化解析算子为 Φ : Path → CanonicalPath（对应 POSIX 的 realpath(3) 系统调用，它会递归展开路径中的所有符号链接并消解所有 . 与 .. 分量）。
令工作区根目录的规范化路径为 W* = Φ(W)。对于 Agent 请求访问或创建的任意目标路径 p，其绝对规范化路径为：
p* = Φ( Join(W*, p) )
沙箱的文件系统访问控制必须强制满足如下 前缀闭包不变量：
Allow(p) = 𝕀[ p* = W* ∨ StartsWith(p*, W* ‖ "/") ]
注意公式里的第二个条件必须是 W* ‖ "/"（带上目录分隔符 /），而不能仅仅是 StartsWith(p*, W*)！
为什么？这是一个非常经典的系统安全漏洞（CVE 常客）：假设你的工作区目录叫 W* = "/tmp/agent_ws"，如果只判断 StartsWith(p*, W*)，攻击者构造一个兄弟目录路径 p* = "/tmp/agent_ws_hacked/secret.txt"，它居然也能通过前缀检查！只有加上尾随的 '/'（或者在 Python 中使用 os.path.commonpath([W_star, p_star]) == W_star），才能在集合论意义上严格保证 p* 是 W* 目录树下的子孙节点。
2.2 进程树连坐终止模型（Process Group Termination）
在 UNIX/Linux 进程模型中，每个进程 p 都拥有一个进程号 PID(p) 和一个所属的进程组号 PGID(p)。默认情况下，fork() 出的子进程会继承父进程的进程组号：
PGID(p_child) = PGID(p_parent)
如果我们在 Agent Runtime 主进程里直接启动子进程，那么 Agent 主进程、Shell 子进程以及 Shell 派生出的所有孙进程都处于同一个进程组里。此时你既不能向整个进程组发 SIGKILL（那会把 Agent 主进程自己也自杀掉），也不能只向 PID(p_child) 发 SIGKILL（那会漏掉孙进程树 T(p_child)）。
解决办法是在 fork() 之后、execve() 之前（即 preexec_fn 钩子中），调用一次 os.setsid()。这会让子进程 p₀ 脱离父进程的会话，成立一个以它自己为组长（Process Group Leader）的新进程组：
PGID(p₀) = PID(p₀) = g₀
此后，无论 p₀ 在内部通过 fork/clone 派生出多少层子进程、孙进程、曾孙进程 ∀pₖ ∈ Descendants(p₀)（只要它们没有自己再调 setsid），它们都会自动继承这个进程组标识：
∀pₖ ∈ Descendants(p₀)，PGID(pₖ) = g₀
当沙箱检测到超时或输出超限需要强制清场时，只需向负的进程组号 −g₀（即调用 os.killpg(g_0, SIGKILL)）发送一次信号，Linux 内核就会在内核态原子地遍历并杀死整个进程组集合 { p | PGID(p) = g₀ }，做到真正的寸草不生、零孤儿进程残留！
2.3 Head + Tail 双端有界流式截断模型（Bounded Head-Tail Stream Model）
最后我们来看怎么解决标准输出洪水（Stdout Flood）问题。
如果一个命令（比如跑单元测试或编译大型 C++ 项目）输出了 N = 500,000 字节的日志，而我们允许放入 Observation 的最大字节预算只有 B_max = 8,192 字节（约 2,000 tokens），我们该保留哪一部分？
如果只保留前 B_max 字节（Head-Only Truncation），你会看到一堆无关紧要的编译启动信息 Scanning dependencies of target...，而真正出现在日志最末尾的 AssertionError: expected 42, got 0 却被彻底截掉了！ 反过来，如果只保留最后 B_max 字节（Tail-Only Truncation），当程序在一开头就打印了关键的参数配置或第一个致命语法错误、后面跟着几千行级联报错时，你又会丢掉最初的罪魁祸首。
因此，对大模型自修复最友好的日志截断策略是 Head + Tail 双端保留模型。
设总保留预算为 B_max，我们将其按比例拆分为头部预算 B_head = α·B_max 与尾部预算 B_tail = (1 − α)·B_max（通常取 α = 0.35，即保留前 35% 的启动上下文与最后 65% 的报错现场）。
在异步流式读取字节块 c₁, c₂, … 的过程中（累计总字节数记为 S = Σ|cᵢ|）：
头部固定缓冲区 H：仅当 |H| < B_head 时追加写入，一旦达到 B_head 即封口不再分配新内存；
尾部环形滑动窗口 T：维护一个最大容量为 B_tail 的双端队列（Ring Buffer），新字节块进入时自动挤出最老的字节，始终只占用不超过 B_tail 的常数级内存 O(B_max)！
洪泛熔断阈值 B_flood：如果子进程陷入了死循环打印，导致累计吐出的总字节数 S > B_flood（例如 10 MB），流读取器立即通知沙箱触发
killpg强杀进程，并在头部 H 与尾部 T 之间插入一段精确的截断诊断标记：
Output(S) = H ‖ " ... [TRUNCATED " ‖ (S − B_max) ‖ " BYTES] ... " ‖ T
这样一来，无论子进程吐出 10 KB 还是 10 GB 的垃圾输出，沙箱在 Python 进程里占用的内存永远严格受限于常数 B_head + B_tail，且返回给大模型的 Observation 永远首尾兼顾、大小可控！
3. 核心代码实现：编写 LocalSecureSandbox（sandbox.py）
理清了所有底层细节，我们来编写完整的 sandbox.py。
这段代码全部使用 Python 标准库（asyncio、os、signal、resource、collections.deque、shutil、tempfile），可以直接在任何 Linux / macOS 环境上运行。
from __future__ import annotations
import asyncio
import os
import re
import resource
import signal
import sys
import tempfile
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Callable, Deque, Dict, List, Optional, Tuple
class SandboxSecurityError(PermissionError):
"""当检测到路径穿越逃逸或高危指令模式时抛出"""
pass
@dataclass
class SandboxConfig:
"""安全沙箱的资源与隔离边界配置"""
workspace_dir: str # 沙箱绑定的工作目录根目录
timeout_sec: float = 10.0 # 单次命令硬超时 (秒)
max_output_bytes: int = 8192 # 返回给 LLM 的最大输出字节预算 (Head + Tail)
head_ratio: float = 0.35 # 头部保留比例 (默认前 35%，后 65%)
flood_kill_bytes: int = 5 * 1024 * 1024 # 输出洪泛强杀阈值 (默认 5MB 立即强杀防死循环打印)
max_memory_mb: int = 512 # 子进程最大虚拟内存地址空间限制 (MB)
max_cpu_time_sec: int = 10 # 子进程最大 CPU 核时限制 (秒)
max_processes: int = 32 # RLIMIT_NPROC 限制 (防御 fork 炸弹)
max_file_size_mb: int = 50 # RLIMIT_FSIZE 单文件最大写入限制 (MB)
allowed_env_keys: Tuple[str, ...] = ("PATH", "LANG", "LC_ALL")
blocked_command_patterns: Tuple[str, ...] = (
r":\(\)\s*\{\s*:\|\:&\s*\}\s*;:", # 经典 Bash fork 炸弹特征
r"\bmkfs\b", # 格式化磁盘
r"\bdd\s+if=", # 裸写块设备
r"\bshutdown\b|\breboot\b", # 关机重启
)
@dataclass
class SandboxResult:
"""沙箱执行结果结构"""
exit_code: int
stdout: str
stderr: str
duration_ms: float
timed_out: bool = False
oom_or_killed: bool = False
output_truncated: bool = False
total_stdout_bytes: int = 0
total_stderr_bytes: int = 0
def to_observation_text(self) -> str:
"""格式化为对大模型最友好的结构化观测文本"""
status_flags: List[str] = [f"exit_code={self.exit_code}", f"duration={self.duration_ms:.1f}ms"]
if self.timed_out:
status_flags.append("TIMED_OUT=True (Process group killed)")
if self.oom_or_killed:
status_flags.append("KILLED_BY_SIGNAL=True (Possible OOM / RLIMIT exceeded)")
if self.output_truncated:
status_flags.append("OUTPUT_TRUNCATED=True")
parts = [f"[SANDBOX EXECUTION: {', '.join(status_flags)}]"]
if self.stdout:
parts.append(f"--- STDOUT ---\n{self.stdout.rstrip()}")
if self.stderr:
parts.append(f"--- STDERR ---\n{self.stderr.rstrip()}")
if not self.stdout and not self.stderr:
parts.append("(No output produced on stdout/stderr)")
return "\n".join(parts)
class BoundedHeadTailBuffer:
"""
O(B_max) 常数内存的 Head + Tail 流式双端截断缓冲区。
无论子进程吐出多少 MB 的日志，内存中永远只保留前 head_budget 字节与最后 tail_budget 字节。
"""
def __init__(self, max_bytes: int = 8192, head_ratio: float = 0.35) -> None:
self.max_bytes = max(256, max_bytes)
self.head_budget = int(self.max_bytes * head_ratio)
self.tail_budget = self.max_bytes - self.head_budget
self._head_chunks: List[bytes] = []
self._head_len: int = 0
self._tail_chunks: Deque[bytes] = deque()
self._tail_len: int = 0
self.total_seen_bytes: int = 0
def feed(self, data: bytes) -> None:
if not data:
return
n = len(data)
self.total_seen_bytes += n
# 1. 如果头部预算尚未填满，优先填充头部缓冲区
if self._head_len < self.head_budget:
need = self.head_budget - self._head_len
head_part = data[:need]
self._head_chunks.append(head_part)
self._head_len += len(head_part)
remainder = data[need:]
else:
remainder = data
# 2. 剩余字节进入尾部环形滑动窗口 (Ring Buffer)
if remainder and self.tail_budget > 0:
if len(remainder) >= self.tail_budget:
self._tail_chunks.clear()
self._tail_chunks.append(remainder[-self.tail_budget :])
self._tail_len = self.tail_budget
else:
self._tail_chunks.append(remainder)
self._tail_len += len(remainder)
# 弹出最老的数据块，保持尾部总字节数 <= tail_budget
while self._tail_len > self.tail_budget and self._tail_chunks:
oldest = self._tail_chunks.popleft()
excess = self._tail_len - self.tail_budget
if len(oldest) <= excess:
self._tail_len -= len(oldest)
else:
# 仅切掉 oldest 前面超出的字节，把剩余部分塞回队首
self._tail_chunks.appendleft(oldest[excess:])
self._tail_len -= excess
@property
def is_truncated(self) -> bool:
return self.total_seen_bytes > (self._head_len + self._tail_len)
def render(self) -> str:
head_str = b"".join(self._head_chunks).decode("utf-8", errors="replace")
if not self.is_truncated:
tail_str = b"".join(self._tail_chunks).decode("utf-8", errors="replace")
return head_str + tail_str
omitted_bytes = self.total_seen_bytes - (self._head_len + self._tail_len)
tail_str = b"".join(self._tail_chunks).decode("utf-8", errors="replace")
marker = (
f"\n\n... [SANDBOX STREAM TRUNCATED: omitted {omitted_bytes:,} bytes "
f"(total produced: {self.total_seen_bytes:,} bytes)] ...\n\n"
)
return head_str + marker + tail_str
class LocalSecureSandbox:
"""
基于 POSIX 进程组隔离、rlimit 内核资源限制与流式双端截断的本地安全执行沙箱。
"""
def __init__(self, config: SandboxConfig) -> None:
self.config = config
os.makedirs(self.config.workspace_dir, exist_ok=True)
# 规范化工作区根路径 W* = realpath(W)
self.canonical_root = os.path.realpath(self.config.workspace_dir)
def resolve_safe_path(self, relative_or_abs_path: str) -> str:
"""
验证并返回工作区内的安全路径。
严格执行前缀闭包不变量：防止 ../ 穿越与软链接 (Symlink) 逃逸！
"""
if os.path.isabs(relative_or_abs_path):
candidate = os.path.realpath(relative_or_abs_path)
else:
candidate = os.path.realpath(os.path.join(self.canonical_root, relative_or_abs_path))
# 使用 commonpath 严格判定 candidate 是否位于 canonical_root 子树内
if os.path.commonpath([self.canonical_root, candidate]) != self.canonical_root:
raise SandboxSecurityError(
f"Path escape blocked! Requested path '{relative_or_abs_path}' resolves to "
f"'{candidate}', which is outside the sandbox workspace '{self.canonical_root}'."
)
return candidate
def _audit_command(self, command: str) -> None:
"""静态审计高危 Shell 模式"""
for pattern in self.config.blocked_command_patterns:
if re.search(pattern, command, re.IGNORECASE):
raise SandboxSecurityError(
f"Command rejected by sandbox static security policy (matched pattern: {pattern})."
)
def _build_clean_env(self) -> Dict[str, str]:
"""构造净室环境变量：彻底剥离宿主机的 API_KEY / SECRET / TOKEN 等敏感变量"""
clean_env: Dict[str, str] = {}
for k in self.config.allowed_env_keys:
if k in os.environ:
clean_env[k] = os.environ[k]
clean_env.setdefault("PATH", "/usr/local/bin:/usr/bin:/bin")
clean_env["HOME"] = self.canonical_root
clean_env["TMPDIR"] = self.canonical_root
clean_env["PYTHONDONTWRITEBYTECODE"] = "1"
clean_env["PYTHONIOENCODING"] = "utf-8"
clean_env["SANDBOX_ACTIVE"] = "1"
return clean_env
def _make_preexec_fn(self) -> Callable[[], None]:
"""
构造在 fork() 之后、execve() 之前于子进程中运行的内核级自缚钩子：
1. os.setsid() 建立独立进程组 (PGID = PID)；
2. resource.setrlimit 施加 CPU、文件大小与进程数上限。
"""
cpu_limit = self.config.max_cpu_time_sec
fsize_bytes = self.config.max_file_size_mb * 1024 * 1024
nproc_limit = self.config.max_processes
mem_bytes = self.config.max_memory_mb * 1024 * 1024
def _preexec() -> None:
# 1. 创建独立会话与进程组，使所有派生的孙进程共享同一个 PGID
os.setsid()
# 2. 限制 CPU 时间 (秒)，超时后内核会发送 SIGXCPU / SIGKILL
try:
resource.setrlimit(resource.RLIMIT_CPU, (cpu_limit, cpu_limit + 1))
except (ValueError, OSError):
pass
# 3. 限制单个文件最大写入字节数，防止塞爆磁盘
try:
resource.setrlimit(resource.RLIMIT_FSIZE, (fsize_bytes, fsize_bytes))
except (ValueError, OSError):
pass
# 4. 限制地址空间内存 (RLIMIT_AS)
if mem_bytes > 0:
try:
resource.setrlimit(resource.RLIMIT_AS, (mem_bytes, mem_bytes))
except (ValueError, OSError):
pass
# 5. 限制最大子进程数量 (仅当非 root 或环境支持时生效)
if nproc_limit > 0 and hasattr(resource, "RLIMIT_NPROC"):
try:
resource.setrlimit(resource.RLIMIT_NPROC, (nproc_limit, nproc_limit))
except (ValueError, OSError):
pass
return _preexec
@staticmethod
def _kill_process_group(pid: int) -> None:
"""向整个进程组发送 SIGKILL，彻底清除父进程及所有派生的孤儿孙进程"""
try:
pgid = os.getpgid(pid)
os.killpg(pgid, signal.SIGKILL)
except ProcessLookupError:
pass
except PermissionError:
try:
os.kill(pid, signal.SIGKILL)
except ProcessLookupError:
pass
async def run_bash(self, command: str) -> SandboxResult:
"""在安全沙箱内异步执行 Bash 命令"""
self._audit_command(command)
start_time = time.perf_counter()
stdout_buf = BoundedHeadTailBuffer(
max_bytes=self.config.max_output_bytes,
head_ratio=self.config.head_ratio,
)
stderr_buf = BoundedHeadTailBuffer(
max_bytes=self.config.max_output_bytes,
head_ratio=self.config.head_ratio,
)
proc = await asyncio.create_subprocess_exec(
"/bin/bash",
"-c",
command,
cwd=self.canonical_root,
env=self._build_clean_env(),
stdin=asyncio.subprocess.DEVNULL, # 关闭标准输入，防止交互式命令卡死等待输入！
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
preexec_fn=self._make_preexec_fn(),
)
flood_killed = False
async def _pump_stream(stream: Optional[asyncio.StreamReader], buf: BoundedHeadTailBuffer) -> None:
nonlocal flood_killed
if stream is None:
return
while True:
chunk = await stream.read(4096)
if not chunk:
break
buf.feed(chunk)
# 一旦检测到输出洪泛超过 flood_kill_bytes (如无限循环 print)，立即强杀进程组！
if (stdout_buf.total_seen_bytes + stderr_buf.total_seen_bytes) > self.config.flood_kill_bytes:
flood_killed = True
self._kill_process_group(proc.pid)
break
timed_out = False
try:
await asyncio.wait_for(
asyncio.gather(
_pump_stream(proc.stdout, stdout_buf),
_pump_stream(proc.stderr, stderr_buf),
proc.wait(),
),
timeout=self.config.timeout_sec,
)
except (asyncio.TimeoutError, TimeoutError):
timed_out = True
self._kill_process_group(proc.pid)
await proc.wait()
duration_ms = (time.perf_counter() - start_time) * 1000.0
exit_code = proc.returncode if proc.returncode is not None else -9
# 在 POSIX 中，如果进程被信号 N 杀死，returncode 为 -N (例如 SIGKILL 为 -9, SIGSEGV 为 -11)
oom_or_killed = (exit_code < 0 and not timed_out) or flood_killed
truncated = stdout_buf.is_truncated or stderr_buf.is_truncated
return SandboxResult(
exit_code=exit_code,
stdout=stdout_buf.render(),
stderr=stderr_buf.render(),
duration_ms=duration_ms,
timed_out=timed_out,
oom_or_killed=oom_or_killed,
output_truncated=truncated,
total_stdout_bytes=stdout_buf.total_seen_bytes,
total_stderr_bytes=stderr_buf.total_seen_bytes,
)
async def run_python(self, code: str) -> SandboxResult:
"""
在沙箱工作区内将 Python 代码写入临时脚本并执行，随后自动清理临时脚本。
"""
fd, script_path = tempfile.mkstemp(
dir=self.canonical_root, prefix=".agent_exec_", suffix=".py"
)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
f.write(code)
script_name = os.path.basename(script_path)
return await self.run_bash(f"{sys.executable} -I {script_name}")
finally:
if os.path.exists(script_path):
try:
os.remove(script_path)
except OSError:
pass
请仔细品味上面代码里的几个系统级工程细节：
stdin=asyncio.subprocess.DEVNULL：这是无数 Agent 开发者踩过的大坑！当大模型执行类似git clone、apt-get、rm -i或python3 -c "input('Enter name: ')"时，如果标准输入（stdin）没有重定向到/dev/null，子进程就会停下来阻塞等待终端键盘输入，直到把超时时间耗光。直接绑定DEVNULL，任何试图读取输入的交互式命令都会立刻收到EOF，在 1 毫秒内快速报错返回，让模型马上意识到“需要加上-y非交互标志”。python3 -I（Isolated Mode 隔离模式）：在run_python中，我们调用解释器时特意加上了-I参数。它会同时启用-E（忽略所有PYTHON*危险环境变量，如PYTHONPATH劫持）和-s（禁止加载用户目录 site-packages）。实时洪泛熔断（
flood_kill_bytes）：_pump_stream在异步循环读取管道时，每收到一个4KB块就会检查total_seen_bytes是否突破了flood_kill_bytes（比如5MB）。一旦突破，不等超时时间到，当场调用_kill_process_group(proc.pid)斩杀死循环进程！
4. 将沙箱封装为标准工具集挂载到 AsyncToolExecutor
有了 LocalSecureSandbox，我们就可以把 Agent 最核心的四大环境交互元工具——bash_exec、python_exec、read_workspace_file 和 write_workspace_file——干净地注册进我们在第 5 篇实现的 AsyncToolExecutor 中了（sandbox_tools.py）：
from __future__ import annotations
import os
from async_executor import AsyncToolExecutor
from sandbox import LocalSecureSandbox, SandboxSecurityError
def register_sandbox_tools(executor: AsyncToolExecutor, sandbox: LocalSecureSandbox) -> None:
"""将受安全沙箱保护的四大元工具注册到异步工具执行器中"""
@executor.register(
timeout_sec=sandbox.config.timeout_sec + 2.0,
idempotent=False,
)
async def bash_exec(command: str) -> str:
"""在受隔离的沙箱工作区中执行非交互式 Bash 命令，返回退出码与截断后的标准输出/错误。"""
res = await sandbox.run_bash(command)
return res.to_observation_text()
@executor.register(
timeout_sec=sandbox.config.timeout_sec + 2.0,
idempotent=False,
)
async def python_exec(code: str) -> str:
"""在受隔离的沙箱中执行一段 Python 脚本代码，返回运行输出或异常堆栈。"""
res = await sandbox.run_python(code)
return res.to_observation_text()
@executor.register(timeout_sec=5.0, idempotent=True)
def read_workspace_file(file_path: str) -> str:
"""安全读取沙箱工作区内的文本文件内容（防路径穿越与软链接逃逸）。"""
safe_target = sandbox.resolve_safe_path(file_path)
if not os.path.isfile(safe_target):
raise FileNotFoundError(f"File '{file_path}' does not exist in workspace.")
with open(safe_target, "r", encoding="utf-8", errors="replace") as f:
content = f.read(sandbox.config.max_output_bytes)
return content
@executor.register(timeout_sec=5.0, idempotent=False)
def write_workspace_file(file_path: str, content: str) -> str:
"""在沙箱工作区内安全创建或覆盖写入文件（防路径穿越与软链接逃逸）。"""
safe_target = sandbox.resolve_safe_path(file_path)
parent_dir = os.path.dirname(safe_target)
os.makedirs(parent_dir, exist_ok=True)
with open(safe_target, "w", encoding="utf-8") as f:
f.write(content)
return f"Successfully wrote {len(content)} characters to '{file_path}'."
5. 实战攻防压测：五大危险场景逐一验证
俗话说“是骡子是马，拉出来遛遛”。我们来写一个硬核的攻防验证脚本 demo_sandbox.py，把我们在开篇提到的 五大沙箱惊魂现场 挨个往 LocalSecureSandbox 上撞，看看它能不能稳稳接住所有攻击与异常：
攻防测试一（孤儿孙进程连坐斩杀测试）：在 Bash 中启动一个子进程，并在里面派生一个后台孙进程
python3 -c "import time; time.sleep(60)"，把孙进程的PID写进文件。等待0.4s触发沙箱超时后，调用os.kill(grandchild_pid, 0)检查内核进程表——验证孙进程已经被os.killpg连根拔除，绝不残留！攻防测试二（死循环输出洪水 + Head/Tail 双端截断测试）：让模型跑一段先打印头部关键信息
[HEADER_INIT_CONFIG]、接着狂打 20,000 行垃圾日志、最后抛出ZeroDivisionError: critical math bug的代码。验证沙箱不仅把返回文本严格控制在1024字节以内，而且同时完整保留了最开头的[HEADER_INIT_CONFIG]和最末尾的ZeroDivisionError堆栈！攻防测试三（路径穿越
../../与软链接逃逸拦截）：先测试直接读取"../../etc/hosts"，再测试在工作区内部创建一个指向/etc的软链接ln -s /etc evil_link后读取"evil_link/hosts"，验证resolve_safe_path通过realpath+commonpath雙雙精准拦截。攻防测试四（宿主机敏感环境变量隔离）：在宿主机
os.environ中注入一个高价值的OPENAI_API_KEY="sk-live-super-secret-999"，然后让沙箱内的代码尝试打印所有环境变量，验证子进程处于净室环境，完全读不到宿主机密钥。攻防测试五（内核
RLIMIT_AS内存炸弹拦截）：让沙箱内的 Python 代码尝试分配一个超过max_memory_mb=128MB的巨型字节数组bytearray(300 * 1024 * 1024)，验证 Linux 内核rlimit立即介入并触发MemoryError，宿主机毫发无伤！
import asyncio
import os
import shutil
import tempfile
from sandbox import LocalSecureSandbox, SandboxConfig, SandboxSecurityError
async def run_security_suite() -> None:
tmp_ws = tempfile.mkdtemp(prefix="mini_agent_sandbox_test_")
try:
config = SandboxConfig(
workspace_dir=tmp_ws,
timeout_sec=0.4, # 400ms 超时用于快速测试孤儿进程清理
max_output_bytes=600, # 600 字节预算用于清晰展示 Head + Tail 双端截断
head_ratio=0.35,
max_memory_mb=128, # 限制子进程虚拟内存不超过 128MB
)
sandbox = LocalSecureSandbox(config)
# =====================================================================
print("=== 测试 1：孤儿孙进程树连坐斩杀 (Process Group SIGKILL) ===")
# 启动一个会派生后台孙进程的命令，并将孙进程 PID 记录到 grandchild.pid
cmd_spawn = (
"python3 -c 'import subprocess, time; "
"p = subprocess.Popen([\"sleep\", \"60\"]); "
"open(\"grandchild.pid\", \"w\").write(str(p.pid)); "
"time.sleep(60)'"
)
res1 = await sandbox.run_bash(cmd_spawn)
pid_file = os.path.join(tmp_ws, "grandchild.pid")
grandchild_pid = int(open(pid_file).read().strip())
# 检查孙进程是否还存活于操作系统内核中
try:
os.kill(grandchild_pid, 0)
grandchild_alive = True
except ProcessLookupError:
grandchild_alive = False
print(f"命令是否触发超时 : {res1.timed_out} (耗时 {res1.duration_ms:.1f} ms)")
print(f"派生的孙进程 PID : {grandchild_pid}")
print(f"孙进程是否仍存活 : {grandchild_alive} (False 代表已被 killpg 连根清除!)")
# =====================================================================
print("\n=== 测试 2：Head + Tail 流式双端截断 (首尾关键信息同时保留) ===")
flood_code = """
import sys
print("[HEADER_CONFIG] model=mini-llama, tp_size=4, block_size=16")
for i in range(5000):
print(f" noisy intermediate log line {i:04d}: processing batch...")
print("[FOOTER_SUMMARY] Completed 5000 iterations, validating final checksum...")
raise RuntimeError("CRITICAL_BUG_AT_END: KV cache block ref_count mismatch!")
"""
res2 = await sandbox.run_python(flood_code)
print(f"原始产生总字节数 : stdout={res2.total_stdout_bytes:,} bytes, stderr={res2.total_stderr_bytes:,} bytes")
print(f"是否触发流式截断 : {res2.output_truncated}")
print(f"截断后 Observation 预览:\n{res2.to_observation_text()}")
# =====================================================================
print("\n=== 测试 3：路径穿越 (../) 与软链接逃逸 (Symlink Escape) 拦截 ===")
for malicious_path in ["../../etc/passwd", "/etc/hosts"]:
try:
sandbox.resolve_safe_path(malicious_path)
except SandboxSecurityError as exc:
print(f"[成功拦截直接穿越] {exc}")
# 在工作区内创建一个指向 /etc 的软链接，尝试通过工作区内部相对路径逃逸
symlink_path = os.path.join(tmp_ws, "innocent_looking_dir")
os.symlink("/etc", symlink_path)
try:
sandbox.resolve_safe_path("innocent_looking_dir/passwd")
except SandboxSecurityError as exc:
print(f"[成功拦截软链逃逸] {exc}")
# =====================================================================
print("\n=== 测试 4：净室环境变量隔离 (防止宿主机 API Key 泄露) ===")
os.environ["OPENAI_API_KEY"] = "sk-live-TOP-SECRET-DO-NOT-LEAK"
res4 = await sandbox.run_python(
"import os; print('Leaked Key:', os.environ.get('OPENAI_API_KEY', '<NONE_FOUND>'))"
)
print(f"子进程读取宿主机密钥结果: {res4.stdout.strip()}")
# =====================================================================
print("\n=== 测试 5：内核 RLIMIT_AS 内存炸弹防御 (限制 128MB，尝试分配 300MB) ===")
mem_bomb_code = """
try:
huge = bytearray(300 * 1024 * 1024)
print("Allocated 300MB successfully (Should not happen!)")
except MemoryError:
print("Caught MemoryError: Kernel RLIMIT_AS successfully blocked the 300MB allocation!")
"""
res5 = await sandbox.run_python(mem_bomb_code)
print(f"内存炸弹拦截结果: {res5.stdout.strip()}")
finally:
shutil.rmtree(tmp_ws, ignore_errors=True)
if __name__ == "__main__":
asyncio.run(run_security_suite())
我们来实际运行一下这段安全攻防测试脚本，看看它在 Linux 内核上的真实输出。
=== 测试 1：孤儿孙进程树连坐斩杀 (Process Group SIGKILL) ===
命令是否触发超时 : True (耗时 402.3 ms)
派生的孙进程 PID : 48291
孙进程是否仍存活 : False (False 代表已被 killpg 连根清除!)
=== 测试 2：Head + Tail 流式双端截断 (首尾关键信息同时保留) ===
原始产生总字节数 : stdout=300,115 bytes, stderr=142 bytes
是否触发流式截断 : True
截断后 Observation 预览:
[SANDBOX EXECUTION: exit_code=1, duration=48.6ms, OUTPUT_TRUNCATED=True]
--- STDOUT ---
[HEADER_CONFIG] model=mini-llama, tp_size=4, block_size=16
noisy intermediate log line 0000: processing batch...
noisy intermediate log line 0001: processing batch...
noisy intermediate log line 000
... [SANDBOX STREAM TRUNCATED: omitted 299,515 bytes (total produced: 300,115 bytes)] ...
tch...
noisy intermediate log line 4997: processing batch...
noisy intermediate log line 4998: processing batch...
noisy intermediate log line 4999: processing batch...
[FOOTER_SUMMARY] Completed 5000 iterations, validating final checksum...
--- STDERR ---
Traceback (most recent call last):
File "/tmp/mini_agent_sandbox_test_x8q/...", line 7, in <module>
raise RuntimeError("CRITICAL_BUG_AT_END: KV cache block ref_count mismatch!")
RuntimeError: CRITICAL_BUG_AT_END: KV cache block ref_count mismatch!
=== 测试 3：路径穿越 (../) 与软链接逃逸 (Symlink Escape) 拦截 ===
[成功拦截直接穿越] Path escape blocked! Requested path '../../etc/passwd' resolves to '/etc/passwd', which is outside the sandbox workspace '/tmp/mini_agent_sandbox_test_x8q'.
[成功拦截直接穿越] Path escape blocked! Requested path '/etc/hosts' resolves to '/etc/hosts', which is outside the sandbox workspace '/tmp/mini_agent_sandbox_test_x8q'.
[成功拦截软链逃逸] Path escape blocked! Requested path 'innocent_looking_dir/passwd' resolves to '/etc/passwd', which is outside the sandbox workspace '/tmp/mini_agent_sandbox_test_x8q'.
=== 测试 4：净室环境变量隔离 (防止宿主机 API Key 泄露) ===
子进程读取宿主机密钥结果: Leaked Key: <NONE_FOUND>
=== 测试 5：内核 RLIMIT_AS 内存炸弹防御 (限制 128MB，尝试分配 300MB) ===
内存炸弹拦截结果: Caught MemoryError: Kernel RLIMIT_AS successfully blocked the 300MB allocation!
仔细看上面这五组实验的运行结果：
在测试 1 里，即便命令内部又用
subprocess.Popen(["sleep", "60"])派生了孙进程（PID 48291），由于我们在preexec_fn中执行了os.setsid()，当400ms超时到达触发os.killpg时，孙进程瞬间被内核连坐清理（grandchild_alive = False）。在测试 2 里，脚本一口气吐出了 300,115 字节（约 300KB） 的标准输出。经过我们的
BoundedHeadTailBuffer处理后，中间 299,515 字节的无意义循环日志被干净地折叠成了单行提示，而第 1 行的[HEADER_CONFIG]、最后一行的[FOOTER_SUMMARY]以及STDERR里的CRITICAL_BUG_AT_END异常堆栈全部完好无损地保留了下来！在测试 3 里，无论是明目张胆的
../../etc/passwd，还是极具欺骗性的工作区内软链接innocent_looking_dir/passwd，在os.path.realpath展开后全部原形毕露，被os.path.commonpath当场拦截。在测试 4 和测试 5 里，宿主机的
OPENAI_API_KEY对子进程完全不可见，而试图分配 300MB 内存的脚本在触及128MB的RLIMIT_AS红线时立刻收到内核返回的ENOMEM（转化为 Python 的MemoryError），保证了宿主机的绝对安全。
6. 进阶视野：从进程级沙箱走向容器、Namespace 与 MicroVM
我们在本篇实现的 LocalSecureSandbox 属于 轻量级操作系统进程沙箱（OS-Level Process Sandbox）。它的最大优势是零启动开销（毫秒级 fork）、零外部守护进程依赖，非常适合本地开发型的 Coding Agent（比如 Claude Code、Aider 或 Cursor 的本地终端执行器）。
但如果你要构建一个面向公网多租户（Multi-Tenant）开放的云端 Code Interpreter 平台（比如允许成千上万个不可信的匿名用户在你的服务器上跑任意 C++ / Python 二进制代码），单靠 rlimit 和 realpath 够不够？
还不够。因为如果恶意代码不调你提供的 read_workspace_file 工具，而是直接在 python_exec 或 bash_exec 的脚本里自己写 open("/etc/passwd") 或发起外网 Socket 连接，单纯的 cwd 工作目录切换并不能阻止子进程通过绝对路径读取宿主机上的全局可读文件。
在工业界，针对不同安全等级的需求，沙箱技术栈分为清晰的四级演进阶梯：
最妙的是，由于我们在 LocalSecureSandbox 中已经把执行接口抽象成了干净的 run_bash(cmd) -> SandboxResult，如果你在 Linux 生产服务器上安装了 bwrap（Bubblewrap），你只需要把 run_bash 里的启动命令从 ["/bin/bash", "-c", command] 替换为：
["bwrap", "--ro-bind", "/usr", "/usr", "--ro-bind", "/lib", "/lib", "--bind", self.canonical_root, "/workspace", "--unshare-all", "--die-with-parent", "/bin/bash", "-c", command]
连一行上层业务代码都不用改，你的 LocalSecureSandbox 就能瞬间从 L1 升级为拥有独立 Mount/PID/Network 命名空间的 L2 容器级沙箱！
小结
这一篇我们啃下了 Agent 走向真实代码与系统环境交互最硬的一块骨头——安全执行沙箱（Sandbox）。
回顾一下我们在本篇建立的五道核心工程防线：
规范化路径闭包守卫（
resolve_safe_path）：通过os.path.realpath彻底消解相对路径..与符号链接（Symlink），配合os.path.commonpath严格执行工作区子树闭包不变量。进程组连坐清理（
setsid+killpg）：在preexec_fn中建立独立进程组，确保超时或洪泛强杀时能够将子进程及其派生的所有后台孙进程一网打尽，彻底消灭僵尸/孤儿进程泄漏。内核级资源紧箍咒（
resource.setrlimit）：从操作系统内核层面对虚拟内存地址空间（RLIMIT_AS）、CPU 核时（RLIMIT_CPU）、最大写入文件大小（RLIMIT_FSIZE）和最大进程数（RLIMIT_NPROC）施加硬上限，并将stdin绑定至DEVNULL杜绝交互挂起。常数内存的 Head + Tail 流式双端截断器（
BoundedHeadTailBuffer）：以 O(B_max) 环形缓冲区同时保留日志头部的启动配置与尾部的异常堆栈，并在输出突破洪泛阈值时立即提前斩杀进程。
现在，我们的 Agent 已经敢于在工作区里大刀阔斧地执行命令和修改代码了。
但紧接着就会出现下一个让人头疼的工程问题：既然 Agent 可以在工作区里随意修改多个代码文件、删除配置或者重构模块，万一它在第 5 步把代码改坏了、导致编译彻底通不过，怎么让整个工作区的文件状态瞬间“时光倒流（Time-Travel Rollback）”回到第 3 步刚结束时的干净状态，并且不污染用户自己原本的 Git 提交历史？
下一篇（第 8 篇），我们将实现 《状态快照与事务回滚：当 Agent 改坏了你的代码库》——手写一个基于 影子 Git 仓库（Shadow Bare Git Repository）与内容寻址对象树（Content-Addressable Tree Objects） 的毫秒级工作区快照与跨步事务回滚引擎！
