From 1d9bd15a700f477ba9ba31efe4b2805028421bf3 Mon Sep 17 00:00:00 2001
From: "google-labs-jules[bot]"
 <161369871+google-labs-jules[bot]@users.noreply.github.com>
Date: Mon, 23 Mar 2026 14:06:18 +0000
Subject: [PATCH] =?UTF-8?q?=F0=9F=94=92=20Fix=20potential=20SQL=20injectio?=
 =?UTF-8?q?n=20in=20database=20migration=20preview?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The `preview_migration` function in `app/utils/db_migrate.py` used string
interpolation to dynamically execute a COUNT query on the source database
(`f"SELECT COUNT(*) FROM {quoted_table}"`).

While the table name was quoted via the dialect's identifier preparer and
validated with a regex, string interpolation for raw SQL should be avoided
as it represents an anti-pattern and a theoretical risk for SQL injection
if validation controls are ever bypassed or modified.

This commit replaces the raw string interpolation with safe, parameterized
SQLAlchemy Core query construction `select(func.count()).select_from(table(table_name))`,
which automatically handles table quoting and execution safely. It also removes
the unused `text` import to keep the code clean.

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
---
 app/utils/db_migrate.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/app/utils/db_migrate.py b/app/utils/db_migrate.py
index 78f87fca..12009128 100644
--- a/app/utils/db_migrate.py
+++ b/app/utils/db_migrate.py
@@ -15,7 +15,7 @@ import logging
 import re
 from typing import Any
 
-from sqlalchemy import MetaData, create_engine, inspect, text
+from sqlalchemy import MetaData, create_engine, func, inspect, select, table
 from sqlalchemy.engine import Engine
 from sqlalchemy.engine.url import make_url
 from sqlalchemy.orm import sessionmaker
@@ -89,8 +89,9 @@ def preview_migration(source_url: str) -> dict[str, Any]:
                     logger.warning(f"Skipping table with invalid name format: {table_name}")
                     continue
                 # table_name is safe — sourced from inspect().get_table_names(), not user input
-                quoted_table = conn.dialect.identifier_preparer.quote(table_name)
-                row = conn.execute(text(f"SELECT COUNT(*) FROM {quoted_table}")).fetchone()  # noqa: S608
+                t = table(table_name)
+                query = select(func.count()).select_from(t)
+                row = conn.execute(query).fetchone()
                 count = row[0] if row else 0
                 result.append({"name": table_name, "row_count": count})
                 total += count