AI Agent 零信任安全架构：从身份认证到权限管控的生产实践

import jwt
import secrets
from datetime import datetime, timedelta
from enum import Flag, auto

class AgentPermission(Flag):
    """Agent 权限标记"""
    READ_DATA = auto()
    WRITE_DATA = auto()
    EXECUTE_CODE = auto()     # 危险：代码执行
    ACCESS_NETWORK = auto()   # 危险：网络访问
    FILE_SYSTEM = auto()      # 危险：文件系统操作
    SYSTEM_CALL = auto()      # 极度危险：系统调用

def issue_agent_token(
    agent_id: str,
    session_id: str,
    permissions: AgentPermission,
    max_duration_minutes: int = 30,
    resource_constraints: dict | None = None,
) -> str:
    """
    为 Agent 会话签发短期 JWT 凭证
    
    关键安全设计：
    1. 短期有效（默认 30 分钟）
    2. 权限显式声明（不继承）
    3. 资源约束绑定（限制可访问的资源范围）
    4. 包含审计追踪信息
    """
    now = datetime.utcnow()
    
    payload = {
        "iss": "agent-platform.example.com",
        "sub": agent_id,
        "sid": session_id,
        "iat": now,
        "exp": now + timedelta(minutes=max_duration_minutes),
        "jti": secrets.token_hex(16),  # 唯一 token ID，用于撤销
        "permissions": [p.name for p in AgentPermission if p in permissions],
        "constraints": {
            "max_api_calls": 1000,           # API 调用次数上限
            "max_tokens": 100_000,           # Token 消耗上限
            "allowed_resources": resource_constraints or ["*"],
            "network_allowlist": ["api.example.com"],  # 网络白名单
            "no_system_calls": AgentPermission.SYSTEM_CALL not in permissions,
        },
        "audit": {
            "created_by": "user@example.com",  # 谁创建了这个 Agent 会话
            "purpose": "code-review-task",       # 用途说明
        }
    }
    
    return jwt.encode(payload, signing_key, algorithm="ES256")

# 使用示例：为代码审查 Agent 签发最小权限 token
review_token = issue_agent_token(
    agent_id="code-reviewer-v2",
    session_id=secrets.token_hex(8),
    permissions=AgentPermission.READ_DATA,  # 只给读权限
    max_duration_minutes=15,
    resource_constraints=["repo:example/app/src/**"],  # 只能读源码
)

{
  "@context": [
    "https://www.w3.org/ns/credentials/v2",
    "https://w3id.org/security/suites/ed25519-2020/v1"
  ],
  "id": "urn:uuid:3f2a7b8c-9d1e-4f5a-b6c7-d8e9f0a1b2c3",
  "type": ["VerifiableCredential", "AgentCapabilityCredential"],
  "issuer": "did:web:agent-platform.example.com",
  "issuanceDate": "2026-06-13T05:00:00Z",
  "expirationDate": "2026-06-13T06:00:00Z",
  "credentialSubject": {
    "id": "did:key:z6Mkf5rGM2atrSes2BxExample",
    "type": "AIAgent",
    "capabilities": [
      {
        "type": "CodeExecution",
        "level": "sandboxed",
        "maxDuration": "30m",
        "allowedLanguages": ["python", "javascript"]
      },
      {
        "type": "APICall",
        "scope": "read-only",
        "rateLimit": "100/min"
      }
    ],
    "trustScore": 0.95,
    "auditLog": "https://audit.example.com/agent/z6Mkf5rGM2atrSes2Bx"
  },
  "proof": {
    "type": "Ed25519Signature2020",
    "created": "2026-06-13T05:00:00Z",
    "verificationMethod": "did:web:agent-platform.example.com#key-1",
    "proofPurpose": "assertionMethod",
    "proofValue": "z58DAdFfa9Skq...base58..."
  }
}

from dataclasses import dataclass
from typing import Any
import json

@dataclass
class AgentContext:
    """Agent 请求上下文"""
    agent_id: str
    session_id: str
    permissions: list[str]
    trust_score: float           # 0.0 - 1.0
    session_age_seconds: int
    api_calls_this_session: int
    tokens_consumed: int
    recent_actions: list[dict]   # 最近 N 次操作

@dataclass
class ResourceContext:
    """资源上下文"""
    resource_type: str           # "api" | "file" | "code_exec" | "network"
    resource_id: str
    sensitivity: str             # "public" | "internal" | "confidential" | "restricted"
    owner: str

