MCP Server Security: Best Practices for Production Deployment

import jwt
import hashlib
import secrets
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
import asyncio
import redis

class MCPSecurityManager:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.jwt_secret = secrets.token_urlsafe(32)
        self.token_expiry = timedelta(hours=1)
        
    async def authenticate_client(self, 
                                client_id: str, 
                                api_key: str, 
                                client_certificate: Optional[str] = None) -> Dict[str, Any]:
        """
        Multi-factor authentication for MCP clients
        """
        # Factor 1: API Key validation
        if not await self.validate_api_key(client_id, api_key):
            raise SecurityError("Invalid API key")
        
        # Factor 2: Client certificate validation (if provided)
        if client_certificate:
            if not await self.validate_client_certificate(client_certificate):
                raise SecurityError("Invalid client certificate")
        
        # Factor 3: Rate limiting check
        if await self.is_rate_limited(client_id):
            raise SecurityError("Rate limit exceeded")
        
        # Generate secure session token
        session_token = await self.create_session_token(client_id)
        
        return {
            "session_token": session_token,
            "expires_at": datetime.utcnow() + self.token_expiry,
            "permissions": await self.get_client_permissions(client_id)
        }
    
    async def validate_api_key(self, client_id: str, api_key: str) -> bool:
        """
        Secure API key validation with timing attack protection
        """
        stored_hash = await self.redis.get(f"api_key:{client_id}")
        if not stored_hash:
            # Prevent timing attacks by computing hash anyway
            hashlib.sha256(api_key.encode()).hexdigest()
            return False
        
        # Constant-time comparison
        provided_hash = hashlib.sha256(api_key.encode()).hexdigest()
        return secrets.compare_digest(stored_hash.decode(), provided_hash)
    
    async def validate_session_token(self, token: str) -> Dict[str, Any]:
        """
        JWT token validation with additional security checks
        """
        try:
            # Decode and verify JWT
            payload = jwt.decode(token, self.jwt_secret, algorithms=['HS256'])
            
            # Check token blacklist
            token_id = payload.get('jti')
            if await self.redis.get(f"blacklist:{token_id}"):
                raise SecurityError("Token has been revoked")
            
            # Verify client is still active
            client_id = payload.get('client_id')
            if not await self.is_client_active(client_id):
                raise SecurityError("Client has been deactivated")
            
            return payload
            
        except jwt.ExpiredSignatureError:
            raise SecurityError("Token has expired")
        except jwt.InvalidTokenError:
            raise SecurityError("Invalid token")
    
    async def create_session_token(self, client_id: str) -> str:
        """
        Create secure JWT session token
        """
        payload = {
            'client_id': client_id,
            'jti': secrets.token_urlsafe(16),  # JWT ID for blacklisting
            'iat': datetime.utcnow(),
            'exp': datetime.utcnow() + self.token_expiry,
            'permissions': await self.get_client_permissions(client_id)
        }
        
        return jwt.encode(payload, self.jwt_secret, algorithm='HS256')
    
    async def check_permission(self, token: str, resource: str, action: str) -> bool:
        """
        Fine-grained permission checking
        """
        payload = await self.validate_session_token(token)
        permissions = payload.get('permissions', [])
        
        # Check exact permission
        if f"{resource}:{action}" in permissions:
            return True
        
        # Check wildcard permissions
        if f"{resource}:*" in permissions or f"*:{action}" in permissions:
            return True
            
        return False

class SecurityError(Exception):
    pass

import re
import html
import bleach
from typing import Any, Dict, List
import sqlparse
from urllib.parse import urlparse

