CWE-89: Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection') - Python

Overview

SQL Injection in Python applications occurs when untrusted user input is incorporated into SQL queries without parameterization. An attacker who controls part of the query text controls what it does: bypassing authentication, reading data out of tables the query never referenced, or destroying rows. Python's database libraries (sqlite3, psycopg2, mysql-connector, etc.) all support parameterized queries as the primary defense.

Common Python SQL Injection Scenarios:

• String concatenation or f-strings building SQL queries
• Using .format() or % formatting with user input
• Raw SQL in Django/Flask without proper escaping
• SQLAlchemy text() queries with string concatenation
• Dynamic ORDER BY, table names, or column names

Popular Python Database Libraries:

• sqlite3: Built-in SQLite database
• psycopg2: PostgreSQL adapter
• mysql-connector-python / PyMySQL: MySQL drivers
• SQLAlchemy: ORM and SQL toolkit
• Django ORM: Django's database abstraction layer

Primary Defence: Use parameterized queries with placeholders (? for sqlite3, %s for psycopg2/MySQL), SQLAlchemy bound parameters, or Django ORM QuerySets which automatically parameterize values.

Common Vulnerable Patterns

String Concatenation

# VULNERABLE - String concatenation in SQL query
import sqlite3

def get_user(username):
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()

    # VULNERABLE - Direct string concatenation
    query = "SELECT * FROM users WHERE username = '" + username + "'"
    cursor.execute(query)

    return cursor.fetchone()

# Attack: username = "admin' OR '1'='1"
# Query becomes: SELECT * FROM users WHERE username = 'admin' OR '1'='1'
# Returns first user (authentication bypass!)

Why this is vulnerable:

• String concatenation allows attackers to inject SQL syntax by closing the string with a quote and adding operators like OR '1'='1, which always evaluates to true, bypassing WHERE conditions to access all records or authenticate without credentials.

f-strings in SQL

# VULNERABLE - f-strings with user input
import psycopg2

def delete_log(log_id):
    conn = psycopg2.connect("dbname=app user=postgres")
    cursor = conn.cursor()

    # VULNERABLE - f-string formatting
    query = f"DELETE FROM logs WHERE id = {log_id}"
    cursor.execute(query)
    conn.commit()

# Attack: log_id = "1 OR 1=1"
# Query becomes: DELETE FROM logs WHERE id = 1 OR 1=1
# Deletes ALL logs!

Why this is vulnerable:

• The f-string interpolates log_id in Python, before the driver ever sees the query, so nothing escapes or validates it.
• 1 OR 1=1 becomes part of the WHERE clause, and the DELETE removes every row in logs instead of one.

.format() Method

# VULNERABLE - Using .format() with SQL
import mysql.connector

def update_user_role(user_id, role):
    conn = mysql.connector.connect(
        host="localhost", user="app", password="pass", database="myapp"
    )
    cursor = conn.cursor()

    # VULNERABLE - .format() with user input
    query = "UPDATE users SET role = '{}' WHERE id = {}".format(role, user_id)
    cursor.execute(query)
    conn.commit()

# Attack: role = "admin' WHERE '1'='1' -- ", user_id = "1"
# Query: UPDATE users SET role = 'admin' WHERE '1'='1' -- ' WHERE id = 1
# The -- comments out the original WHERE, so every user becomes admin

Why this is vulnerable:

• The .format() method performs string substitution before the query reaches the database, allowing attackers to inject quotes and SQL operators like ' OR '1'='1 that escape the role literal and rewrite which rows the UPDATE touches.

Django Raw SQL

# VULNERABLE - Django raw SQL with string formatting
from django.db import connection

def search_products(category):
    with connection.cursor() as cursor:
        # VULNERABLE - % formatting in raw SQL
        query = "SELECT * FROM products WHERE category = '%s'" % category
        cursor.execute(query)
        return cursor.fetchall()

# Attack: category = "electronics' UNION SELECT username, password, null FROM users--"
# Exfiltrates user credentials

Why this is vulnerable:

• Using % formatting with raw SQL performs string substitution outside the database driver's parameterization, allowing UNION injections like ' UNION SELECT password FROM users-- to extract sensitive data from other tables.

SQLAlchemy text() with Concatenation

# VULNERABLE - SQLAlchemy text() with string concat
from sqlalchemy import create_engine, text

