Gist of the Day: Turbo Intruder Cluster Bomb with SmartFiltering

cluster.py

# Gist of the Day: Turbo Intruder Cluster Bomb with SmartFiltering
# Author: Evan Custodio (@defparam)
#
# MIT License
# Copyright 2021 Evan Custodio
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.


####################################
### Helper functions and classes ###
####################################

# This is our make_request function which takes in a method/path/host and returns a fully formed HTTP request with those insertions
def make_request(host, method, path):
    req = '''%s /%s HTTP/1.1
Host: %s
X-Forwarded-For: 127.0.0.1
Accept-Encoding: gzip, deflate
Accept: */*
Content-Type: application/json
Accept-Language: en
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.105 Safari/537.36

''' % (method, path, host)
    return req

# The SmartFilter object analyzes a request from the past history of requests to try and prevent noise
# It does this my imposing a limit on the number of common STATUS+WORDCOUNT that can be shown
class SmartFilter():
    def __init__(self, repeats=10):
        self._db = {} # our data base to keep track of history
        self._repeats = repeats # the number of repeats allowed before muting future responses
    def check(self, status, wordlen):
        key = str(status)+str(wordlen) # We make a directory key by concating status code + number of words
        if key not in self._db: # if we never seen this key before, add it to the dictionary with 1 hit
            self._db[key] = 1
        elif (self._db[key] >= self._repeats): # if the key exists and it reached the repeat maximum mute the response
            return False
        else: # If the key hasn't reached the repeat limit, add to the hit count and allow the response to be shown
            self._db[key] += 1
        return True

# load_filelist helper function takes items from a file and
# loads them into a python list. For dirsearch types of filelists
# it performs extension replacement
def load_filelist(listloc, ext="jsp"):
    with open(listloc) as f:
        paths = f.read().replace("%EXT%",ext).split('\n')
    return paths
####################################

        
##############################
### Global Test Parameters ###
##############################
THE_HOST = "www.<HOST>.com" # the host to evaluate
PATHS = load_filelist("L:\home\websec\git\dirsearch\db\dicc.txt") # the location of your dirsearch wordlist
METHODS = ["GET", "POST", "PUT", "PATCH", ] # list of methods to test
FILTER = SmartFilter(repeats=3) # Only allow repeats of 3 common responses
##############################

def queueRequests(target, wordlists):
    engine = RequestEngine(endpoint="https://%s:443/"%(THE_HOST),
                           concurrentConnections=2,
                           requestsPerConnection=10,
                           resumeSSL=0,
                           timeout=1,
                           pipeline=0,
                           maxRetriesPerRequest=0,
                           engine=Engine.THREADED,
                           )

    # Clusterbomb style of test generation (create a test for every method and every path)
    for method in METHODS:
        for path in PATHS:
            test = make_request(THE_HOST, method, path)
            engine.queue(test,label=method)


# Response Handling
# Attributes of interest: req.status, req.wordcount, req.length and req.response
def handleResponse(req, interesting): 
    # pass on null responses
    if req.length == 4:
        return
    # Only allow repeats of 3 common responses
    if not FILTER.check(req.status, req.wordcount):
        return
    # Only allow non-4XX responses    
    if str(req.status)[0] != "4":
        table.add(req)