Agent 安全实践（Agent Security Practices）

# 基于 pydantic==2.5.0 验证（截至 2026-03）

from typing import Set, Dict, List, Any
from enum import Enum
from dataclasses import dataclass, field
from collections import deque
from datetime import datetime, timedelta
import json


# ========== 1. 权限检查机制 ==========

class Permission(str, Enum):
    """Agent 权限类别——白名单式定义"""
    READ_USER = "read:user"
    WRITE_USER = "write:user"
    DELETE_USER = "delete:user"
    READ_ORDER = "read:order"
    WRITE_ORDER = "write:order"
    DELETE_ORDER = "delete:order"

# 操作到权限的映射
OPERATION_PERMISSION_MAP: Dict[str, Permission] = {
    "read_user": Permission.READ_USER,
    "update_user": Permission.WRITE_USER,
    "delete_user": Permission.DELETE_USER,
    "read_order": Permission.READ_ORDER,
    "update_order": Permission.WRITE_ORDER,
    "delete_order": Permission.DELETE_ORDER,
}

def check_permission(agent_permissions: Set[Permission], operation: str) -> bool:
    """白名单式权限检查：只有明确授予的权限才允许执行"""
    required = OPERATION_PERMISSION_MAP.get(operation)
    if required is None:
        return False  # 未知操作，默认拒绝
    return required in agent_permissions


# ========== 2. 速率限制 + 熔断器 ==========

class RateLimiter:
    """滑动窗口速率限制器"""

    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window = timedelta(seconds=window_seconds)
        self.timestamps: deque = deque()

    def is_allowed(self) -> bool:
        now = datetime.now()
        # 清除窗口外的记录
        while self.timestamps and self.timestamps[0] < now - self.window:
            self.timestamps.popleft()
        if len(self.timestamps) < self.max_requests:
            self.timestamps.append(now)
            return True
        return False


class CircuitBreaker:
    """熔断器：CLOSED → OPEN → HALF_OPEN 三态切换"""

    def __init__(self, failure_threshold: int = 5, timeout_seconds: int = 60):
        self.failure_threshold = failure_threshold
        self.timeout = timedelta(seconds=timeout_seconds)
        self.failure_count = 0
        self.last_failure_time = None
        self.state = "CLOSED"

    def record_success(self):
        self.failure_count = 0
        self.state = "CLOSED"

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"

    def is_available(self) -> bool:
        if self.state == "CLOSED":
            return True
        if self.state == "OPEN" and self.last_failure_time:
            if datetime.now() - self.last_failure_time > self.timeout:
                self.state = "HALF_OPEN"
                return True  # 允许一次试探
            return False
        return True  # HALF_OPEN 放行一次


# ========== 3. 安全执行器（整合三层防护） ==========

@dataclass
class SecureAgentExecutor:
    """集成权限检查、速率限制、熔断器和审计日志的安全执行器"""
    agent_id: str
    permissions: Set[Permission]
    rate_limiter: RateLimiter = field(default_factory=lambda: RateLimiter(10, 60))
    circuit_breaker: CircuitBreaker = field(default_factory=CircuitBreaker)
    audit_log: List[Dict] = field(default_factory=list)

    def execute(self, operation: str, params: Dict[str, Any]) -> Dict:
        # 第一关：权限检查
        if not check_permission(self.permissions, operation):
            self._log(operation, "DENIED", "权限不足")
            return {"status": "DENIED", "reason": "权限不足"}

        # 第二关：速率限制
        if not self.rate_limiter.is_allowed():
            self._log(operation, "RATE_LIMITED", "请求频率超限")
            return {"status": "RATE_LIMITED", "reason": "请求频率超限"}

        # 第三关：熔断器
        if not self.circuit_breaker.is_available():
            self._log(operation, "CIRCUIT_OPEN", "下游故障，熔断器已打开")
            return {"status": "CIRCUIT_OPEN", "reason": "下游故障，熔断器已打开"}

        # 通过三关，执行操作
        self.circuit_breaker.record_success()
        self._log(operation, "SUCCESS", params)
        return {"status": "SUCCESS", "operation": operation, "params": params}

    def _log(self, operation: str, status: str, detail: Any):
        self.audit_log.append({
            "time": datetime.now().isoformat(),
            "agent": self.agent_id,
            "operation": operation,
            "status": status,
            "detail": detail,
        })


# ========== 使用示例 ==========

executor = SecureAgentExecutor(
    agent_id="customer_service_agent",
    permissions={Permission.READ_USER, Permission.READ_ORDER, Permission.WRITE_ORDER},
)

# 有权限的操作 → SUCCESS
print(executor.execute("read_order", {"order_id": "12345"}))
# {'status': 'SUCCESS', 'operation': 'read_order', 'params': {'order_id': '12345'}}

# 无权限的操作 → DENIED
print(executor.execute("delete_user", {"user_id": "67890"}))
# {'status': 'DENIED', 'reason': '权限不足'}

# 审计日志
print(json.dumps(executor.audit_log, indent=2, ensure_ascii=False, default=str))