2017-07-17 73 views
1

我有一個我正在編寫的類,允許用於對postGIS表執行各種空間分析。正因爲如此,用戶必須能夠通過名稱來選擇它們,我從參數中獲得它們。我明白允許這樣的用戶輸入的危險,但我沒有選擇。這是psycopg2代碼注射安全嗎?

我確保使用另一個函數提前清理表名。我通過檢查該參數的輸入字符串是否與數據庫中檢索到的表名稱列表匹配來執行此操作。然後我使用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()就是繞過這一步。但我想獲得其他意見,以瞭解這是否仍然可以注射。

回答

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); 
+0

這是否工作領域呢?表名似乎工作,但我試圖以類似的方式傳遞字段名稱(CREATE TABLE interallocation(pk_id SERIAL PRIMARY KEY,{from_id_field} varchar(50),{to_id_field} varchar(50),distance real); '),但我得到了'TypeError:組合元素必須是可組合的,得到''from_pl_id''而不是''。我通過'id_field = sql.Identifier(from_id_field).as_string(self.conn),'它說在你鏈接的頁面上,字段名稱也算作標識符 – 1saac

+0

好吧,我想我已經明白了。事實證明,現在設置查詢有兩個階段。第一階段是使用表名和列名的「標識符」對象以及值爲「佔位符」的對象進行格式化,這些對象是比較熟悉的'%(價值)'的東西。然後,像平常一樣執行查詢,並用正常的字典填充佔位符值。 – 1saac

+0

不要傳遞'as_string'。一直使用composable到'execute'。我用'as_string'來打印它。 –