FlawOpen

flawopen.com/sql-injection/Python

● CWE-89 · 緊急
セキュリティ研究 · FlawOpen

Python の SQL インジェクション（psycopg2・SQLAlchemy・Django）

cursor.execute() の中で f-string や % を使うと、リクエストパラメータが SQL を書き換え、1 人のユーザーを探す処理がテーブル全件の流出に変わります（CWE-89）。Python の各データベース API での安全な呼び出し方と、プレースホルダーでは扱えない 2 つのケース（カラム名と 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 文とは別に送るため、値が文の構造を変えることはありません。安全なのは値を第 2 引数に渡したときだけで、第 1 引数で f-string や % 式を使うことこそが脆弱性です。[2], [3]
paramstyle: %s, ?, :name
各ドライバーは paramstyle 属性でプレースホルダーの形式を宣言しています。psycopg2 と PyMySQL は %s または %(name)s、sqlite3 は ? または :name です。%s を解釈するのはドライバーであって Python の % 演算子ではないので、文字列を自分でフォーマットしてはいけません。[2], [3], [5]
SQLAlchemy text()
SQLAlchemy で生の SQL 文字列を包む関数です。:email のような名前付きバインドパラメータを使い、値を辞書で execute() に渡せば安全です。渡す文字列がすでに f-string なら危険です。text() は中身を一切エスケープしません。[6], [9]
Django raw() / RawSQL / extra()
Django の ORM を迂回する手段です。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 のリストを 1 つの ? にバインドできないため、結局文字列連結に戻ってしまいがちです。SQLAlchemy では展開可能な bindparam（要素ごとにプレースホルダー 1 つ）、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-string が値を SQL に埋め込む

f"... WHERE id = {user_id}" は 1 本の平坦な文字列を作ります。無害なテスト値 0 OR 1=1 を使うと、テキストは WHERE id = 0 OR 1=1 になり、値がどこで終わるはずだったかを示すものは何もありません。[10], [1], [9]

ステップ 3

ドライバーはパラメータを 1 つも受け取らない

execute(query) は SQL だけで呼ばれるので、psycopg2 はテキストをそのまま送り、PostgreSQL は OR 1=1 をデータではなく WHERE 句の一部として解析します。[3], [1], [9]

ステップ 4

エンドポイントが全行を返す

条件はすべての行で真になるため、1 人のユーザーを探す処理がテーブル全体を返します。psycopg2 はセミコロンで区切った複数の文を 1 回の呼び出しで受け付けます。sqlite3 はこれを拒否するので被害は限定されますが、データの窃取は防げません。[13], [5], [1], [9]

ソースコード比較：脆弱 vs 堅牢化

脆弱な実装
# repository.py: 以下のクエリはどれもリクエストのデータを SQL 文に埋め込んでいる
import sqlite3

from sqlalchemy import text


def get_user(pg_cursor, user_id):
    # psycopg2: ドライバーが見る前に f-string が文を組み立ててしまう
    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):
    # 識別子: ソート列をクエリ文字列から直接受け取っている
    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 はドライバーのプレースホルダー。値は第 2 引数に渡す
    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 リスト: 展開可能なバインドが要素ごとにプレースホルダーを生成する
    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()

エンジニアリング＆システム堅牢化チェックリスト

参考資料