CIRCL
/
AIL-framework
mirror of https://github.com/CIRCL/AIL-framework
2
1
Fork 0
Code Issues Releases Wiki Activity

Merge pull request #396 from stamparm/master 

Improvement of SQLi detection
pull/398/head
Thirion Aurélien 2019-09-19 12:09:28 +02:00 committed by GitHub
parent fed04e8e5a ba46e38125
commit ed217656ec
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 24 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

138
bin/SQLInjectionDetection.py
View File

@ -14,7 +14,6 @@ It test different possibility to makes some sqlInjection.
import time
import datetime
import redis
import string
import urllib.request
import re
from pubsublogger import publisher
@ -22,131 +21,42 @@ from Helper import Process
from packages import Paste
from pyfaup.faup import Faup
# Config Var
regex_injection = []
word_injection = []
word_injection_suspect = []
# Classic atome injection
regex_injection1 = "([[AND |OR ]+[\'|\"]?[0-9a-zA-Z]+[\'|\"]?=[\'|\"]?[0-9a-zA-Z]+[\'|\"]?])"
regex_injection.append(regex_injection1)
# Time-based attack
regex_injection2 = ["SLEEP\([0-9]+", "BENCHMARK\([0-9]+", "WAIT FOR DELAY ", "WAITFOR DELAY"]
regex_injection2 = re.compile('|'.join(regex_injection2))
regex_injection.append(regex_injection2)
# Interesting keyword
word_injection1 = [" IF ", " ELSE ", " CASE ", " WHEN ", " END ", " UNION ", "SELECT ", " FROM ", " ORDER BY ", " WHERE ", " DELETE ", " DROP ", " UPDATE ", " EXEC "]
word_injection.append(word_injection1)
# Database special keywords
word_injection2 = ["@@version", "POW(", "BITAND(", "SQUARE("]
word_injection.append(word_injection2)
# Html keywords
word_injection3 = ["<script>"]
word_injection.append(word_injection3)
# Suspect char
word_injection_suspect1 = ["\'", "\"", ";", "<", ">"]
word_injection_suspect += word_injection_suspect1
# Comment
word_injection_suspect2 = ["--", "#", "/*"]
word_injection_suspect += word_injection_suspect2
# Reference: https://github.com/stamparm/maltrail/blob/master/core/settings.py
SQLI_REGEX = r"information_schema|sysdatabases|sysusers|floor\(rand\(|ORDER BY \d+|\bUNION\s+(ALL\s+)?SELECT\b|\b(UPDATEXML|EXTRACTVALUE)\(|\bCASE[^\w]+WHEN.*THEN\b|\bWAITFOR[^\w]+DELAY\b|\bCONVERT\(|VARCHAR\(|\bCOUNT\(\*\)|\b(pg_)?sleep\(|\bSELECT\b.*\bFROM\b.*\b(WHERE|GROUP|ORDER)\b|\bSELECT \w+ FROM \w+|\b(AND|OR|SELECT)\b.*/\*.*\*/|/\*.*\*/.*\b(AND|OR|SELECT)\b|\b(AND|OR)[^\w]+\d+['\") ]?[=><]['\"( ]?\d+|ODBC;DRIVER|\bINTO\s+(OUT|DUMP)FILE"
def analyse(url, path):
faup.decode(url)
url_parsed = faup.get()
resource_path = url_parsed['resource_path']
query_string = url_parsed['query_string']
result_path = 0
result_query = 0
if resource_path is not None:
## TODO: # FIXME: remove me
try:
resource_path = resource_path.decode()
except:
pass
result_path = is_sql_injection(resource_path)
if query_string is not None:
## TODO: # FIXME: remove me
try:
query_string = query_string.decode()
except:
pass
result_query = is_sql_injection(query_string)
if (result_path > 0) or (result_query > 0):
if is_sql_injection(url.decode()):
faup.decode(url)
url_parsed = faup.get()
paste = Paste.Paste(path)
if (result_path > 1) or (result_query > 1):
print("Detected SQL in URL: ")
print(urllib.request.unquote(url))
to_print = 'SQLInjection;{};{};{};{};{}'.format(paste.p_source, paste.p_date, paste.p_name, "Detected SQL in URL", paste.p_rel_path)
publisher.warning(to_print)
#Send to duplicate
p.populate_set_out(path, 'Duplicate')
print("Detected SQL in URL: ")
print(urllib.request.unquote(url))
to_print = 'SQLInjection;{};{};{};{};{}'.format(paste.p_source, paste.p_date, paste.p_name, "Detected SQL in URL", paste.p_rel_path)
publisher.warning(to_print)
#Send to duplicate
p.populate_set_out(path, 'Duplicate')
msg = 'infoleak:automatic-detection="sql-injection";{}'.format(path)
p.populate_set_out(msg, 'Tags')
#statistics
tld = url_parsed['tld']
if tld is not None:
## TODO: # FIXME: remove me
try:
tld = tld.decode()
except:
pass
date = datetime.datetime.now().strftime("%Y%m")
server_statistics.hincrby('SQLInjection_by_tld:'+date, tld, 1)
else:
print("Potential SQL injection:")
print(urllib.request.unquote(url))
to_print = 'SQLInjection;{};{};{};{};{}'.format(paste.p_source, paste.p_date, paste.p_name, "Potential SQL injection", paste.p_rel_path)
publisher.info(to_print)
msg = 'infoleak:automatic-detection="sql-injection";{}'.format(path)
p.populate_set_out(msg, 'Tags')
#statistics
tld = url_parsed['tld']
if tld is not None:
## TODO: # FIXME: remove me
try:
tld = tld.decode()
except:
pass
date = datetime.datetime.now().strftime("%Y%m")
server_statistics.hincrby('SQLInjection_by_tld:'+date, tld, 1)
# Try to detect if the url passed might be an sql injection by appliying the regex
# defined above on it.
def is_sql_injection(url_parsed):
line = urllib.request.unquote(url_parsed)
line = str.upper(line)
result = []
result_suspect = []
for regex in regex_injection:
temp_res = re.findall(regex, line)
if len(temp_res)>0:
result.append(temp_res)
for word_list in word_injection:
for word in word_list:
temp_res = str.find(line, str.upper(word))
if temp_res!=-1:
result.append(line[temp_res:temp_res+len(word)])
for word in word_injection_suspect:
temp_res = str.find(line, str.upper(word))
if temp_res!=-1:
result_suspect.append(line[temp_res:temp_res+len(word)])
if len(result)>0:
print(result)
return 2
elif len(result_suspect)>0:
print(result_suspect)
return 1
else:
return 0
return re.search(SQLI_REGEX, url_parsed, re.I) is not None
if __name__ == '__main__':