防止SQL注入的SQLAlchemy中text参数传递注意事项

在SQLAlchemy中使用text()构造原生SQL时，防止SQL注入的核心原则只有一条：永远不要用字符串拼接或格式化的方式把用户输入塞进SQL语句里，而是通过text()的bindparams机制或者ORM的params参数来传递参数。很多开发者以为用了ORM就万事大吉，结果在需要写复杂查询时直接用f-string把变量拼进去，这就等于把大门敞开给攻击者。下面我会把所有你需要知道的细节、踩坑点和最佳实践一次性讲透。

一、为什么text()函数容易出现SQL注入

SQLAlchemy的text()函数主要用于执行原生SQL字符串。它本身不是不安全的，但它的设计初衷是让你能写出接近数据库原生语法的SQL，这就意味着如果你不小心，很容易回到"拼接字符串"的老路上。比如你写了这样的代码：

from sqlalchemy import text

username = request.args.get('username')
query = text(f"SELECT * FROM users WHERE name = '{username}'")
result = session.execute(query)

这段代码就是一个典型的SQL注入漏洞。攻击者只需要在username参数中输入 ' OR '1'='1，整个查询逻辑就被篡改了。text()函数本身没有任何防护能力，安全与否完全取决于你怎么传参。

二、正确使用text()传递参数的三种方式

方式一：使用冒号命名参数（推荐）

这是最常用也是最推荐的方式。你在SQL中用冒号加参数名作为占位符，然后通过execute时传入一个字典来绑定值：

from sqlalchemy import text

username = request.args.get('username')
query = text("SELECT * FROM users WHERE name = :username")
result = session.execute(query, {"username": username})

SQLAlchemy会自动对传入的值进行转义和参数化处理，数据库驱动层面会把它当作纯数据而不是SQL代码来执行。这种方式无论你的输入是什么，都不会改变SQL的结构。

方式二：使用问号位置参数

部分数据库驱动（如SQLite）支持问号占位符，写法如下：

query = text("SELECT * FROM users WHERE name = ?")
result = session.execute(query, (username,))

注意这里传的是元组，哪怕只有一个参数也要加逗号。这种方式兼容性稍差，不是所有数据库都支持，但在特定场景下很简洁。

方式三：使用bindparam显式绑定

当你需要对参数指定类型或者设置其他属性时，可以用bindparam：

from sqlalchemy import text, bindparam

query = text("SELECT * FROM users WHERE id = :user_id AND status = :status")
query = query.bindparams(
    bindparam("user_id", type_=Integer),
    bindparam("status", type_=String)
)
result = session.execute(query, {"user_id": 123, "status": "active"})

这种方式在需要明确类型约束、防止类型注入攻击时特别有用。比如id字段如果被传入字符串类型的恶意数据，显式指定Integer类型可以在驱动层就拦截掉。

三、动态SQL拼接时的安全做法

实际开发中经常遇到需要动态构建WHERE条件的场景，比如用户可以选择按姓名、邮箱、年龄等多个字段筛选。这时候很多人会犯一个错误：把字段名也用参数化的方式处理，这是行不通的。SQL参数化只能处理值，不能处理表名、列名、ORDER BY方向等结构性内容。

正确的做法是：结构性内容用白名单校验，值的部分用参数化。例如：

from sqlalchemy import text

# 白名单校验字段名
allowed_columns = {"name", "email", "age", "created_at"}
sort_column = request.args.get("sort", "created_at")
sort_order = request.args.get("order", "desc")

if sort_column not in allowed_columns:
    sort_column = "created_at"
if sort_order not in ("asc", "desc"):
    sort_order = "desc"

# 值的部分参数化
filters = []
params = {}

if name := request.args.get("name"):
    filters.append("name = :name")
    params["name"] = name

if min_age := request.args.get("min_age"):
    filters.append("age >= :min_age")
    params["min_age"] = int(min_age)

where_clause = " AND ".join(filters) if filters else "1=1"
query = text(f"SELECT * FROM users WHERE {where_clause} ORDER BY {sort_column} {sort_order}")
result = session.execute(query, params)

关键点在于：字段名和排序方向通过白名单硬编码校验，绝不直接用用户输入拼接；而所有过滤条件的值都通过params字典参数化传递。这样既保证了灵活性，又堵住了注入漏洞。

四、IN子句和批量参数的处理技巧

当需要用IN子句查询多个值时，比如 WHERE id IN (1, 2, 3)，不能简单地把列表拼成字符串。SQLAlchemy提供了几种安全的处理方式：

# 方式一：展开为多个命名参数
ids = [1, 2, 3, 4, 5]
placeholders = ", ".join([f":id_{i}" for i in range(len(ids))])
query = text(f"SELECT * FROM users WHERE id IN ({placeholders})")
params = {f"id_{i}": id_val for i, id_val in enumerate(ids)}
result = session.execute(query, params)

