python - 这个 psycopg2 代码可以安全地注入(inject)吗？

Question

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

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

我正在尝试自己编写一个注入(inject)以查看这是否是一个问题。我没有为这个变量使用 AsIs() 但我很偏执，想确保它是安全的。我无法传递能够执行注入(inject)的变量(我尝试删除一个名为“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() 可以绕过这一步。但我想得到其他意见，以了解这是否仍然可以注入(inject)。

Answer 1

使用 sql module :

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))

输出:

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