Merge pull request #638 from christianlouis/security-fix-sql-injection-index-mgmt-16594348298290829845

🔒 [security fix] Fix SQL injection in index management queries

app/database.py

@@ -210,8 +210,10 @@ def _run_schema_migrations(engine: Any) -> None:
             if unique_filehash_indexes:
                 logger.info("Migrating files: dropping unique index on 'filehash'")
                 with engine.begin() as conn:
+                    preparer = conn.dialect.identifier_preparer
                     for index in unique_filehash_indexes:
-                        conn.execute(text(f"DROP INDEX IF EXISTS {index['name']}"))
+                        quoted_idx = preparer.quote(index["name"])
+                        conn.execute(text(f"DROP INDEX IF EXISTS {quoted_idx}"))
                 logger.info("Migration complete: unique index on 'filehash' removed")
         except Exception as exc:
             logger.warning(f"Skipping filehash unique index drop: {exc}")
@@ -263,12 +265,16 @@ def _ensure_indexes(engine: Any, inspector: Any) -> None:
     table_names = inspector.get_table_names()
     columns_by_table: dict[str, set[str]] = {}
     with engine.begin() as conn:
+        preparer = conn.dialect.identifier_preparer
         for idx_name, table, column in _PERF_INDEXES:
             if table in table_names:
                 if table not in columns_by_table:
                     columns_by_table[table] = {col["name"] for col in inspector.get_columns(table)}
                 if column in columns_by_table[table]:
-                    conn.execute(text(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table} ({column})"))
+                    quoted_idx = preparer.quote(idx_name)
+                    quoted_table = preparer.quote(table)
+                    quoted_col = preparer.quote(column)
+                    conn.execute(text(f"CREATE INDEX IF NOT EXISTS {quoted_idx} ON {quoted_table} ({quoted_col})"))
 
     logger.info("Performance indexes ensured")
 

tests/test_database.py

@@ -671,6 +671,75 @@ class TestMultiVersionMigrations:
 
         engine.dispose()
 
+    def test_migration_drops_unique_index_with_quoted_name(self, tmp_path):
+        """Test that a unique filehash index whose name requires quoting is dropped correctly.
+
+        Index names containing special characters such as dashes must be quoted by
+        the dialect's identifier_preparer before being interpolated into raw SQL.
+        This test verifies that _run_schema_migrations() handles such names without
+        SQL errors and leaves the underlying table and its data intact.
+        """
+        from sqlalchemy import create_engine, inspect, text
+
+        from app.database import _run_schema_migrations
+
+        db_path = str(tmp_path / "quoted_index.db")
+        engine = create_engine(f"sqlite:///{db_path}")
+
+        # Build a files table that contains all columns the migration expects,
+        # then add a unique index on filehash whose name contains a dash — a
+        # character that requires quoting by the dialect's identifier_preparer.
+        with engine.begin() as conn:
+            conn.execute(
+                text(
+                    "CREATE TABLE files ("
+                    "id INTEGER PRIMARY KEY, "
+                    "filename VARCHAR, "
+                    "filehash VARCHAR, "
+                    "original_file_path VARCHAR, "
+                    "processed_file_path VARCHAR, "
+                    "is_duplicate BOOLEAN DEFAULT FALSE NOT NULL, "
+                    "duplicate_of_id INTEGER)"
+                )
+            )
+            # Index name deliberately contains a dash to exercise the quoting path.
+            conn.execute(text('CREATE UNIQUE INDEX "ix-filehash-unique" ON files (filehash)'))
+            conn.execute(text("INSERT INTO files (filename, filehash) VALUES ('doc.pdf', 'abc123')"))
+
+        # Confirm the index exists before migration.
+        inspector = inspect(engine)
+        pre_indexes = [idx["name"] for idx in inspector.get_indexes("files")]
+        assert "ix-filehash-unique" in pre_indexes
+
+        # Run migration — must not raise despite the special character in the index name.
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", DeprecationWarning)
+            _run_schema_migrations(engine)
+
+        # The unique index on filehash must have been dropped.
+        inspector = inspect(engine)
+        post_indexes = inspector.get_indexes("files")
+        remaining_unique_filehash = [
+            idx for idx in post_indexes if idx.get("unique") and "filehash" in idx.get("column_names", [])
+        ]
+        assert remaining_unique_filehash == [], (
+            f"Expected unique filehash index to be dropped, but found: {remaining_unique_filehash}"
+        )
+
+        # The table and its data must still be intact.
+        files_columns = {col["name"] for col in inspector.get_columns("files")}
+        assert "id" in files_columns
+        assert "filename" in files_columns
+        assert "filehash" in files_columns
+
+        with engine.connect() as conn:
+            row = conn.execute(text("SELECT filename, filehash FROM files WHERE filehash = 'abc123'")).fetchone()
+        assert row is not None
+        assert row[0] == "doc.pdf"
+        assert row[1] == "abc123"
+
+        engine.dispose()
+
     def test_migration_exception_handling(self, tmp_path):
         """Test that migration handles exceptions gracefully for index operations."""
         from sqlalchemy import create_engine, text