SQLAlchemy의 선택적 매개 변수

Question

쿼리 문자열에서 오는 선택적 매개 변수를 사용하여 SQLAlchemy에서 실행될 원시 SQL이 있습니다.

는 내가로 문장을 구축하는 기능을했다 where 절 :

def queryParams(self, params): 
    params = urllib.parse.parse_qs(params) 
    query = "" 
    for (key, val) in params.items(): 
     for value in val: 
      if(re.match("/(\d+(\.\d+)?)/", value) != False or (value == "false" or value == "true")): 
       query += ' AND ' + key + ' = ' + value 
      else: 
       query += ' AND ' + key + ' = ' + "'" + value + "'" 

    return query 

문제 나 쿼리를 실행해야 할 때, 나는 SQL 문자열이 CONCAT해야하고 그것이 좋은 아니에요입니다 SQL 인젝션이 발생했습니다.

s = text(
     'select a.cod_lim_judicial, a.desc_lim_judicial, a.dt_lim_judicial, ' 
     'a.dt_validade, a.nr_mandado, a.tipo_liminar, a.folhas_incidencia, ' 
     'a.num_folha, a.mes_ano_folha, p.mat_servidor, p.cod_depend, a.perc_pa ' 
     'from folha.fl_liminar a, folha.fl_pens_x_liminar p ' 
     'where a.cod_lim_judicial = p.cod_lim_judicial (+)' + self.queryParams(parameters)) 
    result = conn.execute(s).fetchall() 

어떻게 준비된 문을 사용하여 선택적 매개 변수를 사용할 수 있습니까?

미리 감사드립니다!

Answer 1

이 여기에 당신이 무엇을 할 수 있는지의 다소 단순화 된 예입니다 :

예약 된 단어와 혼합 된 경우를 인용 column() 핸들을 사용하고, 열 이름에 잘못된 문자를 이스케이프

In [42]: def queryParams(params): 
    ...:  params = urllib.parse.parse_qs(params) 
    ...:  for key, val in params.items(): 
    ...:   # Trust the implicit conversions: 
    ...:   # https://docs.oracle.com/en/database/oracle/oracle-database/12.2/sqlrf/Data-Type-Comparison-Rules.html#GUID-6DB331B5-0F34-4215-9A20-16AEA9D7FF4B 
    ...:   yield column(key) == val 
    ...: 

하지만해야 여전히 가능성이 화이트리스트하지만. 비교 결과 바인드 된 매개 변수 val과 함께 2 진 SQL 표현식 오브젝트가 생성됩니다. 당신은 당신의 쿼리를 생성하기 위해 Core constructs with text fragments을 결합 할 수 있습니다 :

In [43]: params = 'asdf=1&qwer=true&foo=baz' 

In [44]: s = select([text(""" 
    ...:   a.cod_lim_judicial, a.desc_lim_judicial, a.dt_lim_judicial, 
    ...:   a.dt_validade, a.nr_mandado, a.tipo_liminar, a.folhas_incidencia, 
    ...:   a.num_folha, a.mes_ano_folha, p.mat_servidor, p.cod_depend, a.perc_pa""")]).\ 
    ...:  select_from(
    ...:   outerjoin(
    ...:    text("folha.fl_liminar a"), 
    ...:    text("folha.fl_pens_x_liminar p"), 
    ...:    text("a.cod_lim_judicial = p.cod_lim_judicial"))).\ 
    ...:  where(and_(*queryParams(params))) 
    ...: 


In [45]: print(s) 
SELECT 
     a.cod_lim_judicial, a.desc_lim_judicial, a.dt_lim_judicial, 
     a.dt_validade, a.nr_mandado, a.tipo_liminar, a.folhas_incidencia, 
     a.num_folha, a.mes_ano_folha, p.mat_servidor, p.cod_depend, a.perc_pa 
FROM folha.fl_liminar a LEFT OUTER JOIN folha.fl_pens_x_liminar p ON a.cod_lim_judicial = p.cod_lim_judicial 
WHERE asdf = :asdf_1 AND qwer = :qwer_1 AND foo = :foo_1