From 120002b39408fb610d06c7555abdd40ab40c2ae4 Mon Sep 17 00:00:00 2001
From: "google-labs-jules[bot]"
 <161369871+google-labs-jules[bot]@users.noreply.github.com>
Date: Sat, 14 Mar 2026 09:42:04 +0000
Subject: [PATCH] fix(database): quote identifiers in index management queries
 to prevent SQL injection

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
---
 app/database.py | 10 ++++++++--
 1 file changed, 8 insertions(+), 2 deletions(-)

diff --git a/app/database.py b/app/database.py
index bff3ac08..e551569f 100644
--- a/app/database.py
+++ b/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")