PythonFlask接口防SQL注入与ORM查询安全实践

在Python Flask应用开发中，SQL注入漏洞长期占据OWASP Top 10的前列，根本原因在于开发者将用户输入与SQL指令混为一谈。Flask本身不提供内置的SQL注入防护层，这意味着你写的每一行原生SQL都可能成为攻击者的入口。最典型的攻击场景是登录表单，假设你写了这样的代码：cursor.execute(f"SELECT * FROM users WHERE username='{username}' AND password='{password}'")，攻击者在用户名框输入 admin' -- ，后端拼接出的SQL就变成了 SELECT * FROM users WHERE username='admin' --' AND password='...' ，注释符 -- 直接让密码验证失效，瞬间绕过认证系统。这不是理论推演，而是每天在真实系统中上演的攻击模式。

解决原生SQL注入的方法极其明确：永远不要拼接SQL字符串，只使用参数化查询。在Flask中，无论你用PyMySQL、mysql-connector-python还是psycopg2，底层都支持占位符传参。以PyMySQL为例，正确写法是 cursor.execute("SELECT * FROM users WHERE username=%s AND password=%s", (username, password)) 。这里的 %s 不是Python字符串格式化符号，而是数据库驱动的占位符，驱动层会将参数与SQL指令彻底分离，把参数值当作纯数据处理，而非可执行代码。即便攻击者传入 admin' -- ，数据库也只会把它当作一个奇怪的用户名去匹配，找不到就返回空结果，不会执行任何恶意逻辑。对于SQLite用户，占位符用问号： cursor.execute("SELECT * FROM users WHERE username=? AND password=?", (username, password)) 。PostgreSQL的psycopg2则使用 %s 作为占位符。关键原则是：占位符的位置绝不能用Python的f-string或.format()或百分号格式化替代，那是自毁长城。

很多开发者以为用了SQLAlchemy ORM就天然免疫SQL注入，这是一个危险的误解。ORM确实在绝大多数常规查询中自动处理参数化，当你写 User.query.filter_by(username=username).first() 时，SQLAlchemy内部会将参数安全地绑定到SQL语句中。但问题出在那些需要灵活查询的场景，开发者往往图方便直接写原始SQL片段。SQLAlchemy提供了 text() 函数来执行原生SQL，如果你写成 db.session.execute(text(f"SELECT * FROM users WHERE username='{username}'")) ，这跟直接拼接字符串没有任何区别，ORM的保护机制被你亲手绕过了。正确的做法是使用 text() 的绑定参数功能： db.session.execute(text("SELECT * FROM users WHERE username=:username"), {"username": username}) 。冒号开头的命名参数让SQLAlchemy在底层进行安全绑定，既保留了原生SQL的灵活性，又守住了安全底线。

在实际业务中，前端经常需要向后端传递排序字段或分组字段，比如 /api/users?sort_by=created_at&order=desc 。这类需求无法用常规的占位符处理，因为占位符只能绑定值，不能绑定列名或SQL关键字。很多开发者在这里犯下致命错误，直接拼接列名： f"SELECT * FROM users ORDER BY {sort_by} {order}" 。攻击者只需把 sort_by 改成 (SELECT password FROM users LIMIT 1) 就能窃取敏感数据。解决这个问题必须使用白名单校验。在代码中定义一个允许排序的字段集合： ALLOWED_SORT_COLUMNS = {"id", "username", "email", "created_at"} ，然后检查用户传入的 sort_by 是否在这个集合中，不在就回退到默认排序字段。排序方向同理，只允许 ASC 和 DESC 两个值。代码示例如下：

ALLOWED_SORT_COLUMNS = {"id", "username", "email", "created_at"}
ALLOWED_ORDER = {"asc", "desc"}

sort_by = request.args.get("sort_by", "id")
order = request.args.get("order", "desc").lower()

if sort_by not in ALLOWED_SORT_COLUMNS:
    sort_by = "id"
if order not in ALLOWED_ORDER:
    order = "desc"

query = f"SELECT * FROM users ORDER BY {sort_by} {order}"
db.session.execute(text(query))

