# Security Configuration Module for Context Engineering MCP Platform
import os
import secrets
import hashlib
from typing import Dict, List, Optional
from cryptography.fernet import Fernet
import jwt
from datetime import datetime, timedelta
class SecurityConfig:
"""Centralized security configuration and utilities"""
def __init__(self):
self.jwt_secret = os.getenv('JWT_SECRET', self._generate_jwt_secret())
self.encryption_salt = os.getenv('ENCRYPTION_SALT', self._generate_salt())
self.enable_https = os.getenv('ENABLE_HTTPS', 'false').lower() == 'true'
self.max_request_size = int(os.getenv('MAX_REQUEST_SIZE', '50')) * 1024 * 1024 # MB to bytes
self.max_file_upload_size = int(os.getenv('MAX_FILE_UPLOAD_SIZE', '10')) * 1024 * 1024
self.rate_limit_rpm = int(os.getenv('RATE_LIMIT_RPM', '60'))
self.session_timeout = int(os.getenv('SESSION_TIMEOUT', '3600'))
self._fernet = None
def _generate_jwt_secret(self) -> str:
"""Generate a secure JWT secret if not provided"""
return secrets.token_urlsafe(32)
def _generate_salt(self) -> str:
"""Generate a secure salt for encryption"""
return secrets.token_urlsafe(16)
def get_fernet(self) -> Fernet:
"""Get or create Fernet encryption instance"""
if self._fernet is None:
# Create a key from the encryption salt
key = hashlib.sha256(self.encryption_salt.encode()).digest()
key_b64 = Fernet.generate_key() # Use proper key generation
self._fernet = Fernet(key_b64)
return self._fernet
def encrypt_api_key(self, api_key: str) -> str:
"""Encrypt an API key for secure storage"""
return self.get_fernet().encrypt(api_key.encode()).decode()
def decrypt_api_key(self, encrypted_key: str) -> str:
"""Decrypt an API key"""
return self.get_fernet().decrypt(encrypted_key.encode()).decode()
def generate_jwt_token(self, user_id: str, additional_claims: Optional[Dict] = None) -> str:
"""Generate a JWT token for authentication"""
payload = {
'user_id': user_id,
'exp': datetime.utcnow() + timedelta(seconds=self.session_timeout),
'iat': datetime.utcnow(),
'iss': 'context-engineering-mcp'
}
if additional_claims:
payload.update(additional_claims)
return jwt.encode(payload, self.jwt_secret, algorithm='HS256')
def verify_jwt_token(self, token: str) -> Optional[Dict]:
"""Verify and decode a JWT token"""
try:
payload = jwt.decode(token, self.jwt_secret, algorithms=['HS256'])
return payload
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
return None
def hash_password(self, password: str) -> str:
"""Hash a password with salt"""
return hashlib.pbkdf2_hmac(
'sha256',
password.encode(),
self.encryption_salt.encode(),
100000 # iterations
).hex()
def verify_password(self, password: str, hashed: str) -> bool:
"""Verify a password against its hash"""
return self.hash_password(password) == hashed
def get_cors_origins(self) -> List[str]:
"""Get allowed CORS origins"""
origins = os.getenv('CORS_ORIGINS', 'http://localhost:3000')
return [origin.strip() for origin in origins.split(',')]
def is_safe_path(self, path: str) -> bool:
"""Check if a file path is safe (prevent directory traversal)"""
# Normalize the path and check for directory traversal attempts
normalized = os.path.normpath(path)
return not normalized.startswith('..') and not os.path.isabs(normalized)
def sanitize_filename(self, filename: str) -> str:
"""Sanitize a filename for safe storage"""
import re
# Remove or replace dangerous characters
sanitized = re.sub(r'[^\w\-_\.]', '_', filename)
# Limit length
if len(sanitized) > 255:
name, ext = os.path.splitext(sanitized)
sanitized = name[:255-len(ext)] + ext
return sanitized
def validate_api_key_format(self, api_key: str) -> bool:
"""Validate API key format"""
# Basic validation - adjust based on your API key format
return len(api_key) >= 20 and api_key.replace('-', '').replace('_', '').isalnum()
class SecurityMiddleware:
"""Security middleware for FastAPI applications"""
def __init__(self, security_config: SecurityConfig):
self.config = security_config
self.rate_limit_store = {} # In production, use Redis
async def rate_limit_check(self, client_id: str) -> bool:
"""Check if client is within rate limits"""
now = datetime.utcnow()
minute_key = now.strftime('%Y-%m-%d-%H-%M')
key = f"{client_id}:{minute_key}"
current_count = self.rate_limit_store.get(key, 0)
if current_count >= self.config.rate_limit_rpm:
return False
self.rate_limit_store[key] = current_count + 1
# Clean up old entries (basic cleanup)
if len(self.rate_limit_store) > 10000:
cutoff = (now - timedelta(minutes=2)).strftime('%Y-%m-%d-%H-%M')
keys_to_remove = [k for k in self.rate_limit_store.keys() if k.split(':')[1] < cutoff]
for k in keys_to_remove:
del self.rate_limit_store[k]
return True
def get_client_id(self, request) -> str:
"""Extract client identifier from request"""
# Use API key or IP address as client identifier
api_key = request.headers.get('X-API-Key')
if api_key:
return hashlib.sha256(api_key.encode()).hexdigest()[:16]
# Fallback to IP address
forwarded_for = request.headers.get('X-Forwarded-For')
if forwarded_for:
return forwarded_for.split(',')[0].strip()
return request.client.host if request.client else 'unknown'
# Global security instance
security_config = SecurityConfig()
security_middleware = SecurityMiddleware(security_config)
def get_security_headers() -> Dict[str, str]:
"""Get security headers for HTTP responses"""
return {
'X-Content-Type-Options': 'nosniff',
'X-Frame-Options': 'DENY',
'X-XSS-Protection': '1; mode=block',
'Strict-Transport-Security': 'max-age=31536000; includeSubDomains',
'Content-Security-Policy': "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'",
'Referrer-Policy': 'strict-origin-when-cross-origin'
}
def log_security_event(event_type: str, details: Dict, severity: str = 'INFO'):
"""Log security events for monitoring"""
import logging
logger = logging.getLogger('security')
log_entry = {
'event_type': event_type,
'timestamp': datetime.utcnow().isoformat(),
'severity': severity,
'details': details
}
if severity == 'CRITICAL':
logger.critical(f"Security Event: {log_entry}")
elif severity == 'ERROR':
logger.error(f"Security Event: {log_entry}")
elif severity == 'WARNING':
logger.warning(f"Security Event: {log_entry}")
else:
logger.info(f"Security Event: {log_entry}")
# Example usage and testing
if __name__ == "__main__":
# Test the security configuration
config = SecurityConfig()
# Test API key encryption
test_key = "test-api-key-12345"
encrypted = config.encrypt_api_key(test_key)
decrypted = config.decrypt_api_key(encrypted)
print(f"API Key encryption test: {test_key == decrypted}")
# Test JWT token generation
token = config.generate_jwt_token("user123", {"role": "admin"})
payload = config.verify_jwt_token(token)
print(f"JWT test: {payload is not None}")
# Test password hashing
password = "secure_password_123"
hashed = config.hash_password(password)
verified = config.verify_password(password, hashed)
print(f"Password hashing test: {verified}")
