learningcircuit--local-deep-research/tests/auth_tests/test_sqlcipher_rekey.py

"""
Tests for SQLCipher rekey (password change) functionality.

These tests verify that password changes work correctly and use
consistent key derivation with initial key setup.
"""

import shutil
import tempfile
from pathlib import Path
from unittest.mock import Mock, patch

import pytest

from src.local_deep_research.database.sqlcipher_utils import (
    LEGACY_PBKDF2_SALT,
    apply_cipher_defaults_before_key,
    apply_sqlcipher_pragmas,
    get_sqlcipher_settings,
    set_sqlcipher_key,
    set_sqlcipher_rekey,
    _get_key_from_password,
)
from src.local_deep_research.database.sqlcipher_compat import (
    get_sqlcipher_module,
)


@pytest.fixture
def temp_db_path():
    """Create a temporary database path."""
    temp_dir = tempfile.mkdtemp()
    db_path = Path(temp_dir) / "test_rekey.db"
    yield db_path
    shutil.rmtree(temp_dir)


@pytest.fixture
def sqlcipher_module():
    """Get the SQLCipher module."""
    return get_sqlcipher_module()


@pytest.fixture
def encrypted_db(sqlcipher_module, temp_db_path):
    """Create an encrypted database with test data."""
    conn = sqlcipher_module.connect(str(temp_db_path))
    cursor = conn.cursor()
    # New database, so use creation_mode=True
    apply_cipher_defaults_before_key(cursor)
    set_sqlcipher_key(cursor, "original_password")
    cursor.close()

    conn.execute("CREATE TABLE test_data (id INTEGER PRIMARY KEY, value TEXT)")
    conn.execute("INSERT INTO test_data VALUES (1, 'secret_value_1')")
    conn.execute("INSERT INTO test_data VALUES (2, 'secret_value_2')")
    conn.execute("INSERT INTO test_data VALUES (3, 'secret_value_3')")
    conn.commit()

    return {"conn": conn, "path": temp_db_path, "password": "original_password"}


def _open_existing_db(sqlcipher_module, db_path, password):
    """Helper to open an existing encrypted DB with correct PRAGMA ordering."""
    conn = sqlcipher_module.connect(str(db_path))
    cursor = conn.cursor()
    set_sqlcipher_key(cursor, password)
    apply_sqlcipher_pragmas(cursor, creation_mode=False)
    cursor.close()
    return conn


class TestRekeyFunctionality:
    """Tests for basic rekey operations."""

    def test_rekey_changes_password(self, encrypted_db, sqlcipher_module):
        """Verify rekey successfully changes the database password."""
        conn = encrypted_db["conn"]
        db_path = encrypted_db["path"]

        # Change password
        set_sqlcipher_rekey(conn, "new_password")
        conn.close()

        # Old password should fail
        conn_old = _open_existing_db(
            sqlcipher_module, db_path, "original_password"
        )
        with pytest.raises(Exception):
            conn_old.execute("SELECT * FROM test_data").fetchone()
        conn_old.close()

        # New password should work
        conn_new = _open_existing_db(sqlcipher_module, db_path, "new_password")
        result = conn_new.execute("SELECT * FROM test_data").fetchall()
        assert len(result) == 3, (
            "All data should be accessible with new password"
        )
        conn_new.close()

    def test_rekey_preserves_all_data(self, encrypted_db, sqlcipher_module):
        """Verify all data is preserved after password change."""
        conn = encrypted_db["conn"]
        db_path = encrypted_db["path"]

        # Read data before rekey
        original_data = conn.execute(
            "SELECT id, value FROM test_data ORDER BY id"
        ).fetchall()

        # Change password
        set_sqlcipher_rekey(conn, "new_password")
        conn.close()

        # Read data after rekey
        conn_new = _open_existing_db(sqlcipher_module, db_path, "new_password")
        new_data = conn_new.execute(
            "SELECT id, value FROM test_data ORDER BY id"
        ).fetchall()
        conn_new.close()

        assert original_data == new_data, "Data should be identical after rekey"

    def test_rekey_uses_pbkdf2_derivation(self):
        """Verify rekey uses the same PBKDF2 key derivation as initial key."""
        # Get derived key for a password
        password = "test_password_for_derivation"
        kdf_iterations = get_sqlcipher_settings()["kdf_iterations"]
        derived_key = _get_key_from_password(
            password, LEGACY_PBKDF2_SALT, kdf_iterations
        )

        # Derive again to verify same result (deterministic)
        derived_key_2 = _get_key_from_password(
            password, LEGACY_PBKDF2_SALT, kdf_iterations
        )
        assert derived_key == derived_key_2, (
            "Key derivation must be deterministic"
        )

    def test_rekey_multiple_times(self, encrypted_db, sqlcipher_module):
        """Verify password can be changed multiple times."""
        conn = encrypted_db["conn"]
        db_path = encrypted_db["path"]

        passwords = ["password_1", "password_2", "password_3", "final_password"]

        for new_password in passwords:
            set_sqlcipher_rekey(conn, new_password)

        conn.close()

        # Only the final password should work
        for wrong_password in passwords[:-1]:
            conn_wrong = _open_existing_db(
                sqlcipher_module, db_path, wrong_password
            )
            with pytest.raises(Exception):
                conn_wrong.execute("SELECT * FROM test_data").fetchone()
            conn_wrong.close()

        # Final password works
        conn_final = _open_existing_db(
            sqlcipher_module, db_path, "final_password"
        )
        result = conn_final.execute("SELECT * FROM test_data").fetchall()
        assert len(result) == 3
        conn_final.close()


