""" XSS prevention tests for chat feature. Verifies that XSS payloads are handled safely in chat content, preventing cross-site scripting attacks. """ import json import pytest # ============================================================================= # XSS Payload Collections # ============================================================================= # Common XSS payloads for testing SCRIPT_TAG_PAYLOADS = [ "", "", "", "", "", "", ] EVENT_HANDLER_PAYLOADS = [ "", "", "", "", "
hover me
", "", "", "", "", "
", ] NESTED_XSS_PAYLOADS = [ "</script>", "ipt>alert('XSS')ipt>", "alert(1)>", "<", "", ] ENCODED_XSS_PAYLOADS = [ "<script>alert('XSS')</script>", # HTML entities "%3Cscript%3Ealert('XSS')%3C/script%3E", # URL encoded "\\x3cscript\\x3ealert('XSS')\\x3c/script\\x3e", # Hex encoded "\u003cscript\u003ealert('XSS')\u003c/script\u003e", # Unicode ] DATA_URL_PAYLOADS = [ "click", "", "", ] # ============================================================================= # Content Sanitization Tests # ============================================================================= class TestChatContentSanitization: """Tests verifying XSS payloads are sanitized in chat content.""" @pytest.mark.parametrize("payload", SCRIPT_TAG_PAYLOADS) def test_script_tags_handled_in_message( self, payload, authenticated_client ): """Test that script tags in messages are handled safely.""" # Create session 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"] # Send message with XSS payload response = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": payload, "trigger_research": False}, content_type="application/json", ) # Message should be accepted (we don't reject content, just escape it) assert response.status_code == 200 # Retrieve the message messages_resp = authenticated_client.get( f"/api/chat/sessions/{session_id}/messages" ) messages = json.loads(messages_resp.data)["messages"] # Find our message (should be the last one) user_messages = [m for m in messages if m["role"] == "user"] assert len(user_messages) > 0 # API layer stores raw content; sanitization happens at display layer # Verify the content is stored exactly as submitted (not silently dropped) last_message = user_messages[-1] assert last_message["content"] == payload @pytest.mark.parametrize("payload", EVENT_HANDLER_PAYLOADS) def test_event_handlers_handled_in_message( self, payload, authenticated_client ): """Test that event handlers in messages are handled safely.""" # Create session 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"] # Send message with XSS payload response = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": payload, "trigger_research": False}, content_type="application/json", ) assert response.status_code == 200 @pytest.mark.parametrize("payload", JAVASCRIPT_URL_PAYLOADS) def test_javascript_urls_handled_in_message( self, payload, authenticated_client ): """Test that javascript: URLs in messages are handled safely.""" 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"] response = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": payload, "trigger_research": False}, content_type="application/json", ) assert response.status_code == 200 @pytest.mark.parametrize("payload", NESTED_XSS_PAYLOADS) def test_nested_xss_payloads_handled(self, payload, authenticated_client): """Test that nested/evasion XSS payloads are handled safely.""" 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"] response = authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": payload, "trigger_research": False}, content_type="application/json", ) assert response.status_code == 200 class TestSessionTitleSanitization: """Tests verifying XSS payloads are sanitized in session titles.""" @pytest.mark.parametrize( "payload", SCRIPT_TAG_PAYLOADS[:3] + EVENT_HANDLER_PAYLOADS[:3], ) def test_xss_in_session_title(self, payload, authenticated_client): """Test that XSS payloads in session titles are handled safely.""" # Create session with XSS in initial query (becomes title) create_resp = authenticated_client.post( "/api/chat/sessions", json={"initial_query": payload}, content_type="application/json", ) assert create_resp.status_code == 200 session_id = json.loads(create_resp.data)["session_id"] # Verify session was created session_resp = authenticated_client.get( f"/api/chat/sessions/{session_id}" ) assert session_resp.status_code == 200 @pytest.mark.parametrize( "payload", SCRIPT_TAG_PAYLOADS[:3] + EVENT_HANDLER_PAYLOADS[:3], ) def test_xss_in_title_update(self, payload, authenticated_client): """Test that XSS payloads in title updates are handled safely.""" # Create session create_resp = authenticated_client.post( "/api/chat/sessions", json={"initial_query": "Normal query"}, content_type="application/json", ) session_id = json.loads(create_resp.data)["session_id"] # Update title with XSS payload update_resp = authenticated_client.patch( f"/api/chat/sessions/{session_id}", json={"title": payload}, content_type="application/json", ) # Should either accept (and escape on display) or reject assert update_resp.status_code in [200, 400] class TestContextManagerXSS: """Tests verifying XSS handling in context manager.""" def test_xss_in_messages_handled_in_context(self): """Test that XSS in messages doesn't affect context building.""" from src.local_deep_research.chat.context import ChatContextManager messages = [ { "id": "msg-1", "role": "user", "content": "", "message_type": "query", "research_id": None, }, { "id": "msg-2", "role": "assistant", "content": "Response with ", "message_type": "response", "research_id": "research-1", }, ] manager = ChatContextManager("test-session", messages, {}) # Context building should not crash context = manager.build_research_context() assert isinstance(context, dict) # Findings store the raw content as-is (escaping is the UI layer's job) assert "" in context["accumulated_findings"] def test_xss_in_accumulated_context(self): """Test that XSS in accumulated context is handled safely.""" from src.local_deep_research.chat.context import ChatContextManager accumulated = { "key_entities": [ "", "normal entity", ], "topics": [""], "summary": " summary text", } manager = ChatContextManager("test-session", [], accumulated) # Should handle without crashing entities = manager._get_key_entities() _topics = manager._get_topics() # noqa: F841 assert "", "x", ], ) def test_jinja2_autoescape_escapes_payload(self, app, payload): """A script/HTML payload rendered through the app's Jinja2 env is HTML-escaped, not emitted raw.""" from flask import render_template_string with app.test_request_context(): rendered = render_template_string("{{ value }}", value=payload) # The raw tag must not survive; the escaped form must be present. assert "