learningcircuit--local-deep-research/tests/security/test_security_headers.py

"""Test HTTP security headers."""

import pytest


class TestSecurityHeaders:
    """Test HTTP security headers are properly set."""

    @pytest.fixture
    def test_endpoint(self):
        """Return a test endpoint to check headers."""
        return "/"

    def test_x_frame_options_header(self, client, test_endpoint):
        """Test X-Frame-Options header is set correctly."""
        response = client.get(test_endpoint)
        assert "X-Frame-Options" in response.headers
        assert response.headers["X-Frame-Options"] == "SAMEORIGIN"

    def test_x_content_type_options_header(self, client, test_endpoint):
        """Test X-Content-Type-Options header is set correctly."""
        response = client.get(test_endpoint)
        assert "X-Content-Type-Options" in response.headers
        assert response.headers["X-Content-Type-Options"] == "nosniff"

    def test_referrer_policy_header(self, client, test_endpoint):
        """Test Referrer-Policy header is set correctly."""
        response = client.get(test_endpoint)
        assert "Referrer-Policy" in response.headers
        assert (
            response.headers["Referrer-Policy"]
            == "strict-origin-when-cross-origin"
        )

    def test_permissions_policy_header(self, client, test_endpoint):
        """Test Permissions-Policy header is set correctly."""
        response = client.get(test_endpoint)
        assert "Permissions-Policy" in response.headers
        permissions = response.headers["Permissions-Policy"]
        assert "geolocation=()" in permissions
        assert "microphone=()" in permissions
        assert "camera=()" in permissions

    def test_content_security_policy_header(self, client, test_endpoint):
        """Test Content-Security-Policy header is set correctly."""
        response = client.get(test_endpoint)
        assert "Content-Security-Policy" in response.headers
        csp = response.headers["Content-Security-Policy"]
        # Verify key CSP directives are present
        assert "default-src 'self'" in csp
        assert "script-src" in csp
        assert "style-src" in csp

    def test_hsts_header_not_set_for_http(self, client, test_endpoint):
        """Test HSTS header is not set for HTTP requests."""
        # In test environment, requests are HTTP by default
        response = client.get(test_endpoint)
        # HSTS should NOT be set for non-HTTPS requests
        assert "Strict-Transport-Security" not in response.headers

    def test_hsts_header_set_for_https(self, app, test_endpoint):
        """Test HSTS header is set correctly for HTTPS requests."""
        # Configure app to think we're using HTTPS
        app.config["PREFERRED_URL_SCHEME"] = "https"

        with app.test_client() as https_client:
            response = https_client.get(
                test_endpoint, environ_base={"wsgi.url_scheme": "https"}
            )
            assert "Strict-Transport-Security" in response.headers
            hsts = response.headers["Strict-Transport-Security"]
            assert "max-age=" in hsts
            assert "includeSubDomains" in hsts


class TestSecurityHeadersOnAPIEndpoints:
    """Test security headers are set on API endpoints."""

    def test_security_headers_on_api_endpoint(self, client):
        """Test security headers are present on API endpoints."""
        # Test a health check endpoint that should exist
        response = client.get("/api/health")

        # Security headers should be present even on API endpoints
        assert "X-Frame-Options" in response.headers
        assert "X-Content-Type-Options" in response.headers
        assert "Referrer-Policy" in response.headers

    def test_cors_and_security_headers_coexist(self, client):
        """Test CORS and security headers can coexist on API endpoints."""
        # API endpoints should have both CORS and security headers
        response = client.get("/api/health")

        # CORS headers (if applicable to this endpoint)
        # Note: CORS headers might only be set for certain API routes

        # Security headers should always be present
        assert "X-Frame-Options" in response.headers
        assert "X-Content-Type-Options" in response.headers


class TestSecurityHeadersComprehensive:
    """Comprehensive security header validation."""

    def test_all_critical_security_headers_present(self, client):
        """Test all critical security headers are present."""
        response = client.get("/")

        critical_headers = [
            "Content-Security-Policy",
            "X-Frame-Options",
            "X-Content-Type-Options",
            "Referrer-Policy",
            "Permissions-Policy",
        ]

        for header in critical_headers:
            assert header in response.headers, (
                f"Missing critical header: {header}"
            )

    def test_security_headers_on_authenticated_routes(self, client):
        """Test security headers are present on authenticated routes."""
        # Test auth-related endpoints
        response = client.get("/auth/login")
        assert response.status_code == 200

        assert "X-Frame-Options" in response.headers
        assert "X-Content-Type-Options" in response.headers
        assert "Content-Security-Policy" in response.headers

    def test_security_headers_values_are_secure(self, client):
        """Test security header values follow best practices."""
        response = client.get("/")

        # X-Frame-Options should be DENY or SAMEORIGIN (not ALLOW-FROM)
        xfo = response.headers.get("X-Frame-Options", "")
        assert xfo in ["DENY", "SAMEORIGIN"]

        # X-Content-Type-Options should be nosniff
        assert response.headers.get("X-Content-Type-Options") == "nosniff"

        # Referrer-Policy should not be "unsafe-url" or "no-referrer-when-downgrade"
        rp = response.headers.get("Referrer-Policy", "")
        assert rp not in ["unsafe-url", "no-referrer-when-downgrade"]


