如何使用SQLAlchemy库写出防SQL注入的Raw SQL

Posted 2023-02-28 slvher

Python阵营有很多操作数据库的开源库（安装pip后，可以借助”pip search mysql”查看可用的库列表），其中被使用最多的无疑是MySQLdb，这个库简单易上手。其偏底层的特性为开发者提供灵活性的同时，也对不少新手写出的DB操作代码提出了考验，因为它只支持raw sql，容易导致sql注入攻击。

鉴于此，很多库提供了ORM接口能力，借助OO思想，数据库中的表被映射为Python的类，类的对象代表数据表中的一行记录，所有的DB操作都通过对象方法调用来实现，这些调用在底层被自动转换成SQL语句，在转换过程中，通常会采用parameter bind的方式保证生成的parameterized SQL不存在被注入的风险。

SQLAlchemy就是这样一个具备ORM能力的DB操作Python库，此外，该库还支持开发者执行raw sql，并通过其提供的text对象实现params binding，从而防护SQL注入风险。

备注1：php中的DB操作库（如PDO或MySQLi）支持的prepare/bind_param接口也是业界推荐的预防sql injection的方法，而escape_string只能对单/双引号等特殊字符做简单的替换，它并不能保证防御所有的危险字符。

备注2：SQLAlchemy的官方文档比较多，其架构细节可以参考SQLAlchemy at Architecture of Open Source Applications这篇文章，相信对初学者有不小的帮助。

下面的代码示例用来说明如何借助SQLAlchemy的parameters bind能力来写出能防止sql注入的raw sql。

前提假设

假设我们实现了一个简单的sqlalchemy封装类（dbutil.py），代码如下：

#!/bin/env python
#-*- encoding: utf-8 -*-

from sqlalchemy import create_engine

class DbWrapper(object):
    _db_inst = None
    _db_driver_cfg = 
        'dbtype' : 'mysql',
    

    @classmethod
    def get_db_inst(cls, dbtype = 'mysql', user = '', password = '', host = '127.0.0.1', port = 3306, dbname = '', encoding = 'utf-8'):
        if cls._db_inst is None:
            stmt = '%s://%s:%s@%s:%s/%s' % (cls._db_driver_cfg['dbtype'], user, password, host, port, dbname)
            cls._db_inst = create_engine(stmt, encoding = encoding)
        return cls._db_inst

上面的代码非常简单，在需要操作db时，通过调用dbutil.get_db_inst()并传入db配置就能获取到可以操作db的类的实例。

备注：db实例最好创建一次后保存起来，进程启动后在做必要的初始化工作时就可以先把db实例创建出来且整个进程都可用这个实例访问数据库。这是因为sqlalchemy库是支持connection pool且默认启用的，在大多数情况下，一个db实例足以应对整个进程对db的并发访问需求。

insert示例

下面以insert sql为例说明如何借助sqlalchemy.text写出无sql注入风险的raw sql（假设已经创建出_db_inst实例）。

#!/bin/env python
#-*- encoding: utf-8 -*-

import time
from sqlalchemy import text

def insert_into_xxx_tbl(user_id, user_name, nickname):
    insert_params_dict = 
        'user_id': user_id,
        'user_name': user_name,
        'nickname': nickname,
        'db_insert_time': int(time.time()),
        'db_update_time': int(time.time()),
    

    ## use sqlalchemy bindparams to prevent sql injection
    pre_sql = 'insert into xxx_tbl (user_id, user_name, nickname, db_insert_time, db_update_time) values(:user_id, :user_name, :nickname, :db_insert_time, :db_update_time)'
    bind_sql = text(pre_sql)
    resproxy = _db_inst.connect().execute(bind_sql, insert_params_dict)
    ## return lastid as event_id
    event_id = resproxy.lastrowid
    return event_id

select示例

select的用法与insert类似：借助Python dict构造select sql where条件的kv pairs，利用text()对sql进行参数绑定，调用execute()时传入绑定的sql及真正的参数即可。

#!/bin/env python
#-*- encoding: utf-8 -*-

import time
from sqlalchemy import text

def select_from_xxx_tbl(event_id):
    select_params_dict = 
        'event_id': event_id,
    

    ## use sqlalchemy bindparams to prevent sql injection
    pre_sql = 'select user_id, user_name, nickname from xxx_tbl where event_id = :event_id'
    bind_sql = text(pre_sql)
    resproxy = _db_inst.connect().execute(bind_sql, select_params_dict)
    rows = resproxy.fetchall()
    ret = rows[0]
    ## return (user_id, user_name, nickname)
    return ret

参考资料

1. SQLAlchemy Doc: Using Textual SQL
2. ARCHITECTURAL DOCUMENTATION: SQLAlchemy at Architecture of Open Source Applications
3. StackOverflow: How can I prevent SQL-injection in PHP?