使用SQLAlchemy的text函数拼接查询是否存在SQL注入风险?
你当前的实现确实存在明确的SQL注入风险,直接将用户可控的输入拼接进SQL语句文本,没有任何校验和转义逻辑,恶意用户完全可以构造特殊的range参数实现注入攻击。
以下是兼顾查询灵活性和安全性的优化方案:
- 第一步:对查询字段做白名单校验
字段名无法通过SQL参数绑定处理,所以必须先定义允许作为范围查询条件的字段白名单,只放行合法的模型字段,避免用户传入恶意字段名或SQL语句片段。你可以直接从你的数据模型类中动态获取合法字段,不用硬编码维护。 - 第二步:使用参数绑定传递查询值,禁止直接拼接SQL
不管是用SQLAlchemy原生的ORM查询方法,还是用text()构造自定义查询,都不要直接把用户输入的起止值拼接到SQL字符串中,通过参数绑定的方式传入参数,SQLAlchemy会自动处理特殊字符转义,从根源避免注入风险。
示例代码(推荐使用原生ORM方式,无需手写SQL)
from sqlalchemy import inspect # 你的数据模型类,自行替换为实际的类名 from your_module import DataModel # 动态获取模型所有列作为允许查询的白名单,也可以根据业务需求手动指定 ALLOWED_RANGE_FIELDS = {c.key for c in inspect(DataModel).columns} range_parts = range_str.split(':') if len(range_parts) == 3: field_name, start_val, end_val = range_parts # 校验字段合法性 if field_name not in ALLOWED_RANGE_FIELDS: # 此处根据你的项目逻辑返回参数错误即可 raise ValueError(f"不支持的范围查询字段:{field_name}") # 获取模型对应的列对象 field_obj = getattr(DataModel, field_name) # 直接用ORM原生between方法构造条件,自动处理转义 query = query.filter(field_obj.between(start_val, end_val))
如果确实需要使用text()构造查询的场景
range_parts = range_str.split(':') if len(range_parts) == 3: field_name, start_val, end_val = range_parts # 字段白名单校验步骤不可省略 if field_name not in ALLOWED_RANGE_FIELDS: raise ValueError(f"不支持的范围查询字段:{field_name}") # 使用命名参数占位,通过bindparams传递参数 query = query.filter( text(f"{field_name} BETWEEN :start AND :end") .bindparams(start=start_val, end=end_val) )
通过以上方案,你既可以保留动态字段范围查询的灵活性,也能完全避免SQL注入风险。
内容的提问来源于stack exchange,提问作者Rafael Carrillo
相关产品推荐
相关产品推荐