class TestSecurityHeadersClass:
    """Tests for SecurityHeaders class directly."""

    def test_init_with_flask_app(self):
        """SecurityHeaders can be initialized with Flask app."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"

        sh = SecurityHeaders(app)

        assert sh.app == app

    def test_init_without_app(self):
        """SecurityHeaders can be initialized without app."""
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        sh = SecurityHeaders()

        # Should not have app attribute set
        assert not hasattr(sh, "app") or sh.app is None

    def test_init_app_later(self):
        """SecurityHeaders can use init_app for delayed initialization."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"

        sh = SecurityHeaders()
        sh.init_app(app)

        assert sh.app == app


class TestValidateCorsConfig:
    """Tests for _validate_cors_config method."""

    def test_cors_disabled_passes_validation(self):
        """Should pass validation when CORS is disabled."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"
        app.config["SECURITY_CORS_ENABLED"] = False

        # Should not raise
        sh = SecurityHeaders(app)
        assert sh is not None

    def test_credentials_with_wildcard_raises_error(self):
        """Should raise ValueError for credentials with wildcard origin."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"
        app.config["SECURITY_CORS_ENABLED"] = True
        app.config["SECURITY_CORS_ALLOWED_ORIGINS"] = "*"
        app.config["SECURITY_CORS_ALLOW_CREDENTIALS"] = True

        with pytest.raises(ValueError) as exc_info:
            SecurityHeaders(app)

        assert "Cannot use credentials with wildcard" in str(exc_info.value)

    def test_single_origin_with_credentials_passes(self):
        """Should pass for single origin with credentials."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"
        app.config["SECURITY_CORS_ENABLED"] = True
        app.config["SECURITY_CORS_ALLOWED_ORIGINS"] = "https://example.com"
        app.config["SECURITY_CORS_ALLOW_CREDENTIALS"] = True

        # Should not raise
        sh = SecurityHeaders(app)
        assert sh is not None


class TestCorsHeaders:
    """Tests for _add_cors_headers method."""

    def test_wildcard_origin_sets_star(self):
        """Should set Access-Control-Allow-Origin to * for wildcard."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"
        app.config["SECURITY_CORS_ALLOWED_ORIGINS"] = "*"
        app.config["SECURITY_CORS_ALLOW_CREDENTIALS"] = False

        sh = SecurityHeaders(app)

        with app.test_request_context("/api/test"):
            from flask import make_response

            response = make_response("test")
            response = sh._add_cors_headers(response)

            assert response.headers.get("Access-Control-Allow-Origin") == "*"

    def test_single_origin_sets_that_origin(self):
        """Should set configured single origin."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"
        app.config["SECURITY_CORS_ALLOWED_ORIGINS"] = "https://myapp.com"
        app.config["SECURITY_CORS_ALLOW_CREDENTIALS"] = False

        sh = SecurityHeaders(app)

        with app.test_request_context("/api/test"):
            from flask import make_response

            response = make_response("test")
            response = sh._add_cors_headers(response)

            assert (
                response.headers.get("Access-Control-Allow-Origin")
                == "https://myapp.com"
            )

    def test_multi_origin_reflects_request_origin(self):
        """Should reflect request origin if in whitelist."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"
        app.config["SECURITY_CORS_ALLOWED_ORIGINS"] = (
            "https://a.com,https://b.com"
        )
        app.config["SECURITY_CORS_ALLOW_CREDENTIALS"] = False

        sh = SecurityHeaders(app)

        with app.test_request_context(
            "/api/test", headers={"Origin": "https://b.com"}
        ):
            from flask import make_response

            response = make_response("test")
            response = sh._add_cors_headers(response)

            assert (
                response.headers.get("Access-Control-Allow-Origin")
                == "https://b.com"
            )


