How to Fix AI-Generated SQL Injection Vulnerabilities: A Developer's Security Audit Guide

# VULNERABLE - AI generated this for a "user search" request
def search_users(username):
    query = "SELECT * FROM users WHERE username = '" + username + "'"
    return db.execute(query)

# What happens with input: '; DROP TABLE users; --
# Result: SELECT * FROM users WHERE username = ''; DROP TABLE users; --'

# VULNERABLE - AI generated this for "dynamic filtering"
def get_user_posts(user_id, category):
    query = f"SELECT * FROM posts WHERE user_id = {user_id} AND category = '{category}'"
    return db.execute(query)

# Attack payload: category = "'; DELETE FROM posts WHERE '1'='1"

// VULNERABLE - AI created this for "advanced search"
function buildSearchQuery(filters) {
    let query = "SELECT * FROM products WHERE 1=1";
    
    if (filters.name) {
        query += " AND name LIKE '%" + filters.name + "%'";
    }
    
    if (filters.price) {
        query += " AND price < " + filters.price;
    }
    
    return db.query(query);
}

import re
import os

def find_sql_injection_patterns(directory):
    """
    Scan for common SQL injection patterns in AI-generated code
    Based on patterns I found during manual audits
    """
    
    dangerous_patterns = [
        # String concatenation with user input
        r'["\']SELECT.*[+].*["\']',
        r'["\']INSERT.*[+].*["\']',
        r'["\']UPDATE.*[+].*["\']',
        r'["\']DELETE.*[+].*["\']',
        
        # f-string formatting in SQL
        r'f["\']SELECT.*\{.*\}.*["\']',
        r'f["\']INSERT.*\{.*\}.*["\']',
        
        # .format() method usage
        r'["\']SELECT.*["\']\.format\(',
        r'["\']INSERT.*["\']\.format\(',
        
        # JavaScript template literals
        r'`SELECT.*\$\{.*\}`',
        r'`INSERT.*\$\{.*\}`'
    ]
    
    vulnerable_files = []
    
    for root, dirs, files in os.walk(directory):
        for file in files:
            if file.endswith(('.py', '.js', '.php', '.java')):
                file_path = os.path.join(root, file)
                
                with open(file_path, 'r', encoding='utf-8') as f:
                    content = f.read()
                    
                    for pattern in dangerous_patterns:
                        matches = re.findall(pattern, content, re.IGNORECASE)
                        if matches:
                            vulnerable_files.append({
                                'file': file_path,
                                'pattern': pattern,
                                'matches': matches
                            })
    
    return vulnerable_files

# Usage
vulnerabilities = find_sql_injection_patterns('./src')
print(f"Found {len(vulnerabilities)} potential SQL injection points")

## Manual Review Checklist

### High-Risk Functions (check these first):
- [ ] Login/authentication endpoints
- [ ] Search functionality  
- [ ] Admin panels
- [ ] Reporting/analytics features
- [ ] User profile updates

### Questions I ask for each SQL operation:
1. Where does the input data come from?
2. Is there any input validation before the query?
3. Does the query use parameterized statements?
4. Can an attacker control any part of the SQL structure?
5. What's the blast radius if this gets exploited?

# AI generated this for "user profile updates"
def update_user_profile(user_id, field, value):
    # Looks safe at first glance - user_id is an integer
    query = f"UPDATE users SET {field} = '{value}' WHERE id = {user_id}"
    return db.execute(query)

# The vulnerability: 'field' parameter allows column name injection
# Attack: field = "password = 'hacked' WHERE id = 1; UPDATE users SET email"
# Result: Can modify any user's password

def get_user_by_email(email):
    query = "SELECT * FROM users WHERE email = '" + email + "'"
    return db.execute(query)

def get_user_by_email(email):
    query = "SELECT * FROM users WHERE email = %s"
    return db.execute(query, (email,))

# Test with normal input
result = get_user_by_email("user@example.com")
print(f"Normal test: {len(result)} results")

# Test with injection payload
malicious_input = "'; DROP TABLE users; --"
result = get_user_by_email(malicious_input)
print(f"Injection test: {len(result)} results")

# Expected: Both tests return 0 or 1 result, no database errors

def search_products(filters):
    query = "SELECT * FROM products WHERE 1=1"
    
    if filters.get('name'):
        query += f" AND name LIKE '%{filters['name']}%'"
    
    if filters.get('category'):
        query += f" AND category = '{filters['category']}'"
    
    if filters.get('max_price'):
        query += f" AND price <= {filters['max_price']}"
    
    return db.execute(query)

def search_products(filters):
    query = "SELECT * FROM products WHERE 1=1"
    params = []
    
    if filters.get('name'):
        query += " AND name LIKE %s"
        params.append(f"%{filters['name']}%")
    
    if filters.get('category'):
        query += " AND category = %s"
        params.append(filters['category'])
    
    if filters.get('max_price'):
        # Validate numeric input
        try:
            max_price = float(filters['max_price'])
            query += " AND price <= %s"
            params.append(max_price)
        except ValueError:
            # Log invalid input, don't add to query
            pass
    
    return db.execute(query, params)

