FlawOpen

flawopen.com/sql-injection/Python

● CWE-89 · Kritis
Riset Keamanan · FlawOpen

SQL Injection di Python (psycopg2, SQLAlchemy, Django)

f-string atau % di dalam cursor.execute() membuat parameter request bisa menulis ulang SQL Anda, sehingga pencarian satu pengguna berubah menjadi dump seluruh tabel (CWE-89). Berikut pemanggilan yang aman untuk setiap API database Python, ditambah dua kasus yang tidak bisa ditangani placeholder: nama kolom dan daftar IN.[1], [2], [3], [4], [5], [6], [7], [9]

💡 Penjelasan Sederhana (ELI5)

Bayangkan meja sortir surat dengan kartu instruksi tetap: 'Sortir paket berlabel ___.' Jika petugas menyalin label dari pengirim ke kartu sebelum membacanya, kata-kata tambahan bisa menjadi perintah sortir baru. f-string Python atau pemformatan % melakukan hal itu pada SQL: satu string perintah dibuat sebelum driver menerima nilainya. execute(sql, params) menjaga kartu tetap dan mengirim label secara terpisah sebagai data. Nama kolom adalah bagian dari perintah; pilih dari daftar izin yang ditetapkan di kode.[10], [1], [2], [9]

Konsep Kunci & Istilah

cursor.execute(sql, params)
Pemanggilan dari DB-API 2.0 (PEP 249). Driver mengirim nilai di params terpisah dari teks SQL, sehingga nilai itu tidak pernah bisa mengubah perintahnya. Aman hanya jika nilai berada di argumen kedua; f-string atau ekspresi % di argumen pertama itulah bug-nya.[2], [3]
paramstyle: %s, ?, :name
Setiap driver menyatakan placeholder-nya di atribut paramstyle: psycopg2 dan PyMySQL memakai %s atau %(name)s, sqlite3 memakai ? atau :name. %s dibaca oleh driver, bukan oleh operator % milik Python, jadi jangan pernah memformat string sendiri.[2], [3], [5]
SQLAlchemy text()
Membungkus string SQL mentah untuk SQLAlchemy. Aman jika nilai berupa parameter bind bernama seperti :email yang diberikan ke execute() dalam bentuk dict. Tidak aman jika string yang diberikan sudah berupa f-string: text() tidak meng-escape apa pun di dalamnya.[6], [9]
Django raw() / RawSQL / extra()
Jalan pintas keluar dari ORM Django. raw(sql, params) dan RawSQL(sql, params) hanya aman jika setiap nilai masuk ke params; apa pun yang disisipkan ke sql dikirim Django apa adanya. Dokumentasi Django menyebut extra() sebagai pilihan terakhir, jadi larang saat review.[7], [9]
psycopg2.sql.Identifier
Placeholder hanya membawa nilai, tidak pernah nama tabel atau kolom. sql.Identifier membungkus nama dengan tanda kutip ganda agar tidak bisa keluar dari identifier. Cocokkan dulu nama itu dengan allowlist supaya pengguna tidak bisa memilih kolom yang tidak ingin Anda tampilkan.[4], [9]
IN lists and expanding binds
Di sqlite3, list Python tidak bisa di-bind ke satu ?, sehingga orang kembali menggabungkan string. Gunakan bindparam yang dapat mengembang di SQLAlchemy (satu placeholder per elemen), id = ANY(%s) dengan list di psycopg2, atau buat satu ? per elemen untuk sqlite3.[6], [3], [5]

Pemetaan kerangka kerja

Alur Serangan Langkah demi Langkah

Langkah 1

Nilai request sampai ke lapisan data

Parameter permintaan sampai ke get_user() sebagai teks. View Flask dapat membaca request.args["id"]; view Django membaca request.GET["id"]. View meneruskan teks yang belum divalidasi itu ke get_user().[11], [12], [9]

Langkah 2

f-string menyisipkan nilai ke dalam SQL

f"... WHERE id = {user_id}" menghasilkan satu string datar. Dengan nilai uji yang tidak berbahaya 0 OR 1=1, teksnya menjadi WHERE id = 0 OR 1=1, dan tidak ada penanda di mana nilai itu seharusnya berakhir.[10], [1], [9]

Langkah 3

Driver tidak menerima parameter apa pun

execute(query) dipanggil hanya dengan SQL, sehingga psycopg2 mengirim teks tanpa perubahan dan PostgreSQL mengurai OR 1=1 sebagai bagian dari klausa WHERE, bukan sebagai data.[3], [1], [9]

Langkah 4

Endpoint mengembalikan semua baris

Kondisinya bernilai benar untuk setiap baris, sehingga pencarian satu pengguna mengembalikan seluruh tabel. psycopg2 juga menerima beberapa perintah yang dipisahkan titik koma dalam satu panggilan; sqlite3 menolaknya, yang membatasi kerusakan tetapi tidak mencegah pencurian data.[13], [5], [1], [9]

Kode Sumber: Rentan vs Aman

IMPLEMENTASI RENTAN
# repository.py: setiap kueri di bawah menyisipkan data request ke teks SQL
import sqlite3

from sqlalchemy import text


def get_user(pg_cursor, user_id):
    # psycopg2: f-string menyusun perintah sebelum driver melihatnya
    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: % ini adalah format string Python, bukan placeholder driver
    return conn.execute("SELECT id FROM orders WHERE status = '%s'" % status).fetchall()


def find_by_email(session, email):
    # SQLAlchemy: text() tidak meng-escape isi yang sudah ada di dalam string
    return session.execute(text(f"SELECT id FROM users WHERE email = '{email}'")).all()


def find_users(User, username):
    # Django: raw() mengirim nilai yang disisipkan ke database apa adanya
    return User.objects.raw(f"SELECT * FROM app_user WHERE username = '{username}'")


def get_orders_by_ids(session, ids):
    # daftar IN: string gabungan memungkinkan satu elemen membawa SQL tambahan
    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):
    # identifier: kolom pengurutan diambil langsung dari query string
    pg_cursor.execute(f"SELECT id, email FROM users ORDER BY {sort_column}")
    return pg_cursor.fetchall()
PERBAIKAN AMAN & KUAT
# repository.py: nilai dikirim sebagai parameter bind, nama berasal dari 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 adalah placeholder driver; nilai masuk ke argumen kedua
    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 memakai gaya qmark: placeholder ?, nilai dalam tuple
    return conn.execute("SELECT id FROM orders WHERE status = ?", (status,)).fetchall()


def find_by_email(session, email: str):
    # SQLAlchemy: parameter bind bernama, nilai diberikan saat 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() mengirim params terpisah; filter ORM lebih sederhana lagi
    return User.objects.raw("SELECT * FROM app_user WHERE username = %s", [username])


def get_orders_by_ids(session, ids: list[int]):
    # daftar IN: bind yang mengembang membuat satu placeholder per elemen
    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):
    # identifier: cek allowlist dulu, lalu kutip dengan 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()

Daftar Periksa Penguatan Sistem Rekayasa

Sumber