CWE-89: SQL Injection - Python

Overview

SQL Injection in Python applications occurs when untrusted user input is incorporated into SQL queries without proper sanitization or parameterization. Attackers can exploit this to manipulate query logic, extract sensitive data, modify database contents, or execute administrative operations. 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:

• f-strings embed variables directly
• No escaping or validation
• Can inject SQL operators
• Data destruction

.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
# Makes all users 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 to modify WHERE clauses, escalate privileges, or access unauthorized data.

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, allowing attackers to inject SQL like ' OR 1=1-- to bypass WHERE conditions, access unauthorized records, or modify query logic.

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 = "id; DELETE FROM users WHERE '1'='1'--"
# SQL injection via ORDER BY

Why this is vulnerable:

• ORDER BY columns can't be parameterized traditionally
• String interpolation required
• Allows SQL injection
• No allowlist validation

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:

• User input directly from request
• f-string concatenation
• UNION injection
• Sensitive data exposure

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:

• The ? placeholders are replaced with parameterized values by sqlite3.
• 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.ext.declarative import declarative_base
from sqlalchemy.orm import 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, we validate input against a allowlist of allowed values.
• Only values from the allowlist are used in the query, ensuring no malicious input can reach the SQL.

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.
• The type=int parameter in request.args.get() ensures type safety by automatically converting and validating numeric inputs, length checks limit unbounded input size, and execute() with ? placeholders ensures all user input is parameterized.
• This defense-in-depth approach validates inputs before they reach the database, while parameterization ensures even bypassed validation cannot cause 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.

Verification

After implementing the recommended secure patterns, verify the fix through multiple approaches:

• Manual testing: Submit malicious payloads relevant to this vulnerability and confirm they're handled safely without executing unintended operations
• Code review: Confirm all instances use the secure pattern (parameterized queries, safe APIs, proper encoding) with no string concatenation or unsafe operations
• Static analysis: Use security scanners to verify no new vulnerabilities exist and the original finding is resolved
• Regression testing: Ensure legitimate user inputs and application workflows continue to function correctly
• Edge case validation: Test with special characters, boundary conditions, and unusual inputs to verify proper handling
• Framework verification: If using a framework or library, confirm the recommended APIs are used correctly according to documentation
• Authentication/session testing: Verify security controls remain effective and cannot be bypassed (if applicable to the vulnerability type)
• Rescan: Run the security scanner again to confirm the finding is resolved and no new issues were introduced

Additional Resources

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