engine = create_engine('postgresql://user:pass@localhost/db')

def get_orders(status):
    # VULNERABLE - String concatenation in text()
    query = text("SELECT * FROM orders WHERE status = '" + status + "'")

    with engine.connect() as conn:
        result = conn.execute(query)
        return result.fetchall()

# Attack: status = "pending' OR role='admin'--"
# Bypasses authorization checks

Why this is vulnerable:

• String concatenation with SQLAlchemy's text() bypasses SQLAlchemy's parameter binding: the SQL is already assembled by the time text() receives it, so an injected ' OR 1=1-- drops the WHERE condition and the query returns rows the caller should not see.

Dynamic ORDER BY

# VULNERABLE - User-controlled ORDER BY clause
import sqlite3

def get_users_sorted(sort_column, sort_order):
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()

    # VULNERABLE - Column name and order from user input
    query = f"SELECT * FROM users ORDER BY {sort_column} {sort_order}"
    cursor.execute(query)

    return cursor.fetchall()

# Attack: sort_column injects a subquery, and the row order leaks the answer
#   "(CASE WHEN (SELECT password FROM users WHERE id=1) LIKE 'a%'
#     THEN id ELSE username END)"
# Repeat one character at a time to read the column out of the ordering

Why this is vulnerable:

• An ORDER BY column cannot be a bound parameter, so the value is interpolated and becomes part of the query structure.
• ORDER BY accepts an arbitrary expression, including a subquery, so an attacker does not need to break out of a string literal - there is no quote to escape.
• Nothing constrains the value to a real column name.

Do not expect ; DROP TABLE to work here, and do not conclude it is safe when it does not. Python's sqlite3 rejects a second statement outright:

sqlite3.ProgrammingError: You can only execute one statement at a time.

Stacked statements are a property of the driver, not of the vulnerability. Node's mysql2, for instance, only enables them when multipleStatements is set - so the classic payload fails against code that is fully injectable. Whether any given driver permits a batch has to be checked for that driver; the single-statement subquery above works either way, which is why it is the better test.

Flask with SQLite

# VULNERABLE - Flask route with SQL injection
from flask import Flask, request
import sqlite3

app = Flask(__name__)

@app.route('/user')
def get_user():
    user_id = request.args.get('id')

    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()

    # VULNERABLE - Query parameter directly in SQL
    cursor.execute(f"SELECT * FROM users WHERE id = {user_id}")
    user = cursor.fetchone()

    return str(user)

# Attack: /user?id=-1 UNION SELECT password, null FROM admin_users
# Extracts admin passwords

Why this is vulnerable:

• The id query parameter goes straight from request.args into an f-string, with no type check or escaping between them.
• That lets an attacker append UNION SELECT password, null FROM admin_users, so the route returns passwords from a table the query never referenced. The negative id is doing work: the route returns a single row through fetchone(), so the injected id has to match no real user for the admin row to be the one that comes back.

ORM .raw() Method Abuse

# VULNERABLE - Django ORM .raw() with string formatting
from django.contrib.auth.models import User

def find_users(search_term):
    # VULNERABLE - String interpolation in .raw()
    query = f"SELECT * FROM auth_user WHERE username LIKE '%{search_term}%'"
    users = User.objects.raw(query)

    return list(users)

# Attack: search_term = "%' OR '1'='1"
# Returns all users

Why this is vulnerable:

• Django's .raw() method requires proper parameterization - using % or .format() for string substitution allows LIKE injection with patterns like %' OR '1'='1, enabling attackers to extract all records regardless of the intended filter.

Secure Patterns

Parameterized Queries (sqlite3)

# SECURE - Parameterized queries with sqlite3
import sqlite3

def get_user_secure(username):
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()

    # SECURE - Use ? placeholder for parameters
    query = "SELECT * FROM users WHERE username = ?"
    cursor.execute(query, (username,))

    return cursor.fetchone()

def get_user_by_id_and_role(user_id, role):
    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()

    # SECURE - Multiple parameters
    query = "SELECT * FROM users WHERE id = ? AND role = ?"
    cursor.execute(query, (user_id, role))

    return cursor.fetchone()

Why this works:

• cursor.execute() compiles the statement and binds each ? as a value, so nothing is substituted into the query text.
• The query structure and parameter values are sent separately, ensuring values are treated as data only, preventing SQL injection regardless of input content.

