如何为Pandas创建带动态表名的参数化查询以避免SQL注入？

我使用SQL Server、pandas和pyodbc开发，需要编写一个函数，接收列名、表名、GUID类型列名及GUID值，执行类似如下的SQL查询：

SELECT columnName1 
FROM tableName 
WHERE columnName2 = guidValue

未做注入防护时的可行代码 #

之前不考虑SQL注入风险时，手动给表名添加方括号的函数可以正常工作，我会直接传入完整的WHERE子句作为参数（例如"[Name] = 'Bob Smith'"）：

def get_sql_entity_set_data(entity_set_name, filter_expression=None, query_params='*'):
    connString = get_connection_string()
    conn = pyodbc.connect(connString)
    query = "SELECT " + query_params + " FROM [" + entity_set_name + "]"

    if filter_expression:
        query += " WHERE " + filter_expression

    logger.debug(f'Attempting the following SQL command: {query}.')
    retValue = pd.read_sql(query, conn)

    returnCount = len(retValue.index)

    if returnCount == 0:
        logger.warning('Fetch failed: no data retrieved.')
    else:
        logger.info(f'{returnCount} row(s) retrieved.')

    return retValue

注入防护改造后的异常情况 #

现在想要做SQL注入防护，但遇到了诸多问题：

• 表名和通配符*无法通过参数化方式处理，添加大量分支逻辑也没能解决；
• 尝试用EXEC编写预语句的方案也失败了。

最终我退而求其次，仍将表名直接写入查询，仅用函数包裹方括号，但代码执行后始终返回0条结果——但在SSMS中用相同参数执行查询能得到1条结果。当前的代码如下：

def get_data_with_guid_filter(entity_set_name, filter_col, filter_val):
    connString = get_connection_string()
    conn = pyodbc.connect(connString)
    logger.debug(f'Attempting to fetch all data from {entity_set_name} where {filter_col} = {filter_val}.')
    entity_set_name = parameterize(entity_set_name, "sysid")
    filter_col = parameterize(filter_col, "sysid")
    filter_val = parameterize(filter_val, "guid")
    params_tuple = (filter_col,) + (filter_val,)
    query = f"SELECT * FROM {entity_set_name} WHERE ? = ?"

    retValue = pd.read_sql(query, conn, params=params_tuple)

    returnCount = len(retValue.index)

    if returnCount == 0:
        logger.warning('Fetch failed: no data retrieved.')
    else:
        logger.info(f'{returnCount} row(s) retrieved.')

    return retValue

def parameterize(param, param_type):
    if param_type == "guid":
        return f"(SELECT CONVERT(uniqueidentifier, '{param}'))"
    elif param_type == "string":
        return f"'{param}'"
    elif param_type == "sysid":
        return f"[{param}]"
    else:
        return param

举个具体例子：传入参数'mytable'、'ID'、'123456'（模拟真实GUID），在SSMS中执行SELECT * FROM [mytable] WHERE [ID] = (SELECT CONVERT(uniqueidentifier, '123456'))能得到1条结果，但Python代码执行后返回0结果且无任何报错。

我还尝试了以下修改，结果仍然返回0条数据：

• 将filter_val = parameterize(filter_val, "guid")改为filter_val = parameterize(filter_val, "string")（此前用EXEC方法时曾因GUID列用nvarchar过滤报错）；
• 移除filter_val = parameterize(filter_val, "guid")返回值的外层括号（尽管SSMS中执行似乎需要该括号）。

内容的提问来源于stack exchange，提问作者ysi_d