FlawOpen

flawopen.com/sql-injection/Python

● CWE-89 · Critique
Recherche · FlawOpen

Injection SQL en Python (psycopg2, SQLAlchemy, Django)

Une f-string ou un % dans cursor.execute() laisse un paramètre de requête réécrire votre SQL et transforme la recherche d'un seul utilisateur en extraction de toute la table (CWE-89). Voici l'appel sûr pour chaque API de base de données Python, ainsi que les deux cas que les marqueurs ne couvrent pas : noms de colonnes et listes IN.[1], [2], [3], [4], [5], [6], [7], [9]

💡 Explication en Langage Simple (ELI5)

Imaginez un centre de tri postal avec une fiche d'instructions fixe : « Trier le colis portant l'étiquette ___ ». Si l'on recopie l'étiquette de l'expéditeur dans la fiche avant de la lire, des mots supplémentaires peuvent devenir de nouvelles consignes d'acheminement. Une f-string Python ou le formatage avec % fait cela au SQL : il construit une seule commande avant que le pilote reçoive la valeur. execute(sql, params) garde la fiche fixe et transmet l'étiquette séparément comme donnée. Un nom de colonne fait partie de la commande ; choisissez-le dans une liste d'autorisation définie par le code.[10], [1], [2], [9]

Concepts Clés et Termes

cursor.execute(sql, params)
L'appel de la DB-API 2.0 (PEP 249). Le pilote envoie les valeurs de params séparément du texte SQL, elles ne peuvent donc jamais modifier la requête. L'appel n'est sûr que si les valeurs sont dans le second argument ; une f-string ou une expression % dans le premier argument, c'est précisément la faille.[2], [3]
paramstyle: %s, ?, :name
Chaque pilote déclare son marqueur dans son attribut paramstyle : psycopg2 et PyMySQL utilisent %s ou %(name)s, sqlite3 utilise ? ou :name. Le %s est lu par le pilote, pas par l'opérateur % de Python : ne formatez jamais la chaîne vous-même.[2], [3], [5]
SQLAlchemy text()
Enveloppe une chaîne SQL brute pour SQLAlchemy. Sûr lorsque les valeurs sont des paramètres nommés comme :email, passés à execute() dans un dict. Dangereux si la chaîne fournie est déjà une f-string : text() n'échappe rien de ce qu'elle contient.[6], [9]
Django raw() / RawSQL / extra()
Les portes de sortie de l'ORM de Django. raw(sql, params) et RawSQL(sql, params) ne sont sûrs que si chaque valeur passe par params ; Django transmet tel quel tout ce qui est interpolé dans sql. La documentation de Django qualifie extra() de dernier recours : interdisez-le en revue.[7], [9]
psycopg2.sql.Identifier
Les marqueurs ne transportent que des valeurs, jamais des noms de table ou de colonne. sql.Identifier entoure le nom de guillemets doubles pour qu'il ne puisse pas sortir de l'identifiant. Vérifiez-le d'abord contre une liste autorisée pour que l'utilisateur ne choisisse pas des colonnes que vous ne vouliez pas exposer.[4], [9]
IN lists and expanding binds
Dans sqlite3, une liste Python ne peut pas être liée à un seul ?, d'où le retour à la concaténation. Utilisez un bindparam extensible dans SQLAlchemy (un marqueur par élément), id = ANY(%s) avec une liste dans psycopg2, ou générez un ? par élément pour sqlite3.[6], [3], [5]

Correspondances de référentiels

Déroulement de l'Attaque Étape par Étape

Étape 1

La valeur de la requête atteint la couche de données

Un paramètre de la requête arrive sous forme de texte dans get_user(). Une vue Flask peut lire request.args["id"] ; une vue Django lit request.GET["id"]. La vue transmet ce texte sans validation à get_user().[11], [12], [9]

Étape 2

Une f-string colle la valeur dans le SQL

f"... WHERE id = {user_id}" produit une seule chaîne à plat. Avec la valeur de test inoffensive 0 OR 1=1, le texte devient WHERE id = 0 OR 1=1, et rien n'indique où la valeur était censée s'arrêter.[10], [1], [9]

Étape 3

Le pilote ne reçoit aucun paramètre

execute(query) est appelé avec le SQL seul : psycopg2 envoie le texte sans modification et PostgreSQL analyse OR 1=1 comme une partie de la clause WHERE, et non comme une donnée.[3], [1], [9]

Étape 4

Le endpoint renvoie toutes les lignes

La condition est vraie pour chaque ligne : la recherche d'un utilisateur renvoie toute la table. psycopg2 accepte aussi plusieurs instructions séparées par des points-virgules en un seul appel ; sqlite3 les refuse, ce qui limite les dégâts sans empêcher le vol de données.[13], [5], [1], [9]

Code Source : Vulnérable vs Sécurisé

IMPLÉMENTATION VULNÉRABLE
# repository.py : chaque requête ci-dessous colle des données de la requête HTTP dans le texte SQL
import sqlite3

from sqlalchemy import text


def get_user(pg_cursor, user_id):
    # psycopg2 : la f-string construit l'instruction avant que le pilote ne la voie
    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 : ce % est un formatage de chaîne Python, pas un marqueur du pilote
    return conn.execute("SELECT id FROM orders WHERE status = '%s'" % status).fetchall()


def find_by_email(session, email):
    # SQLAlchemy : text() n'échappe pas ce qui est déjà dans la chaîne
    return session.execute(text(f"SELECT id FROM users WHERE email = '{email}'")).all()


def find_users(User, username):
    # Django : raw() envoie les valeurs interpolées telles quelles à la base
    return User.objects.raw(f"SELECT * FROM app_user WHERE username = '{username}'")


def get_orders_by_ids(session, ids):
    # liste IN : une chaîne jointe permet à un élément de transporter du SQL en plus
    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):
    # identifiant : la colonne de tri vient directement de la query string
    pg_cursor.execute(f"SELECT id, email FROM users ORDER BY {sort_column}")
    return pg_cursor.fetchall()
PATCH SÉCURISÉ ET ROBUSTE
# repository.py : les valeurs voyagent en paramètres liés, les noms viennent d'une liste autorisée
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 est le marqueur du pilote ; la valeur va dans le second argument
    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 utilise le style qmark : marqueurs ?, valeurs dans un tuple
    return conn.execute("SELECT id FROM orders WHERE status = ?", (status,)).fetchall()


def find_by_email(session, email: str):
    # SQLAlchemy : paramètre nommé, valeur fournie au moment de 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() envoie les params à part ; un filtre de l'ORM est encore plus simple
    return User.objects.raw("SELECT * FROM app_user WHERE username = %s", [username])


def get_orders_by_ids(session, ids: list[int]):
    # liste IN : un bind extensible produit un marqueur par élément
    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):
    # identifiant : d'abord la liste autorisée, puis les guillemets via 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()

Liste de Contrôle de Sécurité pour l'Ingénierie

Sources