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

"""
SQL Injection Prevention Tests

Tests that verify SQL injection attacks are properly prevented
through SQLAlchemy ORM usage and proper input sanitization.
"""

import pytest
from datetime import datetime, UTC
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

from tests.test_utils import add_src_to_path

add_src_to_path()

from local_deep_research.database.models import (  # noqa: E402
    Base,
    ResearchHistory,
)


class TestSQLInjectionPrevention:
    """Test SQL injection prevention in database queries."""

    @pytest.fixture
    def test_db_session(self):
        """Create a test database session."""
        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        Session = sessionmaker(bind=engine)
        session = Session()
        yield session
        session.close()
        engine.dispose()

    def test_research_query_with_malicious_id(self, test_db_session):
        """Test that SQLAlchemy ORM prevents SQL injection via filter_by."""
        from local_deep_research.storage.database import (
            DatabaseReportStorage,
        )

        storage = DatabaseReportStorage(test_db_session)

        # Attempt SQL injection through research_id parameter
        malicious_ids = [
            "1' OR '1'='1",
            "1; DROP TABLE research_history;--",
            "1' UNION SELECT * FROM users--",
            "' OR 1=1--",
            "admin'--",
            "1' AND 1=1 UNION SELECT NULL, version()--",
        ]

        for malicious_id in malicious_ids:
            # This should safely return None, not execute SQL injection
            result = storage.get_report(malicious_id)
            assert result is None, (
                f"Malicious ID not properly handled: {malicious_id}"
            )

    def test_query_filter_sql_injection(self, test_db_session):
        """Test that query filtering prevents SQL injection."""
        # Malicious queries that could attempt SQL injection
        malicious_queries = [
            "admin' OR '1'='1",
            "query'; DROP TABLE research_history;--",
            "' OR 1=1--",
            "admin'/*",
            "query' UNION SELECT * FROM research_history--",
        ]

        for query in malicious_queries:
            # Query should handle malicious input safely
            result = (
                test_db_session.query(ResearchHistory)
                .filter_by(query=query)
                .first()
            )
            # Should return None (no match) rather than executing injection
            assert result is None

    def test_search_text_sanitization(self, test_db_session):
        """Test that search/filter text inputs are properly sanitized."""
        # Create test research entry
        research = ResearchHistory(
            id="test-id-123",
            query="Test query",
            mode="auto",
            status="pending",
            created_at=datetime.now(UTC).isoformat(),
            report_content="Test content",
        )
        test_db_session.add(research)
        test_db_session.commit()

        # Malicious search patterns
        malicious_patterns = [
            "%' OR '1'='1",
            "test'; DELETE FROM research_history WHERE '1'='1",
            "' UNION SELECT * FROM users--",
        ]

        for pattern in malicious_patterns:
            # Search using SQLAlchemy ORM (should be safe)
            results = (
                test_db_session.query(ResearchHistory)
                .filter(ResearchHistory.query.like(f"%{pattern}%"))
                .all()
            )
            # Should not execute SQL injection, just search literally
            assert isinstance(results, list)

    def test_raw_sql_uses_parameterized_queries(self, test_db_session):
        """
        Test that if raw SQL is used anywhere, it uses parameterized queries.
        This is a defensive test to catch potential future vulnerabilities.
        """
        # Example of UNSAFE raw SQL (what we should never do):
        # engine.execute(f"SELECT * FROM users WHERE username = '{username}'")

        # Example of SAFE raw SQL (what we should do if raw SQL is needed):
        safe_param = "admin' OR '1'='1"
        result = test_db_session.execute(
            text("SELECT :param as test"), {"param": safe_param}
        )
        row = result.fetchone()

        # The parameter should be treated as literal string, not SQL
        assert row[0] == safe_param
        # Should not return anything else (no injection occurred)

    def test_metadata_json_injection(self, test_db_session):
        """Test that JSON metadata fields prevent injection attacks."""
        from local_deep_research.storage.database import (
            DatabaseReportStorage,
        )

        storage = DatabaseReportStorage(test_db_session)

        # Create research entry
        research = ResearchHistory(
            id="test-metadata-injection",
            query="Test query",
            mode="auto",
            status="pending",
            created_at=datetime.now(UTC).isoformat(),
        )
        test_db_session.add(research)
        test_db_session.commit()

        # Attempt injection through JSON metadata
        malicious_metadata = {
            "key": "'; DROP TABLE research_history;--",
            "nested": {"sql": "1' OR '1'='1"},
            "array": ["admin'--", "user'; DELETE FROM users;"],
        }

        # Save should handle malicious JSON safely
        result = storage.save_report(
            "test-metadata-injection",
            "Test content",
            metadata=malicious_metadata,
        )
        assert result is True

        # Retrieve and verify data is stored literally, not executed
        retrieved = storage.get_report_with_metadata("test-metadata-injection")
        assert retrieved is not None
        assert (
            retrieved["metadata"]["key"] == "'; DROP TABLE research_history;--"
        )

    def test_special_characters_in_inputs(self, test_db_session):
        """Test that special SQL characters are properly escaped."""
        special_chars_inputs = [
            "test'query",
            'test"query',
            "test`query",
            "test\\query",
            "test%query",
            "test_query",
            "test;query",
            "test--query",
            "test/*comment*/query",
        ]

        for input_str in special_chars_inputs:
            # Create research with special characters
            research = ResearchHistory(
                id=f"test-{hash(input_str)}",
                query=input_str,
                mode="auto",
                status="pending",
                created_at=datetime.now(UTC).isoformat(),
            )
            test_db_session.add(research)

        test_db_session.commit()

        # All entries should be saved and retrievable
        count = test_db_session.query(ResearchHistory).count()
        assert count == len(special_chars_inputs)


