AI Agent 企业治理最佳实践：Agent 365 身份安全治理深度解读

# 传统安全模型 vs Agent 治理模型的对比
class TraditionalSecurityModel:
    """传统人类用户安全模型"""
    def authenticate(self, user_id: str, credentials: dict) -> bool:
        # 一次性认证，长期有效
        return verify_credentials(user_id, credentials)
    
    def authorize(self, user_id: str, action: str) -> bool:
        # 静态 RBAC，权限变更需要管理员审批
        role = get_user_role(user_id)
        return role.permissions.contains(action)
    
    def audit(self, user_id: str, action: str, timestamp: datetime):
        # 低频操作，日志量可控
        log_entry = {
            "user": user_id,
            "action": action,
            "timestamp": timestamp.isoformat()
        }
        # 每天约 100-1000 条日志
        write_audit_log(log_entry)


class AgentGovernanceModel:
    """AI Agent 治理模型"""
    def authenticate(self, agent_id: str, context: dict) -> AgentIdentity:
        # 每次调用时验证，考虑上下文
        identity = verify_agent_identity(agent_id, context)
        # 验证 Agent 的模型版本、配置状态、信任等级
        if identity.model_version not in APPROVED_VERSIONS:
            raise AgentVersionError(f"Unapproved model version: {identity.model_version}")
        return identity
    
    def authorize(self, identity: AgentIdentity, action: str, context: dict) -> AuthorizationResult:
        # 动态权限评估，考虑 Agent 状态、任务上下文、风险等级
        risk_score = calculate_risk(identity, action, context)
        if risk_score > identity.max_risk_threshold:
            return AuthorizationResult.DENIED
        elif risk_score > identity.review_threshold:
            return AuthorizationResult.REQUIRES_HUMAN_REVIEW
        return AuthorizationResult.APPROVED
    
    def audit(self, identity: AgentIdentity, action: str, context: dict, result: dict):
        # 高频审计，每秒可能数千条
        audit_entry = {
            "agent_id": identity.id,
            "model_version": identity.model_version,
            "action": action,
            "risk_score": context.get("risk_score"),
            "decision_time_ms": context.get("decision_time"),
            "timestamp": datetime.now().isoformat()
        }
        # 需要实时流处理和异常检测
        stream_to_audit_pipeline(audit_entry)

# Agent 365 治理策略配置示例
agent-governance:
  identity:
    # Agent 身份认证策略
    authentication:
      method: "mutual-tls"  # 双向 TLS 认证
      certificate-rotation: "24h"  # 证书每 24 小时轮换
      model-attestation: true  # 要求模型完整性证明
    
    # Agent 身份属性
    identity-attributes:
      - agent-id          # 唯一标识符
      - model-version     # 模型版本
      - training-data-hash # 训练数据指纹
      - owner-team        # 负责团队
      - risk-tier         # 风险等级（low/medium/high/critical）
  
  permissions:
    # 动态权限管理
    rbac-extension:
      base-role: "agent-basic"      # 基础角色
      context-roles:                # 上下文角色（动态附加）
        - name: "data-reader"
          condition: "task.requires_data_access"
        - name: "email-sender"
          condition: "task.requires_email"
          max-frequency: "10/min"   # 频率限制
    
    # 权限升级流程
    privilege-escalation:
      auto-escalation: false        # 禁止自动提权
      review-required: true         # 提权必须人工审批
      max-duration: "1h"            # 临时权限最长 1 小时
  
  behavior:
    # 行为约束策略
    constraints:
      max-api-calls-per-minute: 100
      max-concurrent-tasks: 5
      forbidden-actions:
        - "database.drop"
        - "system.shutdown"
        - "email.send-to-external"
      allowed-domains:
        - "internal.company.com"
        - "api.trusted-partner.com"

# Agent 身份认证的完整实现流程
import hashlib
import hmac
from dataclasses import dataclass
from enum import Enum
from typing import Optional
from datetime import datetime, timedelta


class IdentityVerificationStatus(Enum):
    VERIFIED = "verified"
    MODEL_MISMATCH = "model_mismatch"
    CONFIG_DRIFT = "config_drift"
    CERT_EXPIRED = "certificate_expired"
    ATTESTATION_FAILED = "attestation_failed"


