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

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

Tests cross-user access prevention and authentication requirements at API layer.
This complements the existing user isolation tests (test_chat_user_isolation.py)
which test at the service layer.
"""

import json


class TestSessionOwnershipSecurity:
    """
    Tests verifying session ownership is enforced at the API layer.

    Note: In LDR's architecture, each user has a separate encrypted database,
    so cross-user access is prevented at the database level. These tests verify
    that the API layer correctly uses the authenticated user's database.
    """

    def test_user_cannot_access_other_user_session_via_api(
        self, authenticated_client, second_user_client
    ):
        """
        Test that one user cannot access another user's session via API.

        Due to LDR's per-user database architecture, "other user's session"
        simply won't exist in the current user's database.
        """
        # Create a session as first user
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "User 1's research question"},
            content_type="application/json",
        )
        assert create_response.status_code == 200
        data = json.loads(create_response.data)
        session_id = data["session_id"]

        # Try to access the session as second user - should get 404 (not found in their DB)
        response = second_user_client.get(f"/api/chat/sessions/{session_id}")
        assert response.status_code == 404
        data = json.loads(response.data)
        assert data["success"] is False
        assert "not found" in data["error"].lower()

    def test_user_cannot_list_other_user_sessions(
        self, authenticated_client, second_user_client
    ):
        """Test that listing sessions only returns the current user's sessions."""
        # Create sessions as first user
        for i in range(3):
            authenticated_client.post(
                "/api/chat/sessions",
                json={"initial_query": f"User 1's query {i}"},
                content_type="application/json",
            )

        # List sessions as second user - should be empty or not contain first user's sessions
        response = second_user_client.get("/api/chat/sessions")
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data["success"] is True
        # Second user should have no sessions (or only their own)
        # The first user's 3 sessions should not appear
        sessions = data["sessions"]
        # Since second_user just registered, they shouldn't have any sessions
        assert len(sessions) == 0

    def test_user_cannot_send_message_to_other_user_session(
        self, authenticated_client, second_user_client
    ):
        """Test that one user cannot send messages to another user's session."""
        # Create session as first user
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "First user's topic"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Try to send message to that session as second user
        response = second_user_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": "Malicious message", "trigger_research": False},
            content_type="application/json",
        )
        # Should get 404 (session not found in their DB)
        assert response.status_code == 404
        data = json.loads(response.data)
        assert data["success"] is False

    def test_session_enumeration_returns_404_not_403(
        self, authenticated_client, second_user_client
    ):
        """
        Test that attempting to access another user's session returns 404, not 403.

        This prevents session enumeration attacks - a 403 would reveal that
        the session exists but belongs to another user, while 404 reveals nothing.
        """
        # Create session as first user
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Secret research"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Access as second user - should get 404, NOT 403
        response = second_user_client.get(f"/api/chat/sessions/{session_id}")
        # The critical assertion: 404 not 403
        assert response.status_code == 404
        # Double-check: response should not reveal the session exists
        data = json.loads(response.data)
        assert "forbidden" not in data.get("error", "").lower()
        assert "permission" not in data.get("error", "").lower()

    def test_user_cannot_update_other_user_session(
        self, authenticated_client, second_user_client
    ):
        """Test that one user cannot update another user's session."""
        # Create session as first user
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Original topic"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Try to update as second user
        hijack_response = second_user_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"title": "Hijacked title"},
            content_type="application/json",
        )
        # Session does not exist in second user's per-user DB, so the route
        # should report 404 — assert the status code so a regression that
        # quietly changes this contract is caught here, not just by the
        # downstream side-effect check.
        assert hijack_response.status_code == 404

        # Verify original user's session is unchanged
        verify_response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}"
        )
        verify_data = json.loads(verify_response.data)
        # Title should still be based on original query, not "Hijacked title"
        assert verify_data["session"]["title"] != "Hijacked title"

    def test_user_cannot_delete_other_user_session(
        self, authenticated_client, second_user_client
    ):
        """Test that one user cannot delete another user's session."""
        # Create session as first user
        create_response = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": "Important research"},
            content_type="application/json",
        )
        session_id = json.loads(create_response.data)["session_id"]

        # Try to delete as second user
        second_user_client.delete(f"/api/chat/sessions/{session_id}")
        # Operation either fails or has no effect (session not in their DB)

        # Verify original user's session still exists
        verify_response = authenticated_client.get(
            f"/api/chat/sessions/{session_id}"
        )
        assert verify_response.status_code == 200
        verify_data = json.loads(verify_response.data)
        assert verify_data["success"] is True
        # Session should still be active
        assert verify_data["session"]["status"] == "active"


