AI Agent 安全运行时：原理与架构设计

from enum import Enum
from dataclasses import dataclass
from typing import Dict, List, Set, Optional
from datetime import datetime, timedelta
import json

class PermissionLevel(Enum):
    """权限等级"""
    READ = "read"           # 只读访问
    WRITE = "write"         # 读写访问
    EXECUTE = "execute"     # 执行权限
    ADMIN = "admin"         # 管理权限

class ResourceScope(Enum):
    """资源作用域"""
    FILESYSTEM = "filesystem"
    NETWORK = "network"
    DATABASE = "database"
    MEMORY = "memory"
    EXTERNAL_API = "external_api"

@dataclass
class PermissionGrant:
    """权限授予"""
    tool_name: str
    resource: ResourceScope
    level: PermissionLevel
    scope_pattern: str       # 资源匹配模式
    expires_at: datetime     # 过期时间
    max_invocations: int = 0 # 最大调用次数（0表示不限）

class PermissionEngine:
    """基于策略的权限引擎"""
    
    def __init__(self):
        self.grants: Dict[str, List[PermissionGrant]] = {}
        self.audit_log: List[Dict] = []
    
    def register_tool_policy(self, tool_name: str, 
                              grants: List[PermissionGrant]):
        """注册工具权限策略"""
        self.grants[tool_name] = grants
    
    def check_permission(self, tool_name: str, 
                          resource: ResourceScope,
                          operation: str,
                          target: str) -> bool:
        """检查操作是否被允许"""
        grants = self.grants.get(tool_name, [])
        now = datetime.now()
        
        for grant in grants:
            if grant.resource != resource:
                continue
            if grant.expires_at < now:
                continue  # 权限已过期
            if grant.level.value not in self._operation_to_level(operation):
                continue
            if not self._match_pattern(grant.scope_pattern, target):
                continue
            
            # 记录审计日志
            self.audit_log.append({
                "tool": tool_name,
                "resource": resource.value,
                "operation": operation,
                "target": target,
                "granted": True,
                "timestamp": now.isoformat()
            })
            return True
        
        # 权限不足，记录拒绝日志
        self.audit_log.append({
            "tool": tool_name,
            "resource": resource.value,
            "operation": operation,
            "target": target,
            "granted": False,
            "timestamp": now.isoformat()
        })
        return False
    
    def _operation_to_level(self, operation: str) -> List[str]:
        """将操作映射到所需权限等级"""
        mapping = {
            "read": ["read", "write", "admin"],
            "write": ["write", "admin"],
            "execute": ["execute", "admin"],
            "delete": ["admin"],
        }
        return mapping.get(operation, [])
    
    def _match_pattern(self, pattern: str, target: str) -> bool:
        """简单的模式匹配"""
        if pattern == "*":
            return True
        if pattern.endswith("*"):
            return target.startswith(pattern[:-1])
        return pattern == target

import logging
import json
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
from enum import Enum

class EventSeverity(Enum):
    INFO = "info"
    WARNING = "warning"
    CRITICAL = "critical"
    ALERT = "alert"

@dataclass
class AgentEvent:
    """Agent 运行时事件"""
    timestamp: str
    agent_id: str
    event_type: str
    severity: str
    tool_name: Optional[str] = None
    action: Optional[str] = None
    result: Optional[str] = None
    context: Optional[Dict] = None
    metadata: Optional[Dict] = None

class AgentAuditor:
    """Agent 运行时审计器"""
    
    def __init__(self, output_dir: str = "/var/log/agent-audit"):
        self.logger = logging.getLogger("agent-audit")
        self.logger.setLevel(logging.INFO)
        handler = logging.FileHandler(f"{output_dir}/audit.log")
        handler.setFormatter(logging.JSONFormatter())
        self.logger.addHandler(handler)
        self.events: List[AgentEvent] = []
    
    def record(self, event: AgentEvent):
        """记录审计事件"""
        self.events.append(event)
        self.logger.info(asdict(event))
        
        # 高危事件立即告警
        if event.severity in ("critical", "alert"):
            self._trigger_alert(event)
    
    def _trigger_alert(self, event: AgentEvent):
        """触发高危告警"""
        alert = {
            "type": "agent_security_alert",
            "severity": event.severity,
            "agent_id": event.agent_id,
            "event": event.event_type,
            "tool": event.tool_name,
            "timestamp": event.timestamp,
            "details": event.context
        }
        self.logger.critical(json.dumps(alert))

