learningcircuit--local-deep-research/tests/chat/test_chat_csrf.py

# allow: no-sut-import — black-box HTTP test; drives real routes through the Flask test client
"""
CSRF protection tests for chat API endpoints.

Two complementary suites:

* ``TestCSRFProtection`` runs against the default no-CSRF ``app`` /
  ``authenticated_client`` fixtures and verifies session-cookie
  authentication (401 on missing cookie).
* ``TestCSRFTokenEnforcement`` runs against the opt-in
  ``app_with_csrf`` / ``csrf_authenticated_client`` fixtures and
  verifies that Flask-WTF enforces the CSRF token on mutating
  chat-API endpoints (400 on missing/invalid token).

Both are required for full coverage: the first proves auth gating,
the second proves CSRF gating. With CSRF globally disabled in the
default fixture, the second suite is the only thing that catches a
regression silently disabling CSRF middleware.
"""

import json


class TestCSRFProtection:
    """
    Tests verifying CSRF protection on API endpoints.

    Note: Flask's API endpoints typically rely on session-based authentication
    rather than CSRF tokens for API routes (JSON requests). CSRF protection
    is more relevant for form-based submissions. These tests document the
    current behavior and verify API authentication is working correctly.
    """

    def test_create_session_without_session_cookie_fails(self, app):
        """Test that creating a session without valid session cookie fails."""
        with app.test_client() as client:
            # Make request without any session/auth
            response = client.post(
                "/api/chat/sessions",
                json={"initial_query": "Test"},
                content_type="application/json",
            )
            # Should require authentication
            assert response.status_code == 401

    def test_send_message_without_session_cookie_fails(self, app):
        """Test that sending a message without valid session cookie fails."""
        with app.test_client() as client:
            response = client.post(
                "/api/chat/sessions/fake-session-id/messages",
                json={"content": "Test message"},
                content_type="application/json",
            )
            assert response.status_code == 401

    def test_update_session_without_session_cookie_fails(self, app):
        """Test that updating a session without valid session cookie fails."""
        with app.test_client() as client:
            response = client.patch(
                "/api/chat/sessions/fake-session-id",
                json={"title": "New title"},
                content_type="application/json",
            )
            assert response.status_code == 401

    def test_delete_session_without_session_cookie_fails(self, app):
        """Test that deleting a session without valid session cookie fails."""
        with app.test_client() as client:
            response = client.delete("/api/chat/sessions/fake-session-id")
            assert response.status_code == 401

    def test_cross_origin_request_with_cookies_handled(
        self, authenticated_client
    ):
        """
        Test that cross-origin requests are handled appropriately.

        In a real attack, the browser would include cookies but the origin
        header would indicate the request is from a different site.
        """
        # Simulate a cross-origin request by adding Origin header
        # Note: This tests the server's CORS configuration

        # This test documents behavior - actual CORS enforcement depends
        # on Flask-CORS or similar middleware configuration
        response = authenticated_client.get(
            "/api/chat/sessions",
            headers={"Origin": "https://malicious-site.com"},
        )

        # The API should still respond (CORS is handled at a different layer)
        # But no sensitive data should leak to the malicious origin
        # This is really a CORS test, not CSRF per se
        assert response.status_code == 200

    def test_api_requires_json_content_type(self, authenticated_client):
        """Test that API endpoints require proper JSON content type for POST/PATCH."""
        # Create a session first
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Try to send data without JSON content type
        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            data="content=Test message",  # Form data, not JSON
            content_type="application/x-www-form-urlencoded",
        )

        # The endpoint expects JSON, so this should fail.
        # 400 (bad request) or 415 (unsupported media type) are acceptable;
        # 500 here would be an unhandled-exception regression.
        assert response.status_code in [400, 415]

    def test_state_changing_endpoints_reject_get_requests(
        self, authenticated_client
    ):
        """
        Test that state-changing endpoints don't accept GET requests.

        This prevents CSRF via simple <img> or <link> tags.
        """
        # Try GET on POST endpoint (create session)
        response = authenticated_client.get("/api/chat/sessions")
        # GET is allowed here because this is also the list endpoint
        assert response.status_code == 200

        # Create a session for further tests
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Try GET on message sending endpoint
        response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        # This is actually the get messages endpoint, so it's allowed
        assert response.status_code == 200

    def test_session_cookie_required_for_api_access(
        self, app, authenticated_client
    ):
        """Test that a valid session cookie is required for API access."""
        # First, make a successful request to get a session ID
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Now try to access without the session cookie using a fresh client
        with app.test_client() as fresh_client:
            response = fresh_client.get(f"/api/chat/sessions/{session_id}")
            # Should fail because no session cookie
            assert response.status_code == 401


class TestAPISecurityHeaders:
    """Tests for security headers on API responses."""

    def test_api_response_content_type(self, authenticated_client):
        """Test that API responses have correct Content-Type."""
        response = authenticated_client.get("/api/chat/sessions")
        assert response.content_type.startswith("application/json")

    def test_successful_api_response_format(self, authenticated_client):
        """Test that successful API responses follow expected format."""
        response = authenticated_client.get("/api/chat/sessions")
        data = json.loads(response.data)

        # Should have success field
        assert "success" in data
        assert data["success"] is True

    def test_error_api_response_format(self, authenticated_client):
        """Test that error API responses follow expected format."""
        response = authenticated_client.get(
            "/api/chat/sessions/non-existent-id"
        )
        data = json.loads(response.data)

        # Should have success field
        assert "success" in data
        assert data["success"] is False
        # Should have error message
        assert "error" in data


