ThreadPool Brute

thread_brute.py

#!/usr/bin/env python3

import subprocess
import traceback
from concurrent.futures import ThreadPoolExecutor

from tqdm import tqdm


def passwords(wordlist: str) -> list:
    with open(wordlist, 'r') as wl:
        passwds = [x.strip() for x in wl.readlines()]
        
    return passwds


def try_connect(password: str):
    try:
        p = subprocess.Popen(["command", "-pass", password], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        output, errors = p.communicate()
        p.wait()

        if "ERROR" not in output.decode() and "ERROR" not in errors.decode():
            print(output.decode().replace("\\n", "\n"))
            print(errors.decode().replace("\\n", "\n"))
            print(password)

    except Exception as e:
        print(traceback.format_exc())
        print(password)
        print(e)


def brute(wordlist: list):
    with ThreadPoolExecutor(max_workers=10) as executor:
        results = list(tqdm(executor.map(try_connect, wordlist), total=len(wordlist), desc='Brute'))


def main():
    passwds = passwords('')
    brute(passwds)


if __name__ == '__main__':
    main()