FlawOpen

flawopen.com/sql-injection/Python

● CWE-89 · Crítica
Pesquisa · FlawOpen

SQL Injection em Python (psycopg2, SQLAlchemy, Django)

Uma f-string ou um % dentro de cursor.execute() permite que um parâmetro da requisição reescreva o seu SQL, transformando a busca de um único usuário no despejo da tabela inteira (CWE-89). Aqui está a chamada segura para cada API de banco de dados do Python, além dos dois casos que placeholders não cobrem: nomes de coluna e listas IN.[1], [2], [3], [4], [5], [6], [7], [9]

💡 Explicação em Linguagem Simples (ELI5)

Imagine uma mesa de triagem postal com um cartão de instruções fixo: 'Separe o pacote com a etiqueta ___'. Se alguém copiar a etiqueta do remetente para dentro do cartão antes de lê-lo, palavras extras podem virar novas ordens de encaminhamento. Uma f-string ou a formatação com % faz isso com o SQL: monta um único comando antes que o driver receba o valor. execute(sql, params) mantém o cartão fixo e envia a etiqueta separadamente, como dado. O nome de uma coluna faz parte do comando; escolha-o em uma lista permitida definida no código.[10], [1], [2], [9]

Conceitos Centrais e Termos

cursor.execute(sql, params)
A chamada da DB-API 2.0 (PEP 249). O driver envia os valores de params separados do texto SQL, então eles nunca alteram o comando. Só é segura quando os valores vão no segundo argumento; uma f-string ou expressão % no primeiro argumento é justamente o bug.[2], [3]
paramstyle: %s, ?, :name
Cada driver declara seu marcador no atributo paramstyle: psycopg2 e PyMySQL usam %s ou %(name)s, o sqlite3 usa ? ou :name. Quem lê o %s é o driver, não o operador % do Python, então nunca formate a string você mesmo.[2], [3], [5]
SQLAlchemy text()
Envolve uma string SQL crua no SQLAlchemy. É segura quando os valores são parâmetros nomeados, como :email, passados a execute() em um dict. É insegura quando a string entregue já é uma f-string: text() não escapa nada do que está dentro dela.[6], [9]
Django raw() / RawSQL / extra()
As saídas de emergência do ORM do Django. raw(sql, params) e RawSQL(sql, params) só são seguras quando todo valor vai em params; o Django envia literalmente o que for interpolado em sql. A documentação do Django chama extra() de último recurso, então proíba-o na revisão.[7], [9]
psycopg2.sql.Identifier
Placeholders carregam apenas valores, nunca nomes de tabela ou coluna. sql.Identifier coloca o nome entre aspas duplas para que ele não escape do identificador. Confira o nome contra uma allowlist antes, para que o usuário não escolha colunas que você não pretendia expor.[4], [9]
IN lists and expanding binds
No sqlite3 não dá para vincular uma lista Python a um único ?, e por isso muita gente volta a concatenar strings. Use um bindparam expansível no SQLAlchemy (um placeholder por elemento), id = ANY(%s) com uma lista no psycopg2, ou gere um ? por elemento no sqlite3.[6], [3], [5]

Mapeamentos de frameworks

Fluxo de Ataque Passo a Paso

Passo 1

O valor da requisição chega à camada de dados

Um parâmetro da requisição chega a get_user() como texto. Uma view Flask pode ler request.args["id"]; uma view Django lê request.GET["id"]. A view repassa esse texto sem validação para get_user().[11], [12], [9]

Passo 2

Uma f-string emenda o valor no SQL

f"... WHERE id = {user_id}" gera uma única string plana. Com o valor de teste inofensivo 0 OR 1=1, o texto vira WHERE id = 0 OR 1=1, e nada indica onde o valor deveria terminar.[10], [1], [9]

Passo 3

O driver não recebe nenhum parâmetro

execute(query) é chamado apenas com o SQL, então o psycopg2 envia o texto sem mudanças e o PostgreSQL interpreta OR 1=1 como parte da cláusula WHERE, e não como dado.[3], [1], [9]

