FlawOpen

flawopen.com/sql-injection/Python

● CWE-89 · Crítica
Investigación · FlawOpen

Inyección SQL en Python (psycopg2, SQLAlchemy, Django)

Una f-string o un % dentro de cursor.execute() permite que un parámetro de la petición reescriba tu SQL y convierta la búsqueda de un solo usuario en el volcado de la tabla completa (CWE-89). Esta es la llamada segura para cada API de base de datos de Python, más los dos casos que los marcadores no cubren: nombres de columna y listas IN.[1], [2], [3], [4], [5], [6], [7], [9]

💡 Explicación en Lenguaje Sencillo (ELI5)

Imagina una mesa de clasificación postal con una tarjeta de instrucciones fija: «Clasifica el paquete con la etiqueta ___». Si alguien copia la etiqueta del remitente dentro de la tarjeta antes de leerla, las palabras adicionales pueden convertirse en nuevas órdenes de reparto. Una f-string o el formato con % hace eso con SQL: construye un único comando antes de que el driver reciba el valor. execute(sql, params) mantiene fija la tarjeta y envía la etiqueta por separado, como dato. El nombre de una columna forma parte del comando; elígelo de una lista permitida definida en el código.[10], [1], [2], [9]

Conceptos Clave y Términos

cursor.execute(sql, params)
La llamada de la DB-API 2.0 (PEP 249). El driver envía los valores de params aparte del texto SQL, así que nunca pueden cambiar la sentencia. Solo es segura si los valores van en el segundo argumento; una f-string o una expresión % en el primer argumento es precisamente el fallo.[2], [3]
paramstyle: %s, ?, :name
Cada driver declara su marcador en el atributo paramstyle: psycopg2 y PyMySQL usan %s o %(name)s, sqlite3 usa ? o :name. El %s lo interpreta el driver, no el operador % de Python, así que nunca formatees la cadena tú mismo.[2], [3], [5]
SQLAlchemy text()
Envuelve una cadena SQL en bruto para SQLAlchemy. Es segura cuando los valores son parámetros con nombre, como :email, pasados a execute() en un dict. Es insegura si la cadena que recibe ya es una f-string: text() no escapa nada de lo que contiene.[6], [9]
Django raw() / RawSQL / extra()
Las vías de escape del ORM de Django. raw(sql, params) y RawSQL(sql, params) solo son seguras si cada valor va en params; Django envía tal cual lo que se interpole en sql. La documentación de Django califica extra() de último recurso, así que prohíbelo en la revisión.[7], [9]
psycopg2.sql.Identifier
Los marcadores solo transportan valores, nunca nombres de tabla o de columna. sql.Identifier pone el nombre entre comillas dobles para que no pueda salirse del identificador. Compruébalo antes contra una lista permitida para que el usuario no elija columnas que no pensabas exponer.[4], [9]
IN lists and expanding binds
En sqlite3 no se puede enlazar una lista de Python a un solo ?, y por eso se vuelve a concatenar cadenas. Usa un bindparam expandible en SQLAlchemy (un marcador por elemento), id = ANY(%s) con una lista en psycopg2, o genera un ? por elemento en sqlite3.[6], [3], [5]

Mapeos de marcos

Flujo de Ataque Paso a Paso

Paso 1

El valor de la petición llega a la capa de datos

Un parámetro de la solicitud llega a get_user() como texto. Una vista de Flask puede leer request.args["id"]; una vista de Django lee request.GET["id"]. La vista pasa ese texto sin validar a get_user().[11], [12], [9]

Paso 2

Una f-string empalma el valor en el SQL

f"... WHERE id = {user_id}" produce una única cadena plana. Con el valor de prueba inofensivo 0 OR 1=1, el texto queda WHERE id = 0 OR 1=1, y nada marca dónde debía terminar el valor.[10], [1], [9]

Paso 3

El driver no recibe ningún parámetro

execute(query) se llama solo con el SQL, así que psycopg2 envía el texto sin cambios y PostgreSQL analiza OR 1=1 como parte de la cláusula WHERE, no como dato.[3], [1], [9]

Paso 4

El endpoint devuelve todas las filas

La condición es verdadera para todas las filas, así que la búsqueda de un usuario devuelve la tabla entera. psycopg2 además acepta varias sentencias separadas por punto y coma en una sola llamada; sqlite3 las rechaza, lo que limita el daño pero no impide el robo de datos.[13], [5], [1], [9]

Código Fuente: Vulnerable vs. Seguro

IMPLEMENTACIÓN VULNERABLE
# repository.py: cada consulta de abajo empalma datos de la petición en el texto SQL
import sqlite3

from sqlalchemy import text


def get_user(pg_cursor, user_id):
    # psycopg2: la f-string construye la sentencia antes de que el driver la vea
    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 % es formateo de cadenas de Python, no un marcador del driver
    return conn.execute("SELECT id FROM orders WHERE status = '%s'" % status).fetchall()


def find_by_email(session, email):
    # SQLAlchemy: text() no escapa lo que ya está dentro de la cadena
    return session.execute(text(f"SELECT id FROM users WHERE email = '{email}'")).all()


def find_users(User, username):
    # Django: raw() envía los valores interpolados a la base de datos tal cual
    return User.objects.raw(f"SELECT * FROM app_user WHERE username = '{username}'")


def get_orders_by_ids(session, ids):
    # lista IN: las cadenas unidas permiten que un elemento lleve SQL adicional
    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: la columna de orden viene directamente de la query string
    pg_cursor.execute(f"SELECT id, email FROM users ORDER BY {sort_column}")
    return pg_cursor.fetchall()
PARCHE SEGURO Y ROBUSTO
# repository.py: los valores viajan como parámetros enlazados, los nombres salen de una lista permitida
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 es el marcador del driver; el valor va en el 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 el estilo qmark: marcadores ?, valores en una tupla
    return conn.execute("SELECT id FROM orders WHERE status = ?", (status,)).fetchall()


def find_by_email(session, email: str):
    # SQLAlchemy: parámetro con nombre, valor aportado al llamar a 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() envía params por separado; un filtro del ORM es aún más simple
    return User.objects.raw("SELECT * FROM app_user WHERE username = %s", [username])


def get_orders_by_ids(session, ids: list[int]):
    # lista IN: un bind expandible genera un marcador 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: primero la lista permitida, luego comillas con 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 Verificación de Seguridad para Ingeniería

Fuentes