在Psycopg中用Python %操作符插入字段名构造查询是否安全？

已有资料明确，用Python的+或%字符串操作符直接拼接SQL值到Psycopg2查询中存在安全风险。但我有个疑问：如果通过cursor.execute(query, values)正确将值字典作为第二个参数传递，此时用%操作符插入字段名来构造动态查询是否安全？

我的需求是从可变长度的字段/值对字典动态生成SQL查询，目前用可信输入能正常运行，但想明确：仅通过修改字典的键或值，是否能发起SQL注入攻击？同时也希望了解这类需求的更优实现方案。

附上当前实现代码：

from psycopg2 import sql

values = {
    "field1": "value1",
    "field2": "value2",
    "field3": "value3",
}

conditions = {
    "field4": "value4",
    "field5": "value5",
}

select_fields = [
    "field1",
    "field3",
]

# 插入记录：使用values字典中的字段和值
insert_query = sql.SQL(
    r"INSERT INTO tablename (%s) VALUES (%s)"
    % (
        ",".join(values),
        ",".join(r"%%(%s)s" % key for key in values)
    )
)
# 等价于：
insert_query = sql.SQL(
    r"INSERT INTO tablename (field1,field2,field3) VALUES (%(field1)s,%(field2)s,%(field3)s)"
)


# 删除符合conditions字典中所有字段/值对的记录
delete_query = sql.SQL(
    r"DELETE FROM tablename WHERE (%s) = (%s)"
    % (
        ",".join(conditions),
        ",".join(r"%%(%s)s" % key for key in conditions),
    )
)
# 等价于：
delete_query = sql.SQL(
    r"DELETE FROM tablename WHERE (field4,field5) = (%(field4)s,%(field5)s)"
)


# 查询符合conditions字典中所有字段/值对的记录
select_query = sql.SQL(
    r"SELECT %s FROM tablename WHERE (%s) = (%s)"
    % (
        ",".join(select_fields),
        ",".join(conditions),
        ",".join(r"%%(%s)s" % key for key in conditions),
    )
)
# 等价于：
select_query = sql.SQL(
    r"SELECT field1,field3 FROM tablename WHERE (field4,field5) = (%(field4)s,%(field5)s)")

# 后续会执行 cursor.execute(a_query, values)，不在本次问题范围内

我曾搭建测试库尝试向字典键/值注入SQL，生成的查询看似有注入条件，但Psycopg执行时持续报错，仍不确定是否真的安全。

一、安全性分析 #

1. 字段名拼接的风险
   直接用Python的%操作符拼接字段名存在SQL注入风险。因为字段名属于SQL标识符（而非参数值），Psycopg的参数化机制只负责处理cursor.execute第二个参数传入的值，不会对拼接进SQL的字段名做任何转义或校验。如果字典的键来自不可信输入（比如用户提交的内容），攻击者可以构造类似"field1); DROP TABLE tablename;--"的键，拼接后会生成恶意SQL语句——你的测试报错只是特定注入语句的执行问题，不代表没有风险。

2. 值传递的安全性
   字典中的值通过cursor.execute的第二个参数传递，这部分是完全安全的。Psycopg会将这些值作为参数绑定到查询中，不会将其解析为SQL代码，因此无论值的内容是什么，都不会引发SQL注入。

二、更优实现方案 #

Psycopg官方提供了sql模块专门用于动态构造包含标识符（字段名、表名等）的SQL，这是安全且推荐的做法：

1. 动态INSERT示例 #

from psycopg2 import sql

values = {
    "field1": "value1",
    "field2": "value2",
    "field3": "value3",
}

# 用sql.Identifier包装字段名，自动处理转义
fields = sql.SQL(',').join(map(sql.Identifier, values.keys()))
# 生成命名占位符
placeholders = sql.SQL(',').join(sql.SQL(f"%({k})s") for k in values.keys())

insert_query = sql.SQL("INSERT INTO tablename ({}) VALUES ({})").format(fields, placeholders)
# 执行时直接传值字典
# cursor.execute(insert_query, values)

2. 动态DELETE示例 #

conditions = {
    "field4": "value4",
    "field5": "value5",
}

cond_fields = sql.SQL(',').join(map(sql.Identifier, conditions.keys()))
cond_placeholders = sql.SQL(',').join(sql.SQL(f"%({k})s") for k in conditions.keys())

delete_query = sql.SQL("DELETE FROM tablename WHERE ({}) = ({})").format(cond_fields, cond_placeholders)
# cursor.execute(delete_query, conditions)

3. 动态SELECT示例 #

select_fields = ["field1", "field3"]
conditions = {
    "field4": "value4",
    "field5": "value5",
}

select_cols = sql.SQL(',').join(map(sql.Identifier, select_fields))
cond_fields = sql.SQL(',').join(map(sql.Identifier, conditions.keys()))
cond_placeholders = sql.SQL(',').join(sql.SQL(f"%({k})s") for k in conditions.keys())

select_query = sql.SQL("SELECT {} FROM tablename WHERE ({}) = ({})").format(select_cols, cond_fields, cond_placeholders)
# cursor.execute(select_query, conditions)

这种方式的核心是用sql.Identifier处理所有SQL标识符，确保它们被正确转义，彻底避免SQL注入风险，同时保持代码的可读性和可维护性。

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