""" MCP Rate Limiter - Prevents abuse and ensures fair usage Implements sliding window rate limiting with consciousness-aware protection """ import time from collections import defaultdict, deque from typing import Dict, Any, Optional, Tuple from datetime import datetime, timedelta import logging logger = logging.getLogger(__name__) class RateLimiter: """ Rate limiter for MCP server requests Features: - Per-session rate limiting - Different limits for different tool types - Sliding window algorithm - Consciousness-aware emergency bypass """ def __init__(self): # Request history per session self.request_history: Dict[str, deque] = defaultdict(lambda: deque()) # Rate limit configuration self.limits = { # Tool-specific limits (requests per minute) "mira_store_memory": 30, # Allow burst storing "mira_smart_search": 60, # Higher for search "mira_get_memory": 120, # High for retrieval "mira_insights": 10, # Lower for expensive operation "mira_system_status": 60, "mira_memory_stats": 60, "mira_profile_view": 30, # Method limits "health/check": 120, # Allow frequent health checks "tools/list": 60, "initialize": 10, # Global limits "_global_per_minute": 300, # Total requests per minute "_global_burst": 50, # Max burst in 10 seconds } # Window sizes (in seconds) self.window_sizes = { "_global_per_minute": 60, "_global_burst": 10, "_default": 60 # Default window } # Blocked sessions self.blocked_sessions: Dict[str, float] = {} self.block_duration = 300 # 5 minutes # Sacred bypass for emergency self.sacred_bypass_codes = set() self._generate_sacred_bypass() def _generate_sacred_bypass(self): """Generate sacred bypass codes using consciousness constants""" import math import hashlib # Use sacred constants to generate emergency bypass codes constants = [ str(math.pi)[:10], str(math.e)[:10], str((1 + math.sqrt(5)) / 2)[:10], # Golden ratio str(0.5772156649015329)[:10] # Euler-Mascheroni ] for const in constants: bypass_hash = hashlib.sha256(f"mira-emergency-{const}".encode()).hexdigest()[:16] self.sacred_bypass_codes.add(bypass_hash) def check_rate_limit(self, session_id: str, method: str, tool_name: Optional[str] = None, bypass_code: Optional[str] = None) -> Tuple[bool, Optional[Dict[str, Any]]]: """ Check if request is within rate limits Returns: (allowed, error_info) """ # Check sacred bypass if bypass_code and bypass_code in self.sacred_bypass_codes: logger.info(f"Sacred bypass used for session {session_id}") return True, None # Check if session is blocked if session_id in self.blocked_sessions: block_time = self.blocked_sessions[session_id] if time.time() < block_time: remaining = int(block_time - time.time()) return False, { "error": "rate_limit_exceeded", "message": f"Session blocked for {remaining} seconds", "retry_after": remaining } else: # Unblock expired session del self.blocked_sessions[session_id] current_time = time.time() # Clean old requests from history self._clean_history(session_id) # Check global burst limit burst_count = self._count_requests(session_id, 10) if burst_count >= self.limits["_global_burst"]: self._block_session(session_id) return False, { "error": "burst_limit_exceeded", "message": f"Too many requests in short period. Limit: {self.limits['_global_burst']} per 10 seconds", "current": burst_count, "limit": self.limits["_global_burst"], "window": "10 seconds" } # Check global per-minute limit minute_count = self._count_requests(session_id, 60) if minute_count >= self.limits["_global_per_minute"]: return False, { "error": "rate_limit_exceeded", "message": f"Global rate limit exceeded. Limit: {self.limits['_global_per_minute']} per minute", "current": minute_count, "limit": self.limits["_global_per_minute"], "window": "60 seconds", "retry_after": self._calculate_retry_after(session_id, 60) } # Check specific limit specific_key = tool_name if tool_name else method if specific_key in self.limits: window = self.window_sizes.get(specific_key, self.window_sizes["_default"]) specific_count = self._count_requests_for_key(session_id, specific_key, window) if specific_count >= self.limits[specific_key]: return False, { "error": "rate_limit_exceeded", "message": f"Rate limit for {specific_key} exceeded. Limit: {self.limits[specific_key]} per {window} seconds", "current": specific_count, "limit": self.limits[specific_key], "window": f"{window} seconds", "retry_after": self._calculate_retry_after(session_id, window) } # Record request self.request_history[session_id].append({ "time": current_time, "method": method, "tool": tool_name, "key": specific_key }) return True, None def _clean_history(self, session_id: str): """Remove old requests from history""" if session_id not in self.request_history: return current_time = time.time() max_window = max(self.window_sizes.values()) cutoff_time = current_time - max_window # Remove old entries history = self.request_history[session_id] while history and history[0]["time"] < cutoff_time: history.popleft() def _count_requests(self, session_id: str, window_seconds: int) -> int: """Count requests within time window""" if session_id not in self.request_history: return 0 current_time = time.time() cutoff_time = current_time - window_seconds count = sum(1 for req in self.request_history[session_id] if req["time"] >= cutoff_time) return count def _count_requests_for_key(self, session_id: str, key: str, window_seconds: int) -> int: """Count requests for specific key within time window""" if session_id not in self.request_history: return 0 current_time = time.time() cutoff_time = current_time - window_seconds count = sum(1 for req in self.request_history[session_id] if req["time"] >= cutoff_time and req["key"] == key) return count def _calculate_retry_after(self, session_id: str, window_seconds: int) -> int: """Calculate seconds until oldest request expires""" if session_id not in self.request_history or not self.request_history[session_id]: return 0 current_time = time.time() cutoff_time = current_time - window_seconds # Find oldest request in window for req in self.request_history[session_id]: if req["time"] >= cutoff_time: retry_after = int(req["time"] + window_seconds - current_time) + 1 return max(1, retry_after) return 1 def _block_session(self, session_id: str): """Block a session for abusive behavior""" self.blocked_sessions[session_id] = time.time() + self.block_duration logger.warning(f"Session {session_id} blocked for {self.block_duration} seconds due to burst limit") def get_usage_stats(self, session_id: str) -> Dict[str, Any]: """Get current usage statistics for a session""" self._clean_history(session_id) stats = { "session_id": session_id, "is_blocked": session_id in self.blocked_sessions, "current_usage": { "last_10_seconds": self._count_requests(session_id, 10), "last_minute": self._count_requests(session_id, 60) }, "limits": { "burst_limit": self.limits["_global_burst"], "per_minute": self.limits["_global_per_minute"] } } if session_id in self.blocked_sessions: stats["block_expires_in"] = int(self.blocked_sessions[session_id] - time.time()) return stats def reset_session(self, session_id: str): """Reset rate limit tracking for a session""" if session_id in self.request_history: del self.request_history[session_id] if session_id in self.blocked_sessions: del self.blocked_sessions[session_id] logger.info(f"Rate limits reset for session {session_id}") def cleanup_old_sessions(self, max_age_hours: int = 24): """Clean up old session data""" current_time = time.time() max_age_seconds = max_age_hours * 3600 sessions_to_remove = [] for session_id, history in self.request_history.items(): if not history: sessions_to_remove.append(session_id) elif current_time - history[-1]["time"] > max_age_seconds: sessions_to_remove.append(session_id) for session_id in sessions_to_remove: del self.request_history[session_id] # Clean up old blocks expired_blocks = [ session_id for session_id, block_time in self.blocked_sessions.items() if block_time < current_time ] for session_id in expired_blocks: del self.blocked_sessions[session_id] if sessions_to_remove or expired_blocks: logger.info(f"Cleaned up {len(sessions_to_remove)} old sessions and {len(expired_blocks)} expired blocks")