SQL Injection Advanced Techniques: Modern Bypass Methods

-- WAF bypass: Encoding
UNION SELECT → UN/**/ION SEL/**/ECT

-- WAF bypass: Comments
SELECT/*comment*/user FROM users

-- WAF bypass: Case variation
SeLeCt UsEr FrOm UsErS

-- Function obfuscation
CHAR(65) → 0x41

Complete SQL Injection Prevention System:

#!/usr/bin/env python3
"""
Production-ready SQL Injection Prevention System
Comprehensive protection with parameterized queries, input validation, and monitoring
"""

import re
import logging
from typing import Any, Optional, List, Dict
from functools import wraps
from dataclasses import dataclass
from enum import Enum

logger = logging.getLogger(__name__)


class SecurityLevel(Enum):
    """Security level for SQL operations."""
    SAFE = "safe"
    WARNING = "warning"
    BLOCKED = "blocked"


@dataclass
class SQLInjectionAttempt:
    """SQL injection attempt detection."""
    pattern: str
    severity: SecurityLevel
    input_value: str
    details: str


class SQLInjectionDetector:
    """Detect SQL injection attempts in input."""
    
    # SQL injection patterns (for detection/monitoring, not prevention)
    INJECTION_PATTERNS = [
        (r"([';]+|(\-\-)|(\#)|(\/\*)|(\*\/))", SecurityLevel.BLOCKED, "SQL comment/terminator"),
        (r"(union|select|insert|update|delete|drop|create|alter|exec|execute)", SecurityLevel.WARNING, "SQL keyword"),
        (r"(\bor\b|\band\b)\s+\d+\s*=\s*\d+", SecurityLevel.BLOCKED, "Boolean-based injection"),
        (r"(\bor\b|\band\b)\s+['\"]\w+['\"]\s*=\s*['\"]\w+['\"]", SecurityLevel.BLOCKED, "String-based injection"),
        (r"(sleep|waitfor|delay|benchmark)", SecurityLevel.BLOCKED, "Time-based injection"),
        (r"(load_file|into\s+outfile|into\s+dumpfile)", SecurityLevel.BLOCKED, "File operation injection"),
        (r"(xp_cmdshell|sp_executesql)", SecurityLevel.BLOCKED, "Command execution"),
    ]
    
    @staticmethod
    def detect(input_value: str) -> Optional[SQLInjectionAttempt]:
        """Detect SQL injection attempt in input.
        
        Args:
            input_value: Input value to check
            
        Returns:
            SQLInjectionAttempt if detected, None otherwise
        """
        if not isinstance(input_value, str):
            return None
        
        input_lower = input_value.lower()
        
        for pattern, severity, description in SQLInjectionDetector.INJECTION_PATTERNS:
            if re.search(pattern, input_lower, re.IGNORECASE):
                return SQLInjectionAttempt(
                    pattern=pattern,
                    severity=severity,
                    input_value=input_value,
                    details=description
                )
        
        return None


class SafeDatabase:
    """Safe database wrapper with SQL injection prevention."""
    
    def __init__(self, connection):
        """Initialize safe database wrapper.
        
        Args:
            connection: Database connection object
        """
        self.conn = connection
        self.detector = SQLInjectionDetector()
    
    def execute_safe(self, query: str, params: tuple = None):
        """Execute parameterized query safely.
        
        Args:
            query: SQL query with placeholders
            params: Parameters for query
            
        Returns:
            Query result
            
        Raises:
            ValueError: If SQL injection detected
        """
        cursor = self.conn.cursor()
        
        # Validate query structure (basic check)
        if not self._is_safe_query(query):
            raise ValueError("Unsafe query structure detected")
        
        # Validate parameters if provided
        if params:
            for param in params:
                if isinstance(param, str):
                    attempt = self.detector.detect(param)
                    if attempt and attempt.severity == SecurityLevel.BLOCKED:
                        logger.warning(f"Blocked SQL injection attempt: {attempt.details}")
                        raise ValueError(f"Potential SQL injection detected: {attempt.details}")
        
        # Execute parameterized query
        try:
            if params:
                cursor.execute(query, params)
            else:
                cursor.execute(query)
            return cursor.fetchall()
        except Exception as e:
            logger.error(f"Database error: {e}")
            raise
    
    def _is_safe_query(self, query: str) -> bool:
        """Check if query structure is safe.
        
        Args:
            query: SQL query string
            
        Returns:
            True if query structure appears safe
        """
        # Check for multiple statements (potential injection)
        if query.count(';') > 1:
            return False
        
        # Check for dangerous keywords in non-parameterized context
        dangerous = ['drop', 'delete', 'truncate', 'alter', 'create']
        query_lower = query.lower()
        
        # Only allow dangerous operations if properly parameterized
        for keyword in dangerous:
            if keyword in query_lower and '?' not in query and '%s' not in query:
                return False
        
        return True


# Example: Safe database operations
import sqlite3

def get_user_safe(user_id: Any):
    """Get user safely with parameterized query.
    
    Args:
        user_id: User ID (will be validated)
        
    Returns:
        User data or None
    """
    # Validate input type
    if not isinstance(user_id, (int, str)):
        raise TypeError("Invalid user_id type")
    
    # Convert to int if string
    if isinstance(user_id, str):
        if not user_id.isdigit():
            raise ValueError("Invalid user_id format")
        user_id = int(user_id)
    
    conn = sqlite3.connect('database.db')
    db = SafeDatabase(conn)
    
    try:
        # Parameterized query - prevents ALL SQL injection
        result = db.execute_safe(
            "SELECT * FROM users WHERE id = ?",
            (user_id,)
        )
        return result[0] if result else None
    finally:
        conn.close()


