SHA256 password cracker - v2 with concurrency

Question

I have updated my previous code with the suggestions. I have also implemented concurrency to speed up the results.

I'd be interested for comments on a more efficient use on concurrency / parallel processing / hyper-threading

import concurrent.futures
import hashlib
import time


def make_next_guess(guess):
    carry = 0
    next_guess = guess
    for i, char in enumerate(next_guess):
        if ord(char) >= ord("z"):
            carry = 1
            next_guess[i] = "!"
        elif ord(char) < ord("z"):
            next_guess[i] = chr(ord(char) + 1)
            carry = 0
        if carry == 0: break
    if carry:
        next_guess.append("!")
    return next_guess


def hash_password(pwrd):
    # Generates hash of original password
    try:
        pwrd = pwrd.encode("UTF-8")
        password = hashlib.sha256()
        password.update(pwrd)
        return password.hexdigest()
    except:
        pass


def find_password(secure_password, guess, integer):
    list_cracked = []
    # Brute force to find original password
    for _ in range(90 ** integer):  # password maximum length 14 and there are 58 characters that can be used
        return_password = "".join(guess)
        if hash_password(return_password) in secure_password:
            # print(f"Another password has been cracked: '{return_password}'")
            list_cracked.append(return_password)
        guess = make_next_guess(guess)
    return f"Finished cracking all passwords of length {integer}", list_cracked, integer


def rainbow_table_check(secure_password):
    global hash
    list_cracked = []
    for password in open("Rainbow.txt", "r", encoding="utf8"):
        try:
            password = password.strip()
            hash = hash_password(password)
        except:
            pass
        if hash in secure_password:
            # print(f"Another password has been cracked: {password}")
            list_cracked.append(password)
    return "Rainbow attack complete - passwords: ", list_cracked, "Rainbow"


def dic_attack(secure_password):
    list_cracked = []
    for password in open("Dic.txt", "r", encoding="utf8"):
        password = password.strip()
        lst = [password.lower(), password.upper(), password.title()]
        for password in lst:
            hash = hash_password(password)
            if hash in secure_password:
                # print(f"Another password has been cracked: {password}")
                list_cracked.append(password)
    return "Dictionary attack complete - passwords: ", list_cracked, "Dictionary"


if __name__ == "__main__":
    all_passwords = []
    start = time.time()
    secure_password = set()
    print("Attempting to crack passwords....")
    password_list = ["ABd", "Abc", "lpo", "J*&", "Njl", "!!!!", "Aqz", "Apple", "Cake", "Biden", "password1"]
    for password in password_list:
        secure_password.add(hash_password(password))
    with concurrent.futures.ProcessPoolExecutor() as executor:
        results = [executor.submit(dic_attack, secure_password),
                   executor.submit(rainbow_table_check, secure_password),
                   executor.submit(find_password, secure_password, ['!'], 1),
                   executor.submit(find_password, secure_password, ['!', '!'], 2),
                   executor.submit(find_password, secure_password, ['!', '!', '!'], 3),
                   executor.submit(find_password, secure_password, ['!', '!', '!', '!'], 4),
                   executor.submit(find_password, secure_password, ['!', '!', '!', '!', '!'], 5)]
        for f in concurrent.futures.as_completed(results):
            message, cracked, method = f.result()
            time_run = f"{round((time.time() - start) // 60)} min {round((time.time() - start) % 60)} sec"
            print(f"{message} : {cracked} - {time_run}")
            all_passwords += cracked
    print(f"Complete list of cracked passwords: {set(tuple(all_passwords))}")
    print(f"This operation took: {round((time.time() - start) // 60)} min {round((time.time() - start) % 60)} sec")

Answer 1

Two observations:

The methods in hashlib expect to receive bytes. In make_next_guess(), use a bytearray instead of a string. That avoids the calls to ''.join(), ord(), ''.encode('UTF-8'), etc. Better yet, make the function a generator that yields the guesses.

Using the else clause on the for i, byte in enumerate(guess): loop simplifies the logic a bit. When the loop is finishes, the else clause is executed. However, a break skips the else clause. Here, if the loop doesn't find any bytes to increase, the else clause add another byte to the length of the guess.

Something like:

def generate_guesses(start):
    ORD_Z = ord('z')
    ORD_BANG = ord("!")
    
    guess = start[:]
    
    while True:
        yield guess
        for i, byte in enumerate(guess):
            if byte < ORD_Z:
                guess[i] += 1
                break
                
            else:
                guess[i] = ORD_BANG
                
        else:
            guess.append(ORD_BANG)

Called like:

for guess in generate_guesses(bytearray(b'xzzzz')):
    ... do something with the guess ...

Possibly, add a stop or count argument that tells when the generator should stop or how many guesses it should generate. Just change the while True: line to check the stop condition.

Second observation is that the last job submitted to the Pool, is 90 times more work than the job before it. Indeed, it is more work than the previous 4 jobs combined (maybe all the other jobs). As a result, you end up with the other jobs finishing sooner and one job running on one processor core for a long time. Try splitting the jobs up into more equal sized chunks to keep all the processor cores busy. For example, the jobs could work on equal sized chunks of the password search space:

'!'    to 'zzz' 
'!!!!' to 'zzz!'
'!!!"' to 'zzz"'
'!!!#' to 'zzz#' these are all chunks of 729k (90*90*90) guesses
     ...
'!!!z' to 'zzzz'