[WIP] Add security headers middleware for enhanced protection

backend/core/security_middleware.py

@@ -0,0 +1,96 @@
+"""
+Security Headers Middleware for FastAPI
+
+This middleware adds security headers to all HTTP responses to protect against
+common web vulnerabilities including XSS, clickjacking, and MIME sniffing attacks.
+
+Security Headers Implemented:
+- Strict-Transport-Security (HSTS): Forces HTTPS connections
+- Content-Security-Policy (CSP): Prevents XSS by restricting resource loading
+- X-Frame-Options: Prevents clickjacking attacks
+- X-Content-Type-Options: Prevents MIME sniffing attacks
+- X-XSS-Protection: Legacy XSS protection for older browsers
+"""
+
+from fastapi import Request
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.types import ASGIApp
+
+
+class SecurityHeadersMiddleware(BaseHTTPMiddleware):
+    """
+    Middleware that adds security headers to all HTTP responses.
+
+    This middleware should be added to the FastAPI application to ensure
+    all responses include appropriate security headers that protect against
+    common web vulnerabilities.
+
+    Example:
+        app = FastAPI()
+        app.add_middleware(SecurityHeadersMiddleware)
+    """
+
+    def __init__(self, app: ASGIApp):
+        """
+        Initialize the security headers middleware.
+
+        Args:
+            app: The ASGI application
+        """
+        super().__init__(app)
+
+    async def dispatch(self, request: Request, call_next):
+        """
+        Process the request and add security headers to the response.
+
+        Args:
+            request: The incoming HTTP request
+            call_next: The next middleware or route handler
+
+        Returns:
+            Response with security headers added
+        """
+        response = await call_next(request)
+
+        # HSTS - HTTP Strict Transport Security
+        # Forces browsers to use HTTPS for all future connections to this domain
+        # max-age=31536000: Valid for 1 year (in seconds)
+        # includeSubDomains: Apply to all subdomains as well
+        response.headers["Strict-Transport-Security"] = (
+            "max-age=31536000; includeSubDomains"
+        )
+
+        # CSP - Content Security Policy
+        # Restricts which resources can be loaded to prevent XSS attacks
+        # default-src 'self': Only allow resources from same origin by default
+        # script-src 'self' 'unsafe-inline': Allow scripts from same origin and inline scripts (needed for some frameworks)
+        # style-src 'self' 'unsafe-inline': Allow styles from same origin and inline styles
+        # img-src 'self' data: https:: Allow images from same origin, data URIs, and HTTPS
+        # font-src 'self' data:: Allow fonts from same origin and data URIs
+        # connect-src 'self' ws: wss:: Allow connections to same origin and WebSocket connections
+        response.headers["Content-Security-Policy"] = (
+            "default-src 'self'; "
+            "script-src 'self' 'unsafe-inline'; "
+            "style-src 'self' 'unsafe-inline'; "
+            "img-src 'self' data: https:; "
+            "font-src 'self' data:; "
+            "connect-src 'self' ws: wss:;"
+        )
+
+        # X-Frame-Options - Clickjacking Protection
+        # DENY: Prevent the page from being displayed in a frame/iframe
+        # This protects against clickjacking attacks where malicious sites
+        # embed your app in an invisible iframe to trick users
+        response.headers["X-Frame-Options"] = "DENY"
+
+        # X-Content-Type-Options - MIME Sniffing Protection
+        # nosniff: Prevents browsers from MIME-sniffing responses away from the declared content-type
+        # This prevents browsers from interpreting files as a different MIME type than declared
+        response.headers["X-Content-Type-Options"] = "nosniff"
+
+        # X-XSS-Protection - Legacy XSS Protection
+        # 1; mode=block: Enable XSS filtering and block the page if attack is detected
+        # Note: This is a legacy header for older browsers. Modern browsers use CSP instead.
+        response.headers["X-XSS-Protection"] = "1; mode=block"
+
+        return response

backend/main.py

@@ -8,6 +8,7 @@
 from backend.api.v1 import api_router
 from backend.websocket.manager import manager
 from backend.core.security import decode_token
+from backend.core.security_middleware import SecurityHeadersMiddleware
 import logging
 
 logger = logging.getLogger(__name__)
@@ -27,6 +28,9 @@
     allow_headers=["*"],
 )
 
+# Security headers middleware
+app.add_middleware(SecurityHeadersMiddleware)
+
 # Include API v1 router
 app.include_router(api_router, prefix=settings.API_V1_PREFIX)
 

backend/tests/core/test_security_middleware.py

