Python Flask SQLalchemy sqlite3 防止数据库搜索中的 SQL 注入

Posted 2023-03-22

【中文标题】Python Flask SQLalchemy sqlite3 防止数据库搜索中的 SQL 注入

【英文标题】：Python Flask SQLalchemy sqlite3 prevent SQL injections in Database search

【发布时间】：2021-11-21 11:12:36

【问题描述】：

我想知道我应该如何更改我的代码以防止它被注入：

import sqlite3

def search_in_database(column,searched_data):
    con = sqlite3.connect('db.sqlite3')
    cursor = con.cursor()
    
    cursor.execute(f"""SELECT
                        id
                        FROM
                        My_library
                        WHERE
                        column LIKE '%searched_data%'
                        ;""")
    all = [i for i in cursor.fetchall()]
    return all

我在网上找到了代码，其中给出了如何做到这一点的示例：

from sqlalchemy.sql import text
# Create a connection conn
stmt = text("""SELECT * FROM users
               WHERE user = :username AND password = :password""")
conn.execute(stmt, prams="username": "foo", "password": "bar")

但在我的 html 文件中，我希望用户可以选择：

他想在标题、作者、published_dates、isbn、语言中搜索的地方...

当他选择要搜索的内容时，他会键入查询。

这种情况下怎么做，避免注入？

我的数据库：

class My_library(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(250))
    authors = db.Column(db.String(100))
    published_date = db.Column(db.Integer)
    isbn_or_identifier = db.Column(db.String(15), unique=True)
    page_count = db.Column(db.String(10000))
    language = db.Column(db.String(3))
    image_links = db.Column(db.String(500))

我还添加了验证器：

from flask_wtf import FlaskForm
from wtforms import SubmitField,StringField
from wtforms.validators import ValidationError,DataRequired,Length, URL
from wtforms.fields.html5 import DateField,IntegerField,DateField,IntegerField, URLField

class AddRecValidators(FlaskForm):

    title = StringField(label=('Title:'),validators=[DataRequired(), Length(min=1,max=50)])
    authors = StringField(label=('Authors:'),validators=[Length(min=1,max=100)])
    published_date = IntegerField(label=('Published date:'),validators=[Length(min=1,max=4)])
    isbn_or_identifier = IntegerField(label=('ISBN:'),validators=[Length(min=1,max=15)])
    page_count = IntegerField(label=('Page count:'),validators=[ Length(min=1,max=10000)])
    language = StringField(label=('Language:'),validators=[ Length(min=1,max=3)])
    image_links = URLField(label=('Image links:'))
    
    submit = SubmitField(label=('Add to library'))

提前感谢您的帮助：D

【参考方案1】：

您可以使用 sqlalchemy 来构建查询。例如：

q = My_library.__table__.select().where(
    My_library.__table__.c.title == "The searched title"
)

但这并不是你想要的。您还可以通过它们的名称来寻址列，如下所示：

q = My_library.__table__.select().where(
    My_library.__table__.c["title"] == "The searched title"
)
# or
q = My_library.__table__.select().where(
    My_library.__table__.c["title"].like("%The searched title%")
)

因此你可以这样做：

q = My_library.__table__.select().where(
    My_library.__table__.c[column].like(f"%searched_data%")
)
cursor.execute(q)

如果您只想要 ID，您可以这样做：

q = sqlalchemy.select([My_library.__table__.c.id]).where(
    My_library.__table__.c[column].like(f"%searched_data%")
)
# you can print(q) to see what it constructed
cursor.execute(q)

那是 SQLAlchemy 查询语言。您正在使用 ORM。我建议你先阅读一些关于烧瓶会话的内容。

仍然可以获取与列名相关的属性，但我不确定这是最有效的方法：

q = session.query(My_library.id).filter(
    getattr(My_library, column).like(f"%searched_data%"),
)