Cracking Zip Password using Python 3 and Multithreading

Hello, world! In my last post on Cracking Zip Password using Python3, you have seen how to crack the password of the zip file using python and a wordlist based on the password policy. Well, the problem that we have experienced is that it takes some time to find the password but eventually it found the password in a few seconds.

If you check the password, it was in the 15% of the file. In that case, it was easy to crack the password in a few seconds, but what if the password is located in the last line of the same dictionary? Well in that case, of course, we will have to look at the entire file and this will take a lot more time in the single-thread execution mode. In this post, I will be using the artefacts from the Cracking ZIP archives II lab provided by AttackDefense for the threaded Bruteforce attack demonstration.

For the twist, you can download both the zip file and the wordlist in your local system and shuffle the contents of the dictionary file using the following commands. Since you can't write to the input file simultaneously while reading, it is recommended to redirect the output to a different file and then later rename the new file with the old file name.

shuf dictionary.txt > dictionary.txt.shuf
mv dictionary.txt.shuf dictionary.txt

Hold your passwords in a queue in the memory

There would be multiple threads reading files at the same time. It might cause unintended behaviour when multiple threads will be accessed from it. So we will first read all the entries from the file and add them into the queue, one by one.

The file is only 8.1 MB in size, so it is reasonable to hold the data in the memory with the thread-safe implementation of the Queue data structure.

Initialize the Queue

You can find the Queue class from the queue module and initialize an object like any other class. Since in this we don't know the number of passwords in thedictionary.txt file is containing (pretending this for now), we can omit the first argument in the constructor. This means that the queue should grow automatically based on the elements pushed into it.

from queue import Queue

passwords = Queue()

Reading the dictionary file and adding it to the queue

Now read the file line by line, strip the contents to make sure no whitespace is there in the string and call .put(item) method on the queue object providing the contents of the current line in the wordlist to add that in the queue at the bottom/end.

with open("dictionary.txt") as wordlist:
	for password in wordlist:
		passwords.put(password.strip())
		pass
	pass

Writing the worker function

Each thread will invoke a function and can pass arguments to that function. This function will execute in separate threads and will not block further execution in the main thread.

def worker(y):
	while not passwords.empty():
		password = passwords.get()
		try:
			z.extractall(path=TMPDIR, pwd=password.encode())
			print("Thread {}: Password Found: {}. Contents are extracted to {}".format(y, password, TMPDIR))
		except:
			print("Thread {}: Tried password: {}".format(y, password))
	print("Thread {}: Failed to crack zip file".format(y))
	pass

Since we are not tracking the length of the queue and element processed or removed from the queue, there should be another way to check whether it contains any other entry or not? Well this is done by a thread-safe function .empty() defined in the Queue class which will return True if there are no further items in the queue to process.

If the queue is not empty then get the first element from it and store in the password variable using thread-safe .get() method. In the queue data structure, the element that is added first will be removed first (aka First in First Out, or FIFO).

Stop the threads if the password is found

Currently, if you see in the worker function, there is a provision to send an event when zip extraction using .extractall() on ZipFile object is successful, this will result in continuing the execution even after that.

Since all the threads depend on whether or not the queue is empty, we can remove all the elements from the queue using the.queue.clear() method. This will then make .empty() function return True value and threads will terminate gracefully.

You need to add the following snippet in the worker function in the try block after the print statement and then return from the function.

with passwords.mutex:
	passwords.queue.clear()
return

Spawning threads and finding the password

Now it's time to create the thread object and spawn the worker function. Internal working of the thread is out of this scope and will be covered in later courses of the system programming. In this section, you will learn how to call the thread and join it with the main thread to wait until workers are completed.

Let's start off by creating 5 threads for now using Thread class from the threading module shipped with python.

threads = []
for _ in range(1, 6):
	thread = Thread(target=worker, args=(_, ), daemon=True)
	thread.start()
	threads.append(thread)

for thread in threads:
	thread.join()
		
msg = "All done! Press any key to exit."
print(msg)	
input()

The target function is the reference to the function to be executed, in this case, it is the worker function and a list of arguments are passed to the args keyword in the constructor. Additionally, we need to pass daemon=True in the constructor which will start this thread in the daemon mode, of course. Now what is daemon mode, it is again out of the scope for this post, you can read more about it here – https://www.geeksforgeeks.org/python-daemon-threads/.

Once the program execution finishes, it is considered a good practice to close all the resource handles. So let's close the ZipFile object.

z.close()

Blessing the python for intuitive output

Once you will run the above code, it will show all the output and it will feel so overwhelming. I thought that what if we can print the output of each thread in the respective line only and if the password is found, then show the thread name which did it best. It also looks like a gamified version.

So after searching a lot, I found that there is a third party module built on top of the ncurses library in the C language known as blessings. You can install it using python3 -m pip install -U --user blessings .

The code starts with instantiating an object of the Terminal class which will set up a virtual terminal on top of the current terminal.

t = Terminal()

Now wrap the creation of the thread under the .fullscreen() method which will clear the screen set up the terminal context and restore the same on leaving.

with t.fullscreen():
	threads = []
	for i in range(1, 6):
		thread = Thread(target=worker, args=(i, ), daemon=True)
		thread.start()
		threads.append(thread)
	
	for thread in threads:
		thread.join()
		
	msg = "All done! Press any key to exit."
	print(msg)	
	input()	

Now if you want to move the cursor inside the fullscreen context, simply call the .move(y, x) method in the print function before printing the actual text. To replace the print function as instructed below.

...
def worker():
...
-        print("Thread {}: Password Found: {}. Contents are extracted to {}".format(y, password, TMPDIR))
+        print(t.move(y, 0) + "Thread {}: Password Found: {}. Contents are extracted to {}".format(y, password, TMPDIR) + t.clear_eol)
...
-        print("Thread {}: Tried password: {}".format(y, password))
+        print(t.move(y, 0) + "Thread {}: Tried password: {}".format(y, password) + t.clear_eol)
...
-    print("Thread {}: Failed to crack zip file".format(y))
+    print(t.move(y, 0) + "Thread {}: Failed to crack zip file".format(y) + t.clear_eol)
...
- print(msg)
+ print(t.move(len(threads) + 1, (t.width - len(msg)) // 2) + msg)

You can find the full snippet code on the GitHub gist here

Brute force the zip file password using multithreading in python 3

Brute force the zip file password using multithreading in python 3 - brute.py

Gist262588213843476

Program in action 🚀

After completing the above code, the program will work as shown below.