class TestCspPolicy:
    """Tests for get_csp_policy method."""

    def test_default_csp_includes_self(self):
        """Default CSP should include 'self' directive."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"

        sh = SecurityHeaders(app)
        csp = sh.get_csp_policy()

        assert "default-src 'self'" in csp

    def test_csp_includes_connect_src(self):
        """CSP should include connect-src directive."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"

        sh = SecurityHeaders(app)
        csp = sh.get_csp_policy()

        assert "connect-src" in csp

    def test_custom_connect_src_used(self):
        """Custom connect-src should be used when configured."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"
        app.config["SECURITY_CSP_CONNECT_SRC"] = "'self' https://api.mysite.com"

        sh = SecurityHeaders(app)
        csp = sh.get_csp_policy()

        assert "https://api.mysite.com" in csp

    def test_csp_includes_media_src(self):
        """CSP should include media-src directive."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"

        sh = SecurityHeaders(app)
        csp = sh.get_csp_policy()

        assert "media-src 'self'" in csp

    def test_csp_includes_child_src(self):
        """CSP should include child-src directive."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"

        sh = SecurityHeaders(app)
        csp = sh.get_csp_policy()

        assert "child-src 'self' blob:" in csp

    def test_csp_includes_manifest_src(self):
        """CSP should include manifest-src directive."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"

        sh = SecurityHeaders(app)
        csp = sh.get_csp_policy()

        assert "manifest-src 'self'" in csp

    def test_csp_includes_frame_ancestors(self):
        """CSP should include frame-ancestors directive (no fallback to default-src)."""
        from flask import Flask
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        app = Flask(__name__)
        app.config["SECRET_KEY"] = "test-key"

        sh = SecurityHeaders(app)
        csp = sh.get_csp_policy()

        # frame-ancestors is required because it doesn't fall back to default-src
        assert "frame-ancestors 'self'" in csp


class TestPermissionsPolicy:
    """Tests for get_permissions_policy method."""

    def test_disables_geolocation(self):
        """Should disable geolocation."""
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        policy = SecurityHeaders.get_permissions_policy()
        assert "geolocation=()" in policy

    def test_disables_camera(self):
        """Should disable camera."""
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        policy = SecurityHeaders.get_permissions_policy()
        assert "camera=()" in policy

    def test_disables_microphone(self):
        """Should disable microphone."""
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        policy = SecurityHeaders.get_permissions_policy()
        assert "microphone=()" in policy


class TestIsApiRoute:
    """Tests for _is_api_route static method."""

    def test_api_prefix_detected(self):
        """Should detect /api/ paths as API routes."""
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        assert SecurityHeaders._is_api_route("/api/v1/data") is True
        assert SecurityHeaders._is_api_route("/api/health") is True

    def test_research_api_detected(self):
        """Should detect /research/api/ paths as API routes."""
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        assert SecurityHeaders._is_api_route("/research/api/status") is True

    def test_non_api_not_detected(self):
        """Should not detect regular paths as API routes."""
        from local_deep_research.security.security_headers import (
            SecurityHeaders,
        )

        assert SecurityHeaders._is_api_route("/") is False
        assert SecurityHeaders._is_api_route("/dashboard") is False
        assert SecurityHeaders._is_api_route("/static/js/app.js") is False


class TestServerHeaderRemoval:
    """Test that Server header is removed from responses."""

    def test_server_header_not_present(self, client):
        """Server header should not leak version information."""
        response = client.get("/")
        assert "Server" not in response.headers

    def test_server_header_not_present_on_api(self, client):
        """Server header should not be present on API responses."""
        response = client.get("/api/health")
        assert "Server" not in response.headers

    def test_server_header_not_present_on_login(self, client):
        """Server header should not be present on login page."""
        response = client.get("/auth/login")
        assert "Server" not in response.headers


class TestCacheControlHeaders:
    """Test Cache-Control headers are properly set."""

    def test_cache_control_on_root(self, client):
        """Root page should have Cache-Control headers."""
        response = client.get("/")
        assert "Cache-Control" in response.headers
        cache_control = response.headers["Cache-Control"]
        assert "no-store" in cache_control
        assert "no-cache" in cache_control

    def test_cache_control_on_login(self, client):
        """Login page should have Cache-Control headers."""
        response = client.get("/auth/login")
        assert "Cache-Control" in response.headers
        cache_control = response.headers["Cache-Control"]
        assert "no-store" in cache_control
        assert "no-cache" in cache_control

    def test_cache_control_on_api(self, client):
        """API endpoints should have Cache-Control headers."""
        response = client.get("/api/health")
        assert "Cache-Control" in response.headers
        cache_control = response.headers["Cache-Control"]
        assert "no-store" in cache_control

    def test_pragma_header_present(self, client):
        """Pragma header should be present for HTTP/1.0 compatibility."""
        response = client.get("/")
        assert "Pragma" in response.headers
        assert response.headers["Pragma"] == "no-cache"

    def test_expires_header_present(self, client):
        """Expires header should be present."""
        response = client.get("/")
        assert "Expires" in response.headers
        assert response.headers["Expires"] == "0"