class TestRekeySpecialPasswords:
    """Tests for rekey with special characters in passwords."""

    def test_rekey_with_quotes(self, encrypted_db, sqlcipher_module):
        """Verify rekey works with quotes in password."""
        conn = encrypted_db["conn"]
        db_path = encrypted_db["path"]

        new_password = "pass'word\"with'quotes\""

        set_sqlcipher_rekey(conn, new_password)
        conn.close()

        conn_new = _open_existing_db(sqlcipher_module, db_path, new_password)
        result = conn_new.execute("SELECT * FROM test_data").fetchone()
        assert result is not None, (
            "Database should be accessible with quoted password"
        )
        conn_new.close()

    def test_rekey_with_unicode(self, encrypted_db, sqlcipher_module):
        """Verify rekey works with unicode characters."""
        conn = encrypted_db["conn"]
        db_path = encrypted_db["path"]

        new_password = "パスワード安全"

        set_sqlcipher_rekey(conn, new_password)
        conn.close()

        conn_new = _open_existing_db(sqlcipher_module, db_path, new_password)
        result = conn_new.execute("SELECT * FROM test_data").fetchone()
        assert result is not None, (
            "Database should be accessible with unicode password"
        )
        conn_new.close()

    def test_rekey_with_backslashes(self, encrypted_db, sqlcipher_module):
        """Verify rekey works with backslashes in password."""
        conn = encrypted_db["conn"]
        db_path = encrypted_db["path"]

        new_password = "pass\\word\\with\\backslashes"

        set_sqlcipher_rekey(conn, new_password)
        conn.close()

        conn_new = _open_existing_db(sqlcipher_module, db_path, new_password)
        result = conn_new.execute("SELECT * FROM test_data").fetchone()
        assert result is not None, (
            "Database should be accessible with backslash password"
        )
        conn_new.close()

    def test_rekey_with_sql_injection_attempt(
        self, encrypted_db, sqlcipher_module
    ):
        """Verify rekey is safe against SQL injection attempts."""
        conn = encrypted_db["conn"]
        db_path = encrypted_db["path"]

        new_password = "'; DROP TABLE test_data; --"

        set_sqlcipher_rekey(conn, new_password)
        conn.close()

        conn_new = _open_existing_db(sqlcipher_module, db_path, new_password)
        result = conn_new.execute("SELECT COUNT(*) FROM test_data").fetchone()
        assert result[0] == 3, (
            "Table should exist and have all data after 'injection' password"
        )
        conn_new.close()


