如何为Python暴力破解(Brute Force)函数实现Multithreading？

Hey there! Since you're new to multithreading and looking to apply it to this brute-force password cracker, let's walk through how to adapt the code step by step. The key idea is to split the work across multiple threads so we can check more password guesses in parallel, plus add safeguards to stop all threads once the password is found.

First, Let's Understand the Limitation of the Original Code #

The original function runs single-threaded: it checks every possible password combination one after another. For longer passwords, this can be extremely slow. Multithreading lets us split this workload so multiple threads can check different sets of combinations at the same time.

Step-by-Step Multithreaded Implementation #

We'll use Python's concurrent.futures.ThreadPoolExecutor (a simpler alternative to manually managing threading.Thread objects) plus thread-safe mechanisms to track progress and stop work once the password is found.

Full Multithreaded Code

import string
import itertools
import threading
from concurrent.futures import ThreadPoolExecutor

# Shared variables (protected by a lock for thread safety)
charset = string.ascii_letters + string.digits
found_password = None
total_attempts = 0
lock = threading.Lock()

def brute_worker(real_password, password_length):
    """Thread worker: checks all combinations of a specific password length"""
    global found_password, total_attempts
    local_attempts = 0

    for guess_tuple in itertools.product(charset, repeat=password_length):
        # First, check if another thread already found the password (avoid wasted work)
        with lock:
            if found_password is not None:
                return
        
        local_attempts += 1
        guess = ''.join(guess_tuple)
        
        if guess == real_password:
            # Update shared variables safely when we find the password
            with lock:
                global total_attempts, found_password
                total_attempts += local_attempts
                found_password = guess
            return
        
        # Batch-update total attempts to reduce lock contention (performance tweak)
        if local_attempts % 1000 == 0:
            with lock:
                total_attempts += 1000
                local_attempts -= 1000
    
    # Add remaining local attempts after loop ends
    with lock:
        total_attempts += local_attempts

def multithreaded_brute(real_password):
    """Main function to launch multithreaded brute-force search"""
    global found_password, total_attempts
    # Reset shared state for each new search
    found_password = None
    total_attempts = 0

    # Use a thread pool (adjust max_workers based on your CPU cores; 4-8 is typical)
    with ThreadPoolExecutor(max_workers=4) as executor:
        # Submit a task for each possible password length (1 to 8)
        futures = [executor.submit(brute_worker, real_password, length) for length in range(1, 9)]
        
        # Wait for tasks to complete, or cancel them early if password is found
        for future in futures:
            with lock:
                if found_password is not None:
                    # Cancel all remaining uncompleted tasks
                    for f in futures:
                        f.cancel()
                    break
    
    if found_password:
        return f"password is {found_password}. found in {total_attempts} guesses."
    else:
        return "password not found."

# Example usage
if __name__ == "__main__":
    apass = "Test123"  # Replace with your random password variable
    print(multithreaded_brute(apass))

Key Concepts Explained #

• Thread Safety: We use a threading.Lock() to protect access to shared variables (found_password and total_attempts). This prevents multiple threads from modifying these variables at the same time, which would cause bugs or incorrect counts.
• Early Termination: Each thread checks if the password has already been found before generating a new guess. Once found, the main thread cancels all remaining tasks to stop unnecessary work.
• Workload Splitting: We split the task by password length—each thread handles all combinations of a specific length. This is a simple, effective way to divide the work without complex logic.
• ThreadPoolExecutor: This manages the thread pool for us, handling thread creation, reuse, and cleanup. You can adjust max_workers to match your CPU core count (too many threads will cause overhead from context switching).

A Quick Note on GIL Limitations #

Python's Global Interpreter Lock (GIL) means CPU-bound tasks like this won't get true parallelism across CPU cores with multithreading. For better performance, you could use multiprocessing instead (which bypasses the GIL), but since you asked for multithreading, this implementation is a solid starting point.

内容的提问来源于stack exchange，提问作者Plus10Voltz