FlawOpen

flawopen.com/sql-injection/Python

● CWE-89 · अति गंभीर
सुरक्षा अनुसंधान · FlawOpen

Python में SQL इंजेक्शन (psycopg2, SQLAlchemy, Django)

cursor.execute() के अंदर f-स्ट्रिंग या % का इस्तेमाल रिक्वेस्ट पैरामीटर को आपका SQL बदलने देता है, और एक यूज़र की खोज पूरी टेबल के डंप में बदल जाती है (CWE-89)। यहाँ Python के हर डेटाबेस API के लिए सुरक्षित कॉल दी गई है, साथ ही वे दो मामले भी जिन्हें प्लेसहोल्डर नहीं संभाल सकते: कॉलम के नाम और IN सूचियाँ।[1], [2], [3], [4], [5], [6], [7], [9]

💡 आसान भाषा में (ELI5)

डाक छँटाई की मेज पर एक स्थिर निर्देश-पर्ची की कल्पना करें: ‘___ लेबल वाले पार्सल को छाँटें।’ यदि कर्मचारी पढ़ने से पहले भेजने वाले का लेबल उसी पर्ची में लिख दे, तो अतिरिक्त शब्द नए छँटाई-निर्देश बन सकते हैं। Python का f-string या % फ़ॉर्मैटिंग SQL के साथ यही करता है: driver को मान मिलने से पहले पूरा command string बन जाता है। execute(sql, params) निर्देश-पर्ची को स्थिर रखता है और लेबल को data के रूप में अलग भेजता है। Column name निर्देश का हिस्सा है; उसे code में तय allowlist से चुनें।[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()
SQLAlchemy में कच्ची SQL स्ट्रिंग को लपेटने वाला फ़ंक्शन। यह तब सुरक्षित है जब मान :email जैसे नामित बाइंड पैरामीटर हों और execute() को dict के रूप में दिए जाएँ। अगर दी गई स्ट्रिंग पहले से f-स्ट्रिंग है तो असुरक्षित है: 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 की सूची को एक ही ? से नहीं बाँधा जा सकता, इसलिए लोग फिर से स्ट्रिंग जोड़ने लगते हैं। 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]

सोर्स कोड: कमज़ोर बनाम सुरक्षित कार्यान्वयन

कमज़ोर कार्यान्वयन
# 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 शैली अपनाता है: ? प्लेसहोल्डर, मान tuple में
    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()

इंजीनियरिंग और सिस्टम सुरक्षा चेकलिस्ट

स्रोत