class TestContentTypeHeaders:
    """Test Content-Type headers are properly set for HTML routes."""

    def test_login_page_has_html_content_type(self, client):
        """Login page should have proper Content-Type."""
        response = client.get("/auth/login")
        content_type = response.headers.get("Content-Type", "")
        assert "text/html" in content_type

    def test_root_page_has_html_content_type(self, client):
        """Root page should have proper Content-Type when redirecting."""
        response = client.get("/")
        # Either redirect (302) or HTML response (200)
        if response.status_code == 200:
            content_type = response.headers.get("Content-Type", "")
            assert "text/html" in content_type

    def test_api_endpoint_has_json_content_type(self, client):
        """API endpoints should have proper Content-Type."""
        response = client.get("/api/health")
        content_type = response.headers.get("Content-Type", "")
        assert "application/json" in content_type


class TestSecurityHeadersResponseCoverage:
    """Pin that security headers fire across every response path Flask
    produces, not just successful HTML routes.

    SecurityHeaders registers app.after_request, which Flask invokes on
    every Response — including error handlers, streaming responses, and
    static file routes. These tests catch silent regressions if anyone
    refactors error handling, adds WSGI middleware that strips headers,
    or returns a raw response that somehow bypasses the after_request
    chain.
    """

    REQUIRED_HEADERS = (
        "Content-Security-Policy",
        "X-Frame-Options",
        "X-Content-Type-Options",
        "Referrer-Policy",
        "Permissions-Policy",
    )

    def _assert_required_headers(self, response):
        missing = [
            h for h in self.REQUIRED_HEADERS if h not in response.headers
        ]
        assert not missing, (
            f"Response (status={response.status_code}) missing security "
            f"headers: {missing}. Present headers: {list(response.headers.keys())}"
        )

    def test_headers_on_unknown_path_404(self, client):
        """404 from the global errorhandler must still carry headers."""
        response = client.get("/this-path-does-not-exist-zzz-12345")
        assert response.status_code == 404
        self._assert_required_headers(response)

    def test_headers_on_unknown_api_path_404(self, client):
        """404 on an /api/ path (JSON branch of the errorhandler)."""
        response = client.get("/api/does-not-exist-zzz-12345")
        assert response.status_code == 404
        self._assert_required_headers(response)

    def test_headers_on_favicon(self, client):
        """The custom /favicon.ico route uses send_from_directory."""
        response = client.get("/favicon.ico")
        # Status may be 200 (file exists) or 404 (file missing in test
        # fixture). Either path must have headers applied.
        self._assert_required_headers(response)

    def test_headers_on_unhandled_exception_500(self, app, client):
        """The 500 errorhandler returns make_response — headers must apply."""
        # TESTING=True normally propagates exceptions; turn that off so
        # the errorhandler runs.
        app.config["PROPAGATE_EXCEPTIONS"] = False

        @app.route("/_test_force_500")
        def _force_500():
            raise RuntimeError("forced exception for header coverage test")

        try:
            response = client.get("/_test_force_500")
            assert response.status_code == 500
            self._assert_required_headers(response)
        finally:
            app.config["PROPAGATE_EXCEPTIONS"] = True

    def test_headers_on_streaming_sse_response(self, app, client):
        """Streaming Response(stream_with_context(...)) — used by SSE
        endpoints in library_routes.py and rag_routes.py."""
        from flask import Response, stream_with_context

        @app.route("/_test_stream_sse")
        def _stream():
            def generate():
                yield "data: one\n\n"
                yield "data: two\n\n"

            return Response(
                stream_with_context(generate()), mimetype="text/event-stream"
            )

        response = client.get("/_test_stream_sse")
        assert response.status_code == 200
        self._assert_required_headers(response)

    def test_headers_on_abort_413_request_too_large(self, app, client):
        """Custom 413 errorhandler returns make_response."""
        from flask import abort

        @app.route("/_test_force_413")
        def _force_413():
            abort(413)

        response = client.get("/_test_force_413")
        assert response.status_code == 413
        self._assert_required_headers(response)