注意，即便经过白名单校验，这里的 sort_by 和 order 已经是安全的固定值，直接拼接进SQL也不会引入注入风险，因为攻击者无法控制这两个变量的实际内容。但更严谨的做法是使用SQLAlchemy的列对象动态构建查询，完全避免字符串拼接： order_column = getattr(User, sort_by) ，然后使用 order_by(order_column.desc() if order == "desc" else order_column.asc()) ，这样连最后一点字符串拼接都消除了。

搜索功能中的LIKE查询是另一个容易翻车的地方。开发者经常写出这样的代码： f"SELECT * FROM products WHERE name LIKE '%{keyword}%'" 。攻击者可以通过构造特殊字符绕过业务逻辑，甚至在某些数据库配置下实现注入。正确的做法仍然是把用户输入当作值来绑定，把百分号通配符拼在参数值上，而不是拼在SQL语句上： db.session.execute(text("SELECT * FROM products WHERE name LIKE :keyword"), {"keyword": f"%{keyword}%"}) 。这样即使用户输入包含单引号或SQL关键字，也完全被当作搜索文本处理。SQLAlchemy ORM的写法更简洁： Product.query.filter(Product.name.like(f"%{keyword}%")).all() ，ORM会自动处理转义。

当业务需要查询多个ID时，比如前端传入 /api/users?ids=1,2,3 ，后端需要生成 WHERE id IN (1,2,3) 的查询。如果直接 split 后拼接字符串，同样存在注入风险。很多开发者不知道的是，IN子句也可以参数化。对于原生SQL，需要动态生成占位符：

ids = [int(i) for i in request.args.get("ids", "").split(",") if i.isdigit()]
if not ids:
    return []
placeholders = ",".join([f":id_{i}" for i in range(len(ids))])
params = {f"id_{i}": ids[i] for i in range(len(ids))}
query = f"SELECT * FROM users WHERE id IN ({placeholders})"
result = db.session.execute(text(query), params).fetchall()

这段代码首先生成 :id_0,:id_1,:id_2 这样的占位符序列，然后把对应的值放入字典，彻底杜绝注入可能。如果使用SQLAlchemy ORM，事情简单得多： User.query.filter(User.id.in_(ids)).all() ，ORM自动处理所有参数绑定。这里额外强调一点，上面的代码先用列表推导式将字符串转换为整数，这一步不仅做了类型安全转换，也天然过滤掉了非数字的恶意输入，属于纵深防御的体现。

有些团队习惯把复杂业务逻辑封装在数据库存储过程中，然后在Flask中调用。存储过程内部如果使用了动态SQL拼接，应用层做再多防护也是徒劳。假设你有一个存储过程 CREATE PROCEDURE search_users(IN keyword VARCHAR(255)) ，内部写了 SET @sql = CONCAT('SELECT * FROM users WHERE username LIKE \'%', keyword, '%\'') ，然后 PREPARE 和 EXECUTE ，这依然是SQL注入的温床。在存储过程中处理动态SQL，必须使用数据库提供的参数化执行机制，比如MySQL的 PREPARE 语句配合 USING 传参，而不是CONCAT拼接。从架构层面讲，我建议尽量减少存储过程中动态SQL的使用，把查询逻辑放在应用层用ORM或参数化SQL实现，让数据库专注于数据存取。

SQL注入不是唯一的安全威胁，ORM使用不当还可能导致数据泄露或性能崩溃。SQLAlchemy的session管理直接影响数据一致性。在Flask中，推荐使用 flask_sqlalchemy 扩展，它自动处理session的创建和销毁，每个请求结束后自动提交或回滚。如果你手动管理session，务必使用 try-except-finally 确保session被正确关闭，否则数据库连接泄漏会逐渐耗尽连接池，最终导致服务不可用。另一个常见问题是批量操作时的性能陷阱：在循环中逐条插入数据会生成大量单独的INSERT语句，正确做法是使用 bulk_insert_mappings 或 bulk_save_objects ，一次性将多条记录发送给数据库，性能提升可达数十倍。但这两种批量方法默认不触发ORM事件，也不加载Python对象关系，需要根据业务场景权衡。

