这个psycopg2代码不会被注入吗？

Question

我正在编写一个类，它允许用户对postGIS表执行各种空间分析。因此，用户必须能够按名称选择它们，这是我从参数中得到的。我理解允许用户这样输入的危险，但我别无选择。在

我确保使用另一个函数提前清理表名。我通过检查该参数的输入字符串是否与从数据库中检索到的表名列表匹配来实现这一点。然后我使用AsIs()传递它，我知道这是不推荐的，但是正如我所说的，我通过查看表名是否是数据库中的现有表来提前验证它。但是我还有一个参数left，一个代表空间坐标系的代码。在

我正试着自己写一个注射剂，看看这是否是个问题。我没有对这个变量使用AsIs()，但我有点偏执，我想确保它是安全的。我无法传递一个能够执行注入的变量（我试图删除一个名为“deletetest”的故事）。在

这是我的代码：

class myClass(object):

    def __init__(self, conn_string, srid):

        self.connString = conn_string
        self.conn = psycopg2.connect(self.connString)
        self.srid = srid

        return None

    def sanitized(self, input_text):

        """
        Makes sure that the input matches an existing table name to make sure that the input name is not an SQL
        injection attempt.  True if the table name is found, False if not.
        :param input_text: String to be sanitized.
        :return: boolean
        """

        query = "SELECT relname FROM pg_class WHERE relkind='r' AND relname !~ '^(pg_|sql_)';"

        cur = self.conn.cursor()
        cur.execute(query)

        for tbl in [i[0] for i in cur.fetchall()]:
            if input_text == tbl:
                return True

        return False

    def interallocate(self, features):

        if self.sanitized(features):

            query = """

                    DROP TABLE IF EXISTS parking_lots_interallocation_result;
                    CREATE TABLE parking_lots_interallocation_result (pk_id SERIAL PRIMARY KEY, from_pl_id varchar(50), to_pl_id varchar(50), distance real);
                    SELECT AddGeometryColumn('public', 'parking_lots_interallocation_result', 'geom', %(srid)s, 'LINESTRING', 2);

                    DROP TABLE IF EXISTS interallocation_duplicate;
                    CREATE TABLE interallocation_duplicate AS TABLE %(features)s;

                    INSERT INTO parking_lots_interallocation_result (from_pl_id, to_pl_id, distance, geom)
                      SELECT
                        %(features)s.pl_id AS from_pl_id,
                        interallocation_duplicate.pl_id AS to_pl_id,
                        ST_Distance(%(features)s.geom, interallocation_duplicate.geom) AS distance,
                        ST_ShortestLine(%(features)s.geom, interallocation_duplicate.geom) AS geom
                      FROM
                        %(features)s
                      LEFT JOIN
                        interallocation_duplicate ON ST_DWithin(%(features)s.geom, interallocation_duplicate.geom, 700)
                      WHERE
                        interallocation_duplicate.pl_id IS NOT NULL AND %(features)s.pl_id != interallocation_duplicate.pl_id
                      ORDER BY
                        %(features)s.pl_id,
                        ST_Distance(%(features)s.geom, interallocation_duplicate.geom);

                    """

            print(query)

            cur = self.conn.cursor()
            cur.execute(query, {
                'features': AsIs(features), # Can use AsIs because we made sure that this string matches an existing table name.
                'srid': self.srid})
            self.conn.commit()

        else:
            raise KeyError('Table {0} was not found.'.format(features))

据我所知，使用cur.execute()应该清理输入，而使用AsIs()可以绕过这一步。但我想听听其他人的意见，看看这是否还可以注射。在

Answer 1

使用^{}：

features = 'Table_Name'
insert_query = sql.SQL("""
INSERT INTO parking_lots_interallocation_result (from_pl_id, to_pl_id, distance, geom)
SELECT
    {0}.pl_id AS from_pl_id,
    interallocation_duplicate.pl_id AS to_pl_id,
    ST_Distance({0}.geom, interallocation_duplicate.geom) AS distance,
    ST_ShortestLine({0}.geom, interallocation_duplicate.geom) AS geom
FROM
    {0}
    LEFT JOIN
    interallocation_duplicate ON ST_DWithin({0}.geom, interallocation_duplicate.geom, 700)
WHERE
    interallocation_duplicate.pl_id IS NOT NULL AND {0}.pl_id != interallocation_duplicate.pl_id
ORDER BY
    {0}.pl_id,
    ST_Distance({0}.geom, interallocation_duplicate.geom);
""")

print (insert_query.format(sql.Identifier(features)).as_string(conn))

输出：

^{pr2}$