时间戳

首页 > 解决方案 > 构造具有多个 AND/OR 条件的安全 SQL 语句

python - 构造具有多个 AND/OR 条件的安全 SQL 语句

问题描述

我有以下问题：我在 Python 代码中有一个非常大的字符串形式的 SQL 语句：

sql = f"""
 *many statements here*
"""

该 SQL 语句的一部分是：

where 1 = 1
     and selector in ('YES', 'NO')
     AND parameter1 = value1
     AND parameter2 = value2.1 OR parameter2 = value2.2
     AND ...

其中那些 AND/OR 语句由 Python 字典以形式给出

{ parameter1: [value1], parameter2: [value2.1, value 2.2], ...}

我编写了一个函数，该函数接受该字典并将其展开为以下形式的字符串：

AND (parameter1 = value1) AND ((parameter2 = value2.1) OR (parameter2 = value2.2)) AND ...

并通过此函数将该字符串插入到大型 SQL 语句中：

where 1 = 1
         and selector in ('YES', 'NO')
         {form_sql_statement_from_dictionary(dictionary)}

但似乎这种方法容易受到 SQL 注入攻击。现在，安全的方法是对大型 SQL 语句进行参数化，但由于我不知道字典中有多少参数和值，所以我不知道如何进行这样的参数化。此外，我无法更改大型 SQL 语句。不知何故，我必须以安全的方式形成该 AND/OR 语句并将其插入到现有字符串中。有什么办法可以做到这一点，而不是尝试管理字典值本身？

完整的 Python 脚本如下所示：

async def query_for_data(
    connection: "PgService", dictionary: Dict[str, Any]
) -> pd.DataFrame:
sql = f"""
*multiple SQL statements*
where 1 = 1
   and selector in ('YES', 'NO')
   {form_sql_statement_from_dictionary(dictionary)}
"""
data = await connection.fetch(sql)
data = pd.DataFrame(res, columns=[k for k in res[0].keys()])
return data

函数如下所示：

    def form_sql_statement_from_dictionary(
    dictionary: Dict[str, Any]) -> str:

    hashvalue = list(dictionary.values())
    scope = hashvalue[0]["scope"]

    dictionary_element_names = list(scope.keys())

    statement_elements = []
    for element_name in dictionary_element_names:
        dictionary_element_values = scope[element_name]

        if len(dictionary_element_values) == 1:
            dictionary_element_value = dictionary_element_values[0]
            statement_element = (
                f"( {prefix}{element_name} = '{dictionary_element_value}' )"
            )
            statement_elements.append(statement_element)
        else:
            statement_or_elements = []
            for dictionary_element_value in dictionary_element_values:
                statement_element = (
                    f"{prefix}{element_name} = '{dictionary_element_value}'"
                )
                statement_or_elements.append(statement_element)
            final_or_statement = "( " + " OR ".join(statement_or_elements) + ")"
            statement_elements.append(final_or_statement)

    final_statement = " AND " + " AND ".join(statement_elements)
    return final_statement

标签： pythonsql

解决方案

在此处找到详细信息：


首先，使用位置参数构建 SQL where 子句，例如...

(x.col1 = %s) AND (x.col2 = %s OR x.col3 = %s)

同时创建这些参数的列表

['foo', 'foo', 'bar']

然后使用参数化查询...

cursor = connection.cursor()
cursor.execute(sql, parameters)
data = await cursor.fetchall()

参数化将为您引用和转义所有参数，因此不会受到 SQL 注入攻击。

但是列名仍被直接替换到查询中。没有内置的方法可以保护您免受这种影响。如果用户可以直接控制这些字符串，他们仍然可以通过这种方式破解您。

因此，您必须通过适合您的用例的任何方式自行监管、验证这些列名称。


总而言之，修改后的 python 看起来像......

def form_sql_statement_from_dictionary(dictionary):
    hashvalue = list(dictionary.values())
    scope = hashvalue[0]["scope"]
    dictionary_element_names = list(scope.keys())
    prefix='ummmmmmmmmm.'
    statement_elements = []
    statement_params   = []
    for element_name in dictionary_element_names:
        statement_elements.append(
            ' OR '.join(
                f"{prefix}{element_name} = %s"
                    for item in scope[element_name]
            )
        )
        statement_params += scope[element_name]
    return '(' + ') AND ('.join(statement_elements) + ')', statement_params


async def query_for_data(
    connection: "PgService", dictionary: Dict[str, Any]
) -> pd.DataFrame:
    sql_where_clause, sql_params = form_sql_statement_from_dictionary(dictionary)
    sql = f"""
    *multiple SQL statements*
    where 1 = 1
       and selector in ('YES', 'NO')
       and {sql_where_clause}
    """
    cursor = connection.cursor()
    cursor.execute(sql, sql_params)
    res = await cursor.fetchall()
    data = pd.DataFrame(res, columns=[k for k in res[0].keys()])
    return data

推荐阅读