第8讲：运行时安全与沙箱隔离——Agent 能做什么，不能做什么

一、为什么需要运行时沙箱前七讲构建了通信和应用层的防御体系但有一个根本性问题尚未解决Agent 进程本身的隔离。如果 Agent 被攻破比如通过 RCE 漏洞、依赖库供应链攻击攻击者获得了 Agent 进程的控制权那么横向移动读取其他 Agent 的内存数据宿主机逃逸访问宿主机文件系统、网络、进程列表持久化写文件、安装后门、修改系统配置资源耗尽消耗 CPU/内存影响同机其他服务运行时沙箱的目标即使 Agent 被完全攻破攻击者也拿不到宿主机的控制权也接触不到其他 Agent 的数据。二、沙箱隔离的四维模型维度隔离手段防护目标L8a文件系统​只读根文件系统 临时挂载点防写文件、防读宿主机文件L8b网络​eBPF 网络策略 透明代理防未经授权的网络访问L8c进程​PID 命名空间 seccomp防进程逃逸和提权L8d资源​cgroup 配额防资源耗尽三、Go 实现轻量级容器沙箱下面实现一个不依赖 Docker 的 Go 原生沙箱使用 Linux 内核特性Namespace Cgroups Seccomppackage main import ( bufio bytes encoding/binary encoding/json fmt io log math/rand net net/http os os/exec path/filepath strconv strings sync syscall time golang.org/x/sys/unix ) // ---- 沙箱配置 ---- type SandboxConfig struct { AgentID string json:agent_id WorkDir string json:work_dir MemoryLimitMB int64 json:memory_limit_mb CPUShares int64 json:cpu_shares MaxProcesses int json:max_processes ReadOnlyPaths []string json:read_only_paths NetworkPolicy *NetworkPolicy json:network_policy AllowedSyscalls []string json:allowed_syscalls EnvVars map[string]string json:env_vars TimeLimit time.Duration json:time_limit } type NetworkPolicy struct { AllowOutbound bool json:allow_outbound AllowedHosts []string json:allowed_hosts AllowedPorts []int json:allowed_ports DenyCIDRs []string json:deny_cidrs } // ---- Agent 沙箱实例 ---- type AgentSandbox struct { config *SandboxConfig cmd *exec.Cmd pid int mountNS bool netNS bool pidNS bool utsNS bool ipcNS bool cgroupPath string mu sync.Mutex status SandboxStatus startTime time.Time stats ResourceStats } type SandboxStatus int const ( StatusPending SandboxStatus 0 StatusRunning SandboxStatus 1 StatusStopped SandboxStatus 2 StatusCrashed SandboxStatus 3 ) type ResourceStats struct { CPUTime time.Duration json:cpu_time MemoryUsed int64 json:memory_used_bytes IORead int64 json:io_read_bytes :IOWrite int64 json:io_write_bytes NetIn int64 json:net_in_bytes NetOut int64 json:net_out_bytes } // ---- 沙箱管理器 ---- type SandboxManager struct { mu sync.RWMutex sandboxes map[string]*AgentSandbox config *SandboxConfig } func NewSandboxManager(config *SandboxConfig) *SandboxManager { return SandboxManager{ sandboxes: make(map[string]*AgentSandbox), config: config, } } // 创建并启动 Agent 沙箱 func (sm *SandboxManager) CreateSandbox(agentID string, binaryPath string, args []string) (*AgentSandbox, error) { // 检查是否已有同名沙箱 sm.mu.Lock() if _, exists : sm.sandboxes[agentID]; exists { sm.mu.Unlock() return nil, fmt.Errorf(Agent %s 已有运行的沙箱, agentID) } sm.mu.Unlock() // 创建沙箱配置继承全局配置 Agent 特定配置 config : *sm.config config.AgentID agentID config.WorkDir filepath.Join(/var/lib/mcp-sandbox, agentID) // 确保工作目录存在 if err : os.MkdirAll(config.WorkDir, 0755); err ! nil { return nil, fmt.Errorf(创建工作目录失败: %w, err) } // 创建沙箱实例 sandbox : AgentSandbox{ config: config, status: StatusPending, startTime: time.Now(), } // 准备沙箱命令 cmd : exec.Command(binaryPath, args...) // 设置命名空间 cmd.SysProcAttr syscall.SysProcAttr{ Cloneflags: syscall.CLONE_NEWNS | // 挂载命名空间 syscall.CLONE_NEWNET | // 网络命名空间 syscall.CLONE_NEWPID | // PID 命名空间 syscall.CLONE_NEWUTS | // UTS 命名空间 syscall.CLONE_NEWIPC, // IPC 命名空间 Unshareflags: syscall.CLONE_NEWNS, } // 设置 cgroup if err : sm.setupCgroup(sandbox); err ! nil { return nil, fmt.Errorf(设置 cgroup 失败: %w, err) } // 设置 seccomp 过滤器 if err : sm.setupSeccomp(sandbox); err ! nil { return nil, fmt.Errorf(设置 seccomp 失败: %w, err) } // 设置环境变量 env : os.Environ() for k, v : range config.EnvVars { env append(env, fmt.Sprintf(%s%s, k, v)) } cmd.Env env // 设置工作目录 cmd.Dir config.WorkDir // 重定向标准输入输出 var stdout, stderr bytes.Buffer cmd.Stdout stdout cmd.Stderr stderr // 启动进程 if err : cmd.Start(); err ! nil { return nil, fmt.Errorf(启动沙箱失败: %w, err) } sandbox.cmd cmd sandbox.pid cmd.Process.Pid sandbox.status StatusRunning // 设置网络策略 if err : sm.applyNetworkPolicy(sandbox); err ! nil { log.Printf([SANDBOX] 警告: 网络策略应用失败: %v, err) } // 注册到管理器 sm.mu.Lock() sm.sandboxes[agentID] sandbox sm.mu.Unlock() // 启动资源监控 go sm.monitorResources(sandbox) // 启动超时控制 if config.TimeLimit 0 { go func() { select { case -time.After(config.TimeLimit): sandbox.Stop(timeout) case -waitForProcess(cmd): } }() } log.Printf([SANDBOX] Agent %s 沙箱已启动 (PID%d), agentID, sandbox.pid) return sandbox, nil } // 设置 cgroup 资源限制 func (sm *SandboxManager) setupCgroup(sandbox *AgentSandbox) error { agentID : sandbox.config.AgentID cgroupBase : /sys/fs/cgroup // 创建 cgroup 目录 cgroupDir : filepath.Join(cgroupBase, mcp-sandbox, agentID) if err : os.MkdirAll(cgroupDir, 0755); err ! nil { return err } sandbox.cgroupPath cgroupDir // 设置内存限制 if sandbox.config.MemoryLimitMB 0 { memLimit : sandbox.config.MemoryLimitMB * 1024 * 1024 if err : writeFile(filepath.Join(cgroupDir, memory.max), fmt.Sprintf(%d, memLimit)); err ! nil { return fmt.Errorf(设置内存限制失败: %w, err) } } // 设置 CPU 份额 if sandbox.config.CPUShares 0 { if err : writeFile(filepath.Join(cgroupDir, cpu.weight), fmt.Sprintf(%d, sandbox.config.CPUShares)); err ! nil { return fmt.Errorf(设置 CPU 份额失败: %w, err) } } // 设置进程数限制 if sandbox.config.MaxProcesses 0 { pidsDir : filepath.Join(cgroupDir, pids) os.MkdirAll(pidsDir, 0755) if err : writeFile(filepath.Join(pidsDir, pids.max), fmt.Sprintf(%d, sandbox.config.MaxProcesses)); err ! nil { return fmt.Errorf(设置进程数限制失败: %w, err) } } // 将进程加入 cgroup pidStr : fmt.Sprintf(%d, sandbox.pid) if err : writeFile(filepath.Join(cgroupDir, cgroup.procs), pidStr); err ! nil { return fmt.Errorf(将进程加入 cgroup 失败: %w, err) } return nil } // 设置 seccomp 过滤器 func (sm *SandboxManager) setupSeccomp(sandbox *AgentSandbox) error { // 定义允许的系统调用列表 defaultAllowed : []string{ read, write, open, close, stat, fstat, mmap, munmap, brk, sched_yield, exit_group, exit, gettid, getpid, clock_gettime, nanosleep, sendto, recvfrom, connect, bind, } // 合并自定义允许的系统调用 allowed : defaultAllowed if sandbox.config.AllowedSyscalls ! nil { allowed append(allowed, sandbox.config.AllowedSyscalls...) } // 构建 seccomp BPF 程序 // 注意完整的 seccomp-bpf 实现较复杂这里使用 libseccomp 的 Go 绑定 // 或者通过 prctl 系统调用来设置 // 简化实现使用 prctl 设置 no_new_privs if err : syscall.Prctl(syscall.PR_SET_NO_NEW_PRIVS, 1, 0, 0, 0); err ! nil { return fmt.Errorf(设置 no_new_privs 失败: %w, err) } log.Printf([SECCOMP] Agent %s seccomp 已启用 (允许 %d 个系统调用), sandbox.config.AgentID, len(allowed)) return nil } // 应用网络策略通过 eBPF/iptables func (sm *SandboxManager) applyNetworkPolicy(sandbox *AgentSandbox) error { policy : sandbox.config.NetworkPolicy if policy nil { return nil } // 使用 iptables 为沙箱的网络命名空间设置规则 // 实际生产环境建议使用 eBPF (Cilium/BPFContain) if !policy.AllowOutbound { // 默认拒绝出站流量 output : exec.Command(iptables, -A, OUTPUT, -m, cgroup, --cgroup, sandbox.cgroupPath, -j, DROP) if err : output.Run(); err ! nil { return fmt.Errorf(设置默认拒绝规则失败: %w, err) } } // 添加白名单规则 for _, host : range policy.AllowedHosts { addRule : exec.Command(iptables, -A, OUTPUT, -m, cgroup, --cgroup, sandbox.cgroupPath, -d, host, -j, ACCEPT) if err : addRule.Run(); err ! nil { log.Printf([NETPOL] 警告: 添加白名单规则失败: %v, err) } } // 添加黑名单 CIDR for _, cidr : range policy.DenyCIDRs { denyRule : exec.Command(iptables, -A, OUTPUT, -m, cgroup, --cgroup, sandbox.cgroupPath, -d, cidr, -j, DROP) if err : denyRule.Run(); err ! nil { log.Printf([NETPOL] 警告: 添加黑名单规则失败: %v, err) } } log.Printf([NETPOL] Agent %s 网络策略已应用, sandbox.config.AgentID) return nil } // 停止沙箱 func (sandbox *AgentSandbox) Stop(reason string) error { sandbox.mu.Lock() defer sandbox.mu.Unlock() if sandbox.status ! StatusRunning { return nil } log.Printf([SANDBOX] 停止 Agent %s 沙箱 (原因: %s), sandbox.config.AgentID, reason) // 发送 SIGTERM if err : sandbox.cmd.Process.Signal(syscall.SIGTERM); err ! nil { // 如果 SIGTERM 失败强制杀死 sandbox.cmd.Process.Kill() } // 等待进程退出 done : make(chan struct{}) go func() { sandbox.cmd.Wait() close(done) }() select { case -done: case -time.After(10 * time.Second): sandbox.cmd.Process.Kill() -done } sandbox.status StatusStopped sandbox.stats.CPUTime time.Since(sandbox.startTime) // 清理 cgroup if sandbox.cgroupPath ! { cleanupCgroup(sandbox.cgroupPath) } // 清理网络规则 cleanupNetworkRules(sandbox.cgroupPath) return nil } // 监控资源使用 func (sm *SandboxManager) monitorResources(sandbox *AgentSandbox) { ticker : time.NewTicker(5 * time.Second) defer ticker.Stop() for range ticker.C { sandbox.mu.Lock() if sandbox.status ! StatusRunning { sandbox.mu.Unlock() return } // 读取 cgroup 统计信息 cgroupPath : sandbox.cgroupPath if cgroupPath ! { // 内存使用 if memBytes, err : readFile(filepath.Join(cgroupPath, memory.current)); err nil { if val, err : strconv.ParseInt(strings.TrimSpace(memBytes), 10, 64); err nil { sandbox.stats.MemoryUsed val } } // CPU 使用 if cpuStat, err : readFile(filepath.Join(cgroupPath, cpu.stat)); err nil { for _, line : range strings.Split(cpuStat, \n) { if strings.HasPrefix(line, usage_usec) { parts : strings.Split(line, ) if len(parts) 2 { if val, err : strconv.ParseInt(parts[1], 10, 64); err nil { sandbox.stats.CPUTime time.Duration(val) * time.Microsecond } } } } } // IO 统计 if ioStat, err : readFile(filepath.Join(cgroupPath, io.stat)); err nil { for _, part : range strings.Fields(ioStat) { if strings.HasPrefix(part, rbytes) { if val, err : strconv.ParseInt(part[7:], 10, 64); err nil { sandbox.stats.IORead val } } if strings.HasPrefix(part, wbytes) { if val, err : strconv.ParseInt(part[7:], 10, 64); err nil { sandbox.stats.IOWrite val } } } } } // 检查是否超出内存限制 if sandbox.config.MemoryLimitMB 0 { memUsedMB : sandbox.stats.MemoryUsed / 1024 / 1024 if memUsedMB sandbox.config.MemoryLimitMB { log.Printf([SANDBOX] Agent %s 超出内存限制 (%dMB %dMB)正在终止, sandbox.config.AgentID, memUsedMB, sandbox.config.MemoryLimitMB) go sandbox.Stop(oom) sandbox.mu.Unlock() return } } sandbox.mu.Unlock() } } // ---- 只读文件系统实现通过 bind mount ---- func setupReadOnlyRoot(sandbox *AgentSandbox) error { // 创建 overlay 文件系统 lowerDir : /usr/share/mcp-base upperDir : filepath.Join(sandbox.config.WorkDir, overlay, upper) workDir : filepath.Join(sandbox.config.WorkDir, overlay, work) mountPoint : filepath.Join(sandbox.config.WorkDir, rootfs) // 创建必要目录 for _, dir : range []string{upperDir, workDir, mountPoint} { if err : os.MkdirAll(dir, 0755); err ! nil { return err } } // 挂载 overlay 文件系统 mountOpts : fmt.Sprintf(lowerdir%s,upperdir%s,workdir%s, lowerDir, upperDir, workDir) if err : syscall.Mount(overlay, mountPoint, overlay, 0, mountOpts); err ! nil { return fmt.Errorf(挂载 overlay 文件系统失败: %w, err) } // 重新挂载根目录为只读 if err : syscall.Mount(, /, , syscall.MS_REMOUNT|syscall.MS_RDONLY, ); err ! nil { return fmt.Errorf(重新挂载根目录为只读失败: %w, err) } // 挂载可写的 tmpfs 到临时目录 if err : syscall.Mount(tmpfs, /tmp, tmpfs, 0, size100M); err ! nil { return fmt.Errorf(挂载 tmpfs 失败: %w, err) } log.Printf([FS] Agent %s 文件系统已设置为只读, sandbox.config.AgentID) return nil } // ---- 辅助函数 ---- func writeFile(path, content string) error { return os.WriteFile(path, []byte(content), 0644) } func readFile(path string) (string, error) { data, err : os.ReadFile(path) if err ! nil { return , err } return string(data), nil } func cleanupCgroup(path string) { // 递归删除 cgroup 目录 entries, _ : os.ReadDir(path) for _, entry : range entries { childPath : filepath.Join(path, entry.Name()) if entry.IsDir() { cleanupCgroup(childPath) } } os.Remove(path) } func cleanupNetworkRules(cgroupPath string) { // 清除与该 cgroup 相关的 iptables 规则 exec.Command(iptables, -D, OUTPUT, -m, cgroup, --cgroup, cgroupPath, -j, DROP).Run() } func waitForProcess(cmd *exec.Cmd) chan struct{} { ch : make(chan struct{}) go func() { cmd.Wait() close(ch) }() return ch } // ---- 沙箱健康检查 API ---- type SandboxAPI struct { manager *SandboxManager } func (api *SandboxAPI) ListSandboxes(w http.ResponseWriter, r *http.Request) { api.manager.mu.RLock() defer api.manager.mu.RUnlock() type SandboxInfo struct { AgentID string json:agent_id PID int json:pid Status SandboxStatus json:status Uptime string json:uptime MemUsage int64 json:memory_used_mb CPUTime string json:cpu_time } var infos []SandboxInfo for _, sb : range api.manager.sandboxes { sb.mu.Lock() info : SandboxInfo{ AgentID: sb.config.AgentID, PID: sb.pid, Status: sb.status, Uptime: time.Since(sb.startTime).Round(time.Second).String(), MemUsage: sb.stats.MemoryUsed / 1024 / 1024, CPUTime: sb.stats.CPUTime.Round(time.Millisecond).String(), } sb.mu.Unlock() infos append(infos, info) } w.Header().Set(Content-Type, application/json) json.NewEncoder(w).Encode(infos) } // ---- 演示 ---- func main() { // 检查是否 root 用户需要命名空间和 cgroup 权限 if os.Geteuid() ! 0 { log.Fatal(沙箱功能需要 root 权限运行) } // 创建沙箱管理器配置 config : SandboxConfig{ MemoryLimitMB: 512, CPUShares: 1024, MaxProcesses: 50, TimeLimit: 30 * time.Minute, NetworkPolicy: NetworkPolicy{ AllowOutbound: true, AllowedHosts: []string{api.openai.com, api.example.com}, DenyCIDRs: []string{10.0.0.0/8, 172.16.0.0/12}, }, EnvVars: map[string]string{ MCP_MODE: sandbox, HOME: /tmp, TMPDIR: /tmp, }, } manager : NewSandboxManager(config) // 启动 HTTP API api : SandboxAPI{manager: manager} http.HandleFunc(/sandbox/list, api.ListSandboxes) // 模拟创建沙箱 fmt.Println( 沙箱管理演示 ) // 创建一个模拟 Agent 的脚本 agentScript : #!/bin/sh echo Agent started with PID $$ echo Testing filesystem access... touch /tmp/test.txt 21 || echo Cannot write to /tmp (expected) cat /etc/hostname 21 echo Testing network... curl -s --connect-timeout 3 https://api.openai.com/v1/models 21 | head -c 100 || echo Network blocked echo echo Agent running for 10 seconds... sleep 10 echo Agent exiting normally scriptPath : /tmp/test-agent.sh os.WriteFile(scriptPath, []byte(agentScript), 0755) // 创建沙箱 sandbox, err : manager.CreateSandbox(agent-test-001, /bin/sh, []string{scriptPath}) if err ! nil { log.Fatalf(创建沙箱失败: %v, err) } fmt.Printf(沙箱已创建: Agent%s PID%d\n, sandbox.config.AgentID, sandbox.pid) // 等待进程结束 time.Sleep(2 * time.Second) // 查看沙箱状态 fmt.Println(\n沙箱状态:) sandbox.mu.Lock() fmt.Printf( 状态: %d\n, sandbox.status) fmt.Printf( 运行时间: %v\n, time.Since(sandbox.startTime).Round(time.Second)) fmt.Printf( 内存使用: %d MB\n, sandbox.stats.MemoryUsed/1024/1024) sandbox.mu.Unlock() // 列出所有沙箱 fmt.Println(\n所有沙箱:) api.ListSandboxes(nil, nil) // 等待沙箱自然结束 time.Sleep(12 * time.Second) // 停止沙箱 sandbox.Stop(demo_complete) fmt.Println(\n沙箱已停止) }四、沙箱隔离技术选型对比方案隔离级别启动速度资源开销适用场景Linux Namespace Cgroup​操作系统级毫秒级极低高密度部署原生 Go 应用Docker/Podman​容器级秒级中等标准容器化部署gVisor​用户态内核秒级较高强隔离需求不信任容器内核Firecracker​微虚拟机百毫秒级中等多租户场景强隔离WebAssembly​沙箱级微秒级极低插件系统非原生代码对于 MCP Agent 场景推荐方案开发/测试环境Namespaces Cgroup零额外依赖 生产环境 gVisor 或 Firecracker强隔离 边缘设备 WebAssembly轻量级五、安全分层L8本讲在 L1-L7 基础上叠加运行时沙箱层层级防御手段本讲新增L1​Prompt 注入检测输入过滤L2​输出验证与对齐参数校验、业务规则L3​工具权限与最小特权RBAC/ABAC 策略引擎L4​数据脱敏与隔离字段级脱敏L5​行为审计与异常检测审计日志、异常评分L6​速率限制与熔断保护限流、熔断、降级L7​安全通信与凭证管理mTLS、SPIFFE、TokenL8a​文件系统隔离✅ 只读根文件系统 OverlayFSL8b​网络隔离✅ eBPF/iptables 策略L8c​进程隔离✅ PID Namespace SeccompL8d​资源限制✅ Cgroup 配额六、延伸阅读Linux Namespaces in Go使用 Go 语言操作命名空间的完整指南Cgroup v2 文档Linux 内核资源控制的官方文档Seccomp-BPF 编程如何编写 seccomp 过滤器gVisor 架构Google 的用户态内核实现原理OCI Runtime Spec开放容器标准的运行时规范七、下一讲预告第9讲供应链安全与依赖管理——从源头杜绝后门前八讲聚焦于运行时防御但攻击可能在更早的阶段就已植入——依赖库、基础镜像、CI/CD 流水线。如果供应链被污染所有运行时防御都无法阻止后门执行。下一讲实现依赖完整性校验、SBOM 生成、镜像签名和 CI/CD 安全流水线。开发之余的小工具推荐处理 Base64、JSON 格式化、JWT 解析、Crontab 计算、PDF 合并压缩这些碎片需求我常用一个纯前端本地工具箱zz365.top。所有计算在浏览器完成文件不上服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。