Python SQLAlchemy如何规避SQL注入风险？-SQL-PHP中文网

sqlalchemy的text()函数需用命名或位置参数绑定防sql注入，动态字段名须白名单校验，filter()/filter_by()应避免字符串拼接，execute()传参禁用字符串格式化。

python sqlalchemy如何规避sql注入风险？

SQLAlchemy的`text()`函数必须手动转义参数

直接拼接字符串进text()是SQL注入高发区。比如text("SELECT * FROM users WHERE name = '" + name + "'")，一旦name含单引号或UNION SELECT就崩。

正确做法是用命名参数或位置参数绑定：

from sqlalchemy import text
# ✅ 命名参数（推荐）
stmt = text("SELECT * FROM users WHERE status = :status AND age > :min_age")
result = conn.execute(stmt, {"status": "active", "min_age": 18})
<h1>✅ 位置参数（注意顺序）</h1><p>stmt = text("SELECT * FROM users WHERE id IN (:id1, :id2)")
result = conn.execute(stmt, {"id1": 101, "id2": 205})</p>

注意：text()本身不解析Python变量，只靠DBAPI底层做参数化，所以:param必须严格匹配字典键名；传错键名会报sqlalchemy.exc.StatementError: (builtins.KeyError) 'xxx'。

`filter()`和`filter_by()`默认安全，但别混用字符串

ORM查询中filter()接受表达式对象（如User.name == value），由SQLAlchemy自动转成参数化语句；filter_by()只支持等值键值对，也安全。

常见错误是把用户输入塞进字符串里再传给filter()：

# ❌ 危险！字符串拼接进filter()
query.filter("name = '" + request.args.get("q") + "'")
<h1>✅ 安全：用表达式</h1><p>name_q = request.args.get("q")
query.filter(User.name == name_q)</p><h1>✅ 安全：用filter_by()（仅限=）</h1><p>query.filter_by(name=name_q)</p>

注意：filter_by()不支持!=、like等操作，强行用filter_by(name__like="%x%")会报AttributeError——那是Flask-SQLAlchemy的旧写法，原生SQLAlchemy不认。

Python 3.14.2

Python 3.14.2是Python编程语言在2025年12月5日发布的稳定版本，属于3.14系列的第二个维护更新。该版本包含了18项修复，重点解决了多进程、数据类及正则表达式等模块的回归问题，并修复了CVE-2025-12084等安全漏洞。此版本标志着自由线程模式（移除GIL）正式获得官方支持，是Python发展的重要里程碑。

下载

原生SQL执行时，`execute()`的参数传递方式决定安全性

即使用了text()，如果调用execute()时传参方式不对，照样中招：

用execute(stmt, {"k": v})或execute(stmt, k=v) → ✅ 安全
用execute(stmt % {"k": v})或execute(stmt.format(...)) → ❌ 等同于字符串拼接
用execute(stmt, [v1, v2])配位置占位符:1, :2 → ⚠️ 少数方言（如Oracle）支持，但PostgreSQL/MySQL不认，易出错

特别提醒：conn.execute("SELECT * FROM t WHERE x = %s", (val,))这种写法看似像psycopg2风格，但在SQLAlchemy 2.0+中已弃用——它绕过了SQLAlchemy的参数处理层，可能触发警告甚至失效。

动态字段名（如ORDER BY列名）无法参数化，必须白名单校验

SQL标准不允许参数化字段名、表名、排序方向（ASC/DESC）。这意味着ORDER BY :col会报错，不是语法问题，是协议限制。

解决方案只有硬校验：

allowed_sort_cols = {"username", "email", "created_at"}
sort_col = request.args.get("sort", "username")
if sort_col not in allowed_sort_cols:
    sort_col = "username"
<h1>✅ 拼接前确认在白名单内</h1><p>stmt = text(f"SELECT * FROM users ORDER BY {sort_col} DESC LIMIT 10")</p>

别试图用getattr(User, col)代替——如果col是用户可控字符串，getattr(User, col)可能抛AttributeError，但更危险的是它可能访问到非预期的混合属性或关系字段，间接暴露数据结构。

最易被忽略的一点：SQL注入不止发生在WHERE子句。动态GROUP BY、HAVING、UNION分支、甚至CTE名称，只要来自外部输入，都得走白名单或正则过滤。别让“参数化”三个字麻痹判断力。