diff --git a/config.json b/config.json new file mode 100644 index 0000000..d24eaef --- /dev/null +++ b/config.json @@ -0,0 +1,3 @@ +{ + "api_key": "" +} diff --git a/install.sh b/install.sh new file mode 100755 index 0000000..29ed2b5 --- /dev/null +++ b/install.sh @@ -0,0 +1,2 @@ +mkdir -p $HOME/.config/pwnedornot +cp config.json $HOME/.config/pwnedornot/config.json diff --git a/pwnedornot.py b/pwnedornot.py index 4c3e921..c0b0907 100755 --- a/pwnedornot.py +++ b/pwnedornot.py @@ -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) - exit() + print(f'\n{R}[!] {C}Keyboard Interrupt.{W}') + exit() \ No newline at end of file