Python: Parameterized Queries

# ============================================================
# WHAT DOES PARAMETERIZED MEAN?
# ============================================================

#  UNSAFE - Direct string concatenation
user_input = input("Enter your name: ")
query = f"SELECT * FROM users WHERE name = '{user_input}'"
cursor.execute(query)

#  SAFE - Parameterized query
user_input = input("Enter your name: ")
query = "SELECT * FROM users WHERE name = %s"
cursor.execute(query, (user_input,))

# ============================================================
# THE THREE KEY PARTS
# ============================================================

# 1. The SQL with placeholders (%s)
query = "SELECT * FROM students WHERE age > %s AND first_name = %s"

# 2. The data (as a tuple)
data = (18, "Rahul")

# 3. Execute with the data separately
cursor.execute(query, data)

# ============================================================
# WHY IT WORKS
# ============================================================

# The database sees:
# - The SQL structure (SELECT * FROM students WHERE age > ? AND first_name = ?)
# - The data separately (18, "Rahul")
# - It never mixes them together

# ============================================================
# THE PROBLEM WITH STRING CONCATENATION
# ============================================================

# Imagine this is your code
def get_user(name):
    # UNSAFE - Using string concatenation
    query = "SELECT * FROM users WHERE name = '" + name + "'"
    cursor.execute(query)
    return cursor.fetchall()

# ============================================================
# WHAT CAN GO WRONG
# ============================================================

# User enters: Rahul
# Query becomes: SELECT * FROM users WHERE name = 'Rahul'
#  Safe - Nothing wrong

# User enters: Rahul' OR '1'='1
# Query becomes: SELECT * FROM users WHERE name = 'Rahul' OR '1'='1'
#  DANGER! This returns ALL users!

# User enters: Rahul'; DROP TABLE users; --
# Query becomes: SELECT * FROM users WHERE name = 'Rahul'; DROP TABLE users; --'
#  DISASTER! The users table gets deleted!

# ============================================================
# HOW ATTACKERS EXPLOIT THIS
# ============================================================

print("""
Common SQL injection attacks:
1. Login bypass: ' OR '1'='1
2. Data theft: ' UNION SELECT * FROM passwords --
3. Data deletion: '; DROP TABLE users --
4. Data modification: '; UPDATE users SET admin=1 WHERE name='admin' --

💀 All of these can destroy your database!
""")

# ============================================================
# THE SAFE WAY
# ============================================================

def get_user_safe(name):
    # SAFE - Using parameterized query
    query = "SELECT * FROM users WHERE name = %s"
    cursor.execute(query, (name,))
    return cursor.fetchall()

# ============================================================
# WHAT HAPPENS WITH PARAMETERIZED QUERIES
# ============================================================

# User enters: Rahul
# Data sent: ('Rahul',)
#  Finds the user named Rahul

# User enters: Rahul' OR '1'='1
# Data sent: ("Rahul' OR '1'='1",)
#  Finds NO user (searches for that exact name)
# The database treats it as TEXT, not as SQL code

# User enters: Rahul'; DROP TABLE users; --
# Data sent: ("Rahul'; DROP TABLE users; --",)
#  Finds NO user (searches for that exact name)
# The database treats it as TEXT, not as SQL code

# ============================================================
# COMPARISON
# ============================================================

print("""
┌─────────────────────────┬─────────────────────────────┐
│ UNSAFE (String concat)  │ SAFE (Parameterized)        │
├─────────────────────────┼─────────────────────────────┤
│ ' OR '1'='1             │ Search for exact name       │
│ → Returns ALL users     │ → Returns NO users          │
├─────────────────────────┼─────────────────────────────┤
│ '; DROP TABLE users; -- │ Search for exact name       │
│ → Table DELETED!        │ → Table SAFE!               │
├─────────────────────────┼─────────────────────────────┤
│ ' UNION SELECT * FROM   │ Search for exact name       │
│ passwords --            │ → Data SAFE!                │
│ → Data STOLEN!          │                             │
└─────────────────────────┴─────────────────────────────┘
""")

# ============================================================
# SELECT - Reading data
# ============================================================

def get_students_by_age(min_age):
    query = "SELECT * FROM students WHERE age > %s"
    cursor.execute(query, (min_age,))
    return cursor.fetchall()

# Usage
students = get_students_by_age(18)

# ============================================================
# INSERT - Adding data
# ============================================================

def add_student(first_name, last_name, age, email):
    query = """
        INSERT INTO students (first_name, last_name, age, email)
        VALUES (%s, %s, %s, %s)
    """
    cursor.execute(query, (first_name, last_name, age, email))
    connection.commit()
    return cursor.lastrowid

# Usage
new_id = add_student("Rahul", "Sharma", 22, "rahul@email.com")

# ============================================================
# UPDATE - Modifying data
# ============================================================

def update_student_age(student_id, new_age):
    query = "UPDATE students SET age = %s WHERE student_id = %s"
    cursor.execute(query, (new_age, student_id))
    connection.commit()
    return cursor.rowcount

# Usage
rows_updated = update_student_age(1, 23)

# ============================================================
# DELETE - Removing data
# ============================================================

def delete_student(student_id):
    query = "DELETE FROM students WHERE student_id = %s"
    cursor.execute(query, (student_id,))
    connection.commit()
    return cursor.rowcount

# Usage
rows_deleted = delete_student(5)

# ============================================================
# MULTIPLE PARAMETERS
# ============================================================

def search_students(first_name, min_age, max_age):
    query = """
        SELECT * FROM students 
        WHERE first_name LIKE %s 
        AND age BETWEEN %s AND %s
    """
    cursor.execute(query, (f"%{first_name}%", min_age, max_age))
    return cursor.fetchall()