class TestDatabaseStorageSecurityIntegration:
    """Integration tests for database storage security."""

    @pytest.fixture
    def test_db_session(self):
        """Create a test database session."""
        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        Session = sessionmaker(bind=engine)
        session = Session()
        yield session
        session.close()
        engine.dispose()

    def test_end_to_end_sql_injection_prevention(self, test_db_session):
        """
        End-to-end test ensuring SQL injection cannot occur through
        the normal application flow.
        """
        from local_deep_research.storage.database import (
            DatabaseReportStorage,
        )

        storage = DatabaseReportStorage(test_db_session)

        # Simulate attacker attempting SQL injection through multiple vectors
        attack_vectors = {
            "research_id": "1' OR '1'='1",
            "content": "'; DROP TABLE research_history;--",
        }

        # Create research with attack vectors
        research = ResearchHistory(
            id=attack_vectors["research_id"],
            query="Normal query",
            mode="auto",
            status="pending",
            created_at=datetime.now(UTC).isoformat(),
        )
        test_db_session.add(research)
        test_db_session.commit()

        # Save report with malicious content
        result = storage.save_report(
            attack_vectors["research_id"],
            attack_vectors["content"],
        )

        # All operations should complete safely
        assert result is True

        # Retrieve and verify all malicious inputs are stored as literals
        retrieved = storage.get_report(attack_vectors["research_id"])
        assert retrieved == attack_vectors["content"]

        # Database should still be intact (no tables dropped)
        tables = test_db_session.execute(
            text("SELECT name FROM sqlite_master WHERE type='table'")
        ).fetchall()
        assert len(tables) > 0  # Tables still exist


@pytest.mark.skip(reason="documentation/placeholder test - not implemented")
def test_orm_provides_sql_injection_protection():
    """
    Verify that SQLAlchemy ORM is being used correctly to prevent SQL injection.
    This is a documentation test that verifies our security assumptions.
    """
    # SQLAlchemy ORM automatically uses parameterized queries
    # This test documents that we rely on ORM for SQL injection prevention

    # Verify that we're using ORM methods, not raw SQL:
    # - query().filter_by() - ✓ Safe (parameterized)
    # - query().filter() - ✓ Safe (parameterized)
    # - session.add() - ✓ Safe (parameterized)
    # - session.query() - ✓ Safe (parameterized)

    # NOT using (these would be unsafe):
    # - session.execute(f"SELECT * FROM table WHERE id = '{user_input}'") - ✗ Unsafe
    # - raw SQL string concatenation - ✗ Unsafe

    # This test passes to document our security model
    assert True