learningcircuit--local-deep-research/tests/security/test_file_upload_validator.py

"""
Tests for FileUploadValidator security module.
"""

from unittest.mock import Mock, patch

from local_deep_research.security.file_upload_validator import (
    FileUploadValidator,
)


class TestFileUploadValidatorConstants:
    """Tests for FileUploadValidator constants."""

    def test_max_file_size_defined(self):
        """MAX_FILE_SIZE is defined and matches the 3 GB server-config default."""
        assert FileUploadValidator.MAX_FILE_SIZE > 0
        assert FileUploadValidator.MAX_FILE_SIZE == 3 * 1024 * 1024 * 1024

    def test_max_files_per_request_defined(self):
        """MAX_FILES_PER_REQUEST is defined and reasonable."""
        assert FileUploadValidator.MAX_FILES_PER_REQUEST > 0
        assert FileUploadValidator.MAX_FILES_PER_REQUEST == 200

    def test_pdf_magic_bytes_correct(self):
        """PDF_MAGIC_BYTES is correct."""
        assert FileUploadValidator.PDF_MAGIC_BYTES == b"%PDF"

    def test_allowed_mime_types_includes_pdf(self):
        """ALLOWED_MIME_TYPES includes PDF."""
        assert "application/pdf" in FileUploadValidator.ALLOWED_MIME_TYPES


class TestResolveMaxFileSize:
    """The per-file cap honors LDR_SECURITY_UPLOAD_MAX_FILE_SIZE_MB."""

    def test_default_when_env_unset(self):
        from local_deep_research.security import file_upload_validator as mod

        with patch.object(mod, "check_env_setting", return_value=None):
            assert (
                mod._resolve_max_file_size()
                == mod._DEFAULT_MAX_FILE_SIZE_MB * 1024 * 1024
            )

    def test_env_value_overrides_default(self):
        from local_deep_research.security import file_upload_validator as mod

        # 750MB override (env vars always arrive as strings).
        with patch.object(mod, "check_env_setting", return_value="750"):
            assert mod._resolve_max_file_size() == 750 * 1024 * 1024

    def test_unparseable_env_falls_back_to_default(self):
        from local_deep_research.security import file_upload_validator as mod

        with patch.object(
            mod, "check_env_setting", return_value="not-a-number"
        ):
            assert (
                mod._resolve_max_file_size()
                == mod._DEFAULT_MAX_FILE_SIZE_MB * 1024 * 1024
            )

    def test_zero_env_value_falls_back_to_default(self):
        """A misconfigured cap of 0 would silently break all uploads."""
        from local_deep_research.security import file_upload_validator as mod

        with patch.object(mod, "check_env_setting", return_value="0"):
            assert (
                mod._resolve_max_file_size()
                == mod._DEFAULT_MAX_FILE_SIZE_MB * 1024 * 1024
            )

    def test_negative_env_value_falls_back_to_default(self):
        """Negative values are nonsensical for a size cap — reject them."""
        from local_deep_research.security import file_upload_validator as mod

        with patch.object(mod, "check_env_setting", return_value="-1"):
            assert (
                mod._resolve_max_file_size()
                == mod._DEFAULT_MAX_FILE_SIZE_MB * 1024 * 1024
            )