# Usage
results = search_students("Ra", 18, 25)

# ============================================================
# SECURE USER AUTHENTICATION SYSTEM
# ============================================================

import mysql.connector
import hashlib  # For password hashing

class SecureUserSystem:
    """A secure user system using parameterized queries"""
    
    def __init__(self, db_config):
        self.db_config = db_config
        self.connection = None
        self.cursor = None
    
    def connect(self):
        try:
            self.connection = mysql.connector.connect(**self.db_config)
            self.cursor = self.connection.cursor()
            return True
        except mysql.connector.Error as e:
            print(f"Connection failed: {e}")
            return False
    
    def hash_password(self, password):
        """Hash a password for secure storage"""
        return hashlib.sha256(password.encode()).hexdigest()
    
    def create_user(self, username, password, email):
        """Create a new user (SECURE)"""
        # Hash the password
        hashed_password = self.hash_password(password)
        
        # Parameterized query
        query = """
            INSERT INTO users (username, password_hash, email, created_at)
            VALUES (%s, %s, %s, NOW())
        """
        try:
            self.cursor.execute(query, (username, hashed_password, email))
            self.connection.commit()
            print(f" User '{username}' created successfully")
            return self.cursor.lastrowid
        except mysql.connector.IntegrityError:
            print(f" Username '{username}' already exists")
            return None
        except mysql.connector.Error as e:
            print(f" Error: {e}")
            self.connection.rollback()
            return None
    
    def login_user(self, username, password):
        """Authenticate a user (SECURE)"""
        hashed_password = self.hash_password(password)
        
        # Parameterized query
        query = """
            SELECT user_id, username, email, created_at
            FROM users
            WHERE username = %s AND password_hash = %s
        """
        self.cursor.execute(query, (username, hashed_password))
        user = self.cursor.fetchone()
        
        if user:
            print(f" Welcome back, {username}!")
            return user
        else:
            print(f" Invalid username or password")
            return None
    
    def update_user_email(self, user_id, new_email):
        """Update user email (SECURE)"""
        query = "UPDATE users SET email = %s WHERE user_id = %s"
        self.cursor.execute(query, (new_email, user_id))
        self.connection.commit()
        return self.cursor.rowcount
    
    def delete_user(self, user_id):
        """Delete a user (SECURE)"""
        query = "DELETE FROM users WHERE user_id = %s"
        self.cursor.execute(query, (user_id,))
        self.connection.commit()
        return self.cursor.rowcount
    
    def search_users(self, search_term):
        """Search for users (SECURE)"""
        query = """
            SELECT user_id, username, email
            FROM users
            WHERE username LIKE %s OR email LIKE %s
        """
        search_pattern = f"%{search_term}%"
        self.cursor.execute(query, (search_pattern, search_pattern))
        return self.cursor.fetchall()
    
    def close(self):
        if self.cursor:
            self.cursor.close()
        if self.connection:
            self.connection.close()

# ============================================================
# DEMONSTRATION
# ============================================================

db_config = {
    "host": "localhost",
    "user": "root",
    "password": "secret",
    "database": "myapp_db"
}

# Create the users table first (if not exists)
# CREATE TABLE IF NOT EXISTS users (
#     user_id INT AUTO_INCREMENT PRIMARY KEY,
#     username VARCHAR(50) UNIQUE NOT NULL,
#     password_hash VARCHAR(255) NOT NULL,
#     email VARCHAR(100) UNIQUE NOT NULL,
#     created_at DATETIME DEFAULT CURRENT_TIMESTAMP
# )

system = SecureUserSystem(db_config)
if system.connect():
    
    # 1. Create users
    print("\n Creating users...")
    system.create_user("rahul_sharma", "password123", "rahul@email.com")
    system.create_user("priya_patel", "secure456", "priya@email.com")
    
    # 2. Login
    print("\n Testing login...")
    system.login_user("rahul_sharma", "password123")
    system.login_user("rahul_sharma", "wrongpassword")
    
    # 3. Search users
    print("\n🔍 Searching for users...")
    results = system.search_users("rahul")
    for user in results:
        print(f"  Found: {user[1]} ({user[2]})")
    
    # 4. Update email
    print("\n Updating email...")
    rows = system.update_user_email(1, "rahul_new@email.com")
    print(f"Updated {rows} user(s)")
    
    system.close()

# ============================================================
# BEST PRACTICES FOR PARAMETERIZED QUERIES
# ============================================================

print("1. ALWAYS USE PARAMETERIZED QUERIES")
print("   - Never use string concatenation for SQL")
print("   - Never use f-strings for SQL")
print("   - Always use %s placeholders")

print("\n2. USE PARAMETERIZED QUERIES EVERYWHERE")
print("   - SELECT queries with user input")
print("   - INSERT queries")
print("   - UPDATE queries")
print("   - DELETE queries")

print("\n3. NEVER TRUST USER INPUT")
print("   - Always treat user input as potentially dangerous")
print("   - Validate input even with parameterized queries")

print("\n4. USE THE LEAST PRIVILEGE PRINCIPLE")
print("   - Database user should have minimal permissions")
print("   - Don't use root/admin for application")

print("\n5. HASH PASSWORDS")
print("   - Never store passwords in plain text")
print("   - Use strong hashing (SHA-256, bcrypt)")

print("\n6. USE ENVIRONMENT VARIABLES")
print("   - Don't hardcode credentials")
print("   - Store them securely")

print("\n7. LOG SUSPICIOUS ACTIVITY")
print("   - Monitor for SQL injection attempts")
print("   - Log errors for debugging")

print("\n8. KEEP YOUR DRIVERS UPDATED")
print("   - Update mysql-connector-python regularly")
print("   - Security patches are important")