如何解决基于Python Flask的二阶SQL注入漏洞问题？

问题根源分析 #

你当前的代码核心问题在于字符串拼接构造SQL，这直接导致了注入风险。而二阶注入的触发点是：get_user_data从数据库取出的USER_ID，未经过安全处理就直接流入check_access的查询中——哪怕这个数据来自数据库，只要后续用它拼接SQL，就可能被攻击者利用之前存入数据库的恶意数据触发注入。

正确解决方案 #

1. 彻底替换字符串拼接，使用SQLAlchemy参数绑定

首先修正get_user_data的写法，完全抛弃str.format，改用SQLAlchemy的参数绑定机制，确保用户输入和数据库返回值都不会直接嵌入SQL语句：

from sqlalchemy import text

# 用:text格式定义参数占位符
USER_ID_QUERY = text("SELECT USER_ID FROM USER WHERE USER_ID = :user_id")

def get_user_data(user_id: str):
    # 通过字典传递参数，由SQLAlchemy安全绑定
    result = db.session.execute(USER_ID_QUERY, {"user_id": user_id})
    # 用fetchone替代fetchall更高效
    user_record = result.fetchone()
    if not user_record:
        raise Exception("User not Found")
    return user_record[0]

2. 处理二阶注入的核心：数据库返回值复用仍需参数绑定

check_access方法是二阶注入的关键环节，必须确保从get_user_data获取的USER_ID依然通过参数绑定传入查询，绝对不能拼接字符串。示例修正如下：

# 同样用text定义带参数占位符的查询
ACCESS_CHECK_QUERY = text("SELECT EXISTS(SELECT 1 FROM ACCESS WHERE USER_ID = :user_id)")

def check_access(user_id_from_db: str):
    # 即使参数来自数据库，也要用参数绑定
    result = db.session.execute(ACCESS_CHECK_QUERY, {"user_id": user_id_from_db})
    has_access = result.scalar()
    return has_access

3. 优先使用ORM查询替代原生SQL

如果业务允许，直接使用SQLAlchemy ORM可以彻底避免手动构造SQL的风险，也更容易被扫描工具识别为安全代码：

# 假设已定义User模型
class User(db.Model):
    user_id = db.Column(db.String, primary_key=True)

def get_user_data(user_id: str):
    user = User.query.filter_by(user_id=user_id).first()
    if not user:
        raise Exception("User not Found")
    return user.user_id

# 对应Access模型
class Access(db.Model):
    user_id = db.Column(db.String, db.ForeignKey('user.user_id'))

def check_access(user_id_from_db: str):
    has_access = Access.query.filter_by(user_id=user_id_from_db).first() is not None
    return has_access

关于CodeGate误报的处理 #

如果已经使用了正确的参数绑定但仍被检测到，可能是扫描工具未识别SQLAlchemy的参数绑定语法。可以：

• 检查工具规则，确认是否支持识别text()配合字典参数的写法
• 尝试用bindparams()显式绑定参数的写法，可能更容易被工具识别：

  result = db.session.execute(USER_ID_QUERY.bindparams(user_id=user_id))
  

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