AI Agent安全治理框架：自主系统的信任边界与管控策略

# Agent权限策略定义 - 使用Cedar策略语言
# 适用于 AWS Verified Permissions / Cedar Agent Framework

permit(
    principal == Agent::"data-analyst",
    action == Action::"query_database",
    resource in Database::"analytics-db"
) when {
    resource.tags.contains("read-only") &&
    context.query_type == "SELECT" &&
    context.row_limit <= 10000 &&
    !context.query.contains("DROP") &&
    !context.query.contains("DELETE")
};

# Agent沙箱 Kubernetes Pod Security Policy
apiVersion: v1
kind: Pod
metadata:
  name: agent-sandbox
  annotations:
    container.apparmor.security.beta.kubernetes.io/agent: unconfined
spec:
  runtimeClassName: gvisor  # 使用gVisor运行时
  containers:
  - name: agent-executor
    image: agent-runtime:v2.1.0
    securityContext:
      readOnlyRootFilesystem: true
      runAsNonRoot: true
      runAsUser: 65534
      allowPrivilegeEscalation: false
      capabilities:
        drop: ["ALL"]
    resources:
      limits:
        cpu: "2"
        memory: "4Gi"
        ephemeral-storage: "1Gi"
      requests:
        cpu: "500m"
        memory: "1Gi"
    env:
    - name: MAX_EXECUTION_TIME
      value: "30"  # 30秒执行超时
    - name: NETWORK_POLICY
      value: "restricted"  # 限制网络访问

import hashlib
import json
from datetime import datetime

class AgentAuditLogger:
    """不可篡改的Agent审计日志"""

    def __init__(self, storage_backend):
        self.backend = storage_backend
        self.chain = []

    def log_action(self, agent_id: str, action: str,
                   tool: str, params: dict, result: dict):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "agent_id": agent_id,
            "action": action,
            "tool": tool,
            "params_hash": hashlib.sha256(
                json.dumps(params, sort_keys=True).encode()
            ).hexdigest(),
            "result_summary": str(result)[:200],
            "previous_hash": self.chain[-1]["hash"] if self.chain else "genesis"
        }
        # 哈希链确保不可篡改
        entry["hash"] = hashlib.sha256(
            json.dumps(entry, sort_keys=True).encode()
        ).hexdigest()
        self.chain.append(entry)
        self.backend.store(entry)
        return entry["hash"]

from enum import Enum
from dataclasses import dataclass

class RiskLevel(Enum):
    LOW = "low"         # 自动执行
    MEDIUM = "medium"   # 记录+通知
    HIGH = "high"       # 暂停等待审批
    CRITICAL = "critical"  # 阻止执行

@dataclass
class ToolCall:
    name: str
    params: dict
    risk_level: RiskLevel

RISK_RULES = {
    "delete_record": RiskLevel.HIGH,
    "send_email": RiskLevel.MEDIUM,
    "execute_code": RiskLevel.HIGH,
    "read_database": RiskLevel.LOW,
    "modify_config": RiskLevel.CRITICAL,
    "transfer_funds": RiskLevel.CRITICAL,
}

async def execute_with_approval(tool_call: ToolCall, agent_id: str):
    risk = RISK_RULES.get(tool_call.name, RiskLevel.MEDIUM)

    if risk == RiskLevel.LOW:
        return await execute_tool(tool_call)

    elif risk == RiskLevel.MEDIUM:
        result = await execute_tool(tool_call)
        await notify_admin(agent_id, tool_call, result)
        return result

    elif risk == RiskLevel.HIGH:
        approval = await request_approval(
            agent_id=agent_id,
            tool_call=tool_call,
            timeout_seconds=1800,  # 30分钟超时
            approvers=["admin@company.com"]
        )
        if approval.granted:
            return await execute_tool(tool_call)
        raise PermissionDenied(f"操作 {tool_call.name} 被拒绝")

    elif risk == RiskLevel.CRITICAL:
        raise PermissionDenied(
            f"操作 {tool_call.name} 为CRITICAL级别，需手动执行"
        )