class TestValidateFileSize:
    """Tests for FileUploadValidator.validate_file_size()."""

    def test_valid_content_length(self):
        """Accepts valid content length."""
        is_valid, error = FileUploadValidator.validate_file_size(1000)
        assert is_valid is True
        assert error is None

    def test_valid_file_content(self):
        """Accepts valid file content size."""
        content = b"x" * 1000
        is_valid, error = FileUploadValidator.validate_file_size(None, content)
        assert is_valid is True
        assert error is None

    def test_content_length_exceeds_max(self):
        """Rejects content length exceeding max."""
        large_size = FileUploadValidator.MAX_FILE_SIZE + 1
        is_valid, error = FileUploadValidator.validate_file_size(large_size)
        assert is_valid is False
        assert "too large" in error.lower()

    def test_file_content_exceeds_max(self):
        """Rejects file content exceeding max.

        Patches the cap to a small value so the test doesn't have to
        allocate gigabytes to cross the real default.
        """
        with patch.object(FileUploadValidator, "MAX_FILE_SIZE", 1024):
            large_content = b"x" * 2048  # 2KB over a 1KB cap
            is_valid, error = FileUploadValidator.validate_file_size(
                None, large_content
            )
        assert is_valid is False
        assert "too large" in error.lower()

    def test_both_none_is_valid(self):
        """Accepts when both parameters are None."""
        is_valid, error = FileUploadValidator.validate_file_size(None, None)
        assert is_valid is True
        assert error is None

    def test_zero_content_length_is_valid(self):
        """Accepts zero content length."""
        is_valid, error = FileUploadValidator.validate_file_size(0)
        assert is_valid is True
        assert error is None

    def test_error_message_includes_size(self):
        """Error message includes file size information."""
        with patch.object(
            FileUploadValidator, "MAX_FILE_SIZE", 50 * 1024 * 1024
        ):
            large_size = 60 * 1024 * 1024  # 60MB, over a patched 50MB cap
            is_valid, error = FileUploadValidator.validate_file_size(large_size)
        assert is_valid is False
        assert "60" in error or "MB" in error


class TestValidateFileCount:
    """Tests for FileUploadValidator.validate_file_count()."""

    def test_valid_file_count(self):
        """Accepts valid file count."""
        is_valid, error = FileUploadValidator.validate_file_count(10)
        assert is_valid is True
        assert error is None

    def test_single_file_valid(self):
        """Accepts single file."""
        is_valid, error = FileUploadValidator.validate_file_count(1)
        assert is_valid is True
        assert error is None

    def test_max_files_valid(self):
        """Accepts exactly max files."""
        is_valid, error = FileUploadValidator.validate_file_count(
            FileUploadValidator.MAX_FILES_PER_REQUEST
        )
        assert is_valid is True
        assert error is None

    def test_exceeds_max_files(self):
        """Rejects count exceeding max."""
        is_valid, error = FileUploadValidator.validate_file_count(
            FileUploadValidator.MAX_FILES_PER_REQUEST + 1
        )
        assert is_valid is False
        assert "too many" in error.lower()

    def test_zero_files_invalid(self):
        """Rejects zero files."""
        is_valid, error = FileUploadValidator.validate_file_count(0)
        assert is_valid is False
        assert "no files" in error.lower()

    def test_negative_files_invalid(self):
        """Rejects negative file count."""
        is_valid, error = FileUploadValidator.validate_file_count(-1)
        assert is_valid is False
        assert "no files" in error.lower()


class TestValidateMimeType:
    """Tests for FileUploadValidator.validate_mime_type()."""

    def test_valid_pdf(self):
        """Accepts valid PDF file."""
        content = b"%PDF-1.4 test content"
        is_valid, error = FileUploadValidator.validate_mime_type(
            "document.pdf", content
        )
        assert is_valid is True
        assert error is None

    def test_valid_pdf_uppercase_extension(self):
        """Accepts PDF with uppercase extension."""
        content = b"%PDF-1.4 test content"
        is_valid, error = FileUploadValidator.validate_mime_type(
            "document.PDF", content
        )
        assert is_valid is True
        assert error is None

    def test_invalid_extension(self):
        """Rejects non-PDF extension."""
        content = b"%PDF-1.4 test content"
        is_valid, error = FileUploadValidator.validate_mime_type(
            "document.txt", content
        )
        assert is_valid is False
        assert "only pdf" in error.lower()

    def test_invalid_magic_bytes(self):
        """Rejects file with wrong magic bytes."""
        content = b"PK\x03\x04"  # ZIP magic bytes
        is_valid, error = FileUploadValidator.validate_mime_type(
            "document.pdf", content
        )
        assert is_valid is False
        assert "signature" in error.lower()

    def test_empty_content(self):
        """Rejects empty file content."""
        is_valid, error = FileUploadValidator.validate_mime_type(
            "document.pdf", b""
        )
        assert is_valid is False
        assert "signature" in error.lower()

    def test_too_short_content(self):
        """Rejects content shorter than magic bytes."""
        is_valid, error = FileUploadValidator.validate_mime_type(
            "document.pdf", b"%PD"
        )
        assert is_valid is False
        assert "signature" in error.lower()