Passo 4

O endpoint devolve todas as linhas

A condição é verdadeira para todas as linhas, então a busca de um usuário devolve a tabela inteira. O psycopg2 também aceita vários comandos separados por ponto e vírgula numa só chamada; o sqlite3 os rejeita, o que limita o estrago mas não impede o roubo de dados.[13], [5], [1], [9]

Código-Fonte: Vulnerável vs. Seguro

IMPLEMENTAÇÃO VULNERÁVEL
# repository.py: toda consulta abaixo emenda dados da requisição no texto SQL
import sqlite3

from sqlalchemy import text


def get_user(pg_cursor, user_id):
    # psycopg2: a f-string monta o comando antes de o driver vê-lo
    pg_cursor.execute(f"SELECT id, email FROM users WHERE id = {user_id}")
    return pg_cursor.fetchone()


def find_orders(conn: sqlite3.Connection, status):
    # sqlite3: este % é formatação de string do Python, não um placeholder do driver
    return conn.execute("SELECT id FROM orders WHERE status = '%s'" % status).fetchall()


def find_by_email(session, email):
    # SQLAlchemy: text() não escapa o que já está dentro da string
    return session.execute(text(f"SELECT id FROM users WHERE email = '{email}'")).all()


def find_users(User, username):
    # Django: raw() envia os valores interpolados ao banco literalmente
    return User.objects.raw(f"SELECT * FROM app_user WHERE username = '{username}'")


def get_orders_by_ids(session, ids):
    # lista IN: strings concatenadas deixam um elemento carregar SQL extra
    id_list = ", ".join(ids)
    return session.execute(text(f"SELECT id FROM orders WHERE id IN ({id_list})")).all()


def list_users(pg_cursor, sort_column):
    # identificador: a coluna de ordenação vem direto da query string
    pg_cursor.execute(f"SELECT id, email FROM users ORDER BY {sort_column}")
    return pg_cursor.fetchall()
PATCH SEGURO E ROBUSTO
# repository.py: valores seguem como parâmetros vinculados, nomes vêm de uma allowlist
import sqlite3

from psycopg2 import sql
from sqlalchemy import bindparam, text

SORTABLE_COLUMNS = {"id", "email", "created_at"}


def get_user(pg_cursor, user_id: int):
    # psycopg2: %s é o placeholder do driver; o valor vai no segundo argumento
    pg_cursor.execute("SELECT id, email FROM users WHERE id = %s", (user_id,))
    return pg_cursor.fetchone()


def find_orders(conn: sqlite3.Connection, status: str):
    # sqlite3 usa o estilo qmark: placeholders ?, valores numa tupla
    return conn.execute("SELECT id FROM orders WHERE status = ?", (status,)).fetchall()


def find_by_email(session, email: str):
    # SQLAlchemy: parâmetro nomeado, valor fornecido na hora do execute()
    stmt = text("SELECT id FROM users WHERE email = :email")
    return session.execute(stmt, {"email": email}).all()


def find_users(User, username: str):
    # Django: raw() envia params separados; um filtro do ORM é ainda mais simples
    return User.objects.raw("SELECT * FROM app_user WHERE username = %s", [username])


def get_orders_by_ids(session, ids: list[int]):
    # lista IN: um bind expansível gera um placeholder por elemento
    stmt = text("SELECT id FROM orders WHERE id IN :ids").bindparams(
        bindparam("ids", expanding=True)
    )
    return session.execute(stmt, {"ids": ids}).all()


def list_users(pg_cursor, sort_column: str):
    # identificador: primeiro a allowlist, depois aspas com sql.Identifier
    if sort_column not in SORTABLE_COLUMNS:
        raise ValueError(f"unsupported sort column: {sort_column!r}")
    query = sql.SQL("SELECT id, email FROM users ORDER BY {}").format(sql.Identifier(sort_column))
    pg_cursor.execute(query)
    return pg_cursor.fetchall()

Lista de Verificação de Segurança para Engenharia

Fontes