class InputValidator:
    def __init__(self):
        # SQL injection patterns
        self.sql_patterns = [
            r'(\bunion\s+select\b)',
            r'(\bselect\s+.*\bfrom\b)',
            r'(\binsert\s+into\b)',
            r'(\bdelete\s+from\b)',
            r'(\bdrop\s+table\b)',
            r'(\balter\s+table\b)',
            r'(--|\#|\/\*|\*\/)',
            r'(\bor\s+1\s*=\s*1\b)',
            r'(\band\s+1\s*=\s*1\b)'
        ]
        
        # Command injection patterns
        self.cmd_patterns = [
            r'[;&|`$(){}[\]<>]',
            r'(\bcat\b|\bls\b|\brm\b|\bmv\b|\bcp\b)',
            r'(\bwget\b|\bcurl\b|\bssh\b|\bftp\b)',
            r'(\bchmod\b|\bchown\b|\bsudo\b|\bsu\b)'
        ]
        
        # Allowed HTML tags for sanitization
        self.allowed_tags = ['b', 'i', 'em', 'strong', 'p', 'br', 'ul', 'ol', 'li']
        
    def validate_tool_input(self, tool_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
        """
        Comprehensive tool input validation
        """
        validated_params = {}
        
        for param_name, param_value in parameters.items():
            # Type validation based on parameter name patterns
            if self._is_numeric_param(param_name):
                validated_params[param_name] = self._validate_numeric(param_value)
            elif self._is_sql_param(param_name):
                validated_params[param_name] = self._validate_sql_query(param_value)
            elif self._is_url_param(param_name):
                validated_params[param_name] = self._validate_url(param_value)
            elif self._is_file_param(param_name):
                validated_params[param_name] = self._validate_file_path(param_value)
            else:
                validated_params[param_name] = self._validate_general_string(param_value)
        
        return validated_params
    
    def _validate_sql_query(self, query: str) -> str:
        """
        Validate and sanitize SQL queries
        """
        if not isinstance(query, str):
            raise ValidationError("SQL query must be a string")
        
        # Check for dangerous SQL patterns
        query_lower = query.lower()
        for pattern in self.sql_patterns:
            if re.search(pattern, query_lower, re.IGNORECASE):
                raise SecurityError(f"Potentially dangerous SQL pattern detected: {pattern}")
        
        # Parse SQL to ensure it's valid
        try:
            parsed = sqlparse.parse(query)
            if not parsed:
                raise ValidationError("Invalid SQL syntax")
        except Exception as e:
            raise ValidationError(f"SQL parsing error: {str(e)}")
        
        # Limit query complexity
        if len(query) > 1000:
            raise ValidationError("SQL query too long")
        
        return query.strip()
    
    def _validate_file_path(self, path: str) -> str:
        """
        Validate file paths to prevent directory traversal
        """
        if not isinstance(path, str):
            raise ValidationError("File path must be a string")
        
        # Check for directory traversal patterns
        if '..' in path or path.startswith('/'):
            raise SecurityError("Directory traversal attempt detected")
        
        # Whitelist allowed characters
        if not re.match(r'^[a-zA-Z0-9._/-]+$', path):
            raise ValidationError("Invalid characters in file path")
        
        return path.strip()
    
    def _validate_url(self, url: str) -> str:
        """
        Validate URLs to prevent SSRF attacks
        """
        if not isinstance(url, str):
            raise ValidationError("URL must be a string")
        
        try:
            parsed = urlparse(url)
            
            # Only allow HTTP/HTTPS
            if parsed.scheme not in ['http', 'https']:
                raise SecurityError("Only HTTP/HTTPS URLs are allowed")
            
            # Block internal/private IP ranges
            if self._is_internal_ip(parsed.hostname):
                raise SecurityError("Internal IP addresses are not allowed")
            
            # Block localhost
            if parsed.hostname in ['localhost', '127.0.0.1', '::1']:
                raise SecurityError("Localhost access is not allowed")
                
        except Exception as e:
            raise ValidationError(f"Invalid URL format: {str(e)}")
        
        return url.strip()
    
    def _validate_general_string(self, value: Any) -> str:
        """
        General string validation and sanitization
        """
        if not isinstance(value, str):
            value = str(value)
        
        # Check for command injection patterns
        for pattern in self.cmd_patterns:
            if re.search(pattern, value, re.IGNORECASE):
                raise SecurityError(f"Potentially dangerous command pattern detected")
        
        # HTML escape for safety
        value = html.escape(value)
        
        # Limit length
        if len(value) > 10000:
            raise ValidationError("Input too long")
        
        return value.strip()
    
    def _is_internal_ip(self, hostname: str) -> bool:
        """
        Check if hostname resolves to internal IP ranges
        """
        import socket
        try:
            ip = socket.gethostbyname(hostname)
            # Check for private IP ranges
            private_ranges = [
                '10.', '172.16.', '172.17.', '172.18.', '172.19.',
                '172.20.', '172.21.', '172.22.', '172.23.', '172.24.',
                '172.25.', '172.26.', '172.27.', '172.28.', '172.29.',
                '172.30.', '172.31.', '192.168.'
            ]
            return any(ip.startswith(range_) for range_ in private_ranges)
        except:
            return True  # Err on the side of caution

class ValidationError(Exception):
    pass

import boto3
import os
from typing import Dict, Optional
import json
import base64
from cryptography.fernet import Fernet
import hvac

class SecretsManager:
    def __init__(self, provider: str = "aws"):
        self.provider = provider
        
        if provider == "aws":
            self.client = boto3.client('secretsmanager')
        elif provider == "vault":
            self.client = hvac.Client(url=os.getenv('VAULT_URL'))
            self.client.token = os.getenv('VAULT_TOKEN')
        elif provider == "local":
            # For development only - use proper secrets manager in production
            self.key = os.getenv('ENCRYPTION_KEY', Fernet.generate_key())
            self.cipher = Fernet(self.key)
    
    async def get_secret(self, secret_name: str) -> Dict[str, str]:
        """
        Retrieve secrets securely from configured provider
        """
        if self.provider == "aws":
            return await self._get_aws_secret(secret_name)
        elif self.provider == "vault":
            return await self._get_vault_secret(secret_name)
        elif self.provider == "local":
            return await self._get_local_secret(secret_name)
    
    async def _get_aws_secret(self, secret_name: str) -> Dict[str, str]:
        """
        Retrieve secret from AWS Secrets Manager
        """
        try:
            response = self.client.get_secret_value(SecretId=secret_name)
            return json.loads(response['SecretString'])
        except Exception as e:
            raise SecretRetrievalError(f"Failed to retrieve AWS secret: {str(e)}")
    
    def encrypt_sensitive_data(self, data: str) -> str:
        """
        Encrypt sensitive data before storage
        """
        if self.provider == "local":
            encrypted_data = self.cipher.encrypt(data.encode())
            return base64.b64encode(encrypted_data).decode()
        else:
            # Use KMS or Vault for production encryption
            return self._encrypt_with_kms(data)
    
    def decrypt_sensitive_data(self, encrypted_data: str) -> str:
        """
        Decrypt sensitive data after retrieval
        """
        if self.provider == "local":
            encrypted_bytes = base64.b64decode(encrypted_data)
            decrypted_data = self.cipher.decrypt(encrypted_bytes)
            return decrypted_data.decode()
        else:
            return self._decrypt_with_kms(encrypted_data)

# Environment-specific configuration
def configure_tls_security():
    """
    Configure TLS settings for production security
    """
    tls_config = {
        "min_version": "TLSv1.3",
        "ciphers": [
            "TLS_AES_256_GCM_SHA384",
            "TLS_CHACHA20_POLY1305_SHA256",
            "TLS_AES_128_GCM_SHA256"
        ],
        "require_client_cert": True,
        "verify_client_cert": True,
        "certificate_chain": "/path/to/cert/chain.pem",
        "private_key": "/path/to/private/key.pem"
    }
    
    return tls_config

import logging
import json
from datetime import datetime
from typing import Dict, Any
import hashlib
import asyncio

class SecurityLogger:
    def __init__(self, log_level=logging.INFO):
        self.logger = logging.getLogger('mcp_security')
        self.logger.setLevel(log_level)
        
        # Structured logging handler
        handler = logging.StreamHandler()
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
    
    def log_authentication_event(self, 
                                client_id: str, 
                                success: bool, 
                                source_ip: str,
                                user_agent: str = None,
                                additional_info: Dict[str, Any] = None):
        """
        Log authentication events with security context
        """
        event = {
            "event_type": "authentication",
            "client_id": self._hash_pii(client_id),  # Hash PII for privacy
            "success": success,
            "source_ip": source_ip,
            "user_agent": user_agent,
            "timestamp": datetime.utcnow().isoformat(),
            "severity": "INFO" if success else "WARNING",
            **additional_info or {}
        }
        
        if success:
            self.logger.info(json.dumps(event))
        else:
            self.logger.warning(json.dumps(event))
    
    def log_permission_violation(self, 
                                client_id: str, 
                                resource: str, 
                                action: str,
                                source_ip: str):
        """
        Log permission violations for security analysis
        """
        event = {
            "event_type": "permission_violation",
            "client_id": self._hash_pii(client_id),
            "resource": resource,
            "action": action,
            "source_ip": source_ip,
            "timestamp": datetime.utcnow().isoformat(),
            "severity": "ERROR"
        }
        
        self.logger.error(json.dumps(event))
    
    def log_suspicious_activity(self, 
                              activity_type: str,
                              details: Dict[str, Any],
                              severity: str = "WARNING"):
        """
        Log suspicious activities for investigation
        """
        event = {
            "event_type": "suspicious_activity",
            "activity_type": activity_type,
            "details": details,
            "timestamp": datetime.utcnow().isoformat(),
            "severity": severity
        }
        
        if severity == "CRITICAL":
            self.logger.critical(json.dumps(event))
        elif severity == "ERROR":
            self.logger.error(json.dumps(event))
        else:
            self.logger.warning(json.dumps(event))
    
    def _hash_pii(self, pii_data: str) -> str:
        """
        Hash PII data for privacy-compliant logging
        """
        return hashlib.sha256(pii_data.encode()).hexdigest()[:12]

class SecurityMonitor:
    def __init__(self, redis_client, logger: SecurityLogger):
        self.redis = redis_client
        self.logger = logger
        self.alert_thresholds = {
            "failed_auth_attempts": 5,
            "permission_violations": 3,
            "rate_limit_violations": 10
        }
    
    async def check_failed_auth_pattern(self, client_id: str, source_ip: str):
        """
        Monitor failed authentication patterns
        """
        key = f"failed_auth:{source_ip}"
        current_failures = await self.redis.incr(key)
        await self.redis.expire(key, 300)  # 5-minute window
        
        if current_failures >= self.alert_thresholds["failed_auth_attempts"]:
            await self._trigger_security_alert(
                "multiple_failed_auth",
                {
                    "source_ip": source_ip,
                    "client_id": client_id,
                    "failure_count": current_failures
                }
            )
    
    async def _trigger_security_alert(self, alert_type: str, details: Dict[str, Any]):
        """
        Trigger immediate security alerts for critical events
        """
        self.logger.log_suspicious_activity(
            alert_type, 
            details, 
            severity="CRITICAL"
        )
        
        # Additional alerting mechanisms (email, Slack, PagerDuty, etc.)
        await self._send_security_notification(alert_type, details)