MCP 的权限分级设计

# mcp_permission_center.py
from typing import List, Dict, Optional
from datetime import datetime

class Attribute:
    def __init__(self, key: str, operator: str, value: str):
        self.key = key
        self.operator = operator
        self.value = value
    
    def to_dict(self) -> Dict:
        return {
            "key": self.key,
            "operator": self.operator,
            "value": self.value
        }

class Permission:
    def __init__(self, perm_id: str, perm_name: str, resource: str, action: str, conditions: Optional[List[Attribute]] = None):
        self.perm_id = perm_id
        self.perm_name = perm_name
        self.resource = resource
        self.action = action
        self.conditions = conditions or []
    
    def to_dict(self) -> Dict:
        return {
            "perm_id": self.perm_id,
            "perm_name": self.perm_name,
            "resource": self.resource,
            "action": self.action,
            "conditions": [cond.to_dict() for cond in self.conditions]
        }
    
    def check_conditions(self, context: Dict) -> bool:
        """检查权限条件是否满足"""
        for cond in self.conditions:
            if cond.key not in context:
                return False
            
            actual_value = context[cond.key]
            expected_value = cond.value
            
            if cond.operator == "eq":
                if actual_value != expected_value:
                    return False
            elif cond.operator == "ne":
                if actual_value == expected_value:
                    return False
            elif cond.operator == "contains":
                if expected_value not in actual_value:
                    return False
            elif cond.operator == "startsWith":
                if not actual_value.startswith(expected_value):
                    return False
            elif cond.operator == "endsWith":
                if not actual_value.endswith(expected_value):
                    return False
            else:
                return False
        
        return True

class Role:
    def __init__(self, role_id: str, role_name: str, permissions: Optional[List[Permission]] = None, parent_role: Optional['Role'] = None):
        self.role_id = role_id
        self.role_name = role_name
        self.permissions = permissions or []
        self.parent_role = parent_role
    
    def to_dict(self) -> Dict:
        return {
            "role_id": self.role_id,
            "role_name": self.role_name,
            "permissions": [perm.to_dict() for perm in self.permissions],
            "parent_role": self.parent_role.role_id if self.parent_role else None
        }
    
    def inherit_from(self, parent: 'Role'):
        """继承父角色的权限"""
        self.parent_role = parent
    
    def override_permission(self, permission: Permission):
        """覆盖或添加权限"""
        for i, perm in enumerate(self.permissions):
            if perm.perm_id == permission.perm_id:
                self.permissions[i] = permission
                return
        self.permissions.append(permission)
    
    def get_effective_permissions(self) -> List[Permission]:
        """获取角色的有效权限（包括继承的权限）"""
        effective_perms = []
        
        # 先获取父角色的权限
        if self.parent_role:
            effective_perms.extend(self.parent_role.get_effective_permissions())
        
        # 然后添加或覆盖自己的权限
        perm_ids = {}
        for perm in effective_perms:
            perm_ids[perm.perm_id] = perm
        
        for perm in self.permissions:
            perm_ids[perm.perm_id] = perm
        
        return list(perm_ids.values())

class User:
    def __init__(self, user_id: str, user_name: str, roles: Optional[List[Role]] = None):
        self.user_id = user_id
        self.user_name = user_name
        self.roles = roles or []
    
    def to_dict(self) -> Dict:
        return {
            "user_id": self.user_id,
            "user_name": self.user_name,
            "roles": [role.role_id for role in self.roles]
        }
    
    def add_role(self, role: Role):
        """添加角色"""
        if role not in self.roles:
            self.roles.append(role)
    
    def remove_role(self, role: Role):
        """移除角色"""
        if role in self.roles:
            self.roles.remove(role)
    
    def get_effective_permissions(self) -> List[Permission]:
        """获取用户的有效权限"""
        effective_perms = []
        perm_ids = {}
        
        for role in self.roles:
            for perm in role.get_effective_permissions():
                perm_ids[perm.perm_id] = perm
        
        return list(perm_ids.values())
    
    def check_permission(self, resource: str, action: str, context: Dict) -> bool:
        """检查用户是否拥有特定资源的特定操作权限"""
        for role in self.roles:
            for perm in role.get_effective_permissions():
                if perm.resource == resource and perm.action == action:
                    if perm.check_conditions(context):
                        return True
        return False

