CVE-2024-25722: Insufficient Information (NVD-CWE-noinfo)
value: qanything_kernel/connector/database/mysql/mysql_client.py in qanything.ai QAnything before 1.2.0 allows SQL Injection.
CWE description: There is insufficient information about the issue to classify it; details are unkown or unspecified.
Published: 2024-02-11T05:15Z
Repository: https://github.com/netease-youdao/QAnything
Language: Python
Files changed: qanything_kernel/connector/database/mysql/mysql_client.py
Commit date: 2024-01-25 17:45:23 +0800
Commit message: fix: Prevent SQL injection attacks.

Diff: diff --git a/qanything_kernel/connector/database/mysql/mysql_client.py b/qanything_kernel/connector/database/mysql/mysql_client.py
index 47dfc27..fac0c71 100644
--- a/qanything_kernel/connector/database/mysql/mysql_client.py
+++ b/qanything_kernel/connector/database/mysql/mysql_client.py
@@ -129,13 +129,16 @@ class KnowledgeBaseManager:
         result = self.execute_query_(query, (user_id,), fetch=True)
         debug_logger.info("check_user_exist {}".format(result))
         return result is not None and len(result) > 0
 
     def check_kb_exist(self, user_id, kb_ids):
-        kb_ids_str = ','.join("'{}'".format(str(x)) for x in kb_ids)
-        query = "SELECT kb_id FROM KnowledgeBase WHERE kb_id IN ({}) AND deleted = 0 AND user_id = %s".format(kb_ids_str)
-        result = self.execute_query_(query, (user_id,), fetch=True)
+        # 使用参数化查询
+        placeholders = ','.join(['%s'] * len(kb_ids))
+        query = "SELECT kb_id FROM KnowledgeBase WHERE kb_id IN ({}) AND deleted = 0 AND user_id = %s".format(
+            placeholders)
+        query_params = kb_ids + [user_id]
+        result = self.execute_query_(query, query_params, fetch=True)
         debug_logger.info("check_kb_exist {}".format(result))
         valid_kb_ids = [kb_info[0] for kb_info in result]
         unvalid_kb_ids = list(set(kb_ids) - set(valid_kb_ids))
         return unvalid_kb_ids
 
@@ -161,30 +164,30 @@ class KnowledgeBaseManager:
                  AND kb_id IN (SELECT kb_id FROM KnowledgeBase WHERE user_id = %s)""".format(file_ids_str)
         # self.execute_query_(query, (kb_id, user_id), commit=True)
         result = self.execute_query_(query, (kb_id, user_id), fetch=True)
         debug_logger.info("check_file_exist {}".format(result))
         return result
-    
+
     def check_file_exist_by_name(self, user_id, kb_id, file_names):
         results = []
         batch_size = 100  # 根据实际情况调整批次大小
 
         # 分批处理file_names
         for i in range(0, len(file_names), batch_size):
-            batch_file_names = file_names[i:i+batch_size]
-            file_names_str = ','.join("'{}'".format(str(x).replace("'", "\\'")) for x in batch_file_names)
+            batch_file_names = file_names[i:i + batch_size]
+            placeholders = ','.join(['%s'] * len(batch_file_names))
             query = """
                 SELECT file_id, file_name, file_size, status FROM File 
                 WHERE deleted = 0
                 AND file_name IN ({})
                 AND kb_id = %s 
                 AND kb_id IN (SELECT kb_id FROM KnowledgeBase WHERE user_id = %s)