def get_user_data(user_id, columns):
    # AI generated this for "flexible user data API"
    column_list = ", ".join(columns)
    query = f"SELECT {column_list} FROM users WHERE id = %s"
    return db.execute(query, (user_id,))

def get_user_data(user_id, columns):
    # Whitelist allowed columns
    allowed_columns = {
        'id', 'username', 'email', 'created_at', 
        'last_login', 'profile_image', 'bio'
    }
    
    # Filter to only allowed columns
    safe_columns = [col for col in columns if col in allowed_columns]
    
    if not safe_columns:
        safe_columns = ['id', 'username']  # Default safe selection
    
    column_list = ", ".join(safe_columns)
    query = f"SELECT {column_list} FROM users WHERE id = %s"
    return db.execute(query, (user_id,))

def process_bulk_order(order_data):
    for item in order_data['items']:
        # AI generated this complex inventory check
        query = f"""
            UPDATE inventory 
            SET quantity = quantity - {item['quantity']}
            WHERE product_id = {item['product_id']} 
            AND warehouse = '{item['warehouse']}'
        """
        db.execute(query)

def process_bulk_order(order_data):
    try:
        db.begin_transaction()
        
        for item in order_data['items']:
            # Validate inputs first
            product_id = int(item['product_id'])  # Raises ValueError if invalid
            quantity = int(item['quantity'])      # Raises ValueError if invalid
            
            # Whitelist warehouse names
            allowed_warehouses = {'main', 'west', 'east', 'central'}
            if item['warehouse'] not in allowed_warehouses:
                raise ValueError(f"Invalid warehouse: {item['warehouse']}")
            
            query = """
                UPDATE inventory 
                SET quantity = quantity - %s
                WHERE product_id = %s AND warehouse = %s
            """
            db.execute(query, (quantity, product_id, item['warehouse']))
        
        db.commit()
        
    except (ValueError, DatabaseError) as e:
        db.rollback()
        raise BusinessLogicError(f"Order processing failed: {e}")

import sqlite3
from contextlib import contextmanager

@contextmanager
def test_database():
    """Create isolated test database for injection testing"""
    conn = sqlite3.connect(':memory:')
    cursor = conn.cursor()
    
    # Create test schema
    cursor.execute('''
        CREATE TABLE users (
            id INTEGER PRIMARY KEY,
            username TEXT,
            email TEXT,
            password TEXT
        )
    ''')
    
    # Insert test data
    cursor.execute("INSERT INTO users VALUES (1, 'admin', 'admin@test.com', 'secret')")
    cursor.execute("INSERT INTO users VALUES (2, 'user', 'user@test.com', 'password')")
    
    yield conn
    conn.close()

def test_injection_vulnerability(query_function):
    """Test if a query function is vulnerable to SQL injection"""
    
    injection_payloads = [
        "'; DROP TABLE users; --",
        "' OR '1'='1",
        "' UNION SELECT password FROM users --",
        "admin'--",
        "' OR 1=1 /*"
    ]
    
    with test_database() as db:
        for payload in injection_payloads:
            try:
                result = query_function(db, payload)
                
                # Check if we got unexpected results
                if isinstance(result, list) and len(result) > 1:
                    print(f"⚠️  POTENTIAL VULNERABILITY: Payload '{payload}' returned {len(result)} results")
                
                # Check if we can still query the users table (wasn't dropped)
                test_query = db.execute("SELECT COUNT(*) FROM users").fetchone()
                if test_query[0] != 2:
                    print(f"🚨 CRITICAL: Database structure modified by payload '{payload}'")
                    
            except Exception as e:
                # SQL errors might indicate injection attempts were blocked
                print(f"✅ Payload '{payload}' properly rejected: {e}")

# Example usage
def vulnerable_login(db, username):
    query = f"SELECT * FROM users WHERE username = '{username}'"
    return db.execute(query).fetchall()

def secure_login(db, username):
    query = "SELECT * FROM users WHERE username = ?"
    return db.execute(query, (username,)).fetchall()

print("Testing vulnerable function:")
test_injection_vulnerability(vulnerable_login)

print("\nTesting secure function:")
test_injection_vulnerability(secure_login)

## AI Code Security Review Checklist

### Before approving any database interaction code:

**SQL Injection Prevention:**
- [ ] No string concatenation or f-strings in SQL queries
- [ ] All user input uses parameterized queries
- [ ] Dynamic column/table names use whitelisting
- [ ] Input validation exists for all parameters

**Input Validation:**
- [ ] Data types are validated before database operations
- [ ] String lengths are limited and checked
- [ ] Numeric ranges are validated
- [ ] Special characters are handled properly

**Error Handling:**
- [ ] Database errors don't expose schema information
- [ ] Transaction rollback is implemented for multi-step operations
- [ ] Logging captures security events without exposing sensitive data

**Business Logic:**
- [ ] Authorization checks exist for data access
- [ ] Rate limiting is considered for query-heavy endpoints
- [ ] Data sanitization happens before storage and display