Flask条件查询接口出现SQL注入，使用参数化查询:写法的解决方案（附带企业级开发实际例子与经验分享）-EW帮帮网

背景：

一个接口出现了SQL注入，条件查询场景下出现，形如下图

解决问题时，我们先要问，什么是SQL注入？

下面的资料有助于针对SQL注入是什么、如何验证SQL注入解决成功了，提供一些思路，建议一看，花费时间2-4min

Flask：SQLAlchemy查询是否容易受到注入攻击|极客教程 (geek-docs.com)

从0到1，SQL注入（sql十大注入类型）收藏这一篇就够了，技术解析与实战演练 - FreeBuf网络安全行业门户

过程：

此前SQL注入的错误的代码如下，like '%{0}%'这个写法就很有问题，这个代码的含义与直接使用+号来拼接sql字符串造成的SQL注入，感觉没有什么区别，我们即将对这个代码进行修复

    currentPage = util.params(request=request, key='page', default=1, type=int)
    limit = util.params(request=request, key='limit', default=20, type=int)
    name = util.params(request=request, key='name', default=0)
    code = util.params(request=request, key='code', default=0)
    water_id = util.params(request=request, key='water_id', default=0)
    filter_list = []
    filter_list.append("water_name !='{}'".format('测试'))
    if name:
        filter_list.append("station_name like '%{0}%' or outfall_name like '%{0}%'".format(name))
    if code:
        filter_list.append("station_code like '%{0}%' or outfall_code like '%{0}%'".format(name))
    if water_id:
        filter_list.append("water_id = {}".format(water_id))



    filter_str = " and ".join(filter_list)
    if filter_list:
        filter_str = "where " + filter_str
    try:
        sql_str = "select * from v_monitor_point {} order by water_id offset {} rows fetch next {} rows only".format(
            filter_str, (currentPage - 1) * limit, limit)
        count_sql = "select count(point_id) from v_monitor_point {}".format(filter_str)
        monitor_point_list = db.session.execute(text(sql_str)).fetchall()
        count = db.session.execute(text(count_sql)).fetchone()
        count = count[0] if count else 0
    except Exception:
        error_info = traceback.format_exc()
        logger_error.error(error_info)
        return res.ResponseJson(res.RetCode.DB_ERROR)
    info_list = []
    facility_sn_data = db.session.query(FacilitySn.id, FacilitySn.facility_sn, FacilitySn.facility_code).all()

思路如使用ORM的查询过滤器，如这种，但是这种在我的背景代码里面不适用，我考虑使用参数化查询，如:占位符

username = request.args.get('username')
users = User.query.filter(or_(User.username == username, User.email == username)).all()

具体的使用方案思路如下，即使用参数化查询

修改成功的demo

    currentPage = util.params(request=request, key='page', default=1, type=int)
    limit = util.params(request=request, key='limit', default=20, type=int)
    name = util.params(request=request, key='name', default=0)
    code = util.params(request=request, key='code', default=0)
    water_id = util.params(request=request, key='water_id', default=0)
	
	
	filter_list = []
    params = {}
    # 动态条件
    if name:
        filter_list.append("station_name like :name or outfall_name like :name")
        params['name'] = f"%{name}%"
    if code:
        filter_list.append("station_code like :code or outfall_code like :code")
        params['code'] = f"%{code}%"
    if water_id:
        filter_list.append("water_id = :water_id")
        params['water_id'] = water_id
    filter_str = " and ".join(filter_list)
    if filter_list:
        filter_str = "where " + filter_str
        # SQL 查询字符串
    sql_str = f"select * from v_monitor_point {filter_str} order by water_id offset :offset rows fetch next :limit rows only"
    count_sql = f"select count(point_id) from v_monitor_point {filter_str}"
    try:
        # 执行查询
        monitor_point_list = db.session.execute(text(sql_str), {
            **params,
            'offset': (currentPage - 1) * limit,
            'limit': limit
        }).fetchall()

        count = db.session.execute(text(count_sql), params).fetchone()
        count = count[0] if count else 0
    except Exception:
        error_info = traceback.format_exc()
        logger_error.error(error_info)
        return res.ResponseJson(res.RetCode.DB_ERROR)
    # 其他处理代码
    info_list = []
    facility_sn_data = db.session.query(FacilitySn.id, FacilitySn.facility_sn, FacilitySn.facility_code).all()

修复成功的截图，这个截图证明SQL注入初步修复成功

实际上我反馈给了专业的网安同学，请他帮忙验证了一下修改是否成功，结果修改是成功的