class TestRekeyWithSQLAlchemy:
    """Tests for rekey via SQLAlchemy connections."""

    def test_rekey_via_sqlalchemy(self, temp_db_path):
        """Verify rekey works through SQLAlchemy connection."""
        from sqlalchemy import create_engine, event, text

        from src.local_deep_research.database.sqlcipher_utils import (
            apply_cipher_defaults_before_key,
            apply_sqlcipher_pragmas,
            set_sqlcipher_key,
            set_sqlcipher_rekey,
        )
        from src.local_deep_research.database.sqlcipher_compat import (
            get_sqlcipher_module,
        )

        sqlcipher_module = get_sqlcipher_module()

        # Create database with raw connection first
        raw_conn = sqlcipher_module.connect(str(temp_db_path))
        cursor = raw_conn.cursor()
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "original_password")
        cursor.close()
        raw_conn.execute(
            "CREATE TABLE test (id INTEGER PRIMARY KEY, data TEXT)"
        )
        raw_conn.execute("INSERT INTO test VALUES (1, 'test_data')")
        raw_conn.commit()
        raw_conn.close()

        # Open with SQLAlchemy and rekey
        engine = create_engine(
            f"sqlite:///{temp_db_path}",
            module=sqlcipher_module,
        )

        @event.listens_for(engine, "connect")
        def set_cipher(dbapi_conn, connection_record):
            cursor = dbapi_conn.cursor()
            set_sqlcipher_key(cursor, "original_password")
            apply_sqlcipher_pragmas(cursor, creation_mode=False)
            cursor.close()

        with engine.connect() as conn:
            # Verify data is accessible
            result = conn.execute(
                text("SELECT data FROM test WHERE id = 1")
            ).scalar()
            assert result == "test_data"

            # Rekey via SQLAlchemy connection
            set_sqlcipher_rekey(conn, "new_password")
            conn.commit()

        engine.dispose()

        # Verify new password works with raw connection
        raw_conn2 = _open_existing_db(
            sqlcipher_module, temp_db_path, "new_password"
        )
        result = raw_conn2.execute(
            "SELECT data FROM test WHERE id = 1"
        ).fetchone()
        assert result[0] == "test_data", (
            "Data should be accessible with new password"
        )
        raw_conn2.close()


class TestSetSqlcipherRekey:
    """Unit tests for set_sqlcipher_rekey verifying it uses PBKDF2."""

    def test_rekey_calls_get_key_from_password(self):
        """Verify rekey uses get_key_from_password (PBKDF2), not raw encoding."""
        mock_conn = Mock()
        mock_conn.execute.side_effect = [TypeError("not sqlalchemy"), None]

        with patch(
            "src.local_deep_research.database.sqlcipher_utils.get_key_from_password",
            return_value=b"\xab\xcd\xef",
        ) as mock_get_key:
            set_sqlcipher_rekey(mock_conn, "new_password")
            mock_get_key.assert_called_once_with("new_password", db_path=None)

    def test_rekey_does_not_use_raw_utf8_hex(self):
        """Verify rekey does NOT use password.encode().hex() (raw UTF-8)."""
        mock_conn = Mock()
        mock_conn.execute.side_effect = [TypeError("not sqlalchemy"), None]

        with patch(
            "src.local_deep_research.database.sqlcipher_utils.get_key_from_password",
            return_value=b"\xab\xcd\xef",
        ):
            set_sqlcipher_rekey(mock_conn, "test_password")

        # The SQL should contain the PBKDF2-derived hex, not raw password hex
        rekey_call = mock_conn.execute.call_args_list[-1][0][0]
        raw_hex = "test_password".encode().hex()
        assert raw_hex not in rekey_call, (
            "Rekey should NOT use raw UTF-8 hex encoding"
        )
        assert "abcdef" in rekey_call, "Rekey should use PBKDF2-derived hex key"

    def test_rekey_works_with_sqlalchemy_connection(self):
        """Verify rekey works through SQLAlchemy connection (text() wrapper)."""
        mock_conn = Mock()

        with patch(
            "src.local_deep_research.database.sqlcipher_utils.get_key_from_password",
            return_value=b"\xab\xcd\xef",
        ):
            set_sqlcipher_rekey(mock_conn, "new_password")

        mock_conn.execute.assert_called_once()