project initialization

backend/security/headers.py

@@ -0,0 +1,504 @@
+"""
+Security headers and Content Security Policy (CSP) management for enhanced security.
+"""
+
+from django.conf import settings
+from django.http import HttpResponse
+from django.middleware.security import SecurityMiddleware as DjangoSecurityMiddleware
+from django.utils.deprecation import MiddlewareMixin
+from django.core.exceptions import MiddlewareNotUsed
+import re
+import json
+from typing import Dict, List, Optional, Set, Tuple
+from urllib.parse import urlparse, urlunparse
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class SecurityHeadersMiddleware(MiddlewareMixin):
+    """
+    Enhanced security middleware with comprehensive security headers and CSP policies.
+    """
+
+    def __init__(self, get_response):
+        super().__init__(get_response)
+
+        # Initialize CSP configuration
+        self.csp_config = getattr(settings, 'CSP_CONFIG', self._get_default_csp_config())
+
+        # Initialize security headers
+        self.security_headers = getattr(settings, 'SECURITY_HEADERS', self._get_default_security_headers())
+
+        # Initialize allowed domains
+        self.allowed_domains = set(getattr(settings, 'ALLOWED_DOMAINS', []))
+
+        # Initialize nonce generator
+        self.nonce_generator = CSPNonceGenerator()
+
+    def process_response(self, request, response):
+        """
+        Add security headers to the response.
+        """
+        # Skip for static files and media
+        if self._should_skip_headers(request):
+            return response
+
+        # Add security headers
+        for header, value in self.security_headers.items():
+            response[header] = value
+
+        # Add CSP header
+        csp_value = self._generate_csp_header(request)
+        if csp_value:
+            response['Content-Security-Policy'] = csp_value
+
+        # Add Report-Only CSP in development
+        if settings.DEBUG and getattr(settings, 'CSP_REPORT_ONLY', False):
+            response['Content-Security-Policy-Report-Only'] = csp_value
+
+        # Add feature policy
+        feature_policy = self._generate_feature_policy()
+        if feature_policy:
+            response['Feature-Policy'] = feature_policy
+
+        # Add permissions policy
+        permissions_policy = self._generate_permissions_policy()
+        if permissions_policy:
+            response['Permissions-Policy'] = permissions_policy
+
+        # Add HSTS header in production
+        if not settings.DEBUG and getattr(settings, 'SECURE_HSTS_SECONDS', 0):
+            response['Strict-Transport-Security'] = self._generate_hsts_header()
+
+        return response
+
+    def _should_skip_headers(self, request) -> bool:
+        """
+        Determine if security headers should be skipped for this request.
+        """
+        # Skip for static files
+        if request.path.startswith(settings.STATIC_URL):
+            return True
+
+        # Skip for media files
+        if request.path.startswith(settings.MEDIA_URL):
+            return True
+
+        # Skip for health checks
+        if request.path.startswith('/health/'):
+            return True
+
+        # Skip for metrics
+        if request.path.startswith('/metrics/'):
+            return True
+
+        return False
+
+    def _get_default_csp_config(self) -> Dict[str, List[str]]:
+        """
+        Get default CSP configuration.
+        """
+        return {
+            'default-src': ["'self'"],
+            'script-src': ["'self'", "'unsafe-inline'", "'unsafe-eval'"],
+            'style-src': ["'self'", "'unsafe-inline'"],
+            'img-src': ["'self'", "data:", "https:"],
+            'font-src': ["'self'", "data:"],
+            'connect-src': ["'self'"],
+            'media-src': ["'self'"],
+            'object-src': ["'none'"],
+            'frame-src': ["'self'"],
+            'frame-ancestors': ["'self'"],
+            'form-action': ["'self'"],
+            'base-uri': ["'self'"],
+            'manifest-src': ["'self'"],
+            'worker-src': ["'self'"],
+            'child-src': ["'self'"],
+            'prefetch-src': ["'self'"],
+            'require-trusted-types-for': ["'script'"],
+            'trusted-types': ["'default'"],
+            'upgrade-insecure-requests': [],
+            'block-all-mixed-content': [],
+            'report-uri': ['/csp-report-endpoint/'],
+            'report-to': ['csp-endpoint'],
+        }
+
+    def _get_default_security_headers(self) -> Dict[str, str]:
+        """
+        Get default security headers.
+        """
+        return {
+            'X-Content-Type-Options': 'nosniff',
+            'X-Frame-Options': 'DENY',
+            'X-XSS-Protection': '1; mode=block',
+            'Referrer-Policy': 'strict-origin-when-cross-origin',
+            'X-Permitted-Cross-Domain-Policies': 'none',
+            'Clear-Site-Data': '"cache", "cookies", "storage"',
+            'Cross-Origin-Opener-Policy': 'same-origin',
+            'Cross-Origin-Embedder-Policy': 'require-corp',
+            'Cross-Origin-Resource-Policy': 'same-origin',
+        }
+
+    def _generate_csp_header(self, request) -> str:
+        """
+        Generate CSP header value based on configuration.
+        """
+        directives = []
+
+        for directive, sources in self.csp_config.items():
+            if sources:
+                # Add nonce for script and style directives
+                if directive in ['script-src', 'style-src'] and "'unsafe-inline'" in sources:
+                    nonce = self.nonce_generator.get_nonce()
+                    sources.remove("'unsafe-inline'")
+                    sources.append(f"'nonce-{nonce}'")
+
+                # Join sources
+                source_list = ' '.join(sources)
+                directives.append(f"{directive} {source_list}")
+
+        return '; '.join(directives)
+
+    def _generate_feature_policy(self) -> str:
+        """
+        Generate Feature Policy header.
+        """
+        policies = [
+            'camera none',
+            'microphone none',
+            'geolocation none',
+            'payment none',
+            'usb none',
+            'magnetometer none',
+            'gyroscope none',
+            'accelerometer none',
+            'fullscreen self',
+            'document-domain none',
+            'sync-xhr self',
+            'usb none',
+        ]
+
+        return ', '.join(policies)
+
+    def _generate_permissions_policy(self) -> str:
+        """
+        Generate Permissions Policy header.
+        """
+        policies = [
+            'camera=()',
+            'microphone=()',
+            'geolocation=()',
+            'payment=()',
+            'usb=()',
+            'magnetometer=()',
+            'gyroscope=()',
+            'accelerometer=()',
+            'fullscreen=(self)',
+            'document-domain=()',
+            'sync-xhr=(self)',
+            'usb=()',
+        ]
+
+        return ', '.join(policies)
+
+    def _generate_hsts_header(self) -> str:
+        """
+        Generate HSTS header.
+        """
+        max_age = getattr(settings, 'SECURE_HSTS_SECONDS', 31536000)
+        include_subdomains = getattr(settings, 'SECURE_HSTS_INCLUDE_SUBDOMAINS', True)
+        preload = getattr(settings, 'SECURE_HSTS_PRELOAD', False)
+
+        header = f'max-age={max_age}'
+
+        if include_subdomains:
+            header += '; includeSubDomains'
+
+        if preload:
+            header += '; preload'
+
+        return header
+
+
+class CSPNonceGenerator:
+    """
+    Generator for CSP nonces.
+    """
+
+    def __init__(self):
+        self._nonces = set()
+        self._max_nonces = 1000  # Prevent memory leaks
+
+    def get_nonce(self) -> str:
+        """
+        Generate a new nonce.
+        """
+        import secrets
+
+        # Clean up old nonces if we have too many
+        if len(self._nonces) > self._max_nonces:
+            self._nonces.clear()
+
+        # Generate new nonce
+        nonce = secrets.token_urlsafe(16)
+        self._nonces.add(nonce)
+
+        return nonce
+
+    def is_valid_nonce(self, nonce: str) -> bool:
+        """
+        Check if a nonce is valid.
+        """
+        return nonce in self._nonces
+
+    def clear_nonces(self):
+        """
+        Clear all nonces.
+        """
+        self._nonces.clear()
+
+
+class CSPReportHandler:
+    """
+    Handler for CSP violation reports.
+    """
+
+    def __init__(self):
+        self.logger = logging.getLogger('security.csp')
+
+    def handle_report(self, report_data: Dict):
+        """
+        Handle CSP violation report.
+        """
+        try:
+            # Log the violation
+            self.logger.warning(
+                f"CSP Violation: {report_data.get('document-uri')} - "
+                f"{report_data.get('violated-directive')} - "
+                f"{report_data.get('blocked-uri')}"
+            )
+
+            # Send to monitoring system
+            self._send_to_monitoring(report_data)
+
+            # Store for analysis
+            self._store_violation(report_data)
+
+        except Exception as e:
+            self.logger.error(f"Error handling CSP report: {e}")
+
+    def _send_to_monitoring(self, report_data: Dict):
+        """
+        Send violation report to monitoring system.
+        """
+        try:
+            from monitoring.alerts import alert_manager
+            from monitoring.alerts import Alert, AlertSeverity, AlertCategory
+
+            alert = Alert(
+                title="CSP Violation",
+                description=f"CSP violation detected: {report_data.get('violated-directive')}",
+                severity=AlertSeverity.WARNING,
+                category=AlertCategory.SECURITY,
+                metadata={
+                    'document_uri': report_data.get('document-uri'),
+                    'violated_directive': report_data.get('violated-directive'),
+                    'blocked_uri': report_data.get('blocked-uri'),
+                    'line_number': report_data.get('line-number'),
+                    'column_number': report_data.get('column-number'),
+                }
+            )
+
+            alert_manager.trigger_alert(alert)
+
+        except Exception as e:
+            self.logger.error(f"Error sending CSP report to monitoring: {e}")
+
+    def _store_violation(self, report_data: Dict):
+        """
+        Store violation for analysis.
+        """
+        try:
+            from django.core.cache import cache
+
+            # Store recent violations
+            cache_key = f"csp_violations_{report_data.get('document-uri', 'unknown')}"
+            violations = cache.get(cache_key, [])
+
+            violations.append({
+                'timestamp': report_data.get('timestamp'),
+                'violated_directive': report_data.get('violated-directive'),
+                'blocked_uri': report_data.get('blocked-uri'),
+                'line_number': report_data.get('line-number'),
+                'column_number': report_data.get('column-number'),
+            })
+
+            # Keep only last 100 violations
+            if len(violations) > 100:
+                violations = violations[-100:]
+
+            cache.set(cache_key, violations, timeout=86400)  # 24 hours
+
+        except Exception as e:
+            self.logger.error(f"Error storing CSP violation: {e}")
+
+
+class SecurityHeaderValidator:
+    """
+    Validator for security headers.
+    """
+
+    def __init__(self):
+        self.logger = logging.getLogger('security.headers')
+
+    def validate_headers(self, response: HttpResponse) -> Dict[str, bool]:
+        """
+        Validate security headers in response.
+        """
+        results = {}
+
+        # Validate CSP header
+        results['csp'] = self._validate_csp_header(response)
+
+        # Validate HSTS header
+        results['hsts'] = self._validate_hsts_header(response)
+
+        # Validate other security headers
+        results['x_content_type_options'] = self._validate_x_content_type_options(response)
+        results['x_frame_options'] = self._validate_x_frame_options(response)
+        results['x_xss_protection'] = self._validate_x_xss_protection(response)
+        results['referrer_policy'] = self._validate_referrer_policy(response)
+
+        return results
+
+    def _validate_csp_header(self, response: HttpResponse) -> bool:
+        """
+        Validate CSP header.
+        """
+        csp_header = response.get('Content-Security-Policy')
+        if not csp_header:
+            self.logger.warning("Missing CSP header")
+            return False
+
+        # Check for required directives
+        required_directives = ['default-src', 'script-src', 'style-src']
+        for directive in required_directives:
+            if f"{directive} " not in csp_header:
+                self.logger.warning(f"Missing required CSP directive: {directive}")
+                return False
+
+        return True
+
+    def _validate_hsts_header(self, response: HttpResponse) -> bool:
+        """
+        Validate HSTS header.
+        """
+        hsts_header = response.get('Strict-Transport-Security')
+        if not hsts_header:
+            self.logger.warning("Missing HSTS header")
+            return False
+
+        # Check for max-age
+        if 'max-age=' not in hsts_header:
+            self.logger.warning("HSTS header missing max-age")
+            return False
+
+        return True
+
+    def _validate_x_content_type_options(self, response: HttpResponse) -> bool:
+        """
+        Validate X-Content-Type-Options header.
+        """
+        header = response.get('X-Content-Type-Options')
+        if header != 'nosniff':
+            self.logger.warning("Invalid X-Content-Type-Options header")
+            return False
+
+        return True
+
+    def _validate_x_frame_options(self, response: HttpResponse) -> bool:
+        """
+        Validate X-Frame-Options header.
+        """
+        header = response.get('X-Frame-Options')
+        if header not in ['DENY', 'SAMEORIGIN']:
+            self.logger.warning("Invalid X-Frame-Options header")
+            return False
+
+        return True
+
+    def _validate_x_xss_protection(self, response: HttpResponse) -> bool:
+        """
+        Validate X-XSS-Protection header.
+        """
+        header = response.get('X-XSS-Protection')
+        if header != '1; mode=block':
+            self.logger.warning("Invalid X-XSS-Protection header")
+            return False
+
+        return True
+
+    def _validate_referrer_policy(self, response: HttpResponse) -> bool:
+        """
+        Validate Referrer-Policy header.
+        """
+        header = response.get('Referrer-Policy')
+        valid_policies = [
+            'no-referrer',
+            'no-referrer-when-downgrade',
+            'origin',
+            'origin-when-cross-origin',
+            'same-origin',
+            'strict-origin',
+            'strict-origin-when-cross-origin',
+            'unsafe-url'
+        ]
+
+        if header not in valid_policies:
+            self.logger.warning("Invalid Referrer-Policy header")
+            return False
+
+        return True
+
+
+class SecurityHeaderMiddleware(SecurityHeadersMiddleware):
+    """
+    Enhanced security middleware with Malaysian-specific security considerations.
+    """
+
+    def __init__(self, get_response):
+        super().__init__(get_response)
+
+        # Initialize Malaysian-specific security headers
+        self.malaysian_headers = getattr(settings, 'MALAYSIAN_SECURITY_HEADERS', {
+            'X-Malaysian-Data-Protection': 'PDPA-Compliant',
+            'X-Malaysian-Privacy-Policy': '/privacy-policy/',
+            'X-Malaysian-Contact': '/contact/',
+        })
+
+        # Initialize validator
+        self.validator = SecurityHeaderValidator()
+
+        # Initialize report handler
+        self.report_handler = CSPReportHandler()
+
+    def process_response(self, request, response):
+        """
+        Add security headers with Malaysian-specific considerations.
+        """
+        # Call parent method
+        response = super().process_response(request, response)
+
+        # Add Malaysian-specific headers
+        for header, value in self.malaysian_headers.items():
+            response[header] = value
+
+        # Validate headers
+        if settings.DEBUG:
+            validation_results = self.validator.validate_headers(response)
+            for header, is_valid in validation_results.items():
+                if not is_valid:
+                    self.validator.logger.warning(f"Invalid {header} header")
+
+        return response