"""Core Middleware Components Middleware components for the Adtlas Core module. Provides essential middleware for request processing, security, and system monitoring. Middleware Components: - RequestMiddleware: Core request processing and context management - SystemHealthMiddleware: System health monitoring - AuditMiddleware: Audit logging for system operations - MaintenanceModeMiddleware: Maintenance mode handling - SecurityHeadersMiddleware: Additional security headers - RequestLoggingMiddleware: Request/response logging - PerformanceMiddleware: Performance monitoring - CacheMiddleware: Core caching functionality Usage: Add to MIDDLEWARE setting in Django settings: MIDDLEWARE = [ ... 'apps.core.middleware.RequestMiddleware', 'apps.core.middleware.SystemHealthMiddleware', 'apps.core.middleware.AuditMiddleware', 'apps.core.middleware.MaintenanceModeMiddleware', 'apps.core.middleware.SecurityHeadersMiddleware', 'apps.core.middleware.RequestLoggingMiddleware', 'apps.core.middleware.PerformanceMiddleware', 'apps.core.middleware.CacheMiddleware', ... ] Configuration: Configure middleware behavior in Django settings: # Core middleware settings CORE_MIDDLEWARE_ENABLED = True MAINTENANCE_MODE = False AUDIT_LOGGING_ENABLED = True PERFORMANCE_MONITORING_ENABLED = True REQUEST_LOGGING_ENABLED = True # Security settings SECURITY_HEADERS_ENABLED = True CUSTOM_SECURITY_HEADERS = { 'X-Custom-Header': 'value' } # Performance settings PERFORMANCE_THRESHOLD_MS = 1000 SLOW_REQUEST_LOGGING = True """ import time import logging import uuid from typing import Optional, Dict, Any from django.conf import settings from django.core.cache import cache from django.http import HttpRequest, HttpResponse, JsonResponse from django.utils import timezone from django.utils.deprecation import MiddlewareMixin from django.contrib.auth.models import AnonymousUser from django.urls import resolve from django.db import connection from django.template.response import TemplateResponse # Configure logging logger = logging.getLogger(__name__) # BASE MIDDLEWARE class BaseCoreMiddleware(MiddlewareMixin): """ Base middleware class for core components. Provides common functionality for all core middleware: - Configuration management - Logging utilities - Error handling - Context management Attributes: enabled: Whether middleware is enabled config: Middleware configuration """ def __init__(self, get_response=None): """ Initialize base core middleware. Args: get_response: Django get_response callable """ super().__init__(get_response) self.get_response = get_response self.enabled = getattr(settings, 'CORE_MIDDLEWARE_ENABLED', True) self.config = self._load_config() self._initialize_middleware() def _load_config(self) -> Dict[str, Any]: """ Load middleware configuration. Returns: Dictionary containing middleware configuration """ try: # Default configuration default_config = { 'enabled': True, 'debug': getattr(settings, 'DEBUG', False), 'log_level': 'INFO', } # Get middleware-specific config config_key = f'{self.__class__.__name__.upper()}_CONFIG' middleware_config = getattr(settings, config_key, {}) # Merge configurations config = {**default_config, **middleware_config} return config except Exception as e: logger.error(f"Failed to load middleware config: {e}") return {'enabled': True, 'debug': False, 'log_level': 'INFO'} def _initialize_middleware(self): """ Initialize middleware components. Override in subclasses for specific initialization. """ pass def _is_enabled(self) -> bool: """ Check if middleware is enabled. Returns: True if middleware is enabled, False otherwise """ return self.enabled and self.config.get('enabled', True) def _log_middleware_action( self, action: str, request: HttpRequest, extra_data: Optional[Dict[str, Any]] = None, level: str = 'INFO' ): """ Log middleware action. Args: action: Action being performed request: HTTP request object extra_data: Additional data to log level: Log level """ try: log_data = { 'action': action, 'middleware': self.__class__.__name__, 'path': request.path, 'method': request.method, 'user': str(request.user) if hasattr(request, 'user') else 'Anonymous', 'timestamp': timezone.now().isoformat(), } if extra_data: log_data.update(extra_data) getattr(logger, level.lower())(f"Middleware {action}", extra=log_data) except Exception as e: logger.error(f"Middleware logging failed: {e}") # REQUEST MIDDLEWARE class RequestMiddleware(BaseCoreMiddleware): """ Core request processing middleware. Provides: - Request ID generation - Request context management - Request timing - User context setup """ def process_request(self, request: HttpRequest) -> Optional[HttpResponse]: """ Process incoming request. Args: request: HTTP request object Returns: None to continue processing, HttpResponse to short-circuit """ if not self._is_enabled(): return None try: # Generate unique request ID request.request_id = str(uuid.uuid4()) # Set request start time request.start_time = time.time() # Set request context request.context = { 'request_id': request.request_id, 'start_time': request.start_time, 'path': request.path, 'method': request.method, } # Resolve URL pattern try: url_match = resolve(request.path) request.context['url_name'] = url_match.url_name request.context['app_name'] = url_match.app_name request.context['namespace'] = url_match.namespace except Exception: pass self._log_middleware_action('request_processed', request) except Exception as e: logger.error(f"Request middleware error: {e}") return None def process_response(self, request: HttpRequest, response: HttpResponse) -> HttpResponse: """ Process outgoing response. Args: request: HTTP request object response: HTTP response object Returns: Modified HTTP response object """ if not self._is_enabled(): return response try: # Add request ID to response headers if hasattr(request, 'request_id'): response['X-Request-ID'] = request.request_id # Calculate request duration if hasattr(request, 'start_time'): duration = time.time() - request.start_time response['X-Response-Time'] = f"{duration:.3f}s" self._log_middleware_action('response_processed', request, { 'status_code': response.status_code, 'duration_ms': round(duration * 1000, 2), }) except Exception as e: logger.error(f"Response middleware error: {e}") return response # SYSTEM HEALTH MIDDLEWARE class SystemHealthMiddleware(BaseCoreMiddleware): """ System health monitoring middleware. Monitors: - Database connectivity - Cache availability - Memory usage - Response times """ def __init__(self, get_response=None): super().__init__(get_response) self.health_check_interval = self.config.get('health_check_interval', 60) # seconds self.last_health_check = 0 def process_request(self, request: HttpRequest) -> Optional[HttpResponse]: """ Process request and check system health. Args: request: HTTP request object Returns: None to continue, HttpResponse if system is unhealthy """ if not self._is_enabled(): return None try: current_time = time.time() # Perform health check if interval has passed if current_time - self.last_health_check > self.health_check_interval: health_status = self._check_system_health() self.last_health_check = current_time # Store health status in cache cache.set('system_health_status', health_status, timeout=300) # Log health status self._log_middleware_action('health_check_performed', request, { 'health_status': health_status }) except Exception as e: logger.error(f"System health middleware error: {e}") return None def _check_system_health(self) -> Dict[str, Any]: """ Check system health status. Returns: Dictionary containing health status information """ health_status = { 'timestamp': timezone.now().isoformat(), 'status': 'healthy', 'checks': {} } try: # Check database connectivity with connection.cursor() as cursor: cursor.execute("SELECT 1") health_status['checks']['database'] = 'healthy' except Exception as e: health_status['checks']['database'] = f'unhealthy: {str(e)}' health_status['status'] = 'unhealthy' try: # Check cache availability cache.set('health_check', 'test', timeout=10) cache.get('health_check') health_status['checks']['cache'] = 'healthy' except Exception as e: health_status['checks']['cache'] = f'unhealthy: {str(e)}' health_status['status'] = 'degraded' return health_status # AUDIT MIDDLEWARE class AuditMiddleware(BaseCoreMiddleware): """ Audit logging middleware. Logs: - User actions - System operations - Security events - Data modifications """ def __init__(self, get_response=None): super().__init__(get_response) self.audit_enabled = getattr(settings, 'AUDIT_LOGGING_ENABLED', True) self.sensitive_paths = self.config.get('sensitive_paths', [ '/admin/', '/api/auth/', '/accounts/' ]) def process_request(self, request: HttpRequest) -> Optional[HttpResponse]: """ Process request for audit logging. Args: request: HTTP request object Returns: None to continue processing """ if not self._is_enabled() or not self.audit_enabled: return None try: # Check if path should be audited should_audit = any(request.path.startswith(path) for path in self.sensitive_paths) if should_audit or request.method in ['POST', 'PUT', 'PATCH', 'DELETE']: audit_data = { 'user_id': request.user.id if hasattr(request, 'user') and not isinstance(request.user, AnonymousUser) else None, 'username': str(request.user) if hasattr(request, 'user') and not isinstance(request.user, AnonymousUser) else 'Anonymous', 'ip_address': self._get_client_ip(request), 'user_agent': request.META.get('HTTP_USER_AGENT', ''), 'path': request.path, 'method': request.method, 'query_params': dict(request.GET), 'timestamp': timezone.now().isoformat(), } # Store audit data for response processing request.audit_data = audit_data self._log_middleware_action('audit_request', request, audit_data) except Exception as e: logger.error(f"Audit middleware error: {e}") return None def process_response(self, request: HttpRequest, response: HttpResponse) -> HttpResponse: """ Process response for audit logging. Args: request: HTTP request object response: HTTP response object Returns: HTTP response object """ if not self._is_enabled() or not self.audit_enabled: return response try: if hasattr(request, 'audit_data'): audit_data = request.audit_data audit_data.update({ 'status_code': response.status_code, 'response_size': len(response.content) if hasattr(response, 'content') else 0, }) # Log audit event self._log_middleware_action('audit_response', request, audit_data) # Store in audit log if model exists try: from apps.core.models import AuditLog AuditLog.objects.create( user_id=audit_data.get('user_id'), action=f"{audit_data['method']} {audit_data['path']}", ip_address=audit_data.get('ip_address'), user_agent=audit_data.get('user_agent'), details=audit_data ) except Exception: pass # Fail silently if AuditLog model doesn't exist except Exception as e: logger.error(f"Audit response middleware error: {e}") return response def _get_client_ip(self, request: HttpRequest) -> str: """ Get client IP address from request. Args: request: HTTP request object Returns: Client IP address """ x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip = x_forwarded_for.split(',')[0] else: ip = request.META.get('REMOTE_ADDR') return ip # MAINTENANCE MODE MIDDLEWARE class MaintenanceModeMiddleware(BaseCoreMiddleware): """ Maintenance mode middleware. Features: - Enable/disable maintenance mode - Custom maintenance page - Whitelist IP addresses - Staff user bypass """ def __init__(self, get_response=None): super().__init__(get_response) self.maintenance_mode = getattr(settings, 'MAINTENANCE_MODE', False) self.whitelist_ips = self.config.get('whitelist_ips', []) self.maintenance_template = self.config.get('maintenance_template', 'core/maintenance.html') def process_request(self, request: HttpRequest) -> Optional[HttpResponse]: """ Process request for maintenance mode. Args: request: HTTP request object Returns: Maintenance response if in maintenance mode, None otherwise """ if not self.maintenance_mode: return None try: # Check if user should bypass maintenance mode if self._should_bypass_maintenance(request): return None # Return maintenance response context = { 'maintenance_message': 'The system is currently under maintenance. Please try again later.', 'estimated_completion': self.config.get('estimated_completion'), } self._log_middleware_action('maintenance_mode_active', request) return TemplateResponse(request, self.maintenance_template, context, status=503) except Exception as e: logger.error(f"Maintenance mode middleware error: {e}") return None def _should_bypass_maintenance(self, request: HttpRequest) -> bool: """ Check if request should bypass maintenance mode. Args: request: HTTP request object Returns: True if should bypass, False otherwise """ # Check if user is staff if hasattr(request, 'user') and request.user.is_authenticated and request.user.is_staff: return True # Check if IP is whitelisted client_ip = self._get_client_ip(request) if client_ip in self.whitelist_ips: return True # Check if accessing admin or health check endpoints if request.path.startswith('/admin/') or request.path.startswith('/health/'): return True return False def _get_client_ip(self, request: HttpRequest) -> str: """ Get client IP address from request. Args: request: HTTP request object Returns: Client IP address """ x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR') if x_forwarded_for: ip = x_forwarded_for.split(',')[0] else: ip = request.META.get('REMOTE_ADDR') return ip # Export middleware classes __all__ = [ 'BaseCoreMiddleware', 'RequestMiddleware', 'SystemHealthMiddleware', 'AuditMiddleware', 'MaintenanceModeMiddleware', ]