Parameterized Queries (psycopg2)

# SECURE - PostgreSQL parameterized queries
import psycopg2

def delete_log_secure(log_id):
    conn = psycopg2.connect("dbname=app user=postgres")
    cursor = conn.cursor()

    # SECURE - Use %s placeholder (NOT % formatting!)
    query = "DELETE FROM logs WHERE id = %s"
    cursor.execute(query, (log_id,))
    conn.commit()

def insert_user(username, email, role):
    conn = psycopg2.connect("dbname=app user=postgres")
    cursor = conn.cursor()

    # SECURE - Named parameters
    query = """
        INSERT INTO users (username, email, role)
        VALUES (%(username)s, %(email)s, %(role)s)
        RETURNING id
    """
    cursor.execute(query, {
        'username': username,
        'email': email,
        'role': role
    })

    user_id = cursor.fetchone()[0]
    conn.commit()
    return user_id

Why this works:

• psycopg2's %s syntax creates placeholders (not Python % formatting!).
• Named parameters like %(username)s map dictionary keys to placeholders.
• The library sends the query and values separately, preventing injection.

MySQL with Parameterization

# SECURE - MySQL parameterized queries
import mysql.connector

def update_user_role_secure(user_id, role):
    # Validate role against allowlist
    allowed_roles = ['user', 'moderator', 'admin']
    if role not in allowed_roles:
        raise ValueError(f"Invalid role. Must be one of: {allowed_roles}")

    conn = mysql.connector.connect(
        host="localhost", user="app", password="pass", database="myapp"
    )
    cursor = conn.cursor()

    # SECURE - Use %s placeholder
    query = "UPDATE users SET role = %s WHERE id = %s"
    cursor.execute(query, (role, user_id))
    conn.commit()

Why this works:

• mysql.connector uses %s placeholders for parameterization (distinct from Python % formatting).
• The allowlist validation ensures only allowed values reach the query, and parameterization prevents SQL syntax injection.

Django ORM (Recommended)

# SECURE - Django ORM methods (type-safe)
from django.contrib.auth.models import User

def search_products_secure(category):
    # SECURE - Django ORM automatically parameterizes
    products = Product.objects.filter(category=category)
    return list(products)

def get_user_by_credentials(username, email):
    # SECURE - Q objects are safe
    from django.db.models import Q
    user = User.objects.filter(
        Q(username=username) & Q(email=email)
    ).first()
    return user

def get_expensive_products(min_price):
    # SECURE - ORM comparison operators
    products = Product.objects.filter(price__gte=min_price)
    return products

Why this works:

• Django ORM translates method calls like filter() into parameterized SQL automatically.
• Field lookups (e.g., price__gte) generate safe SQL with parameters, ensuring user input is never concatenated into queries.

SQLAlchemy ORM

# SECURE - SQLAlchemy ORM (type-safe)
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50))
    email = Column(String(100))
    role = Column(String(20))

engine = create_engine('postgresql://user:pass@localhost/db')
Session = sessionmaker(bind=engine)

def get_users_by_role_secure(role):
    session = Session()

    # SECURE - ORM automatically parameterizes
    users = session.query(User).filter(User.role == role).all()

    return users

def search_users(search_term):
    session = Session()

    # SECURE - LIKE with parameterization
    users = session.query(User).filter(
        User.username.like(f'%{search_term}%')
    ).all()

    return users

Why this works:

• SQLAlchemy ORM generates parameterized SQL from Python expressions.
• The .like() method creates a parameterized LIKE clause where the pattern is a parameter, not a concatenated string, preventing injection.

SQLAlchemy text() with Bindparams

# SECURE - SQLAlchemy text() with proper parameterization
from sqlalchemy import create_engine, text

engine = create_engine('postgresql://user:pass@localhost/db')

def get_orders_secure(status):
    # SECURE - Use :param syntax for parameters
    query = text("SELECT * FROM orders WHERE status = :status")

    with engine.connect() as conn:
        result = conn.execute(query, {"status": status})
        return result.fetchall()