-            """.format(file_names_str)
-            
-            # 这里假设 execute_query_ 是一个可以执行SQL查询并提交或获取结果的方法
-            batch_result = self.execute_query_(query, (kb_id, user_id), fetch=True)
-            debug_logger.info("check_file_exist_by_name batch {}: {}".format(i//batch_size, batch_result))
+            """.format(placeholders)
+
+            query_params = batch_file_names + [kb_id, user_id]
+            batch_result = self.execute_query_(query, query_params, fetch=True)
+            debug_logger.info("check_file_exist_by_name batch {}: {}".format(i // batch_size, batch_result))
             results.extend(batch_result)
 
         return results
 
     # 对外接口不需要增加用户，新建知识库的时候增加用户就可以了
@@ -209,23 +212,27 @@ class KnowledgeBaseManager:
         query = "SELECT user_id FROM User"
         return self.execute_query_(query, (), fetch=True)
 
     # [知识库] 获取指定kb_ids的知识库
     def get_knowledge_base_name(self, kb_ids):
-        kb_ids_str = ','.join("'{}'".format(str(x)) for x in kb_ids)
-        query = "SELECT user_id, kb_id, kb_name FROM KnowledgeBase WHERE kb_id IN ({}) AND deleted = 0".format(kb_ids_str)
-        return self.execute_query_(query, (), fetch=True)
+        # 使用参数化查询
+        placeholders = ','.join(['%s'] * len(kb_ids))
+        query = "SELECT user_id, kb_id, kb_name FROM KnowledgeBase WHERE kb_id IN ({}) AND deleted = 0".format(placeholders)
+        query_params = kb_ids
+        return self.execute_query_(query, query_params, fetch=True)
 
     # [知识库] 删除指定知识库
     def delete_knowledge_base(self, user_id, kb_ids):
-        # 删除知识库
-        kb_ids_str = ','.join("'{}'".format(str(x)) for x in kb_ids)
-        query = "UPDATE KnowledgeBase SET deleted = 1 WHERE user_id = %s AND kb_id IN ({})".format(kb_ids_str)
-        self.execute_query_(query, (user_id,), commit=True)
-        # 删除知识库下面的文件
-        query = """UPDATE File SET deleted = 1 WHERE kb_id IN ({}) AND kb_id IN (SELECT kb_id FROM KnowledgeBase WHERE user_id = %s)""".format(kb_ids_str)
-        self.execute_query_(query, (user_id,), commit=True)
+        # 使用参数化查询
+        placeholders = ','.join(['%s'] * len(kb_ids))
+        query = "UPDATE KnowledgeBase SET deleted = 1 WHERE user_id = %s AND kb_id IN ({})".format(placeholders)
+        query_params = [user_id] + kb_ids
+        self.execute_query_(query, query_params, commit=True)
+
+        # 更新文件的删除状态也需要使用参数化查询
+        query = "UPDATE File SET deleted = 1 WHERE kb_id IN ({}) AND kb_id IN (SELECT kb_id FROM KnowledgeBase WHERE user_id = %s)".format(placeholders)
+        self.execute_query_(query, query_params, commit=True)
     
     # [知识库] 重命名知识库
     def rename_knowledge_base(self, user_id, kb_id, kb_name):
         query = "UPDATE KnowledgeBase SET kb_name = %s WHERE kb_id = %s AND user_id = %s"
         self.execute_query_(query, (kb_name, kb_id, user_id), commit=True)

Vulnerable code:         kb_ids_str = ','.join("'{}'".format(str(x)) for x in kb_ids)
        query = "SELECT kb_id FROM KnowledgeBase WHERE kb_id IN ({}) AND deleted = 0 AND user_id = %s".format(kb_ids_str)
        result = self.execute_query_(query, (user_id,), fetch=True)
    
            batch_file_names = file_names[i:i+batch_size]
            file_names_str = ','.join("'{}'".format(str(x).replace("'", "\\'")) for x in batch_file_names)
            """.format(file_names_str)
            
            # 这里假设 execute_query_ 是一个可以执行SQL查询并提交或获取结果的方法
            batch_result = self.execute_query_(query, (kb_id, user_id), fetch=True)
            debug_logger.info("check_file_exist_by_name batch {}: {}".format(i//batch_size, batch_result))
        kb_ids_str = ','.join("'{}'".format(str(x)) for x in kb_ids)
        query = "SELECT user_id, kb_id, kb_name FROM KnowledgeBase WHERE kb_id IN ({}) AND deleted = 0".format(kb_ids_str)
        return self.execute_query_(query, (), fetch=True)
        # 删除知识库
        kb_ids_str = ','.join("'{}'".format(str(x)) for x in kb_ids)
        query = "UPDATE KnowledgeBase SET deleted = 1 WHERE user_id = %s AND kb_id IN ({})".format(kb_ids_str)
        self.execute_query_(query, (user_id,), commit=True)
        # 删除知识库下面的文件
        query = """UPDATE File SET deleted = 1 WHERE kb_id IN ({}) AND kb_id IN (SELECT kb_id FROM KnowledgeBase WHERE user_id = %s)""".format(kb_ids_str)
        self.execute_query_(query, (user_id,), commit=True)
Fixed code:         # 使用参数化查询
        placeholders = ','.join(['%s'] * len(kb_ids))
        query = "SELECT kb_id FROM KnowledgeBase WHERE kb_id IN ({}) AND deleted = 0 AND user_id = %s".format(
            placeholders)
        query_params = kb_ids + [user_id]
        result = self.execute_query_(query, query_params, fetch=True)

            batch_file_names = file_names[i:i + batch_size]
            placeholders = ','.join(['%s'] * len(batch_file_names))
            """.format(placeholders)

            query_params = batch_file_names + [kb_id, user_id]
            batch_result = self.execute_query_(query, query_params, fetch=True)
            debug_logger.info("check_file_exist_by_name batch {}: {}".format(i // batch_size, batch_result))
        # 使用参数化查询
        placeholders = ','.join(['%s'] * len(kb_ids))
        query = "SELECT user_id, kb_id, kb_name FROM KnowledgeBase WHERE kb_id IN ({}) AND deleted = 0".format(placeholders)
        query_params = kb_ids
        return self.execute_query_(query, query_params, fetch=True)
        # 使用参数化查询
        placeholders = ','.join(['%s'] * len(kb_ids))
        query = "UPDATE KnowledgeBase SET deleted = 1 WHERE user_id = %s AND kb_id IN ({})".format(placeholders)
        query_params = [user_id] + kb_ids
        self.execute_query_(query, query_params, commit=True)

        # 更新文件的删除状态也需要使用参数化查询
        query = "UPDATE File SET deleted = 1 WHERE kb_id IN ({}) AND kb_id IN (SELECT kb_id FROM KnowledgeBase WHERE user_id = %s)".format(placeholders)
        self.execute_query_(query, query_params, commit=True)