class PermissionCenter:
    def __init__(self):
        self.roles = {}
        self.users = {}
        self.permissions = {}
    
    def create_role(self, role_id: str, role_name: str, parent_role_id: Optional[str] = None) -> Role:
        """创建角色"""
        parent_role = self.roles.get(parent_role_id)
        role = Role(role_id, role_name, parent_role=parent_role)
        self.roles[role_id] = role
        return role
    
    def get_role(self, role_id: str) -> Optional[Role]:
        """获取角色"""
        return self.roles.get(role_id)
    
    def create_permission(self, perm_id: str, perm_name: str, resource: str, action: str, conditions: Optional[List[Attribute]] = None) -> Permission:
        """创建权限"""
        permission = Permission(perm_id, perm_name, resource, action, conditions)
        self.permissions[perm_id] = permission
        return permission
    
    def get_permission(self, perm_id: str) -> Optional[Permission]:
        """获取权限"""
        return self.permissions.get(perm_id)
    
    def create_user(self, user_id: str, user_name: str) -> User:
        """创建用户"""
        user = User(user_id, user_name)
        self.users[user_id] = user
        return user
    
    def get_user(self, user_id: str) -> Optional[User]:
        """获取用户"""
        return self.users.get(user_id)
    
    def assign_role(self, user_id: str, role_id: str):
        """分配角色给用户"""
        user = self.get_user(user_id)
        role = self.get_role(role_id)
        if user and role:
            user.add_role(role)
    
    def assign_permission(self, role_id: str, perm_id: str):
        """分配权限给角色"""
        role = self.get_role(role_id)
        permission = self.get_permission(perm_id)
        if role and permission:
            role.override_permission(permission)
    
    def check_permission(self, user_id: str, resource: str, action: str, context: Dict) -> bool:
        """检查用户权限"""
        user = self.get_user(user_id)
        if user:
            return user.check_permission(resource, action, context)
        return False

# mcp_dynamic_permission.py
from typing import List, Dict, Any
from mcp_permission_center import PermissionCenter

class MCPDynamicPermissionNegotiator:
    def __init__(self, permission_center: PermissionCenter):
        self.permission_center = permission_center
        
    def negotiate_permissions(self, client_id: str, available_tools: List[Dict], context: Dict) -> List[Dict]:
        """
        动态权限协商：根据用户权限过滤可用工具
        
        Args:
            client_id: 客户端ID
            available_tools: 可用工具列表
            context: 协商上下文
            
        Returns:
            过滤后的工具列表
        """
        negotiated_tools = []
        
        for tool in available_tools:
            tool_id = tool.get("tool_id")
            if not tool_id:
                continue
            
            # 检查工具级权限
            if self.permission_center.check_permission(
                user_id=client_id,
                resource=f"tool:{tool_id}",
                action="execute",
                context=context
            ):
                
                # 检查工具参数级权限
                filtered_params = []
                for param in tool.get("parameters", []):
                    param_name = param.get("name")
                    if self.permission_center.check_permission(
                        user_id=client_id,
                        resource=f"tool:{tool_id}:param:{param_name}",
                        action="access",
                        context=context
                    ):
                        filtered_params.append(param)
                
                # 更新工具的参数列表
                tool_with_perm = tool.copy()
                tool_with_perm["parameters"] = filtered_params
                negotiated_tools.append(tool_with_perm)
        
        return negotiated_tools
    
    def adjust_permissions(self, client_id: str, tool_id: str, action: str, context: Dict, adjustment: Dict) -> bool:
        """
        动态调整权限
        
        Args:
            client_id: 客户端ID
            tool_id: 工具ID
            action: 操作类型
            context: 上下文
            adjustment: 权限调整参数
            
        Returns:
            权限调整是否成功
        """
        # 根据调整参数动态生成临时权限
        temp_perm_id = f"temp_{client_id}_{tool_id}_{action}_{datetime.now().timestamp()}"
        temp_perm = self.permission_center.create_permission(
            perm_id=temp_perm_id,
            perm_name=f"临时权限: {tool_id}.{action}",
            resource=f"tool:{tool_id}",
            action=action,
            conditions=[Attribute(k, "eq", v) for k, v in context.items()]
        )
        
        # 分配临时权限给用户
        temp_role_id = f"temp_role_{client_id}"
        temp_role = self.permission_center.create_role(
            role_id=temp_role_id,
            role_name=f"临时角色: {client_id}"
        )
        
        self.permission_center.assign_permission(temp_role_id, temp_perm_id)
        self.permission_center.assign_role(client_id, temp_role_id)
        
        return True

{
  "audit_id": "audit_1234567890",
  "timestamp": "2026-01-01T12:00:00Z",
  "event_type": "tool_call",
  "actor_type": "user",
  "actor_id": "client_001",
  "resource_type": "tool",
  "resource_id": "file_reader",
  "action": "execute",
  "context": {
    "server_id": "server_001",
    "tool_id": "file_reader",
    "parameters": {
      "file_path": "/data/test.txt",
      "mode": "read"
    },
    "ip_address": "192.168.1.100"
  },
  "result": "success",
  "status_code": 200,
  "message": "工具调用成功",
  "duration_ms": 100
}

# mcp_audit_system.py
from typing import Dict, List
from datetime import datetime
import json
import logging

