FlawOpen

flawopen.com/sql-injection/Python

● CWE-89 · Критический
Исследования · FlawOpen

SQL-инъекции в Python (psycopg2, SQLAlchemy, Django)

f-строка или оператор % внутри cursor.execute() позволяют параметру запроса переписать ваш SQL и превращают поиск одного пользователя в выгрузку всей таблицы (CWE-89). Здесь — безопасный вызов для каждого API баз данных в Python и два случая, которые плейсхолдеры не закрывают: имена столбцов и списки IN.[1], [2], [3], [4], [5], [6], [7], [9]

💡 Простыми словами (ELI5)

Представьте почтовый сортировочный участок с неизменной карточкой-инструкцией: «Отсортировать посылку с ярлыком ___». Если перед чтением вписать ярлык отправителя прямо в инструкцию, дополнительные слова могут стать новыми указаниями по сортировке. Python f-string или форматирование через % делает это с SQL: сначала собирает одну команду, и только потом драйвер получает значение. execute(sql, params) сохраняет команду неизменной и передаёт ярлык отдельно как данные. Имя столбца входит в саму команду — выбирайте его из заданного кодом списка разрешённых значений.[10], [1], [2], [9]

Ключевые понятия и термины

cursor.execute(sql, params)
Вызов из DB-API 2.0 (PEP 249). Драйвер передаёт значения из params отдельно от текста SQL, поэтому они никогда не меняют сам запрос. Безопасно это лишь тогда, когда значения стоят во втором аргументе; f-строка или выражение с % в первом аргументе — это и есть уязвимость.[2], [3]
paramstyle: %s, ?, :name
Каждый драйвер объявляет свой плейсхолдер в атрибуте paramstyle: psycopg2 и PyMySQL используют %s или %(name)s, sqlite3 — ? или :name. %s обрабатывает драйвер, а не оператор % в Python, поэтому никогда не форматируйте строку сами.[2], [3], [5]
SQLAlchemy text()
Обёртка для «сырой» SQL-строки в SQLAlchemy. Безопасна, когда значения — именованные параметры вроде :email, переданные в execute() словарём. Опасна, если переданная строка уже является f-строкой: text() ничего внутри неё не экранирует.[6], [9]
Django raw() / RawSQL / extra()
Обходные пути мимо ORM в Django. raw(sql, params) и RawSQL(sql, params) безопасны, только если каждое значение передано в params; всё, что подставлено в sql, Django отправляет как есть. Документация Django называет extra() крайней мерой, так что запрещайте его на ревью.[7], [9]
psycopg2.sql.Identifier
Плейсхолдеры переносят только значения, но не имена таблиц или столбцов. sql.Identifier заключает имя в двойные кавычки, и оно не может выйти за пределы идентификатора. Перед этим сверяйте имя с белым списком, чтобы пользователь не выбрал столбцы, которые вы не собирались открывать.[4], [9]
IN lists and expanding binds
В sqlite3 список Python нельзя привязать к одному ?, поэтому разработчики возвращаются к склейке строк. Используйте в SQLAlchemy расширяемый bindparam (по плейсхолдеру на элемент), в psycopg2 — id = ANY(%s) со списком, а для sqlite3 генерируйте по одному ? на элемент.[6], [3], [5]

Связи со стандартами

Пошаговый сценарий атаки

Шаг 1

Значение из запроса попадает в слой данных

Параметр запроса поступает в get_user() как текст. Представление Flask может читать request.args["id"], а представление Django читает request.GET["id"]. Представление передаёт этот непроверенный текст в get_user().[11], [12], [9]

Шаг 2

f-строка вклеивает значение в SQL

f"... WHERE id = {user_id}" даёт одну плоскую строку. С безобидным тестовым значением 0 OR 1=1 текст превращается в WHERE id = 0 OR 1=1, и ничто не отмечает, где значение должно было закончиться.[10], [1], [9]

Шаг 3

Драйвер не получает ни одного параметра

execute(query) вызывается только с SQL, поэтому psycopg2 отправляет текст без изменений, а PostgreSQL разбирает OR 1=1 как часть условия WHERE, а не как данные.[3], [1], [9]

Шаг 4

Эндпоинт возвращает все строки

Условие истинно для каждой строки, и поиск одного пользователя возвращает всю таблицу. psycopg2 к тому же принимает несколько операторов через точку с запятой в одном вызове; sqlite3 их отклоняет, что ограничивает ущерб, но не мешает краже данных.[13], [5], [1], [9]

Исходный код: Уязвимый vs Защищённый вариант

УЯЗВИМАЯ РЕАЛИЗАЦИЯ
# repository.py: каждый запрос ниже вклеивает данные запроса в текст SQL
import sqlite3

from sqlalchemy import text


def get_user(pg_cursor, user_id):
    # psycopg2: f-строка собирает оператор до того, как его увидит драйвер
    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: этот % — форматирование строк Python, а не плейсхолдер драйвера
    return conn.execute("SELECT id FROM orders WHERE status = '%s'" % status).fetchall()


def find_by_email(session, email):
    # SQLAlchemy: text() не экранирует то, что уже находится в строке
    return session.execute(text(f"SELECT id FROM users WHERE email = '{email}'")).all()


def find_users(User, username):
    # Django: raw() отправляет подставленные значения в базу как есть
    return User.objects.raw(f"SELECT * FROM app_user WHERE username = '{username}'")


def get_orders_by_ids(session, ids):
    # список IN: склеенная строка позволяет элементу пронести лишний SQL
    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):
    # идентификатор: столбец сортировки берётся прямо из query string
    pg_cursor.execute(f"SELECT id, email FROM users ORDER BY {sort_column}")
    return pg_cursor.fetchall()
БЕЗОПАСНЫЙ ИСПРАВЛЕННЫЙ ВАРИАНТ
# repository.py: значения идут связанными параметрами, имена — из белого списка
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 — плейсхолдер драйвера; значение идёт во втором аргументе
    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 использует стиль qmark: плейсхолдеры ?, значения в кортеже
    return conn.execute("SELECT id FROM orders WHERE status = ?", (status,)).fetchall()


def find_by_email(session, email: str):
    # SQLAlchemy: именованный параметр, значение передаётся при вызове 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() передаёт params отдельно; фильтр ORM ещё проще
    return User.objects.raw("SELECT * FROM app_user WHERE username = %s", [username])


def get_orders_by_ids(session, ids: list[int]):
    # список IN: расширяемый bind даёт по плейсхолдеру на элемент
    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):
    # идентификатор: сначала белый список, потом кавычки через 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()

Чек-лист по защите системы для инженеров

Источники