Python psycopg2实现PostgreSQL WHERE IN安全参数化查询防注入

你当前手动拼接引号、插入值到SQL字符串的写法确实存在明确的SQL注入风险，只要filter_data表的Value字段中包含单引号或者恶意SQL片段，就会触发语法错误或者注入攻击。psycopg2本身原生支持参数化查询处理列表类型的IN条件，完全不需要手动拼接值。

改造要点 #

• 从数据库A取出待匹配值之后，不要做任何加单引号、字符串拼接、手动转义的操作，直接保留原始值组成Python列表即可
• 构造SQL时根据列表长度动态生成对应数量的%s占位符，不要把值直接拼进SQL语句
• 把值列表作为参数传给execute方法，由psycopg2驱动内部完成安全转义，从底层隔离SQL语法和数据参数，彻底避免注入风险

可直接替换的代码 #

from psycopg2 import sql

# 可顺手修复原函数中重复创建游标的冗余问题
def select_command_postgres_no_argument(conn, query):
    with conn.cursor() as cur:
        cur.execute(query)
        return cur.fetchall()

def select_command_postgres_with_argument(conn, query, sql_args=()):
    with conn.cursor() as cur:
        cur.execute(query, sql_args)
        return cur.fetchall()

# 从数据库A读取待匹配的商品列表
commodity_records = select_command_postgres_no_argument(postgres_conn, "SELECT Value FROM filter_data")
commodity_list = [row[0] for row in commodity_records]

# 安全构造IN查询语句，自动生成对应数量的占位符
query = sql.SQL("SELECT * FROM forecast_data WHERE Commodity IN ({})").format(
    sql.SQL(', ').join(sql.Placeholder() * len(commodity_list))
)

# 传入参数执行查询，所有值由驱动自动安全转义
final_result = select_command_postgres_with_argument(postgres_conn_b, query, commodity_list)

方案说明 #

• 这种写法下，所有传入的参数值都会被psycopg2按照PostgreSQL通信协议做严格转义，参数和SQL语句结构是分开传输解析的，哪怕值中包含单引号、DROP TABLE之类的恶意内容，也只会被当成普通的字段值处理，完全不会触发注入。
• 不需要自行实现任何转义逻辑，驱动内置的转义规则和数据库版本完全适配，不存在自行转义常见的绕过漏洞。
• 自动适配任意长度的匹配列表，不需要手动维护占位符数量。

禁止使用自行替换单引号、正则过滤特殊字符之类的自实现转义方案，这类方案几乎都存在可被绕过的风险，数据库驱动提供的原生参数化能力是唯一安全的选择。

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