class AuditEvent:
    def __init__(self, event_type: str, actor_type: str, actor_id: str, 
                 resource_type: str, resource_id: str, action: str, 
                 context: Dict, result: str, status_code: int, 
                 message: str, duration_ms: int = 0):
        self.audit_id = f"audit_{datetime.now().timestamp()}_{id(self)}"
        self.timestamp = datetime.now().isoformat()
        self.event_type = event_type
        self.actor_type = actor_type
        self.actor_id = actor_id
        self.resource_type = resource_type
        self.resource_id = resource_id
        self.action = action
        self.context = context
        self.result = result
        self.status_code = status_code
        self.message = message
        self.duration_ms = duration_ms
    
    def to_dict(self) -> Dict:
        return {
            "audit_id": self.audit_id,
            "timestamp": self.timestamp,
            "event_type": self.event_type,
            "actor_type": self.actor_type,
            "actor_id": self.actor_id,
            "resource_type": self.resource_type,
            "resource_id": self.resource_id,
            "action": self.action,
            "context": self.context,
            "result": self.result,
            "status_code": self.status_code,
            "message": self.message,
            "duration_ms": self.duration_ms
        }
    
    def to_json(self) -> str:
        return json.dumps(self.to_dict(), ensure_ascii=False)

class AuditCollector:
    def __init__(self):
        self.events = []
    
    def collect(self, event: AuditEvent):
        """收集审计事件"""
        self.events.append(event)
    
    def flush(self) -> List[AuditEvent]:
        """刷新审计事件，返回当前收集的事件并清空列表"""
        events = self.events.copy()
        self.events.clear()
        return events

class AuditStorage:
    def __init__(self, storage_path: str):
        self.storage_path = storage_path
        # 初始化存储目录
        os.makedirs(storage_path, exist_ok=True)
    
    def store(self, event: AuditEvent):
        """存储审计事件"""
        # 按日期分文件存储
        date_str = datetime.now().strftime("%Y-%m-%d")
        file_path = os.path.join(self.storage_path, f"audit_{date_str}.log")
        
        with open(file_path, "a", encoding="utf-8") as f:
            f.write(event.to_json() + "\n")
    
    def bulk_store(self, events: List[AuditEvent]):
        """批量存储审计事件"""
        for event in events:
            self.store(event)

class AuditAnalyzer:
    def __init__(self, storage: AuditStorage):
        self.storage = storage
    
    def query_events(self, query: Dict) -> List[AuditEvent]:
        """查询审计事件"""
        # 实现审计事件查询逻辑
        pass
    
    def analyze_patterns(self, time_range: Dict) -> Dict:
        """分析审计事件模式"""
        # 实现审计事件模式分析
        pass
    
    def detect_anomalies(self, baseline: Dict) -> List[AuditEvent]:
        """检测异常审计事件"""
        # 实现异常检测逻辑
        pass

class AuditSystem:
    def __init__(self, storage_path: str):
        self.collector = AuditCollector()
        self.storage = AuditStorage(storage_path)
        self.analyzer = AuditAnalyzer(self.storage)
        self.logger = logging.getLogger("mcp_audit")
    
    def log_event(self, event_type: str, actor_type: str, actor_id: str, 
                  resource_type: str, resource_id: str, action: str, 
                  context: Dict, result: str, status_code: int, 
                  message: str, duration_ms: int = 0):
        """记录审计事件"""
        event = AuditEvent(
            event_type=event_type,
            actor_type=actor_type,
            actor_id=actor_id,
            resource_type=resource_type,
            resource_id=resource_id,
            action=action,
            context=context,
            result=result,
            status_code=status_code,
            message=message,
            duration_ms=duration_ms
        )
        
        # 收集事件
        self.collector.collect(event)
        
        # 立即存储到文件
        self.storage.store(event)
        
        # 记录到日志
        self.logger.info(f"Audit: {event.to_json()}")
    
    def log_tool_call(self, client_id: str, server_id: str, tool_id: str, 
                      parameters: Dict, result: str, status_code: int, 
                      message: str, duration_ms: int = 0):
        """记录工具调用审计事件"""
        self.log_event(
            event_type="tool_call",
            actor_type="client",
            actor_id=client_id,
            resource_type="tool",
            resource_id=tool_id,
            action="execute",
            context={
                "server_id": server_id,
                "parameters": parameters
            },
            result=result,
            status_code=status_code,
            message=message,
            duration_ms=duration_ms
        )
    
    def log_permission_change(self, admin_id: str, change_type: str, 
                              target_type: str, target_id: str, 
                              permission: Dict, result: str, 
                              status_code: int, message: str):
        """记录权限变更审计事件"""
        self.log_event(
            event_type="permission_change",
            actor_type="admin",
            actor_id=admin_id,
            resource_type=target_type,
            resource_id=target_id,
            action=change_type,
            context={
                "permission": permission
            },
            result=result,
            status_code=status_code,
            message=message
        )
    
    def flush(self):
        """刷新审计事件"""
        events = self.collector.flush()
        self.storage.bulk_store(events)
    
    def query(self, query: Dict) -> List[AuditEvent]:
        """查询审计事件"""
        return self.analyzer.query_events(query)
    
    def analyze(self, time_range: Dict) -> Dict:
        """分析审计事件"""
        return self.analyzer.analyze_patterns(time_range)
    
    def detect_anomalies(self, baseline: Dict) -> List[AuditEvent]:
        """检测异常审计事件"""
        return self.analyzer.detect_anomalies(baseline)

# 安装依赖
pip install mcp-permission-center

# 配置权限系统
cp config.example.yaml config.yaml
# 编辑配置文件
vim config.yaml

# 启动权限管理中心
mcp-permission-center --config config.yaml