企业级 AI Agent 系统开发实战
轻量级沙箱技术深度研究报告

# Dockerfile - AI Agent 代码执行环境
FROM python:3.11-slim

# 创建非 root 用户
RUN useradd -m -u 1000 sandbox

# 安装必要依赖
RUN apt-get update && apt-get install -y --no-install-recommends \
    python3-pip \
    python3-dev \
    && rm -rf /var/lib/apt/lists/*

# 设置工作目录
WORKDIR /sandbox

# 复制代码（限制可执行文件）
COPY --chown=sandbox:sandbox . /sandbox/

# 切换到非 root 用户
USER sandbox

# 设置资源限制
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD python3 -c "print('OK')"

CMD ["python3", "main.py"]

import docker
import uuid
import asyncio
from typing import Optional, Tuple
from dataclasses import dataclass

@dataclass
class SandboxConfig:
    """沙箱配置"""
    image: str = "python:3.11-slim"
    memory_limit: str = "512m"
    cpu_quota: int = 50000  # 50% CPU
    pids_limit: int = 50
    network_disabled: bool = True
    read_only: bool = True
    timeout: int = 60

class DockerSandbox:
    """Docker 代码执行沙箱"""
    
    def __init__(self, config: SandboxConfig = None):
        self.config = config or SandboxConfig()
        self.client = docker.from_client()
        self.container: Optional[docker.models.containers.Container] = None
    
    async def create(self) -> bool:
        """创建沙箱容器"""
        try:
            self.container = self.client.containers.run(
                self.config.image,
                "tail -f /dev/null",
                detach=True,
                remove=True,
                name=f"sandbox-{uuid.uuid4().hex[:8]}",
                # 资源限制
                mem_limit=self.config.memory_limit,
                cpu_quota=self.config.cpu_quota,
                pids_limit=self.config.pids_limit,
                # 安全配置
                network_disabled=self.config.network_disabled,
                read_only=self.config.read_only,
                tmpfs={"/tmp": "rw,noexec,nosuid,size=100m"},
                # 安全选项
                security_opt=["no-new-privileges:true"],
                cap_drop=["ALL"],
            )
            return True
        except Exception as e:
            print(f"Failed to create sandbox: {e}")
            return False
    
    async def execute(self, code: str) -> Tuple[int, str, str]:
        """执行代码"""
        if not self.container:
            raise RuntimeError("Sandbox not created")
        
        # 写入代码文件
        await self._write_file("/tmp/code.py", code)
        
        # 执行代码（带超时）
        try:
            result = self.container.exec_run(
                "python3 /tmp/code.py",
                demux=True,
                timeout=self.config.timeout
            )
            
            stdout = result.output[0].decode() if result.output[0] else ""
            stderr = result.output[1].decode() if result.output[1] else ""
            
            return result.exit_code, stdout, stderr
        except Exception as e:
            return -1, "", str(e)
    
    async def _write_file(self, path: str, content: str):
        """写入文件到容器"""
        import tarfile
        import io
        
        tar_stream = io.BytesIO()
        with tarfile.open(fileobj=tar_stream, mode="w") as tar:
            tarinfo = tarfile.TarInfo(name=path.lstrip("/"))
            tarinfo.size = len(content.encode())
            tar.addfile(tarinfo, io.BytesIO(content.encode()))
        
        tar_stream.seek(0)
        self.container.put_archive("/", tar_stream)
    
    async def destroy(self):
        """销毁沙箱"""
        if self.container:
            self.container.stop()
            self.container = None

# firecracker-config.json
{
  "boot-source": {
    "kernel_image_path": "/path/to/vmlinux",
    "boot_args": "console=ttyS0 reboot=k panic=1 pci=off"
  },
  "drives": [
    {
      "drive_id": "rootfs",
      "path_on_host": "/path/to/rootfs.img",
      "is_root_device": true,
      "partuuid": "66d9f393-833d-4c1f-a540-3a8a9e8f5b5c",
      "is_read_only": true
    }
  ],
  "machine-config": {
    "vcpu_count": 1,
    "mem_size_mib": 512,
    "ht_enabled": false
  },
  "network-interfaces": [
    {
      "iface_id": "eth0",
      "host_dev_name": "tap0",
      "allow_mmds_requests": false
    }
  ],
  "logger": {
    "log_path": "/var/log/firecracker.log",
    "level": "Info"
  }
}

import fcntl
import socket
import json
import subprocess
from pathlib import Path
from typing import Optional

class FirecrackerSandbox:
    """Firecracker MicroVM 沙箱"""
    
    def __init__(self, vm_id: str, config: dict):
        self.vm_id = vm_id
        self.config = config
        self.socket_path = Path(f"/tmp/firecracker-{vm_id}.socket")
        self.process: Optional[subprocess.Popen] = None
    
    def start(self) -> bool:
        """启动 MicroVM"""
        try:
            # 启动 Firecracker 进程
            self.process = subprocess.Popen(
                [
                    "firecracker",
                    "--api-sock", str(self.socket_path),
                    "--config-file", self.config["config_path"]
                ],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE
            )
            
            # 等待 socket 就绪
            for _ in range(50):
                if self.socket_path.exists():
                    break
                time.sleep(0.1)
            
            return True
        except Exception as e:
            print(f"Failed to start Firecracker: {e}")
            return False
    
    def _api_call(self, method: str, path: str, body: dict = None) -> dict:
        """调用 Firecracker API"""
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.connect(str(self.socket_path))
        
        request = f"{method} {path} HTTP/1.1\r\n"
        if body:
            request += f"Content-Length: {len(json.dumps(body))}\r\n"
            request += "\r\n" + json.dumps(body)
        else:
            request += "\r\n"
        
        sock.sendall(request.encode())
        response = sock.recv(4096).decode()
        sock.close()
        
        return json.loads(response.split("\r\n\r\n", 1)[1])
    
    def execute(self, code: str, timeout: int = 60) -> dict:
        """执行代码（通过 guest API）"""
        # 通过 MMDS 或 virtio-serial 传递代码到 VM 内执行
        # 实际实现需要 VM 内的 agent 配合
        pass
    
    def stop(self):
        """停止 MicroVM"""
        if self.process:
            self.process.terminate()
            self.process.wait()
            self.socket_path.unlink(missing_ok=True)

# 1. 编写 Go 应用
// main.go
package main

import (
    "fmt"
    "net/http"
)

func handler(w http.ResponseWriter, r *http.Request) {
    fmt.Fprintf(w, "Hello from Unikernel!")
}

func main() {
    http.HandleFunc("/", handler)
    http.ListenAndServe(":8080", nil)
}

# 2. 编译为 Unikernel 镜像
$ ops build -t nanos main.go

# 3. 运行镜像
$ ops run -p 8080 main

# 4. 镜像大小对比
传统 Docker 镜像：~50MB
Nanos Unikernel:  ~8MB

from enum import Enum
from typing import Optional, Type
from abc import ABC, abstractmethod

class SecurityLevel(Enum):
    """安全等级"""
    LOW = "low"      # 可信代码，Docker
    MEDIUM = "medium" # 半可信，Docker+ 限制
    HIGH = "high"    # 不可信，MicroVM
    CRITICAL = "critical" # 高危，Unikernel

class SandboxProvider(ABC):
    """沙箱提供者接口"""
    
    @abstractmethod
    async def create(self, config: dict) -> "Sandbox":
        pass
    
    @abstractmethod
    async def destroy(self, sandbox: "Sandbox"):
        pass

class SandboxScheduler:
    """沙箱调度器"""
    
    def __init__(self):
        self.providers: dict[SecurityLevel, SandboxProvider] = {
            SecurityLevel.LOW: DockerProvider(),
            SecurityLevel.MEDIUM: DockerProvider(security_enhanced=True),
            SecurityLevel.HIGH: FirecrackerProvider(),
            SecurityLevel.CRITICAL: UnikernelProvider()
        }
    
    async def schedule(self, code: str, context: dict) -> Sandbox:
        """根据代码和上下文选择合适的沙箱"""
        # 1. 评估安全等级
        security_level = await self._assess_security(code, context)
        
        # 2. 选择沙箱提供者
        provider = self.providers[security_level]
        
        # 3. 创建沙箱
        config = self._build_config(security_level, context)
        sandbox = await provider.create(config)
        
        return sandbox
    
    async def _assess_security(self, code: str, context: dict) -> SecurityLevel:
        """评估代码安全等级"""
        # 基于代码特征、来源、用户信任度等评估
        if context.get("trusted_source"):
            return SecurityLevel.LOW
        elif self._contains_dangerous_patterns(code):
            return SecurityLevel.HIGH
        else:
            return SecurityLevel.MEDIUM

{
  "defaultAction": "SCMP_ACT_ERRNO",
  "architectures": ["SCMP_ARCH_X86_64"],
  "syscalls": [
    {
      "names": [
        "accept", "access", "alarm", "bind",
        "brk", "close", "connect", "dup",
        "execve", "exit", "fstat", "getpid",
        "getuid", "listen", "mmap", "open",
        "read", "recvfrom", "sendto", "socket",
        "write"
      ],
      "action": "SCMP_ACT_ALLOW"
    }
  ]
}