class BehaviorAnalyzer:
    """Agent 行为分析引擎"""
    
    def __init__(self, auditor: AgentAuditor):
        self.auditor = auditor
        self.baseline: Dict[str, Dict] = {}
    
    def detect_anomaly(self, agent_id: str, 
                        current_behavior: Dict) -> List[str]:
        """检测异常行为"""
        anomalies = []
        baseline = self.baseline.get(agent_id, {})
        
        # 检查工具调用频率
        tool_calls = current_behavior.get("tool_calls", {})
        baseline_calls = baseline.get("tool_calls", {})
        for tool, count in tool_calls.items():
            avg = baseline_calls.get(tool, {}).get("avg_per_hour", 0)
            if avg > 0 and count > avg * 3:
                anomalies.append(
                    f"工具 {tool} 调用频率异常: {count}/h "
                    f"(基准: {avg}/h)"
                )
        
        # 检查新工具使用
        known_tools = set(baseline_calls.keys())
        current_tools = set(tool_calls.keys())
        new_tools = current_tools - known_tools
        if new_tools:
            anomalies.append(f"使用了未知工具: {new_tools}")
        
        # 记录异常事件
        if anomalies:
            self.auditor.record(AgentEvent(
                timestamp=datetime.now().isoformat(),
                agent_id=agent_id,
                event_type="behavior_anomaly",
                severity="warning" if len(anomalies) < 3 else "critical",
                action="anomaly_detection",
                result=json.dumps(anomalies),
                context=current_behavior
            ))
        
        return anomalies

import asyncio
import uuid
from typing import Dict, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta

@dataclass
class SandboxInstance:
    """沙盒实例"""
    id: str
    agent_id: str
    runtime: str  # "docker", "gvisor", "firecracker"
    status: str = "pending"
    created_at: datetime = field(default_factory=datetime.now)
    destroyed_at: Optional[datetime] = None
    resource_limits: Dict = field(default_factory=dict)
    granted_permissions: Dict = field(default_factory=dict)

class RuntimeOrchestrator:
    """安全运行时编排引擎"""
    
    def __init__(self):
        self.sandboxes: Dict[str, SandboxInstance] = {}
        self.max_concurrent = 10
    
    async def create_sandbox(self, agent_id: str,
                              security_level: str = "standard") -> str:
        """创建沙盒实例"""
        if len(self.sandboxes) >= self.max_concurrent:
            raise RuntimeError("达到最大并发沙盒数")
        
        sandbox_id = str(uuid.uuid4())[:8]
        
        # 根据安全等级选择运行时
        if security_level == "high":
            runtime = "gvisor"
        elif security_level == "critical":
            runtime = "firecracker"
        else:
            runtime = "docker"
        
        sandbox = SandboxInstance(
            id=sandbox_id,
            agent_id=agent_id,
            runtime=runtime,
            resource_limits=self._get_resource_limits(security_level),
            granted_permissions=self._get_initial_permissions(
                agent_id, security_level
            ),
        )
        
        self.sandboxes[sandbox_id] = sandbox
        await self._provision_container(sandbox)
        sandbox.status = "running"
        
        return sandbox_id
    
    async def execute_tool(self, sandbox_id: str, 
                            tool_call: Dict) -> Dict:
        """在沙盒中执行工具调用"""
        sandbox = self.sandboxes.get(sandbox_id)
        if not sandbox or sandbox.status != "running":
            return {"error": "沙盒不可用"}
        
        # 策略检查
        if not self._check_tool_policy(sandbox, tool_call):
            return {"error": "权限不足", "tool": tool_call["name"]}
        
        # 在隔离容器中执行
        result = await self._run_in_container(
            sandbox, tool_call
        )
        
        # 输出检查
        if not self._check_output_safety(result):
            return {"error": "输出包含不安全内容"}
        
        return result
    
    async def destroy_sandbox(self, sandbox_id: str):
        """销毁沙盒实例"""
        sandbox = self.sandboxes.get(sandbox_id)
        if sandbox:
            await self._cleanup_container(sandbox)
            sandbox.status = "destroyed"
            sandbox.destroyed_at = datetime.now()
    
    def _get_resource_limits(self, level: str) -> Dict:
        limits = {
            "standard": {"cpu": "1.0", "memory": "512m", "timeout": 60},
            "high": {"cpu": "2.0", "memory": "1g", "timeout": 120},
            "critical": {"cpu": "4.0", "memory": "2g", "timeout": 300},
        }
        return limits.get(level, limits["standard"])
    
    async def _provision_container(self, sandbox: SandboxInstance):
        """实际创建容器"""
        pass
    
    async def _run_in_container(self, sandbox: SandboxInstance,
                                 tool_call: Dict) -> Dict:
        """在容器中执行工具"""
        pass
    
    async def _cleanup_container(self, sandbox: SandboxInstance):
        """清理容器资源"""
        pass
    
    def _check_tool_policy(self, sandbox: SandboxInstance,
                           tool_call: Dict) -> bool:
        """检查工具调用是否符合策略"""
        return True
    
    def _check_output_safety(self, result: Dict) -> bool:
        """检查输出是否安全"""
        return True
    
    def _get_initial_permissions(self, agent_id: str,
                                  level: str) -> Dict:
        return {"filesystem": "read", "network": False}