@dataclass
class AgentIdentity:
    agent_id: str                    # Agent 唯一标识符
    model_name: str                  # 模型名称
    model_version: str               # 模型版本号
    model_hash: str                  # 模型权重文件 SHA-256
    config_hash: str                 # 配置哈希值
    certificate_serial: str          # 证书序列号
    risk_tier: str                   # 风险等级
    issued_at: datetime              # 身份签发时间
    expires_at: datetime             # 身份过期时间


class AgentIdentityVerifier:
    """Agent 365 身份认证器"""
    
    def __init__(self, identity_registry, tee_client, cert_authority):
        self.registry = identity_registry
        self.tee = tee_client
        self.ca = cert_authority
    
    def verify_identity(self, agent_runtime: dict) -> tuple[AgentIdentity, IdentityVerificationStatus]:
        """
        多步骤 Agent 身份验证流程
        
        返回: (AgentIdentity, 验证状态)
        """
        # 步骤 1: 证书验证
        cert_status = self._verify_certificate(agent_runtime)
        if cert_status != IdentityVerificationStatus.VERIFIED:
            return None, cert_status
        
        # 步骤 2: 模型完整性证明
        model_status = self._verify_model_attestation(agent_runtime)
        if model_status != IdentityVerificationStatus.VERIFIED:
            return None, model_status
        
        # 步骤 3: 配置一致性校验
        config_status = self._verify_config_consistency(agent_runtime)
        if config_status != IdentityVerificationStatus.VERIFIED:
            return None, config_status
        
        # 所有验证通过，构建 Agent 身份对象
        approved_profile = self.registry.get_profile(agent_runtime["agent_id"])
        
        identity = AgentIdentity(
            agent_id=agent_runtime["agent_id"],
            model_name=approved_profile.model_name,
            model_version=approved_profile.model_version,
            model_hash=approved_profile.model_hash,
            config_hash=approved_profile.config_hash,
            certificate_serial=agent_runtime["cert_serial"],
            risk_tier=approved_profile.risk_tier,
            issued_at=datetime.now(),
            expires_at=datetime.now() + timedelta(hours=24)
        )
        
        return identity, IdentityVerificationStatus.VERIFIED
    
    def _verify_model_attestation(self, runtime: dict) -> IdentityVerificationStatus:
        """验证模型完整性证明"""
        attestation = runtime.get("model_attestation")
        if not attestation:
            return IdentityVerificationStatus.ATTESTATION_FAILED
        
        # 通过 TEE 验证模型哈希
        expected_hash = self.registry.get_model_hash(runtime["agent_id"])
        is_valid = self.tee.verify_model_hash(
            attestation["proof"],
            attestation["measurement"],
            expected_hash
        )
        
        if not is_valid:
            return IdentityVerificationStatus.MODEL_MISMATCH
        return IdentityVerificationStatus.VERIFIED
    
    def _verify_config_consistency(self, runtime: dict) -> IdentityVerificationStatus:
        """验证配置一致性"""
        current_config_hash = hashlib.sha256(
            runtime["config_json"].encode()
        ).hexdigest()
        
        expected_hash = self.registry.get_config_hash(runtime["agent_id"])
        
        if current_config_hash != expected_hash:
            return IdentityVerificationStatus.CONFIG_DRIFT
        return IdentityVerificationStatus.VERIFIED

// Agent 身份上下文传递（TypeScript 实现）
// 在微服务架构中，Agent 身份需要在服务间传递

interface AgentIdentityToken {
  sub: string;              // Agent ID
  model: string;            // 模型名称
  model_version: string;    // 模型版本
  model_hash: string;       // 模型哈希
  config_hash: string;      // 配置哈希
  risk_tier: 'low' | 'medium' | 'high' | 'critical';
  scope: string[];          // 允许的数据范围
  iat: number;              // 签发时间戳
  exp: number;              // 过期时间戳
  jti: string;              // 唯一令牌 ID
}

interface AgentRequestContext {
  identity: AgentIdentityToken;
  task_id: string;          // 当前任务 ID
  correlation_id: string;   // 关联 ID（用于审计追踪）
  parent_agent?: string;    // 如果是子 Agent，记录父 Agent
  inherited_permissions?: string[];  // 继承的权限
}