@@ -0,0 +1,198 @@
+"""
+Tests for Security Headers Middleware
+
+Validates that all security headers are correctly added to HTTP responses
+and that the middleware doesn't break existing application functionality.
+"""
+
+from fastapi.testclient import TestClient
+
+from backend.models.organization import Organization
+
+
+class TestSecurityHeadersMiddleware:
+    """Test suite for security headers middleware"""
+
+    def test_health_endpoint_has_all_security_headers(self, client: TestClient):
+        """Test that health check endpoint includes all security headers"""
+        # Act
+        response = client.get("/health")
+
+        # Assert
+        assert response.status_code == 200
+
+        # Check all security headers are present
+        assert "Strict-Transport-Security" in response.headers
+        assert "Content-Security-Policy" in response.headers
+        assert "X-Frame-Options" in response.headers
+        assert "X-Content-Type-Options" in response.headers
+        assert "X-XSS-Protection" in response.headers
+
+    def test_hsts_header_has_correct_value(self, client: TestClient):
+        """Test HSTS header enforces HTTPS with 1 year max-age"""
+        # Act
+        response = client.get("/health")
+
+        # Assert
+        hsts_header = response.headers.get("Strict-Transport-Security")
+        assert hsts_header == "max-age=31536000; includeSubDomains"
+
+        # Verify it includes required directives
+        assert "max-age=31536000" in hsts_header  # 1 year in seconds
+        assert "includeSubDomains" in hsts_header
+
+    def test_csp_header_has_correct_value(self, client: TestClient):
+        """Test CSP header restricts resource loading appropriately"""
+        # Act
+        response = client.get("/health")
+
+        # Assert
+        csp_header = response.headers.get("Content-Security-Policy")
+
+        # Verify all required CSP directives are present
+        assert "default-src 'self'" in csp_header
+        assert "script-src 'self' 'unsafe-inline'" in csp_header
+        assert "style-src 'self' 'unsafe-inline'" in csp_header
+        assert "img-src 'self' data: https:" in csp_header
+        assert "font-src 'self' data:" in csp_header
+        assert "connect-src 'self' ws: wss:" in csp_header
+
+    def test_x_frame_options_prevents_clickjacking(self, client: TestClient):
+        """Test X-Frame-Options header prevents framing"""
+        # Act
+        response = client.get("/health")
+
+        # Assert
+        assert response.headers.get("X-Frame-Options") == "DENY"
+
+    def test_x_content_type_options_prevents_mime_sniffing(self, client: TestClient):
+        """Test X-Content-Type-Options prevents MIME sniffing"""
+        # Act
+        response = client.get("/health")
+
+        # Assert
+        assert response.headers.get("X-Content-Type-Options") == "nosniff"
+
+    def test_x_xss_protection_header_is_present(self, client: TestClient):
+        """Test X-XSS-Protection header is present for legacy browsers"""
+        # Act
+        response = client.get("/health")
+
+        # Assert
+        assert response.headers.get("X-XSS-Protection") == "1; mode=block"
+
+    def test_root_endpoint_has_security_headers(self, client: TestClient):
+        """Test that root endpoint also has security headers"""
+        # Act
+        response = client.get("/")
+
+        # Assert
+        assert response.status_code == 200
+        assert "Strict-Transport-Security" in response.headers
+        assert "Content-Security-Policy" in response.headers
+        assert "X-Frame-Options" in response.headers
+
+    def test_api_endpoints_have_security_headers(
+        self, client: TestClient, sample_organization: Organization
+    ):
+        """Test that API endpoints have security headers"""
+        # Arrange
+        payload = {
+            "email": "sectest@test.com",
+            "name": "Security Test User",
+            "password": "SecurePass123!",
+            "organization_slug": sample_organization.slug,
+        }
+
+        # Act
+        response = client.post("/api/v1/auth/register", json=payload)
+
+        # Assert
+        assert response.status_code == 201
+        assert "Strict-Transport-Security" in response.headers
+        assert "Content-Security-Policy" in response.headers
+        assert "X-Frame-Options" in response.headers
+        assert "X-Content-Type-Options" in response.headers
+        assert "X-XSS-Protection" in response.headers
+
+    def test_error_responses_have_security_headers(self, client: TestClient):
+        """Test that error responses also include security headers"""
+        # Act - Request non-existent endpoint
+        response = client.get("/api/v1/nonexistent")
+
+        # Assert
+        assert response.status_code == 404
+        assert "Strict-Transport-Security" in response.headers
+        assert "Content-Security-Policy" in response.headers
+        assert "X-Frame-Options" in response.headers
+
+    def test_authenticated_endpoints_have_security_headers(
+        self, client: TestClient, auth_headers: dict
+    ):
+        """Test that authenticated endpoints include security headers"""
+        # Act - Try to access authenticated endpoint (will fail but should have headers)
+        response = client.get("/api/v1/users/me", headers=auth_headers)
+
+        # Assert - Response should have security headers regardless of success
+        assert "Strict-Transport-Security" in response.headers
+        assert "Content-Security-Policy" in response.headers
+
+    def test_security_headers_do_not_break_json_responses(
+        self, client: TestClient, sample_organization: Organization
+    ):
+        """Test that security headers don't interfere with JSON response parsing"""
+        # Arrange
+        payload = {
+            "email": "jsontest@test.com",
+            "name": "JSON Test User",
+            "password": "JsonPass123!",
+            "organization_slug": sample_organization.slug,
+        }
+
+        # Act
+        response = client.post("/api/v1/auth/register", json=payload)
+
+        # Assert - Response should be valid JSON with security headers
+        assert response.status_code == 201
+        data = response.json()  # Should parse without error
+        assert data["email"] == "jsontest@test.com"
+        assert "Strict-Transport-Security" in response.headers
+
+    def test_all_security_headers_present_together(self, client: TestClient):
+        """Test that all 5 security headers are present in every response"""
+        # Act
+        response = client.get("/health")
+
+        # Assert - Count security headers
+        security_headers = [
+            "Strict-Transport-Security",
+            "Content-Security-Policy",
+            "X-Frame-Options",
+            "X-Content-Type-Options",
+            "X-XSS-Protection",
+        ]
+
+        for header in security_headers:
+            assert header in response.headers, f"Missing security header: {header}"
+
+    def test_middleware_processes_requests_correctly(self, client: TestClient):
+        """Test that middleware doesn't break request processing"""
+        # Act
+        response = client.get("/health")
+
+        # Assert
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "healthy"
+        # Verify security headers are added (count the 5 specific headers)
+        security_headers = [
+            "Strict-Transport-Security",
+            "Content-Security-Policy",
+            "X-Frame-Options",
+            "X-Content-Type-Options",
+            "X-XSS-Protection",
+        ]
+        present_headers = [h for h in security_headers if h in response.headers]
+        assert (
+            len(present_headers) == 5
+        ), f"Expected 5 security headers, found {len(present_headers)}"