即使防住了SQL注入，如果ORM查询不加选择地返回所有字段，也可能造成敏感信息过度暴露。比如用户表包含密码哈希、手机号、身份证号等字段，在列表查询接口中绝不应该返回这些字段。SQLAlchemy提供了多种方式控制返回字段：使用 with_entities() 指定查询列，或者定义模型时使用 deferred() 将敏感字段标记为延迟加载，只有显式访问时才从数据库查询。更彻底的方案是使用DTO（数据传输对象）模式，在ORM查询结果和API响应之间加一层转换，只暴露前端需要的字段。这层转换可以用 marshmallow 或 pydantic 实现，不仅做字段过滤，还能做类型校验和序列化。代码示例：

from marshmallow import Schema, fields

class UserPublicSchema(Schema):
    id = fields.Int()
    username = fields.Str()
    created_at = fields.DateTime()

# 在视图函数中
users = User.query.filter_by(is_active=True).all()
schema = UserPublicSchema(many=True)
return jsonify(schema.dump(users))

这样即使User模型包含password_hash字段，API响应中也绝不会出现，因为Schema根本没定义它。这是一种纵深防御思路，即使ORM层查询了多余字段，序列化层也会把它们过滤掉。

安全防护不能只靠代码层面的防御，还需要可观测性来发现潜在攻击。在Flask应用中，建议开启SQLAlchemy的查询日志记录，将SQL语句和执行参数输出到日志系统。通过配置 SQLALCHEMY_ECHO=True 或更精细地使用Python的logging模块设置 sqlalchemy.engine 的日志级别为INFO，可以捕获所有实际执行的SQL语句。当监控系统发现异常模式时，比如短时间内大量包含UNION、SELECT、DROP等关键字的查询，或者单个IP尝试多种不同的注入载荷，就能触发告警。这不是预防措施，而是检测和响应措施，是安全体系中不可或缺的一环。结合Flask的请求钩子 before_request 和 after_request ，可以记录每个请求的IP、User-Agent、请求参数和执行耗时，为事后审计提供完整链路。

Flask生态中大量第三方扩展内部也执行SQL查询，它们的安全性取决于开发者的编码习惯。使用 flask-admin 或 flask-appbuilder 这类自动生成管理后台的扩展时，务必检查它们的默认配置是否开启了安全的查询模式。某些旧版本的扩展可能存在已知的SQL注入漏洞，定期更新依赖版本是基本的安全实践。可以通过 pip-audit 或 safety 工具扫描项目依赖中的已知漏洞。另外，不少开发者喜欢从网上复制粘贴代码片段来解决特定问题，这些片段往往为了简洁而牺牲安全性，比如直接拼接SQL字符串的示例代码。引入任何外部代码前，务必审查其中的数据库查询方式。

单靠参数化查询或ORM不足以构成完整的安全防线。一个健壮的防注入体系至少包含四层：第一层是输入校验，在数据进入应用逻辑之前，用白名单或正则表达式过滤明显的恶意载荷，比如拒绝包含分号、UNION、SELECT等SQL关键字的输入（注意不要误伤正常业务，校验规则需要精细设计）。第二层是参数化查询，确保所有SQL操作都通过占位符绑定参数。第三层是最小权限原则，数据库连接账号只授予必要的权限，比如Web应用账号通常只需要SELECT、INSERT、UPDATE、DELETE权限，不应拥有DROP、ALTER等DDL权限，这样即使攻击者绕过了前两层，破坏范围也受到限制。第四层是Web应用防火墙（WAF），在流量入口处检测和拦截SQL注入攻击模式。四层防御互为补充，任何一层被突破都有后续机制兜底。

安全是一个持续对抗的过程，今天的防护措施明天可能就出现新的绕过技术。保持对OWASP最新动态的关注，定期对代码进行安全审计，使用自动化工具如Bandit扫描Python代码中的安全漏洞，把SQL注入的防护意识融入团队的开发规范中，远比记住几个代码片段更重要。你写的每一行数据库查询代码，都是在构建系统的安全边界。