class TestValidatePdfStructure:
    """Tests for FileUploadValidator.validate_pdf_structure()."""

    def test_valid_pdf_structure(self, mock_pdf_content):
        """Accepts valid PDF structure."""
        with patch(
            "local_deep_research.security.file_upload_validator.pdfplumber"
        ) as mock_pdfplumber:
            mock_pdf = Mock()
            mock_page = Mock()
            mock_page.width = 612
            mock_page.height = 792
            mock_pdf.pages = [mock_page]
            mock_pdfplumber.open.return_value.__enter__ = Mock(
                return_value=mock_pdf
            )
            mock_pdfplumber.open.return_value.__exit__ = Mock(
                return_value=False
            )

            is_valid, error = FileUploadValidator.validate_pdf_structure(
                "document.pdf", mock_pdf_content
            )
            assert is_valid is True
            assert error is None

    def test_pdf_no_pages(self, mock_pdf_content):
        """Rejects PDF with no pages."""
        with patch(
            "local_deep_research.security.file_upload_validator.pdfplumber"
        ) as mock_pdfplumber:
            mock_pdf = Mock()
            mock_pdf.pages = []
            mock_pdfplumber.open.return_value.__enter__ = Mock(
                return_value=mock_pdf
            )
            mock_pdfplumber.open.return_value.__exit__ = Mock(
                return_value=False
            )

            is_valid, error = FileUploadValidator.validate_pdf_structure(
                "document.pdf", mock_pdf_content
            )
            assert is_valid is False
            assert "no pages" in error.lower()

    def test_pdf_parse_error(self, mock_pdf_content):
        """Rejects PDF that cannot be parsed."""
        with patch(
            "local_deep_research.security.file_upload_validator.pdfplumber"
        ) as mock_pdfplumber:
            mock_pdfplumber.open.side_effect = Exception("PDF parsing error")

            is_valid, error = FileUploadValidator.validate_pdf_structure(
                "corrupt.pdf", mock_pdf_content
            )
            assert is_valid is False
            assert "invalid" in error.lower() or "corrupted" in error.lower()

    def test_pdf_none_pages(self, mock_pdf_content):
        """Rejects PDF with None pages."""
        with patch(
            "local_deep_research.security.file_upload_validator.pdfplumber"
        ) as mock_pdfplumber:
            mock_pdf = Mock()
            mock_pdf.pages = None
            mock_pdfplumber.open.return_value.__enter__ = Mock(
                return_value=mock_pdf
            )
            mock_pdfplumber.open.return_value.__exit__ = Mock(
                return_value=False
            )

            is_valid, error = FileUploadValidator.validate_pdf_structure(
                "document.pdf", mock_pdf_content
            )
            assert is_valid is False
            assert "no pages" in error.lower()