class AgentABACPolicyEngine:
    """Agent 属性基访问控制引擎"""
    
    def evaluate(self, agent: AgentContext, resource: ResourceContext, action: str) -> bool:
        """
        评估 Agent 是否有权执行操作
        返回 True = 允许, False = 拒绝
        """
        # 规则 1: 基础权限检查
        required_perm = f"{resource.resource_type}:{action}"
        if required_perm not in agent.permissions:
            return False
        
        # 规则 2: 敏感度 + 信任分数
        sensitivity_thresholds = {
            "public": 0.0,
            "internal": 0.5,
            "confidential": 0.8,
            "restricted": 0.95,
        }
        if agent.trust_score < sensitivity_thresholds.get(resource.sensitivity, 1.0):
            return False
        
        # 规则 3: 会话时长限制（防止 Agent 失控长时间运行）
        max_session_seconds = 3600  # 1 小时
        if agent.session_age_seconds > max_session_seconds:
            return False
        
        # 规则 4: API 调用频率限制
        if agent.api_calls_this_session > 5000:
            return False
        
        # 规则 5: 异常行为检测（组合权限风险）
        if self._detect_anomalous_pattern(agent):
            return False
        
        # 规则 6: 资源配额检查
        if agent.tokens_consumed > 500_000:
            return False
        
        return True
    
    def _detect_anomalous_pattern(self, agent: AgentContext) -> bool:
        """
        检测异常行为模式
        例如：短时间内大量文件访问 + 网络请求 = 可能的数据外泄
        """
        recent = agent.recent_actions[-20:]  # 最近 20 次操作
        file_reads = sum(1 for a in recent if a.get("type") == "file_read")
        network_calls = sum(1 for a in recent if a.get("type") == "network")
        
        # 如果最近 20 次操作中 15+ 是文件读取 且 5+ 是网络调用 → 可疑
        if file_reads >= 15 and network_calls >= 5:
            return True
        
        return False

apiVersion: v1
kind: Pod
metadata:
  name: agent-sandbox
  labels:
    app: ai-agent
    agent-id: "code-reviewer-v2"
spec:
  runtimeClassName: gvisor  # 使用 gVisor 运行时
  containers:
  - name: agent-runtime
    image: agent-platform/runtime:2026.06
    resources:
      limits:
        cpu: "2"
        memory: "4Gi"
        ephemeral-storage: "10Gi"
      requests:
        cpu: "500m"
        memory: "1Gi"
    env:
    - name: AGENT_SESSION_ID
      valueFrom:
        fieldRef:
          fieldPath: metadata.uid
    - name: AGENT_MAX_DURATION
      value: "1800"  # 30 分钟硬限制
    - name: AGENT_NETWORK_POLICY
      value: "allowlist-only"
    securityContext:
      readOnlyRootFilesystem: true
      allowPrivilegeEscalation: false
      runAsNonRoot: true
      runAsUser: 65532  # nonroot
      capabilities:
        drop: ["ALL"]
    volumeMounts:
    - name: workspace
      mountPath: /workspace
    - name: agent-config
      mountPath: /etc/agent
      readOnly: true
  volumes:
  - name: workspace
    emptyDir:
      sizeLimit: "5Gi"
  - name: agent-config
    configMap:
      name: agent-runtime-config
---
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
  name: agent-network-isolation
spec:
  podSelector:
    matchLabels:
      app: ai-agent
  policyTypes:
  - Egress
  egress:
  - to:
    - ipBlock:
        cidr: 10.0.0.0/8  # 仅允许访问内部网络
    ports:
    - port: 443
      protocol: TCP
  - to:
    - namespaceSelector:
        matchLabels:
          name: allowed-apis
    ports:
    - port: 443

{
  "eventId": "evt_a1b2c3d4e5f6",
  "timestamp": "2026-06-13T05:00:00.123Z",
  "agent": {
    "id": "code-reviewer-v2",
    "sessionId": "sess_x7y8z9",
    "version": "2.1.0",
    "trustScore": 0.92
  },
  "action": {
    "type": "tool_call",
    "tool": "code_executor",
    "method": "run_python",
    "parameters": {
      "code_hash": "sha256:abc123...",
      "code_lines": 15,
      "imports": ["os", "subprocess"],
      "network_calls": false,
      "file_operations": ["read:/workspace/src/app.py"]
    }
  },
  "result": {
    "status": "success",
    "duration_ms": 234,
    "output_size_bytes": 1024,
    "exit_code": 0
  },
  "context": {
    "triggeredBy": "user@example.com",
    "task": "code-review-pr-1234",
    "step": 7,
    "totalSteps": 12
  },
  "risk": {
    "level": "medium",
    "flags": ["subprocess_import", "file_read_outside_workspace"],
    "autoBlocked": false
  }
}