MCP Security Model¶

# Client certificate authentication
import ssl

context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
context.check_hostname = False
context.verify_mode = ssl.CERT_REQUIRED
context.load_cert_chain('/path/to/client.crt', '/path/to/client.key')
context.load_verify_locations('/path/to/ca.crt')

transport = HttpTransport(
    url="https://secure-mcp-server.com",
    ssl_context=context
)

import jwt
from datetime import datetime, timedelta

# Generate JWT token
payload = {
    'sub': '[email protected]',
    'iat': datetime.utcnow(),
    'exp': datetime.utcnow() + timedelta(hours=1),
    'scopes': ['mcp:tools', 'mcp:resources']
}

token = jwt.encode(payload, 'secret-key', algorithm='HS256')

# Use in MCP request
headers = {
    'Authorization': f'Bearer {token}'
}

class CustomAuthenticator:
    async def authenticate(self, request):
        # Custom authentication logic
        token = request.headers.get('X-Custom-Auth')
        if not self.validate_token(token):
            raise AuthenticationError("Invalid token")

        return UserContext(
            user_id=self.extract_user_id(token),
            permissions=self.get_permissions(token)
        )

# ABAC Policy Example
class ABACPolicy:
    def evaluate(self, subject, resource, action, environment):
        # Time-based access
        if environment.time.hour < 9 or environment.time.hour > 17:
            if action in ['tools:call', 'resources:write']:
                return False

        # Resource-specific rules
        if resource.type == 'database' and resource.name == 'production':
            return subject.role in ['admin', 'senior-developer']

        # IP-based restrictions
        if subject.ip_address not in environment.allowed_ips:
            return False

        return True

# Secure implementation
@server.tool()
async def query_database(query: str, parameters: list = None):
    # Validate query pattern
    if not re.match(r'^SELECT\s+', query, re.IGNORECASE):
        raise ValueError("Only SELECT queries allowed")

    # Use parameterized query
    async with database.connection() as conn:
        result = await conn.execute(query, parameters or [])
        return result.fetchall()

# Insecure - DON'T DO THIS
# await conn.execute(f"SELECT * FROM users WHERE id = {user_id}")

import os
from pathlib import Path

@server.tool()
async def read_file(file_path: str):
    # Validate path is within allowed directory
    allowed_path = Path("/var/data")
    requested_path = Path(file_path).resolve()

    if not requested_path.is_relative_to(allowed_path):
        raise SecurityError("Path traversal attempt detected")

    # Additional filename validation
    if '..' in file_path or file_path.startswith('/'):
        raise SecurityError("Invalid file path")

    return await read_file_content(requested_path)

import asyncio
from collections import defaultdict
from time import time

class RateLimiter:
    def __init__(self, max_requests=100, window_seconds=3600):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.buckets = defaultdict(list)

    async def check_rate_limit(self, client_id: str):
        now = time()
        bucket = self.buckets[client_id]

        # Remove expired entries
        bucket[:] = [t for t in bucket if now - t < self.window_seconds]

        if len(bucket) >= self.max_requests:
            raise RateLimitError("Rate limit exceeded")

        bucket.append(now)

class SlidingWindowRateLimit:
    def __init__(self, max_requests=1000, window_minutes=60):
        self.max_requests = max_requests
        self.window_seconds = window_minutes * 60
        self.request_log = defaultdict(list)

    def is_allowed(self, client_id: str) -> bool:
        now = time()
        requests = self.request_log[client_id]

        # Clean old requests
        cutoff = now - self.window_seconds
        self.request_log[client_id] = [t for t in requests if t > cutoff]

        if len(self.request_log[client_id]) >= self.max_requests:
            return False

        self.request_log[client_id].append(now)
        return True

import logging
import json
from datetime import datetime

class AuditLogger:
    def __init__(self):
        self.logger = logging.getLogger('mcp.audit')
        handler = logging.FileHandler('/var/log/mcp/audit.log')
        formatter = logging.Formatter('%(asctime)s %(message)s')
        handler.setFormatter(formatter)
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def log_request(self, client_id, method, params, client_ip):
        audit_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'event_type': 'mcp_request',
            'client_id': client_id,
            'client_ip': client_ip,
            'method': method,
            'params': self.sanitize_params(params),
            'request_id': generate_request_id()
        }
        self.logger.info(json.dumps(audit_entry))

    def log_tool_execution(self, tool_name, client_id, success, duration):
        audit_entry = {
            'timestamp': datetime.utcnow().isoformat(),
            'event_type': 'tool_execution',
            'tool_name': tool_name,
            'client_id': client_id,
            'success': success,
            'duration_ms': duration,
            'request_id': get_current_request_id()
        }
        self.logger.info(json.dumps(audit_entry))

import syslog

class SIEMIntegration:
    def __init__(self, siem_host, siem_port):
        syslog.openlog("mcp-server")

    def send_security_event(self, event_type, severity, details):
        message = {
            'event_type': event_type,
            'severity': severity,
            'timestamp': datetime.utcnow().isoformat(),
            'details': details
        }

        syslog.syslog(syslog.LOG_WARNING, json.dumps(message))

from cryptography.fernet import Fernet

class DataEncryption:
    def __init__(self, key: bytes):
        self.cipher = Fernet(key)

    def encrypt_sensitive_data(self, data: str) -> str:
        return self.cipher.encrypt(data.encode()).decode()

    def decrypt_sensitive_data(self, encrypted_data: str) -> str:
        return self.cipher.decrypt(encrypted_data.encode()).decode()

# Usage
encryption = DataEncryption(Fernet.generate_key())
encrypted_token = encryption.encrypt_sensitive_data("api-token-value")

import re

class DataAnonymizer:
    @staticmethod
    def mask_email(email: str) -> str:
        local, domain = email.split('@')
        return f"{local[0]}***@{domain}"

    @staticmethod
    def mask_credit_card(cc_number: str) -> str:
        return f"****-****-****-{cc_number[-4:]}"

    @staticmethod
    def remove_pii(text: str) -> str:
        # Remove Social Security Numbers
        text = re.sub(r'\b\d{3}-\d{2}-\d{4}\b', '[SSN]', text)

        # Remove phone numbers
        text = re.sub(r'\b\d{3}-\d{3}-\d{4}\b', '[PHONE]', text)

        return text