SQL 注入保护是否内置于 SQLAlchemy 的 ORM 或核心中?
Is SQL injection protection built into SQLAlchemy's ORM or Core?
我正在开发 aiohttp server application, and I just saw that 。所以,我想知道:如果我的应用程序只能使用 SQLAlchemy 的核心,它是否仍然可以免受 SQL 注入攻击?
我的代码如下:
async def add_sensor(db_engine, name):
async with db_engine.acquire() as connection:
query = model.Sensor.__table__.insert().values(name=name)
await connection.execute(query)
对 this related question 中接受的答案的评论让我怀疑:
you can still use execute() or other literal data that will NOT be
escaped by SQLAlchemy.
那么,在我的代码中使用 execute()
,上面的引用是否意味着我的代码不安全?一般而言:是否只能通过 SQLAlchemy ORM 层来防止 SQL 注入,就像您最终将启动的核心层一样 execute()
?
在您上面的示例中,我没有看到任何提供给数据库查询的变量。由于没有用户提供的输入,因此也没有 Sql 注入可能。
即使有用户提供的值,只要您不使用带有 sqlalchemy 的手写 sql 语句,而是尽可能使用 orm 模型方法 (model.Sensor.__table__.select()
)在您的示例中可以看出,您可以安全地防止 Sql 注入。
最后,这一切都是为了明确地告诉 sqlalchemy 应该使用哪些列和表来 select 和插入数据 from/to 并将其与正在发送的数据分开插入或 selected。切勿将数据字符串与查询字符串组合在一起,并始终使用 sqlalchemy orm 模型对象来描述您的查询。
错误的方式(Sql 可注射):
Session.execute("select * form users where name = %s" % request.GET['name'])
好方法(不可 Sql 注射):
Session.execute(model.users.__table__.select().where(model.users.name == request.GET['name']))
我正在开发 aiohttp server application, and I just saw that
我的代码如下:
async def add_sensor(db_engine, name):
async with db_engine.acquire() as connection:
query = model.Sensor.__table__.insert().values(name=name)
await connection.execute(query)
对 this related question 中接受的答案的评论让我怀疑:
you can still use execute() or other literal data that will NOT be escaped by SQLAlchemy.
那么,在我的代码中使用 execute()
,上面的引用是否意味着我的代码不安全?一般而言:是否只能通过 SQLAlchemy ORM 层来防止 SQL 注入,就像您最终将启动的核心层一样 execute()
?
在您上面的示例中,我没有看到任何提供给数据库查询的变量。由于没有用户提供的输入,因此也没有 Sql 注入可能。
即使有用户提供的值,只要您不使用带有 sqlalchemy 的手写 sql 语句,而是尽可能使用 orm 模型方法 (model.Sensor.__table__.select()
)在您的示例中可以看出,您可以安全地防止 Sql 注入。
最后,这一切都是为了明确地告诉 sqlalchemy 应该使用哪些列和表来 select 和插入数据 from/to 并将其与正在发送的数据分开插入或 selected。切勿将数据字符串与查询字符串组合在一起,并始终使用 sqlalchemy orm 模型对象来描述您的查询。
错误的方式(Sql 可注射):
Session.execute("select * form users where name = %s" % request.GET['name'])
好方法(不可 Sql 注射):
Session.execute(model.users.__table__.select().where(model.users.name == request.GET['name']))