From ba46e3812547b578e432e79d114b1127a0ef86bb Mon Sep 17 00:00:00 2001
From: Miroslav Stampar <miroslav.stampar@gmail.com>
Date: Thu, 19 Sep 2019 12:08:11 +0200
Subject: [PATCH] Improvement of SQLi detection

---
 bin/SQLInjectionDetection.py | 138 ++++++-----------------------------
 1 file changed, 24 insertions(+), 114 deletions(-)

diff --git a/bin/SQLInjectionDetection.py b/bin/SQLInjectionDetection.py
index ffb21003..f46a740e 100755
--- a/bin/SQLInjectionDetection.py
+++ b/bin/SQLInjectionDetection.py
@@ -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__':