このpsycopg2コードは注射から安全ですか？

Question

私が書いているクラスでは、postGISテーブルでさまざまな空間解析を行うことができます。このため、ユーザーはパラメータを使用して名前から選択する必要があります。私はこのようなユーザー入力を許可する危険を理解していますが、私は選択肢がありません。

別の機能を使用してテーブル名を事前にサニタイズするようにしてください。私は、そのパラメータの入力文字列が、データベースから取得したテーブル名のリストと一致するかどうかを調べることによってこれを行います。それから私はAsIs()を使用して渡しますが、これはお勧めできませんが、私が言ったように、データベースの既存のテーブルであるかどうかを確認することで、テーブル名を事前に確認しました。しかし、空間座標系を表すコードはまだ残っています。

私はこれが問題であるかどうかを確認するために自分で注射を書こうとしています。私はこの変数に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

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