云端大规模 Agent 沙箱：多租户隔离、持久化、弹性调度与合规治理

Agent 沙箱架构全景图：

┌─────────────────────────────────────────────────────────────┐
│                    Agent Sandbox Platform                   │
├──────────┬──────────┬──────────┬────────────────────────────┤
│ 多租户隔离│ 状态持久化│ 弹性调度 │       合规治理             │
│          │          │          │                            │
│ • 计算   │ • 会话   │ • K8s    │ • 审计日志                 │
│   隔离   │   状态   │   CRD    │ • PII 脱敏                 │
│ • 数据   │ • 文件   │ • Warm   │ • Crypto-                  │
│   隔离   │   系统   │   Pool   │   Shredding                │
│ • 身份   │ • 向量   │ • HPA    │ • 数据主权                 │
│   权限   │   索引   │ • 预测   │ • OWASP/MITRE              │
│ • 速率   │ • 长期   │   伸缩   │                            │
│   限制   │   记忆   │          │                            │
└──────────┴──────────┴──────────┴────────────────────────────┘

传统 Web 应用：
  用户请求 → 应用服务器（预编译代码）→ 数据库
  └─ 代码是开发者写的，经过审查，可信

AI Agent 系统：
  用户请求 → LLM 推理 → 生成代码 → 沙箱执行 → 调用工具
  └─ 代码是 LLM 实时生成的，未经审查，不可信

Firecracker 架构：

┌─────────────────────────────────────────────┐
│  Host OS (Linux)                            │
│  ┌───────────┐  ┌───────────┐  ┌─────────┐ │
│  │ MicroVM 1 │  │ MicroVM 2 │  │ MicroVM N│ │
│  │ ┌───────┐ │  │ ┌───────┐ │  │ ┌──────┐ │ │
│  │ │Kernel │ │  │ │Kernel │ │  │ │Kernel│ │ │
│  │ │(独立) │ │  │ │(独立) │ │  │ │(独立)│ │ │
│  │ └───────┘ │  │ └───────┘ │  │ └──────┘ │ │
│  │ ┌───────┐ │  │ ┌───────┐ │  │ ┌──────┐ │ │
│  │ │Agent  │ │  │ │Agent  │ │  │ │Agent │ │ │
│  │ │代码   │ │  │ │代码   │ │  │ │代码  │ │ │
│  │ └───────┘ │  │ └───────┘ │  │ └──────┘ │ │
│  └───────────┘  └───────────┘  └─────────┘ │
│  ┌───────────────────────────────────────┐  │
│  │       KVM (Hardware Virtualization)   │  │
│  └───────────────────────────────────────┘  │
└─────────────────────────────────────────────┘

每个 MicroVM：
  • 独立内核 — 无共享内核逃逸风险
  • ~5MB 内存开销 — 单机可跑数千个
  • ~150ms 冷启动 — 接近容器速度
  • 最小设备模型 — 攻击面极小

from dataclasses import dataclass, field
from typing import Optional

@dataclass
class SandboxSecurityConfig:
    """Agent 沙箱安全配置 — Day 1 基线"""

    # 文件系统
    readonly_rootfs: bool = True
    writable_tmpfs: bool = True
    tmpfs_size_mb: int = 512

    # 网络
    network_mode: str = "restricted"  # restricted | isolated | full
    egress_allowlist: list[str] = field(default_factory=list)
    dns_policy: str = "None"  # 自定义 DNS，防止 DNS 隧道

    # 资源限制
    cpu_limit: str = "2"
    memory_limit: str = "4Gi"
    pids_limit: int = 256  # 防止 fork bomb

    # 系统调用
    seccomp_profile: str = "RuntimeDefault"
    apparmor_profile: Optional[str] = "runtime/default"

    # 权限
    run_as_non_root: bool = True
    allow_privilege_escalation: bool = False
    drop_capabilities: list[str] = field(
        default_factory=lambda: ["ALL"]
    )
    add_capabilities: list[str] = field(default_factory=list)

    # 生命周期
    max_lifetime_seconds: int = 86400  # 24h 最大存活
    idle_timeout_seconds: int = 1800   # 30min 空闲超时

    def validate(self) -> bool:
        """验证安全配置是否符合基线要求"""
        if self.allow_privilege_escalation:
            raise ValueError("特权提升必须禁止")
        if "ALL" not in self.drop_capabilities:
            raise ValueError("必须 drop ALL capabilities")
        if not self.readonly_rootfs:
            raise ValueError("根文件系统必须只读")
        return True

# generated by hugo AI

Workspace 数据流：

  用户上传文件
       │
       ▼
┌──────────────────┐
│  Ingestion       │  文件立即标记 Workspace ID
│  • Tag: ws_abc   │  元数据写入: { workspace_id, tenant_id, hash }
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│  RAG Pipeline    │  Embeddings 存入 tenant namespace
│  • Chunk & Embed │  Vector DB: namespace="tenant_acme"
│  • Index         │  Metadata: { tenant_id: "acme", ws_id: "ws_abc" }
└────────┬─────────┘
         │
         ▼