class TestAuthenticationRequired:
    """Tests verifying authentication is required for chat endpoints."""

    def test_chat_page_redirects_without_login(self, app):
        """Test that chat page redirects to login when not authenticated."""
        with app.test_client() as client:
            response = client.get("/chat/")
            # Should redirect to login
            assert response.status_code == 302
            assert "/auth/login" in response.location

    def test_chat_page_with_session_id_redirects_without_login(self, app):
        """Test that chat page with session ID redirects to login when not authenticated."""
        with app.test_client() as client:
            response = client.get("/chat/some-session-id")
            assert response.status_code == 302
            assert "/auth/login" in response.location

    def test_create_session_api_requires_authentication(self, app):
        """Test that creating a session requires authentication."""
        with app.test_client() as client:
            response = client.post(
                "/api/chat/sessions",
                json={"initial_query": "Test query"},
                content_type="application/json",
            )
            assert response.status_code == 401
            data = json.loads(response.data)
            assert "error" in data
            assert "authentication" in data["error"].lower()

    def test_list_sessions_api_requires_authentication(self, app):
        """Test that listing sessions requires authentication."""
        with app.test_client() as client:
            response = client.get("/api/chat/sessions")
            assert response.status_code == 401
            data = json.loads(response.data)
            assert "error" in data
            assert "authentication" in data["error"].lower()

    def test_get_session_api_requires_authentication(self, app):
        """Test that getting a session requires authentication."""
        with app.test_client() as client:
            response = client.get("/api/chat/sessions/some-id")
            assert response.status_code == 401
            data = json.loads(response.data)
            assert "error" in data
            assert "authentication" in data["error"].lower()

    def test_update_session_api_requires_authentication(self, app):
        """Test that updating a session requires authentication."""
        with app.test_client() as client:
            response = client.patch(
                "/api/chat/sessions/some-id",
                json={"title": "New title"},
                content_type="application/json",
            )
            assert response.status_code == 401
            data = json.loads(response.data)
            assert "error" in data
            assert "authentication" in data["error"].lower()

    def test_delete_session_api_requires_authentication(self, app):
        """Test that deleting a session requires authentication."""
        with app.test_client() as client:
            response = client.delete("/api/chat/sessions/some-id")
            assert response.status_code == 401
            data = json.loads(response.data)
            assert "error" in data
            assert "authentication" in data["error"].lower()

    def test_get_messages_api_requires_authentication(self, app):
        """Test that getting messages requires authentication."""
        with app.test_client() as client:
            response = client.get("/api/chat/sessions/some-id/messages")
            assert response.status_code == 401
            data = json.loads(response.data)
            assert "error" in data
            assert "authentication" in data["error"].lower()

    def test_send_message_api_requires_authentication(self, app):
        """Test that sending messages requires authentication."""
        with app.test_client() as client:
            response = client.post(
                "/api/chat/sessions/some-id/messages",
                json={"content": "Test message"},
                content_type="application/json",
            )
            assert response.status_code == 401
            data = json.loads(response.data)
            assert "error" in data
            assert "authentication" in data["error"].lower()


class TestSessionIdValidation:
    """Tests for session ID validation to prevent injection attacks."""

    def test_session_id_with_path_traversal_rejected(
        self, authenticated_client
    ):
        """Test that session IDs with path traversal are handled safely."""
        malicious_ids = [
            "../../../etc/passwd",
            "..%2F..%2Fetc%2Fpasswd",
            "session-id/../../admin",
            "valid-id/../other-id",
        ]

        for session_id in malicious_ids:
            response = authenticated_client.get(
                f"/api/chat/sessions/{session_id}"
            )
            # Path-traversal attempts should be normalized by Flask routing
            # and surface as 404 (no such session) or 400 (invalid id).
            # 500 would be an unhandled-exception regression. Importantly,
            # the response body must never echo system file paths.
            assert response.status_code in [404, 400]
            data = json.loads(response.data)
            assert "etc" not in str(data).lower()
            assert "passwd" not in str(data).lower()

    def test_session_id_with_null_bytes_rejected(self, authenticated_client):
        """Test that session IDs with null bytes are handled safely."""
        # URL-encoded null byte
        response = authenticated_client.get(
            "/api/chat/sessions/valid-id%00.txt"
        )
        # Null-byte-tainted IDs should be rejected with 404 or 400.
        # 500 would be an unhandled-exception regression.
        assert response.status_code in [404, 400]

    def test_session_id_with_sql_injection_handled(self, authenticated_client):
        """Test that SQL injection attempts in session ID are handled safely."""
        injection_attempts = [
            "' OR '1'='1",
            "'; DROP TABLE chat_sessions; --",
            "1 UNION SELECT * FROM users--",
        ]

        for injection in injection_attempts:
            response = authenticated_client.get(
                f"/api/chat/sessions/{injection}"
            )
            # Should return 404 (not found) - the injection string is just a literal ID
            assert response.status_code == 404
            data = json.loads(response.data)
            # Should not contain SQL error messages
            assert "syntax" not in str(data).lower()
            assert "sqlite" not in str(data).lower()
            assert "database" not in str(data).lower()