class TestAuthenticationState:
    """Tests verifying authentication state handling."""

    def test_expired_session_handled_gracefully(self, app, temp_data_dir):
        """Test that expired/invalid session cookies are handled gracefully."""
        # This simulates what happens when a user's session expires
        with app.test_client() as client:
            # Manually set a session cookie that won't be valid
            with client.session_transaction() as sess:
                sess["username"] = "nonexistent_user"
                sess["session_id"] = "fake_session"

            # Try to make an API call
            response = client.get("/api/chat/sessions")

            # Should get an auth error (401), not a server error (500)
            assert response.status_code == 401

    def test_concurrent_requests_maintain_auth(self, authenticated_client):
        """Test that concurrent requests maintain proper authentication."""
        # Create multiple sessions rapidly
        session_ids = []
        for i in range(3):
            response = authenticated_client.post(
                "/api/chat/sessions",
                json={"initial_query": f"Query {i}"},
                content_type="application/json",
            )
            assert response.status_code == 200
            data = json.loads(response.data)
            session_ids.append(data["session_id"])

        # Verify all sessions were created for the same user
        response = authenticated_client.get("/api/chat/sessions")
        data = json.loads(response.data)

        # All created sessions should be in the list
        listed_ids = [s["id"] for s in data["sessions"]]
        for sid in session_ids:
            assert sid in listed_ids


class TestCSRFTokenEnforcement:
    """CSRF-token enforcement tests against the CSRF-enabled app fixture.

    These run against ``csrf_authenticated_client`` (uses
    ``app_with_csrf`` from tests/conftest.py), where Flask-WTF's
    CSRF middleware is active. They prove that mutating chat-API
    endpoints reject requests whose ``X-CSRFToken`` header is missing
    or wrong, and accept requests bearing the valid session-bound
    token. Without these, the suite cannot detect a regression that
    silently disables CSRF protection.
    """

    def test_create_session_without_csrf_token_rejected(
        self, csrf_authenticated_client
    ):
        client, _token = csrf_authenticated_client
        response = client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
        )
        # Flask-WTF returns 400 on missing token (default behavior).
        assert response.status_code == 400

    def test_create_session_with_invalid_csrf_token_rejected(
        self, csrf_authenticated_client
    ):
        client, _token = csrf_authenticated_client
        response = client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
            headers={"X-CSRFToken": "definitely-not-a-real-token"},
        )
        assert response.status_code == 400

    def test_create_session_with_valid_csrf_token_accepted(
        self, csrf_authenticated_client
    ):
        client, token = csrf_authenticated_client
        response = client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
            headers={"X-CSRFToken": token},
        )
        # Sanity check: the fixture is not accidentally rejecting valid requests.
        assert response.status_code == 200, (
            f"valid-token request was rejected ({response.status_code}); "
            f"the fixture or the route is broken: "
            f"{response.data.decode()[:300]}"
        )

    def test_send_message_without_csrf_token_rejected(
        self, csrf_authenticated_client
    ):
        client, token = csrf_authenticated_client

        # Create a session via the valid-token path so we have an id.
        create_resp = client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
            headers={"X-CSRFToken": token},
        )
        assert create_resp.status_code == 200, (
            f"setup failed: {create_resp.status_code} "
            f"{create_resp.data.decode()[:300]}"
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # POST a message WITHOUT the CSRF token — must be rejected.
        response = client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "Test message"},
            content_type="application/json",
        )
        assert response.status_code == 400

    def test_delete_session_without_csrf_token_rejected(
        self, csrf_authenticated_client
    ):
        client, token = csrf_authenticated_client

        create_resp = client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
            headers={"X-CSRFToken": token},
        )
        assert create_resp.status_code == 200
        session_id = json.loads(create_resp.data)["session_id"]

        # DELETE without token — must be rejected.
        response = client.delete(f"/api/chat/sessions/{session_id}")
        assert response.status_code == 400

    def test_update_session_without_csrf_token_rejected(
        self, csrf_authenticated_client
    ):
        """PATCH /api/chat/sessions/<id> is a state-changing endpoint;
        a regression that exempted it from CSRF would silently strip
        protection from session renames. Mirror the DELETE coverage."""
        client, token = csrf_authenticated_client

        create_resp = client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
            headers={"X-CSRFToken": token},
        )
        assert create_resp.status_code == 200
        session_id = json.loads(create_resp.data)["session_id"]

        # PATCH without token — must be rejected.
        response = client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"title": "Renamed"},
            content_type="application/json",
        )
        assert response.status_code == 400

    def test_generate_title_without_csrf_token_rejected(
        self, csrf_authenticated_client
    ):
        """POST /api/chat/sessions/<id>/generate-title regenerates a
        durable session attribute via an LLM round-trip; same CSRF
        risk surface as PATCH/update."""
        client, token = csrf_authenticated_client

        create_resp = client.post(
            "/api/chat/sessions",
            json={"initial_query": "Test"},
            content_type="application/json",
            headers={"X-CSRFToken": token},
        )
        assert create_resp.status_code == 200
        session_id = json.loads(create_resp.data)["session_id"]

        # POST generate-title without token — must be rejected.
        response = client.post(
            f"/api/chat/sessions/{session_id}/generate-title",
            json={"query": "What is X?"},
            content_type="application/json",
        )
        assert response.status_code == 400