┌──────────────────┐
│  Agent Retrieval │  搜索强制限定 namespace
│  • Search tool   │  query.filter = { tenant_id: "acme" }
│  • Namespace     │  返回结果二次验证 tenant_id
│    filter        │
└──────────────────┘

身份传播链：

  用户 (User Identity)
    │ OIDC Token
    ▼
┌──────────────────┐
│  API Gateway     │  ← 验证用户身份，注入 tenant_id
│  • 鉴权          │
│  • Tenant 解析   │
└────────┬─────────┘
         │ Authorization Envelope:
         │ { user_id, tenant_id, scope, expires_at }
         ▼
┌──────────────────┐
│  Agent Runtime   │  ← 获取短期工作负载凭证
│  • Workload ID   │     (STS AssumeRole / OIDC Federation)
│  • Token 交换    │
└────────┬─────────┘
         │ Short-lived Credential (TTL < 15min)
         ▼
┌──────────────────┐
│  Tool Execution  │  ← 凭证限定工具权限范围
│  • Scoped MCP    │
│  • Egress ACL    │
└──────────────────┘

import os
from dataclasses import dataclass

@dataclass
class ScopedToolContext:
    """限定范围的工具执行上下文"""

    tenant_id: str
    workspace_id: str
    user_id: str
    allowed_paths: list[str]
    allowed_domains: list[str]
    max_output_bytes: int = 1_000_000

    def validate_file_access(self, path: str) -> bool:
        """验证文件访问是否在允许范围内"""
        resolved = os.path.realpath(path)
        return any(
            resolved.startswith(allowed)
            for allowed in self.allowed_paths
        )

    def validate_network_access(self, url: str) -> bool:
        """验证网络访问是否在白名单内"""
        from urllib.parse import urlparse
        host = urlparse(url).hostname
        return any(
            host.endswith(domain)
            for domain in self.allowed_domains
        )

# generated by hugo AI

限流层级：

┌──────────────────────────────────────────┐
│  全局限流                                │
│  • 总并发 Sandbox 数                     │
│  • 总 GPU/CPU 配额                       │
├──────────────────────────────────────────┤
│  租户限流                                │
│  • 每租户最大 Sandbox 数                 │
│  • 每租户 LLM API 调用 QPS               │
│  • 每租户网络带宽                        │
├──────────────────────────────────────────┤
│  会话限流                                │
│  • 单会话最大执行时间                    │
│  • 单会话最大内存                        │
│  • 单会话最大文件 I/O                    │
└──────────────────────────────────────────┘

from dataclasses import dataclass
from enum import Enum

class SandboxState(Enum):
    RUNNING = "running"
    PAUSED = "paused"
    TERMINATED = "terminated"

@dataclass
class SandboxLifecycle:
    """Agent 沙箱生命周期管理 — Pause/Resume 模式"""

    sandbox_id: str
    state: SandboxState = SandboxState.RUNNING
    timeout_ms: int = 30 * 60 * 1000  # 30 分钟空闲超时
    auto_resume: bool = True

    async def create(self, template: str) -> str:
        """创建沙箱实例"""
        # sandbox = await e2b.Sandbox.create(
        #     template=template,
        #     timeout_ms=self.timeout_ms,
        #     lifecycle={
        #         "onTimeout": "pause",
        #         "autoResume": self.auto_resume,
        #     }
        # )
        # return sandbox.sandbox_id
        ...

    async def pause(self) -> None:
        """暂停沙箱 — 保存文件系统 + 内存状态"""
        # await sandbox.pause()
        self.state = SandboxState.PAUSED

    async def resume(self) -> None:
        """恢复沙箱 — 从暂停时的状态继续"""
        # sandbox = await e2b.Sandbox.connect(self.sandbox_id)
        self.state = SandboxState.RUNNING

    async def connect_or_create(self) -> str:
        """连接已有沙箱或创建新的 — 自动恢复暂停的沙箱"""
        # 如果沙箱存在（包括暂停状态），自动连接/恢复
        # 否则创建新的
        ...

# generated by hugo AI

from dataclasses import dataclass

@dataclass
class VectorIsolationConfig:
    """向量数据库多租户隔离配置"""

    tenant_id: str
    isolation_mode: str = "namespace"  # namespace | shard | instance

    def get_collection_name(self, workspace_id: str) -> str:
        """根据隔离模式生成集合名"""
        if self.isolation_mode == "namespace":
            # 共享实例，逻辑隔离 — 适合大多数租户
            return f"tenant_{self.tenant_id}_vectors"
        elif self.isolation_mode == "shard":
            # 分片隔离，物理边界 — 适合 B2B 客户
            return f"shard_{self.tenant_id}_{workspace_id}"
        else:
            # 独立实例，最强隔离 — 适合金融/医疗等强合规场景
            return f"instance_{self.tenant_id}"

    def build_query_filter(self) -> dict:
        """构建强制租户过滤条件 — 所有查询必须携带"""
        return {
            "tenant_id": self.tenant_id,
            # 防御性编程：即使 namespace 隔离，也加 metadata filter
        }