class TestValidateUpload:
    """Tests for FileUploadValidator.validate_upload()."""

    def test_valid_upload(self, mock_pdf_content):
        """Accepts valid file upload."""
        with patch(
            "local_deep_research.security.file_upload_validator.pdfplumber"
        ) as mock_pdfplumber:
            mock_pdf = Mock()
            mock_page = Mock()
            mock_page.width = 612
            mock_page.height = 792
            mock_pdf.pages = [mock_page]
            mock_pdfplumber.open.return_value.__enter__ = Mock(
                return_value=mock_pdf
            )
            mock_pdfplumber.open.return_value.__exit__ = Mock(
                return_value=False
            )

            is_valid, error = FileUploadValidator.validate_upload(
                "document.pdf", mock_pdf_content
            )
            assert is_valid is True
            assert error is None

    def test_upload_file_too_large(self):
        """Rejects upload that's too large."""
        with patch.object(FileUploadValidator, "MAX_FILE_SIZE", 1024):
            large_content = b"%PDF-1.4" + b"x" * 2048  # 2KB over 1KB cap
            is_valid, error = FileUploadValidator.validate_upload(
                "document.pdf", large_content
            )
        assert is_valid is False
        assert "too large" in error.lower()

    def test_upload_wrong_extension(self, mock_pdf_content):
        """Rejects upload with wrong extension."""
        is_valid, error = FileUploadValidator.validate_upload(
            "document.txt", mock_pdf_content
        )
        assert is_valid is False
        assert "only pdf" in error.lower()

    def test_upload_wrong_magic_bytes(self):
        """Rejects upload with wrong magic bytes."""
        content = b"not a pdf"
        is_valid, error = FileUploadValidator.validate_upload(
            "document.pdf", content
        )
        assert is_valid is False
        assert "signature" in error.lower()

    def test_upload_corrupted_pdf(self, mock_pdf_content):
        """Rejects corrupted PDF."""
        with patch(
            "local_deep_research.security.file_upload_validator.pdfplumber"
        ) as mock_pdfplumber:
            mock_pdfplumber.open.side_effect = Exception("Corrupted")

            is_valid, error = FileUploadValidator.validate_upload(
                "document.pdf", mock_pdf_content
            )
            assert is_valid is False
            assert "invalid" in error.lower() or "corrupted" in error.lower()

    def test_upload_with_content_length(self, mock_pdf_content):
        """Validates with content length header."""
        with patch(
            "local_deep_research.security.file_upload_validator.pdfplumber"
        ) as mock_pdfplumber:
            mock_pdf = Mock()
            mock_page = Mock()
            mock_page.width = 612
            mock_page.height = 792
            mock_pdf.pages = [mock_page]
            mock_pdfplumber.open.return_value.__enter__ = Mock(
                return_value=mock_pdf
            )
            mock_pdfplumber.open.return_value.__exit__ = Mock(
                return_value=False
            )

            is_valid, error = FileUploadValidator.validate_upload(
                "document.pdf",
                mock_pdf_content,
                content_length=len(mock_pdf_content),
            )
            assert is_valid is True
            assert error is None

    def test_upload_content_length_mismatch_large(self, mock_pdf_content):
        """Rejects when content length indicates too large."""
        # Content length header says file is too large
        is_valid, error = FileUploadValidator.validate_upload(
            "document.pdf",
            mock_pdf_content,
            content_length=FileUploadValidator.MAX_FILE_SIZE + 1,
        )
        assert is_valid is False
        assert "too large" in error.lower()


class TestSecurityScenarios:
    """Integration tests for security scenarios."""

    def test_disguised_executable(self):
        """Rejects executable disguised as PDF."""
        # Windows executable magic bytes
        content = b"MZ" + b"\x00" * 100
        is_valid, error = FileUploadValidator.validate_upload(
            "malware.pdf", content
        )
        assert is_valid is False

    def test_zip_bomb_prevention(self):
        """Rejects suspiciously large claimed size."""
        # Content length claims enormous size
        content = b"%PDF-1.4 small content"
        is_valid, error = FileUploadValidator.validate_upload(
            "bomb.pdf",
            content,
            content_length=10 * 1024 * 1024 * 1024,  # 10GB
        )
        assert is_valid is False

    def test_polyglot_file_rejection(self):
        """Rejects files that start with valid PDF but have wrong extension."""
        content = b"%PDF-1.4 this is actually executable code"
        is_valid, error = FileUploadValidator.validate_upload(
            "script.js", content
        )
        assert is_valid is False

    def test_html_in_pdf_extension(self):
        """Rejects HTML file with PDF extension."""
        content = b"<html><script>alert('xss')</script></html>"
        is_valid, error = FileUploadValidator.validate_upload(
            "attack.pdf", content
        )
        assert is_valid is False
        assert "signature" in error.lower()