// 中间件：在每个请求中验证 Agent 身份
async function agentIdentityMiddleware(
  req: Request,
  next: NextFunction
): Promise<Response> {
  const token = req.headers.get('X-Agent-Identity');
  if (!token) {
    return new Response('Missing Agent identity', { status: 401 });
  }
  
  // 验证 JWT 签名和有效期
  const identity = await verifyAgentToken(token);
  if (!identity) {
    return new Response('Invalid Agent identity', { status: 401 });
  }
  
  // 检查模型版本是否在白名单中
  if (!APPROVED_MODEL_VERSIONS.includes(identity.model_version)) {
    return new Response('Unapproved model version', { status: 403 });
  }
  
  // 将身份注入请求上下文
  req.context = {
    ...req.context,
    agent: identity
  };
  
  return next(req);
}

# Agent 动态权限管理实现
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime, timedelta
from typing import Optional, Set
import uuid


class PermissionLevel(Enum):
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    ADMIN = "admin"


class PermissionScope:
    """权限范围定义"""
    def __init__(self, resource_type: str, resource_id: str, level: PermissionLevel):
        self.resource_type = resource_type
        self.resource_id = resource_id
        self.level = level
    
    def __hash__(self):
        return hash((self.resource_type, self.resource_id, self.level))
    
    def __eq__(self, other):
        if not isinstance(other, PermissionScope):
            return False
        return (self.resource_type == other.resource_type and
                self.resource_id == other.resource_id and
                self.level == other.level)


@dataclass
class TaskPermissionGrant:
    """任务级权限授予"""
    grant_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    agent_id: str = ""
    task_id: str = ""
    permissions: Set[PermissionScope] = field(default_factory=set)
    granted_at: datetime = field(default_factory=datetime.now)
    expires_at: Optional[datetime] = None
    approved_by: Optional[str] = None  # 审批人（应急权限需要）
    reason: Optional[str] = None       # 授予原因


class DynamicPermissionManager:
    """动态权限管理器"""
    
    def __init__(self, base_permissions: dict):
        # Agent ID -> 基础权限集
        self.base_permissions = base_permissions
        # grant_id -> TaskPermissionGrant
        self.active_grants: dict[str, TaskPermissionGrant] = {}
    
    def analyze_task_permissions(self, task_definition: dict) -> Set[PermissionScope]:
        """分析任务所需的权限"""
        required_permissions = set()
        
        # 解析任务工具链
        for tool in task_definition.get("tools", []):
            tool_permissions = self._resolve_tool_permissions(tool)
            required_permissions.update(tool_permissions)
        
        # 解析数据访问需求
        for data_source in task_definition.get("data_sources", []):
            data_permissions = self._resolve_data_permissions(data_source)
            required_permissions.update(data_permissions)
        
        # 应用最小权限原则
        return self._apply_least_privilege(required_permissions, task_definition)
    
    def grant_task_permissions(
        self, agent_id: str, task_id: str, 
        permissions: Set[PermissionScope],
        duration_minutes: int = 30
    ) -> TaskPermissionGrant:
        """授予任务级权限"""
        grant = TaskPermissionGrant(
            agent_id=agent_id,
            task_id=task_id,
            permissions=permissions,
            expires_at=datetime.now() + timedelta(minutes=duration_minutes)
        )
        self.active_grants[grant.grant_id] = grant
        return grant
    
    def check_permission(self, agent_id: str, requested: PermissionScope) -> bool:
        """检查 Agent 是否有指定权限"""
        # 检查基础权限
        base = self.base_permissions.get(agent_id, set())
        if requested in base:
            return True
        
        # 检查活动授权
        now = datetime.now()
        for grant in self.active_grants.values():
            if (grant.agent_id == agent_id and
                grant.expires_at and grant.expires_at > now and
                requested in grant.permissions):
                return True
        
        return False
    
    def revoke_task_permissions(self, grant_id: str):
        """回收任务权限"""
        if grant_id in self.active_grants:
            del self.active_grants[grant_id]
    
    def cleanup_expired_grants(self):
        """清理过期的授权"""
        now = datetime.now()
        expired = [
            gid for gid, grant in self.active_grants.items()
            if grant.expires_at and grant.expires_at < now
        ]
        for gid in expired:
            del self.active_grants[gid]

