在 python 的 SQL Alchemy 中使用quoted_name安全的方式来参数化表名和字段吗？-解网

问：

我花了很多时间寻找解决方案，以在 SQL Alchemy 纯文本 SQL 查询中为 SQL Server 参数化表名和字段名。我偶然发现了几个 stackoverflow 问题和其他资源，例如：

我不喜欢上面的答案，因为它只是从字符串构建查询，这对于SQL注入攻击来说是proun

我知道可以通过创建表对象来做到这一点（我过去就是这样做的），但它需要声明数据库的整个架构，这是很多不可扩展的代码。sqlalchemy.ext.declarative.declarative_base

在SQL Server上没有太多运气的情况下，我使用Postgres找到了解决方案。因此，从这里开始，我开始在 SQL Alchemy 中寻找等价物。我找到了quoted_name。据我所知，它可以作为防止SQL注入的标识符。但真的是这样吗？有人可以确认它使用安全吗？psycopg2psycopg2.sql.Identifier

下面的代码示例，返回传入表中的行数：

    def count_rows(self, table_name: str) -> int:

        query_base = "SELECT COUNT(*) FROM {}"
        query_params = [quoted_name(table_name, True)]
        query = text((query_base).format(*query_params))

        with self.engine.connect() as con:
            result = con.execute(query).fetchone()

        return result[0]

python sql-server sqlalchemy 准备语句 sql 注入

我个人认为，“最安全”的方法是将模式和表的名称作为参数传递给 SQL，然后在其中创建动态语句，验证对象是否存在对象，并用正确引用值。当然，如果你只是想对表中的行进行计数，那么有一些非动态的SQL方法可以做到这一点，这些方法已经足够准确了。sysQUOTENAME

0赞 Aaron Bertrand 4/6/2022

这似乎是该函数的唯一意图，使表引用免受令人讨厌的文本的影响。但是我同意@Larnu：我更愿意在引擎中执行此操作，您还可以在尝试使用它构建查询之前轻松检查是否存在。如果您真的只是从用户指定的任何表中获取计数，请参阅坏习惯：以艰难的方式计算行数，其中显示了如何以更安全、更有效的方式执行此操作。table_namesys.tables

0赞 M.wol 4/6/2022

这个函数只是在 python 函数中使用表名作为参数的示例。我的模块中还有更多内容。你能详细说明@Larnu吗？这在 python 函数中会是什么样子？

0赞 Thom A 4/6/2022

不幸的是，我没有使用过sqlalchemy库，所以它在Python中的外观如何，我不是100%确定（而且因为我在办公室没有Python环境，所以不想猜测）。但是，从 SQL 的角度来看，查询将类似于我在此处演示的：验证对象

答：

0赞 Paddy Alton 4/8/2022 #1

我没有从文档中得到这样的印象，这就是预期的目的。我的理解是，它适用于使用列或表名的非标准命名约定的情况，需要引用才能使它们起作用。quoted_name

我认为有两种可能的解决方案：

1. 对允许的表名进行完全控制

f"SELECT COUNT(*) FROM {table_name}"如果您不允许用户在不进行过滤的情况下提供，则很好。table_name

例如，您可以简单地拥有

    ...
    allowed = ["table_1", ..., "table_N"]
    if table_name not in allowed:
        raise ValueError(f"Table name must be one of {allowed}. Received {table_name}.")

当然，还有很多其他方法可以做到这一点。但这个想法是将用户输入映射到允许的值，拒绝不允许的值，或者两者兼而有之。

2. 反映模式

你提到

我知道可以通过创建表对象来做到这一点（我过去就是这样做的），但它需要声明数据库的整个架构，这是很多不可扩展的代码。sqlalchemy.ext.declarative.declarative_base

事实并非如此。您可以“反映”现有数据库的架构，如下所示：

from sqlalchemy import create_engine, func, select, MetaData


class YourClass:
    def __init__(self, db_connection_string: str):
        """
        __init__ for YourClass

        (for example)

        """
        self.engine = create_engine(db_connection_string)
        self.metadata = MetaData(bind=self.engine)

        MetaData.reflect(self.metadata)

    def count_rows(self, table_name: str) -> int:
        """
        count_rows

        Returns the COUNT of the rows for a given table

        """
        table = self.metadata.tables[table_name]

        result = select([func.count()]).select_from(table).scalar()

        return result

值得注意的是，如果数据库中不存在，此方法也会引发异常。table_name

替代语法 - 为了获得完整的 ORM 优点，请使用：sessionmaker

from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import sessionmaker


class YourClass:
    def __init__(self, db_connection_string: str):
        self.engine = create_engine(db_connection_string)
        self.Session = sessionmaker(bind=self.engine)
        self.metadata = MetaData(bind=self.engine)

        MetaData.reflect(self.metadata)

    def count_rows(self, table_name: str) -> int:
        table = self.metadata.tables[table_name]
        # if you want a new session every call:
        with self.Session.begin() as session: 
            return session.query(table).count()

上一个：SQL 注入 mssql 节点 .js [重复]

下一个：如何使用带参数的表名的 Select 查询？[复制]

在 python 的 SQL Alchemy 中使用quoted_name安全的方式来参数化表名和字段吗？

Is using quoted_name safe way for parametrizing table name and fields in python's SQL Alchemy?

评论

1. 对允许的表名进行完全控制

2. 反映模式