# generated by hugo AI

apiVersion: sandbox.x-k8s.io/v1alpha1
kind: Sandbox
metadata:
  name: agent-workspace-user-1234
spec:
  # 安全运行时 — 使用 gVisor 做内核隔离
  runtimeClassName: gvisor
  # 持久化工作空间
  workspace:
    storageClassName: ssd
    size: 10Gi
    mountPath: /workspace
  # 空闲时自动缩到零
  lifecycle:
    idleTimeout: 30m
    scaleToZero: true
  # 网络策略 — 只允许出站白名单
  networkPolicy:
    egress:
      - to:
          - cidr: 10.0.0.0/8  # 内网 API
      - to:
          - cidr: 0.0.0.0/0
        ports:
          - port: 443  # 只允许 HTTPS

apiVersion: sandbox.x-k8s.io/v1alpha1
kind: SandboxWarmPool
spec:
  templateRef:
    name: python-agent-template
  minReady: 10
  maxReady: 50
  scalingPolicy: predictive  # 基于历史并发预测预热

弹性伸缩三层架构：

  用户请求
     │
     ▼
┌────────────┐     ┌──────────────┐     ┌──────────────┐
│  API       │────▶│  Sandbox     │────▶│  Agent       │
│  Gateway   │     │  Orchestrator│     │  Runtime     │
│            │     │              │     │  (MicroVM)   │
│ • 路由     │     │ • Claim/     │     │              │
│ • 限流     │     │   Release    │     │ • LLM 推理   │
│ • 鉴权     │     │ • Warm Pool  │     │ • 代码执行   │
└────────────┘     │ • 生命周期   │     │ • 工具调用   │
                   └──────┬───────┘     └──────────────┘
                          │
                   ┌──────▼───────┐
                   │  Kubernetes  │
                   │  HPA +       │
                   │  Cluster     │
                   │  Autoscaler  │
                   └──────────────┘

Crypto-Shredding 架构：

┌─────────────────────────────────────────────┐
│  KMS (Key Management Service)               │
│                                             │
│  KEK_tenant_A  KEK_tenant_B  KEK_tenant_C   │
│  (Key Encrypting Key per tenant)            │
└──────────────┬──────────────────────────────┘
               │ 加密
               ▼
┌─────────────────────────────────────────────┐
│  DEK (Data Encrypting Keys)                 │
│  ┌──────────┐ ┌──────────┐ ┌────────────┐  │
│  │Session DB│ │Vector DB │ │Object Store│  │
│  │(加密存储) │ │(加密存储) │ │(加密存储)  │  │
│  └──────────┘ └──────────┘ └────────────┘  │
└─────────────────────────────────────────────┘

删除租户 A 的数据：
  1. 销毁 KEK_tenant_A
  2. 所有用该 KEK 加密的数据立即不可读
  3. 无需物理删除分散的数据块
  4. 满足 GDPR "Right to be Forgotten"

from dataclasses import dataclass

@dataclass
class AgentAuditEntry:
    """Agent 审计日志条目 — SOC 2 合规要求"""

    timestamp: str
    tenant_id: str
    user_id: str
    sandbox_id: str

    # 事件类型
    event_type: str  # tool_call | file_access | memory_retrieval | network_request

    # 事件详情
    action: str  # 具体操作，如 "read_file", "search_vectors"
    resource: str  # 访问的资源路径或 ID
    result: str  # success | denied | error

    # 安全上下文
    workload_identity: str  # Agent 的工作负载身份
    source_ip: str
    tenant_context_verified: bool  # 租户上下文是否通过验证

    # 可选：敏感信息脱敏后的摘要
    sanitized_summary: str = ""

    def to_json(self) -> dict:
        """序列化为 JSON — 写入审计日志系统"""
        return {
            k: v for k, v in self.__dict__.items()
            if v is not None
        }

# generated by hugo AI

┌──────────────┬──────────────┬──────────────┬──────────────┐
│              │  自建 (K8s)  │  E2B         │  Blaxel      │
├──────────────┼──────────────┼──────────────┼──────────────┤
│ 隔离技术     │ 自选         │ Firecracker  │ MicroVM      │
│ 持久化       │ 自行设计     │ Pause/Resume │ 永续沙箱     │
│ 恢复速度     │ 取决于实现   │ ~1s          │ <25ms        │
│ 空闲成本     │ 需自行优化   │ 暂停零成本   │ 待机零成本   │
│ 最大会话时长 │ 无限制       │ 24h (Pro)    │ 无限         │
│ 多租户       │ 自行实现     │ 支持         │ 原生支持     │
│ 合规支持     │ 自行实现     │ 基础         │ ZDR 原生     │
│ 运维复杂度   │ 高           │ 低           │ 低           │
│ 适合场景     │ 大规模/定制  │ 快速启动     │ 生产级 Agent │
└──────────────┴──────────────┴──────────────┴──────────────┘