# Agent 行为异常检测实现
from collections import defaultdict, deque
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import math


@dataclass
class BehaviorMetric:
    """行为指标"""
    metric_name: str          # 指标名称
    current_value: float      # 当前值
    baseline_mean: float      # 基线均值
    baseline_std: float       # 基线标准差
    window_size: int          # 窗口大小
    history: deque = field(default_factory=lambda: deque(maxlen=1000))
    
    @property
    def z_score(self) -> float:
        """计算 Z 分数（偏离基线的标准差倍数）"""
        if self.baseline_std == 0:
            return 0 if self.current_value == self.baseline_mean else float('inf')
        return abs(self.current_value - self.baseline_mean) / self.baseline_std
    
    def is_anomalous(self, threshold: float = 3.0) -> bool:
        """判断是否异常（Z 分数超过阈值）"""
        return self.z_score > threshold


class AgentBehaviorMonitor:
    """Agent 行为监控器"""
    
    def __init__(self, anomaly_threshold: float = 3.0):
        self.threshold = anomaly_threshold
        # agent_id -> metric_name -> BehaviorMetric
        self.metrics: Dict[str, Dict[str, BehaviorMetric]] = defaultdict(dict)
        # agent_id -> 行为基线统计
        self.baselines: Dict[str, Dict[str, dict]] = defaultdict(dict)
    
    def record_action(self, agent_id: str, action: str, context: dict):
        """记录 Agent 行为"""
        timestamp = datetime.now()
        
        # 更新各类指标
        self._update_call_frequency(agent_id, action, timestamp)
        self._update_data_access_pattern(agent_id, action, context)
        self._update_resource_usage(agent_id, action, context)
        
        # 实时异常检测
        anomalies = self._detect_anomalies(agent_id)
        if anomalies:
            self._handle_anomalies(agent_id, anomalies)
    
    def _update_call_frequency(self, agent_id: str, action: str, timestamp: datetime):
        """更新调用频率指标"""
        metric_key = f"call_frequency:{action}"
        
        if metric_key not in self.metrics[agent_id]:
            # 初始化指标（使用前 14 天的数据作为基线）
            baseline = self.baselines[agent_id].get(metric_key, {})
            self.metrics[agent_id][metric_key] = BehaviorMetric(
                metric_name=metric_key,
                current_value=1.0,
                baseline_mean=baseline.get("mean", 0),
                baseline_std=baseline.get("std", 1.0)
            )
        
        metric = self.metrics[agent_id][metric_key]
        metric.current_value += 1
        metric.history.append(timestamp)
    
    def _detect_anomalies(self, agent_id: str) -> List[tuple]:
        """检测异常行为"""
        anomalies = []
        
        for metric_name, metric in self.metrics[agent_id].items():
            if metric.is_anomalous(self.threshold):
                anomalies.append((
                    metric_name,
                    metric.current_value,
                    metric.baseline_mean,
                    metric.z_score
                ))
        
        return anomalies
    
    def _handle_anomalies(self, agent_id: str, anomalies: List[tuple]):
        """处理检测到的异常"""
        for metric_name, current, baseline, z_score in anomalies:
            risk_level = self._classify_risk(z_score)
            
            if risk_level == "critical":
                # 紧急异常：立即终止 Agent
                self._trigger_emergency_response(agent_id, metric_name, z_score)
            elif risk_level == "high":
                # 高风险：暂停 Agent 并告警
                self._pause_agent_and_alert(agent_id, metric_name, z_score)
            elif risk_level == "medium":
                # 中风险：发送通知
                self._send_notification(agent_id, metric_name, z_score)
            # 低风险：仅记录日志
    
    def _classify_risk(self, z_score: float) -> str:
        """根据 Z 分数分类风险等级"""
        if z_score > 6.0:
            return "critical"
        elif z_score > 4.0:
            return "high"
        elif z_score > 3.0:
            return "medium"
        return "low"