def complex_query_secure(min_amount, max_amount):
    # SECURE - Multiple parameters
    query = text("""
        SELECT * FROM orders 
        WHERE amount BETWEEN :min_amount AND :max_amount
        ORDER BY created_at DESC
    """)

    with engine.connect() as conn:
        result = conn.execute(query, {
            "min_amount": min_amount,
            "max_amount": max_amount
        })
        return result.fetchall()

Why this works:

• SQLAlchemy's text() with :param syntax creates named parameter placeholders.
• When you pass a dictionary to execute(), SQLAlchemy binds values as parameters, sending them separately from the query to prevent injection.

Dynamic ORDER BY with Allowlist

# SECURE - Dynamic ORDER BY with column allowlist
import sqlite3

def get_users_sorted_secure(sort_column, sort_order):
    # SECURE - Allowlist allowed columns
    allowed_columns = ['id', 'username', 'email', 'created_at']
    allowed_orders = ['ASC', 'DESC']

    if sort_column not in allowed_columns:
        raise ValueError(f"Invalid column. Allowed: {allowed_columns}")

    if sort_order.upper() not in allowed_orders:
        raise ValueError(f"Invalid order. Allowed: {allowed_orders}")

    conn = sqlite3.connect('app.db')
    cursor = conn.cursor()

    # SECURE - Use validated allowlist values (can't parameterize ORDER BY)
    query = f"SELECT * FROM users ORDER BY {sort_column} {sort_order.upper()}"
    cursor.execute(query)

    return cursor.fetchall()

Why this works:

• ORDER BY column and direction cannot be parameterized.
• Instead, the column and the direction are checked against allowlists first, so the only strings that ever reach the f-string are the four column names and the two sort directions written into the function.

Flask with Parameterization

# SECURE - Flask with parameterized queries
from flask import Flask, request, jsonify
import sqlite3

app = Flask(__name__)

def get_db():
    conn = sqlite3.connect('app.db')
    conn.row_factory = sqlite3.Row
    return conn

@app.route('/user')
def get_user_secure():
    user_id = request.args.get('id', type=int)

    if not user_id:
        return jsonify({'error': 'Invalid user ID'}), 400

    conn = get_db()
    cursor = conn.cursor()

    # SECURE - Parameterized query
    cursor.execute("SELECT id, username, email FROM users WHERE id = ?", (user_id,))
    user = cursor.fetchone()

    if not user:
        return jsonify({'error': 'User not found'}), 404

    return jsonify(dict(user))

@app.route('/search')
def search_users_secure():
    search_term = request.args.get('q', '')

    if len(search_term) > 50:
        return jsonify({'error': 'Search term too long'}), 400

    conn = get_db()
    cursor = conn.cursor()

    # SECURE - Parameterized LIKE query
    cursor.execute(
        "SELECT id, username FROM users WHERE username LIKE ?",
        (f'%{search_term}%',)
    )
    users = [dict(row) for row in cursor.fetchall()]

    return jsonify(users)

Why this works:

• Flask routes combine parameterized queries with input validation at the application layer.
• request.args.get('id', type=int) converts the parameter and yields None for anything non-numeric, and the length check bounds the search term - but the ? placeholders are what stop injection, so a gap in either check does not become a SQL injection.
• The sqlite3.Row factory allows dict-like access to results without extra manual mapping.
• Error handling avoids leaking internal details that could aid attackers.

Common Pitfalls

• Confusing the driver's %s placeholder syntax (psycopg2, mysql.connector) with Python's % string-formatting operator. Writing "... WHERE id = %s" % user_id applies old-style string formatting client-side before the query ever reaches the driver - the exact vulnerability the placeholder syntax exists to prevent - even though the resulting SQL text looks identical to the safe version.
• Calling Django's .raw() or .extra() with an f-string or .format()-built query instead of using the params argument. .raw(query, params) parameterizes correctly, but .raw(f"...{value}...") builds the SQL client-side first, so the params protection never applies to that value.
• Writing a SQLAlchemy text() query where most values use :name bind parameters but one value - often a table name, column name, or something assembled from configuration - is still concatenated into the SQL string before being passed to text().
• Allowlisting a sort column and order for one query function, then reusing the raw sort_column/sort_order request values in a second endpoint or export/report path that copies the query logic without re-adding the allowlist check.

Additional Resources

• CWE-89: SQL Injection
• Django Query Security
• OWASP SQL Injection
• Python DB-API Parameterization
• SQLAlchemy SQL Injection Protection