这种方式适合ID数量不多的场景。如果ID列表很长（比如上千个），可以用另一种方式：

# 方式二：使用bindparam配合expanding
from sqlalchemy import bindparam

query = text("SELECT * FROM users WHERE id IN :ids")
result = session.execute(query, {"ids": tuple(ids)})

不过要注意，expanding参数在某些数据库驱动下可能有数量限制或者性能问题，具体需要看你使用的数据库。PostgreSQL对数组参数支持较好，MySQL则需要用展开方式更稳妥。

五、常见的错误写法和隐蔽的注入点

除了明显的字符串拼接，还有几种容易被忽视的注入场景：

1. 用字符串的format方法或%格式化

# 错误！
query = text("SELECT * FROM users WHERE name = '%s'" % username)
# 错误！
query = text("SELECT * FROM users WHERE name = '{}'".format(username))

这两种写法和f-string本质一样，都是在Python层面完成字符串拼接后才发给数据库，完全绕过了参数化机制。

2. 在LIKE查询中直接拼接通配符

# 错误！
query = text(f"SELECT * FROM users WHERE name LIKE '%{keyword}%'")

# 正确！
query = text("SELECT * FROM users WHERE name LIKE :pattern")
result = session.execute(query, {"pattern": f"%{keyword}%"})

通配符的拼接应该在Python层面做好，然后把完整的模式字符串作为参数传入，而不是在SQL语句里拼。

3. 用concat函数拼接SQL片段

有些开发者觉得用数据库的concat函数拼接就安全了，其实不然。如果concat的参数来自用户输入，依然存在注入风险。参数化永远是第一选择。

六、text()与ORM查询的安全对比

很多人会问：既然ORM的查询方式天然参数化，为什么还要用text()？原因很简单：ORM不是万能的。当你需要写窗口函数、CTE（公共表表达式）、复杂的子查询、数据库特定函数（如PostgreSQL的jsonb操作、MySQL的GROUP_CONCAT）时，ORM的表达能力有限，必须回到原生SQL。

但要记住一个原则：能用ORM解决的就用ORM，必须用text()的时候严格参数化。两者的安全边界是一样的——都依赖参数化查询来防注入，区别只在于ORM帮你自动做了这件事，而text()需要你手动做。

# ORM方式（自动参数化）
from sqlalchemy.orm import Session

users = session.query(User).filter(User.name == username).all()

# text()方式（需要手动参数化）
query = text("SELECT * FROM users WHERE name = :name")
users = session.execute(query, {"name": username}).fetchall()

七、连接池和session层面的安全配置

除了代码层面的参数化，还有一些基础设施层面的配置值得注意。SQLAlchemy的Engine和Session在创建时可以设置一些安全相关的参数：

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+psycopg2://user:pass@host/db",
    # 开启语句日志，方便审计
    echo=False,
    # 设置连接池大小，防止资源耗尽攻击
    pool_size=10,
    max_overflow=20,
    # 设置语句超时
    connect_args={"connect_timeout": 10}
)

虽然这些配置不直接防SQL注入，但合理的超时和连接池控制可以在一定程度上减轻注入攻击带来的性能损耗和资源耗尽风险。

八、审计和测试：如何验证你的代码没有注入漏洞

写完代码不代表就安全了。建议在开发阶段做以下几件事：

第一，开启SQLAlchemy的echo模式，打印所有实际执行的SQL语句，检查参数是否正确绑定而不是被拼进了字符串。

第二，写单元测试时专门构造恶意输入，比如包含单引号、双引号、分号、注释符的字符串，验证查询结果是否符合预期且没有报错或异常行为。

第三，使用静态代码分析工具扫描项目中所有text()调用，检查是否存在字符串拼接的模式。很多IDE插件和CI工具都支持这种检测。

九、总结：记住这几条铁律

第一，text()中的SQL语句永远用冒号参数或bindparam传值，绝不拼接。第二，表名、列名等结构性内容用白名单校验，不要参数化。第三，LIKE、IN、批量操作等特殊场景都有对应的安全写法，不要图省事。第四，能用ORM就用ORM，必须用原生SQL时保持警惕。第五，上线前做安全测试，不要心存侥幸。

SQL注入是Web安全中最古老但依然最致命的漏洞之一。SQLAlchemy给了你足够强大的工具来防御它，但工具不会自动保护你——只有你自己写对了代码，安全才真正存在。把参数化查询当成肌肉记忆，每次写text()的时候都先问自己一句：这个值是参数化传进去的吗？做到这一点，你就已经挡住了绝大多数注入攻击。

防止SQL注入的SQLAlchemy中text参数传递注意事项

服务指南

帮助中心