def search_users_safe(search_term: str):
    """Search users safely with parameterized query.
    
    Args:
        search_term: Search term (will be validated)
        
    Returns:
        List of matching users
    """
    # Input validation
    if not isinstance(search_term, str):
        raise TypeError("Search term must be string")
    
    if len(search_term) > 100:
        raise ValueError("Search term too long")
    
    # Sanitize (remove dangerous characters, but parameterization is the real protection)
    search_term = search_term.strip()
    
    conn = sqlite3.connect('database.db')
    db = SafeDatabase(conn)
    
    try:
        # Parameterized query with LIKE pattern
        # Note: LIKE still uses parameterization, preventing injection
        result = db.execute_safe(
            "SELECT * FROM users WHERE username LIKE ?",
            (f"%{search_term}%",)
        )
        return result
    finally:
        conn.close()


# Flask integration with SQL injection protection
from flask import Flask, request, jsonify

app = Flask(__name__)


@app.route('/api/users/<user_id>', methods=['GET'])
def get_user_endpoint(user_id):
    """Get user endpoint with SQL injection protection."""
    try:
        user = get_user_safe(user_id)
        if user:
            return jsonify({
                'id': user[0],
                'username': user[1],
                'email': user[2]
            })
        else:
            return jsonify({'error': 'User not found'}), 404
    except (ValueError, TypeError) as e:
        return jsonify({'error': str(e)}), 400
    except Exception as e:
        logger.error(f"Error retrieving user: {e}")
        return jsonify({'error': 'Internal server error'}), 500


@app.route('/api/users/search', methods=['GET'])
def search_users_endpoint():
    """Search users endpoint with SQL injection protection."""
    search_term = request.args.get('q', '')
    
    try:
        users = search_users_safe(search_term)
        return jsonify({
            'results': [
                {'id': u[0], 'username': u[1], 'email': u[2]}
                for u in users
            ]
        })
    except (ValueError, TypeError) as e:
        return jsonify({'error': str(e)}), 400
    except Exception as e:
        logger.error(f"Error searching users: {e}")
        return jsonify({'error': 'Internal server error'}), 500


# Advanced: ORM-based protection (SQLAlchemy example)
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):
    """User model with ORM protection."""
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)


# ORM automatically uses parameterized queries
def get_user_orm(user_id: int):
    """Get user using ORM (automatically safe)."""
    engine = create_engine('sqlite:///database.db')
    Session = sessionmaker(bind=engine)
    session = Session()
    
    try:
        # ORM automatically uses parameterized queries
        user = session.query(User).filter(User.id == user_id).first()
        return user
    finally:
        session.close()

#!/usr/bin/env python3
"""
Unit tests for SQL Injection Prevention System
"""

import pytest
import sqlite3
from sql_injection_prevention import (
    SafeDatabase, SQLInjectionDetector, SecurityLevel
)

class TestSQLInjectionPrevention:
    """Tests for SQL injection prevention."""
    
    @pytest.fixture
    def db_connection(self):
        conn = sqlite3.connect(':memory:')
        conn.execute('CREATE TABLE users (id INTEGER, name TEXT)')
        conn.execute('INSERT INTO users VALUES (1, "test")')
        return conn
    
    @pytest.fixture
    def safe_db(self, db_connection):
        return SafeDatabase(db_connection)
    
    def test_safe_query(self, safe_db):
        """Test safe parameterized query."""
        result = safe_db.execute_safe(
            "SELECT * FROM users WHERE id = ?",
            (1,)
        )
        assert len(result) > 0
    
    def test_injection_detection(self):
        """Test SQL injection detection."""
        detector = SQLInjectionDetector()
        attempt = detector.detect("' OR '1'='1")
        assert attempt is not None
        assert attempt.severity == SecurityLevel.BLOCKED
    
    def test_blocked_injection(self, safe_db):
        """Test that injection attempts are blocked."""
        with pytest.raises(ValueError):
            safe_db.execute_safe(
                "SELECT * FROM users WHERE name = ?",
                ("' OR '1'='1",)
            )

if __name__ == "__main__":
    pytest.main([__file__, "-v"])

#!/usr/bin/env python3
"""
SQL Injection Prevention System Cleanup
Production-ready cleanup and resource management
"""

import logging
import sqlite3

logger = logging.getLogger(__name__)

class SQLInjectionPreventionCleanup:
    """Handles cleanup operations."""
    
    def __init__(self, safe_db):
        self.safe_db = safe_db
    
    def cleanup_test_data(self):
        """Clean up test data from database."""
        try:
            # In production, would have proper cleanup queries
            logger.info("Cleaning up test data")
            # Example: self.safe_db.execute_safe("DELETE FROM test_table WHERE test_flag = ?", (True,))
        except Exception as e:
            logger.error(f"Error cleaning up test data: {e}")
    
    def cleanup_connections(self):
        """Clean up database connections."""
        try:
            if hasattr(self.safe_db, 'conn'):
                self.safe_db.conn.close()
                logger.info("Database connection closed")
        except Exception as e:
            logger.error(f"Error closing connection: {e}")
    
    def cleanup(self):
        """Perform complete cleanup."""
        logger.info("Starting SQL injection prevention cleanup")
        self.cleanup_test_data()
        self.cleanup_connections()
        logger.info("SQL injection prevention cleanup complete")