v1.2.9 - code optimization

Former-commit-id: ad807fdf629e6048ddfbc95e06eb922762508beb
This commit is contained in:
thewhiteh4t
2021-08-10 07:23:00 +05:30
parent a17f169719
commit 2b7fe35af0
3 changed files with 221 additions and 198 deletions
+3
View File
@@ -0,0 +1,3 @@
{
"api_key": ""
}
Executable
+2
View File
@@ -0,0 +1,2 @@
mkdir -p $HOME/.config/pwnedornot
cp config.json $HOME/.config/pwnedornot/config.json
+215 -197
View File
@@ -1,30 +1,59 @@
#!/usr/bin/env python3
import os
import re
import sys
import json
import time
import argparse
import requests
from argparse import ArgumentParser
ap = ArgumentParser()
ap.add_argument('-e', '--email', required=False,
help='Email Address You Want to Test')
ap.add_argument('-f', '--file', required=False,
help='Load a File with Multiple Email Addresses')
ap.add_argument('-d', '--domain', required=False,
help='Filter Results by Domain Name')
ap.add_argument('-n', '--nodumps', required=False, action='store_true',
help='Only Check Breach Info and Skip Password Dumps')
ap.add_argument('-l', '--list', required=False, action='store_true',
help='Get List of all pwned Domains')
ap.add_argument('-c', '--check', required=False,
help='Check if your Domain is pwned')
arg = ap.parse_args()
addr = arg.email
file = arg.file
domain = arg.domain
nodumps = arg.nodumps
list_domain = arg.list
check_domain = arg.check
R = '\033[31m' # red
G = '\033[32m' # green
C = '\033[36m' # cyan
W = '\033[0m' # white
version = '1.2.8'
version = '1.2.9'
key = ''
useragent = ''
start = ''
def banner():
if sys.platform == 'win32':
os.system('cls')
else:
os.system('clear')
import requests
from os import getenv
from re import search
from time import time, sleep
from json import loads, dumps
home = getenv('HOME')
conf_path = home + '/.config/pwnedornot/config.json'
response_codes = {
200: "OK",
400: "Bad request — the account does not comply with an acceptable format (i.e. it's an empty string)",
401: "Unauthorised — either no API key was provided or it wasn't valid",
403: "Forbidden — no user agent has been specified in the request",
404: "Not Pwned",
429: "Too many requests — the rate limit has been exceeded",
503: "Service unavailable — usually returned by Cloudflare if the underlying service is not available",
}
def banner():
banner = r'''
______ _ __ __
____ _ ______ ___ ____/ / __ \_____/ | / /___ / /_
@@ -32,32 +61,38 @@ def banner():
/ /_/ / |/ |/ / / / / __/ /_/ / /_/ / / / /| / /_/ / /_
/ .___/|__/|__/_/ /_/\___/\__,_/\____/_/ /_/ |_/\____/\__/
/_/
'''
'''
print(G + banner + W)
print(G + '[>]' + C + ' Created by : ' + W + 'thewhiteh4t')
print(G + '[>]' + C + ' Version : ' + W + version + '\n')
print(f'{G}[>]{C} Created by : {W}thewhiteh4t')
print(f'{G}[>]{C} Version : {W}{version}\n')
def api_key():
global key, useragent
try:
with open('key.txt', 'r') as keyfile:
key = keyfile.readline()
key = key.strip()
print(G + '[+]' + C + ' API Key Found...' + W + '\n')
with open(conf_path, 'r') as config:
json_cnf = loads(config.read())
key = json_cnf['api_key']
if len(key) > 0:
print(f'{G}[+] {C}API Key Found...{W}\n')
useragent = {'User-Agent': 'pwnedOrNot', 'hibp-api-key': key}
except FileNotFoundError:
print(R + '[-]' + C + ' API Key Not Found...' + W + '\n')
print(G + '[+]' + C + ' Get your API Key : ' + W + 'https://haveibeenpwned.com/API/Key' + '\n')
enter_key = input(G + '[+]' + C + ' Enter your API Key : ' + W)
enter_key = enter_key.strip()
with open('key.txt', 'w') as keyfile:
keyfile.write(enter_key)
key_path = os.getcwd() + '/key.txt'
print(G + '[+]' + C + ' Saved API Key in : ' + W + key_path + '\n')
else:
print(f'{R}[-] {C}API Key Not Found...{W}\n')
print(f'{G}[+] {C}Get your API Key : {W}https://haveibeenpwned.com/API/Key \n')
enter_key = input(f'{G}[+]' + C + ' Enter your API Key : ' + W)
enter_key = enter_key.strip()
with open(conf_path, 'w') as keyfile:
key_dict = {'api_key': enter_key}
json_data = dumps(key_dict)
keyfile.write(json_data)
print(f'{G}[+] {C}Saved API Key in : {W}{conf_path}\n')
def main():
global addr, start
start = time.time()
start = time()
banner()
api_key()
if list_domain is True:
domains_list()
@@ -68,115 +103,116 @@ def main():
elif addr is not None and domain is None:
check()
elif file is not None and domain is None:
print (G + '[+]' + C + ' Reading Emails Addresses from ' + W + '{}'.format(file) + '\n')
print(f'{G}[+] {C}Reading Emails Addresses from {W}{file}\n')
with open(file) as dict:
for line in dict:
line = line.strip()
addr = line
if addr != '':
check()
time.sleep(1.6)
sleep(1.6)
elif file != None and domain != None:
print(G + '[+]' + C + ' Reading Emails Addresses from ' + W + '{}'.format(file) + '\n')
print(G + '[+]' + C + ' Domain : ' + W + domain)
print(f'{G}[+] {C}Reading Emails Addresses from {W}{file}\n')
print(f'{G}[+] {C}Domain : {W}{domain}')
with open(file) as dict:
for line in dict:
line = line.strip()
addr = line
if addr != '':
filtered_check()
time.sleep(1.6)
sleep(1.6)
else:
print('\n' + R + '[-]' + C + ' Error : Atleast 1 Argument is Required, Try : python3 pwnedornot.py -h' + W)
print(f'{R}[-] {C}Error : {W}Atleast 1 Argument is Required, Try : {G}python3 pwnedornot.py -h{W}')
exit()
def check():
print(G + '[+]' + C + ' Checking Breach status for ' + W + '{}'.format(addr), end = '')
rqst = requests.get('https://haveibeenpwned.com/api/v3/breachedaccount/{}'.format(addr), headers=useragent, params={'truncateResponse': 'false'}, timeout=10)
print(f'{G}[+] {C}Checking Breach status for {W}{addr}', end = '')
rqst = requests.get(
f'https://haveibeenpwned.com/api/v3/breachedaccount/{addr}',
headers=useragent,
params={'truncateResponse': 'false'},
timeout=10
)
sc = rqst.status_code
if sc == 200:
print(G + ' [ pwned ]' + W)
json_out = rqst.content.decode('utf-8', 'ignore')
simple_out = json.loads(json_out)
for item in simple_out:
print( '\n'
+ G + '[+]' + C + ' Breach : ' + W + str(item['Title']) + '\n'
+ G + '[+]' + C + ' Domain : ' + W + str(item['Domain']) + '\n'
+ G + '[+]' + C + ' Date : ' + W + str(item['BreachDate']) + '\n'
+ G + '[+]' + C + ' Fabricated : ' + W + str(item['IsFabricated']) + '\n'
+ G + '[+]' + C + ' Verified : ' + W + str(item['IsVerified']) + '\n'
+ G + '[+]' + C + ' Retired : ' + W + str(item['IsRetired']) + '\n'
+ G + '[+]' + C + ' Spam : ' + W + str(item['IsSpamList']))
if nodumps is not True:
dump()
elif sc == 404:
print(R + ' [ Not Breached ]' + W)
if nodumps is not True:
dump()
elif sc == 503:
print('\n')
print(R + '[-]' + C + ' Error 503 : ' + W + 'Request Blocked by Cloudflare DDoS Protection')
elif sc == 403:
print('\n')
print(R + '[-]' + C + ' Error 403 : ' + W + 'Request Blocked by haveibeenpwned API')
print('\n-------------------------------------------------')
print('Email This Complete Response at troy@troyhunt.com')
print('-------------------------------------------------\n')
print(rqst.text)
else:
print('\n')
print(R + '[-]' + C + ' An Unknown Error Occurred')
print(rqst.text)
for code, desc in response_codes.items():
if sc == code:
if sc == 200:
print(f' {G}[ pwned ]{W}')
json_out = rqst.content.decode('utf-8', 'ignore')
simple_out = loads(json_out)
print(f'\n{G}[+] {C}Total Breaches : {W}{len(simple_out)}')
for item in simple_out:
print(f'\n' \
f'{G}[+] {C}Breach : {W}{str(item["Title"])} \n' \
f'{G}[+] {C}Domain : {W}{str(item["Domain"])} \n' \
f'{G}[+] {C}Date : {W}{str(item["BreachDate"])} \n' \
f'{G}[+] {C}Fabricated : {W}{str(item["IsFabricated"])} \n' \
f'{G}[+] {C}Verified : {W}{str(item["IsVerified"])} \n' \
f'{G}[+] {C}Retired : {W}{str(item["IsRetired"])} \n' \
f'{G}[+] {C}Spam : {W}{str(item["IsSpamList"])}'
)
if nodumps != True:
dump()
elif sc == 404:
print(f' {R}[ not pwned ]{W}')
if nodumps != True:
dump()
else:
print(f'\n\n{R}[-] {C}Status {code} : {W}{desc}')
def filtered_check():
print('\n' + G + '[+]' + C + ' Checking Breach status for ' + W + '{}'.format(addr), end='')
rqst = requests.get('https://haveibeenpwned.com/api/v3/breachedaccount/{}?domain={}'.format(addr, domain), headers=useragent, params={'truncateResponse': 'false'}, verify=True, timeout=10)
print(f'\n{G}[+] {C}Checking Breach status for {W}{addr}', end='')
rqst = requests.get(
f'https://haveibeenpwned.com/api/v3/breachedaccount/{addr}?domain={domain}',
headers=useragent,
params={'truncateResponse': 'false'},
verify=True,
timeout=10
)
sc = rqst.status_code
if sc == 200:
print(G + ' [ pwned ]' + W)
json_out = rqst.content.decode('utf-8', 'ignore')
simple_out = json.loads(json_out)
for code, desc in response_codes.items():
if sc == code:
if sc == 200:
print(f' {G}[ pwned ]{W}')
json_out = rqst.content.decode('utf-8', 'ignore')
simple_out = loads(json_out)
for item in simple_out:
print( '\n'
+ G + '[+]' + C + ' Breach : ' + W + str(item['Title']) + '\n'
+ G + '[+]' + C + ' Domain : ' + W + str(item['Domain']) + '\n'
+ G + '[+]' + C + ' Date : ' + W + str(item['BreachDate']) + '\n'
+ G + '[+]' + C + ' Fabricated : ' + W + str(item['IsFabricated']) + '\n'
+ G + '[+]' + C + ' Verified : ' + W + str(item['IsVerified']) + '\n'
+ G + '[+]' + C + ' Retired : ' + W + str(item['IsRetired']) + '\n'
+ G + '[+]' + C + ' Spam : ' + W + str(item['IsSpamList']))
if nodumps is not True:
dump()
elif sc == 404:
print(R + ' [ Not Breached ]' + W)
if nodumps is not True:
dump()
elif sc == 503:
print('\n')
print(R + '[-]' + C + ' Error 503 : ' + W + 'Request Blocked by Cloudflare DDoS Protection')
elif sc == 403:
print('\n')
print(R + '[-]' + C + ' Error 403 : ' + W + 'Request Blocked by Cloudflare')
else:
print('\n')
print(R + '[-]' + C + ' An Unknown Error Occurred')
print(rqst.text)
for item in simple_out:
print(f'\n' \
f'{G}[+] {C}Breach : {W}{str(item["Title"])} \n' \
f'{G}[+] {C}Domain : {W}{str(item["Domain"])} \n' \
f'{G}[+] {C}Date : {W}{str(item["BreachDate"])} \n' \
f'{G}[+] {C}Fabricated : {W}{str(item["IsFabricated"])} \n' \
f'{G}[+] {C}Verified : {W}{str(item["IsVerified"])} \n' \
f'{G}[+] {C}Retired : {W}{str(item["IsRetired"])} \n' \
f'{G}[+] {C}Spam : {W}{str(item["IsSpamList"])}'
)
if nodumps is not True:
dump()
elif sc == 404:
print(f' {R}[ not pwned ]{W}')
if nodumps is not True:
dump()
else:
print(f'\n{R}[-] {C}Status {code} : {W}{desc}')
def dump():
dumplist = []
print('\n' + G + '[+]' + C + ' Looking for Dumps...' + W, end = '')
rqst = requests.get('https://haveibeenpwned.com/api/v3/pasteaccount/{}'.format(addr), headers=useragent, timeout=10)
print(f'\n{G}[+] {C}Looking for Dumps...{W}', end = '')
rqst = requests.get(
f'https://haveibeenpwned.com/api/v3/pasteaccount/{addr}',
headers=useragent,
timeout=10
)
sc = rqst.status_code
if sc != 200:
print(R + ' [ No Dumps Found ]' + W)
print(f' {R}[ No Dumps Found ]{W}')
else:
print(G + ' [ Dumps Found ]' + W + '\n')
print(f' {G}[ Dumps Found ]{W}\n')
json_out = rqst.content.decode('utf-8', 'ignore')
simple_out = json.loads(json_out)
simple_out = loads(json_out)
for item in simple_out:
if (item['Source']) == 'Pastebin':
@@ -187,9 +223,9 @@ def dump():
sc = page.status_code
if sc == 200:
dumplist.append(url)
print(G + '[+]' + C + ' Dumps Found : ' + W + str(len(dumplist)), end='\r')
print(f'{G}[+] {C}Dumps Found : {W}{len(dumplist)}', end='\r')
if len(dumplist) == 0:
print(R + '[-]' + C + ' Dumps are not Accessible...' + W)
print(f'{R}[-] {C}Dumps are not Accessible...{W}')
except requests.exceptions.ConnectionError:
pass
elif (item['Source']) == 'AdHocUrl':
@@ -199,118 +235,100 @@ def dump():
sc = page.status_code
if sc == 200:
dumplist.append(url)
print(G + '[+]' + C + ' Dumps Found : ' + W + str(len(dumplist)), end='\r')
print(f'{G}[+] {C}Dumps Found : {W}{len(dumplist)}', end='\r')
if len(dumplist) == 0:
print(R + '[-]' + C + ' Dumps are not Accessible...' + W)
except requests.exceptions.ConnectionError:
print(f'{R}[-] {C}Dumps are not Accessible...{W}')
except Exception:
pass
if len(dumplist) != 0:
print('\n\n' + G + '[+]' + C + ' Passwords:' + W + '\n')
print(f'\n\n{G}[+] {C}Passwords : {W}\n')
for entry in dumplist:
time.sleep(1.1)
sleep(1.1)
try:
page = requests.get(entry, timeout=10)
dict = page.content.decode('utf-8', 'ignore')
passwd = re.search('{}:(\w+)'.format(addr), dict)
passwd = search('{}:(\w+)'.format(addr), dict)
if passwd:
print(G + '[+] ' + W + passwd.group(1))
print(f'{G}[+] {W}{passwd.group(1)}')
elif not passwd:
for line in dict.splitlines():
passwd = re.search('(.*{}.*)'.format(addr), line)
passwd = search('(.*{}.*)'.format(addr), line)
if passwd:
print(G + '[+] ' + W + passwd.group(0))
print(f'{G}[+] {W}{passwd.group(0)}')
except requests.exceptions.ConnectionError:
pass
def domains_list():
domains = []
print(G + '[+]' + C + ' Fetching List of Breached Domains...' + W + '\n')
rqst = requests.get('https://haveibeenpwned.com/api/v3/breaches', headers=useragent, timeout=10)
print(f'{G}[+] {C}Fetching List of Breached Domains...{W}\n')
rqst = requests.get(
'https://haveibeenpwned.com/api/v3/breaches',
headers=useragent,
timeout=10
)
sc = rqst.status_code
if sc == 200:
json_out = rqst.content.decode('utf-8', 'ignore')
simple_out = json.loads(json_out)
for item in simple_out:
domain_name = item['Domain']
if len(domain_name) != 0:
print (G + '[+] ' + W + str(domain_name))
domains.append(domain_name)
print('\n' + G + '[+]' + C + ' Total : ' + W + str(len(domains)))
elif sc == 503:
print(R + '[-]' + C + ' Error 503 : ' + W + 'Request Blocked by Cloudflare DDoS Protection')
elif sc == 403:
print(R + '[-]' + C + ' Error 403 : ' + W + 'Request Blocked by Cloudflare')
else:
print(R + '[-]' + C + ' An Unknown Error Occurred')
print(rqst.text)
for code, desc in response_codes.items():
if sc == code:
if sc == 200:
json_out = rqst.content.decode('utf-8', 'ignore')
simple_out = loads(json_out)
for item in simple_out:
domain_name = item['Domain']
if len(domain_name) != 0:
print(G + '[+] ' + W + str(domain_name))
domains.append(domain_name)
print(f'\n{G}[+] {C}Total : {W}{len(domains)}')
else:
print(f'\n{R}[-] {C}Status {code} : {W}{desc}')
def domain_check():
print(G + '[+]' + C + ' Domain Name : ' + W + check_domain, end = '')
rqst = requests.get('https://haveibeenpwned.com/api/v3/breaches?domain={}'.format(check_domain), headers=useragent, timeout=10)
print(f'{G}[+] {C}Domain Name : {W}{check_domain}', end = '')
rqst = requests.get(
f'https://haveibeenpwned.com/api/v3/breaches?domain={check_domain}',
headers=useragent,
timeout=10
)
sc = rqst.status_code
if sc == 200:
json_out = rqst.content.decode('utf-8', 'ignore')
simple_out = json.loads(json_out)
if len(simple_out) != 0:
print(G + ' [ pwned ]' + W)
for item in simple_out:
print( '\n'
+ G + '[+]' + C + ' Breach : ' + W + str(item['Title']) + '\n'
+ G + '[+]' + C + ' Domain : ' + W + str(item['Domain']) + '\n'
+ G + '[+]' + C + ' Date : ' + W + str(item['BreachDate']) + '\n'
+ G + '[+]' + C + ' Pwn Count : ' + W + str(item['PwnCount']) + '\n'
+ G + '[+]' + C + ' Fabricated : ' + W + str(item['IsFabricated']) + '\n'
+ G + '[+]' + C + ' Verified : ' + W + str(item['IsVerified']) + '\n'
+ G + '[+]' + C + ' Retired : ' + W + str(item['IsRetired']) + '\n'
+ G + '[+]' + C + ' Spam : ' + W + str(item['IsSpamList']) + '\n'
+ G + '[+]' + C + ' Data Types : ' + W + str(item['DataClasses']))
else:
print(R + ' [ Not Breached ]' + W)
elif sc == 503:
print('\n')
print(R + '[-]' + C + ' Error 503 : ' + W + 'Request Blocked by Cloudflare DDoS Protection')
elif sc == 403:
print('\n')
print(R + '[-]' + C + ' Error 403 : ' + W + 'Request Blocked by Cloudflare')
else:
print('\n')
print(R + '[-]' + C + ' An Unknown Error Occurred')
print(rqst.text)
for code, desc in response_codes.items():
if sc == code:
if sc == 200:
json_out = rqst.content.decode('utf-8', 'ignore')
simple_out = loads(json_out)
print(simple_out)
if len(simple_out) != 0:
print(f' {G}[ pwned ]{W}')
for item in simple_out:
print(f'\n' \
f'{G}[+] {C}Breach : {W}{str(item["Title"])}\n' \
f'{G}[+] {C}Domain : {W}{str(item["Domain"])}\n' \
f'{G}[+] {C}Date : {W}{str(item["BreachDate"])}\n' \
f'{G}[+] {C}Pwn Count : {W}{str(item["PwnCount"])}\n' \
f'{G}[+] {C}Fabricated : {W}{str(item["IsFabricated"])}\n' \
f'{G}[+] {C}Verified : {W}{str(item["IsVerified"])}\n' \
f'{G}[+] {C}Retired : {W}{str(item["IsRetired"])}\n' \
f'{G}[+] {C}Spam : {W}{str(item["IsSpamList"])}\n' \
f'{G}[+] {C}Data Types : {W}{str(item["DataClasses"])}'
)
else:
print(f' {R}[ Not Breached ]{W}')
elif sc == 404:
print(f' {R}[ Not Breached ]{W}')
else:
print(f'\n{R}[-] {C}Status {code} : {W}{desc}')
def quit():
global start
print('\n' + G + '[+]' + C + ' Completed in ' + W + str(time.time()-start) + C + ' seconds.' + W)
print(f'\n{G}[+] {C}Completed in {W}{str(time()-start)} {C}seconds.{W}')
exit()
try:
banner()
ap = argparse.ArgumentParser()
ap.add_argument('-e', '--email', required=False,
help='Email Address You Want to Test')
ap.add_argument('-f', '--file', required=False,
help='Load a File with Multiple Email Addresses')
ap.add_argument('-d', '--domain', required=False,
help='Filter Results by Domain Name')
ap.add_argument('-n', '--nodumps', required=False, action='store_true',
help='Only Check Breach Info and Skip Password Dumps')
ap.add_argument('-l', '--list', required=False, action='store_true',
help='Get List of all pwned Domains')
ap.add_argument('-c', '--check', required=False,
help='Check if your Domain is pwned')
arg = ap.parse_args()
addr = arg.email
file = arg.file
domain = arg.domain
nodumps = arg.nodumps
list_domain = arg.list
check_domain = arg.check
api_key()
main()
quit()
if __name__ == "__main__":
main()
quit()
else:
pass
except KeyboardInterrupt:
print('\n' + R + '[!]' + C + ' Keyboard Interrupt.' + W)
print(f'\n{R}[!] {C}Keyboard Interrupt.{W}')
exit()