FlawOpen

flawopen.com/sql-injection/Python

● CWE-89 · Kritisch
Sicherheitsforschung · FlawOpen

SQL-Injection in Python verhindern (psycopg2, SQLAlchemy, Django)

Ein f-String oder ein % in cursor.execute() lässt einen Request-Parameter Ihr SQL umschreiben und macht aus der Abfrage eines einzelnen Nutzers einen Dump der ganzen Tabelle (CWE-89). Hier steht der sichere Aufruf für jede Python-Datenbank-API, dazu die zwei Fälle, die Platzhalter nicht abdecken: Spaltennamen und IN-Listen.[1], [2], [3], [4], [5], [6], [7], [9]

💡 Einfache Erklärung (ELI5)

Stellen Sie sich eine Postsortierung mit einer festen Anweisungskarte vor: „Sortiere das Paket mit dem Etikett ___.“ Wird das Etikett des Absenders vor dem Lesen in die Anweisung kopiert, können zusätzliche Wörter zu neuen Sortierbefehlen werden. Ein Python-f-string oder %-Formatierung macht genau das mit SQL: Der gesamte Befehl entsteht, bevor der Treiber den Wert erhält. execute(sql, params) lässt die Anweisung unverändert und übermittelt das Etikett getrennt als Datenwert. Ein Spaltenname gehört zur Anweisung; wählen Sie ihn aus einer im Code festgelegten Allowlist.[10], [1], [2], [9]

Kernkonzepte & Begriffe

cursor.execute(sql, params)
Der Aufruf aus DB-API 2.0 (PEP 249). Der Treiber schickt die Werte aus params getrennt vom SQL-Text, sodass sie die Anweisung nie verändern können. Sicher ist das nur, wenn die Werte im zweiten Argument stehen; ein f-String oder ein %-Ausdruck im ersten Argument ist genau der Fehler.[2], [3]
paramstyle: %s, ?, :name
Jeder Treiber gibt seinen Platzhalter im Attribut paramstyle an: psycopg2 und PyMySQL verwenden %s oder %(name)s, sqlite3 verwendet ? oder :name. Das %s wertet der Treiber aus, nicht Pythons %-Operator – formatieren Sie den String also nie selbst.[2], [3], [5]
SQLAlchemy text()
Kapselt einen rohen SQL-String für SQLAlchemy. Sicher, wenn Werte benannte Bind-Parameter wie :email sind und als Dict an execute() gehen. Unsicher, wenn der übergebene String schon ein f-String ist: text() maskiert nichts von dem, was darin steht.[6], [9]
Django raw() / RawSQL / extra()
Die Notausgänge aus Djangos ORM. raw(sql, params) und RawSQL(sql, params) sind nur sicher, wenn jeder Wert in params steht; alles, was in sql eingesetzt wird, schickt Django unverändert weiter. Die Django-Dokumentation nennt extra() ein letztes Mittel, also im Review verbieten.[7], [9]
psycopg2.sql.Identifier
Platzhalter transportieren nur Werte, nie Tabellen- oder Spaltennamen. sql.Identifier setzt den Namen in doppelte Anführungszeichen, sodass er nicht aus dem Bezeichner ausbrechen kann. Prüfen Sie ihn vorher gegen eine Allowlist, damit Nutzer keine Spalten wählen, die Sie nicht freigeben wollten.[4], [9]
IN lists and expanding binds
In sqlite3 lässt sich eine Python-Liste nicht an ein einzelnes ? binden, deshalb wird oft wieder verkettet. Verwenden Sie in SQLAlchemy einen expandierenden bindparam (ein Platzhalter pro Element), in psycopg2 id = ANY(%s) mit einer Liste, oder erzeugen Sie für sqlite3 ein ? pro Element.[6], [3], [5]

Framework-Zuordnungen

Schritt-für-Schritt Angriffsablauf

Schritt 1

Der Request-Wert erreicht die Datenschicht

Ein Anfrageparameter erreicht get_user() als Text. Eine Flask-View kann request.args["id"] lesen; eine Django-View liest request.GET["id"]. Die View übergibt diesen ungeprüften Text an get_user().[11], [12], [9]

Schritt 2

Ein f-String fügt den Wert ins SQL ein

f"... WHERE id = {user_id}" erzeugt einen einzigen flachen String. Mit dem harmlosen Testwert 0 OR 1=1 wird daraus WHERE id = 0 OR 1=1, und nichts markiert, wo der Wert enden sollte.[10], [1], [9]

Schritt 3

Der Treiber erhält keine Parameter

execute(query) wird nur mit dem SQL aufgerufen, also schickt psycopg2 den Text unverändert, und PostgreSQL parst OR 1=1 als Teil der WHERE-Klausel statt als Daten.[3], [1], [9]

Schritt 4

Der Endpunkt liefert jede Zeile

Die Bedingung ist für jede Zeile wahr, also liefert die Abfrage eines Nutzers die ganze Tabelle. psycopg2 akzeptiert außerdem mehrere durch Semikolon getrennte Anweisungen in einem Aufruf; sqlite3 lehnt sie ab, was den Schaden begrenzt, den Datendiebstahl aber nicht verhindert.[13], [5], [1], [9]

Quellcode: Verwundbar vs. Sicher

VERWUNDBARE IMPLEMENTIERUNG
# repository.py: jede Abfrage unten fügt Request-Daten in den SQL-Text ein
import sqlite3

from sqlalchemy import text


def get_user(pg_cursor, user_id):
    # psycopg2: der f-String baut die Anweisung, bevor der Treiber sie sieht
    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: dieses % ist Pythons String-Formatierung, kein Treiber-Platzhalter
    return conn.execute("SELECT id FROM orders WHERE status = '%s'" % status).fetchall()


def find_by_email(session, email):
    # SQLAlchemy: text() maskiert nicht, was schon im String steht
    return session.execute(text(f"SELECT id FROM users WHERE email = '{email}'")).all()


def find_users(User, username):
    # Django: raw() schickt eingesetzte Werte unverändert an die Datenbank
    return User.objects.raw(f"SELECT * FROM app_user WHERE username = '{username}'")


def get_orders_by_ids(session, ids):
    # IN-Liste: zusammengefügte Strings lassen ein Element zusätzliches SQL mitbringen
    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):
    # Bezeichner: die Sortierspalte kommt direkt aus dem Query-String
    pg_cursor.execute(f"SELECT id, email FROM users ORDER BY {sort_column}")
    return pg_cursor.fetchall()
GEHÄRTETER SICHERHEITS-PATCH
# repository.py: Werte reisen als gebundene Parameter, Namen kommen aus einer 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 ist der Platzhalter des Treibers; der Wert steht im zweiten 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 nutzt den qmark-Stil: ?-Platzhalter, Werte in einem Tupel
    return conn.execute("SELECT id FROM orders WHERE status = ?", (status,)).fetchall()


def find_by_email(session, email: str):
    # SQLAlchemy: benannter Bind-Parameter, Wert erst beim execute()-Aufruf
    stmt = text("SELECT id FROM users WHERE email = :email")
    return session.execute(stmt, {"email": email}).all()


def find_users(User, username: str):
    # Django: raw() schickt params getrennt; ein ORM-Filter ist noch einfacher
    return User.objects.raw("SELECT * FROM app_user WHERE username = %s", [username])


def get_orders_by_ids(session, ids: list[int]):
    # IN-Liste: ein expandierender Bind erzeugt einen Platzhalter pro Element
    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):
    # Bezeichner: erst Allowlist, dann mit sql.Identifier quoten
    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()

Checkliste für Engineering & Systemsicherheit

Quellen