From 7a750b51601fbdf9aec172e54c7778121590cde4 Mon Sep 17 00:00:00 2001
From: Sebastion <sebastion@sebastion.dev>
Date: Tue, 28 Jul 2026 23:10:27 +0100
Subject: [PATCH] fix(db): use parameterized queries for Kingbase
 get_tables/get_fields (CWE-89)

---
 backend/apps/db/db.py     | 4 ++--
 backend/apps/db/db_sql.py | 6 +++---
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/backend/apps/db/db.py b/backend/apps/db/db.py
index 8f27432ce..9ba8739b6 100644
--- a/backend/apps/db/db.py
+++ b/backend/apps/db/db.py
@@ -527,7 +527,7 @@ def get_tables(ds: CoreDatasource):
                 return res_list
         elif equals_ignore_case(ds.type, 'kingbase'):
             with get_driver_connection(ds) as conn, conn.cursor() as cursor:
-                cursor.execute(sql.format(sql_param))
+                cursor.execute(sql, (sql_param,))
                 res = cursor.fetchall()
                 res_list = [TableSchema(*item) for item in res]
                 return res_list
@@ -577,7 +577,7 @@ def get_fields(ds: CoreDatasource, table_name: str = None):
                 return res_list
         elif equals_ignore_case(ds.type, 'kingbase'):
             with get_driver_pool(ds).connection() as conn, conn.cursor() as cursor:
-                cursor.execute(sql.format(p1, p2))
+                cursor.execute(sql, (p1, p2) if p2 is not None and p2 != "" else (p1,))
                 res = cursor.fetchall()
                 res_list = [ColumnSchema(*item) for item in res]
                 return res_list
diff --git a/backend/apps/db/db_sql.py b/backend/apps/db/db_sql.py
index 496075378..aeeb89727 100644
--- a/backend/apps/db/db_sql.py
+++ b/backend/apps/db/db_sql.py
@@ -154,7 +154,7 @@ def get_table_sql(ds: CoreDatasource, conf: DatasourceConf, db_version: str = ''
                    pg_namespace n ON n.oid = c.relnamespace
                        LEFT JOIN
                    pg_description d ON d.objoid = c.oid AND d.objsubid = 0
-              WHERE n.nspname = '{0}'
+              WHERE n.nspname = %s
                 AND c.relkind IN ('r', 'v', 'p', 'm')
                 AND c.relname NOT LIKE 'pg_%'
                 AND c.relname NOT LIKE 'sql_%'
@@ -308,11 +308,11 @@ def get_field_sql(ds: CoreDatasource, conf: DatasourceConf, table_name: str = No
                             pg_catalog.pg_class c ON a.attrelid = c.oid
                                 JOIN
                             pg_catalog.pg_namespace n ON n.oid = c.relnamespace
-                       WHERE n.nspname = '{0}'
+                       WHERE n.nspname = %s
                          AND a.attnum > 0
                          AND NOT a.attisdropped \
                        """
-        sql2 = " AND c.relname = '{1}'" if table_name is not None and table_name != "" else ""
+        sql2 = " AND c.relname = %s" if table_name is not None and table_name != "" else ""
         return sql1 + sql2, conf.dbSchema, table_name
     elif equals_ignore_case(ds.type, "es"):
         return "", None, None