From ddf46c6bbe29df83e45ce4d1796f02ac0f7e163f Mon Sep 17 00:00:00 2001 From: ekultek Date: Sat, 25 Nov 2017 19:59:09 -0600 Subject: [PATCH] minor edits to some of the attacks, also created an interactive pause that you will be able to skip or exit from --- lib/attacks/admin_panel_finder/__init__.py | 87 ++++++------ lib/attacks/clickjacking_scan/__init__.py | 4 +- lib/attacks/gist_lookup/__init__.py | 84 +++++------ lib/attacks/nmap_scan/__init__.py | 30 ++-- lib/attacks/sqlmap_scan/__init__.py | 3 + lib/attacks/whois_lookup/whois.py | 57 ++++---- lib/attacks/xss_scan/__init__.py | 156 +++++++++++---------- 7 files changed, 223 insertions(+), 198 deletions(-) diff --git a/lib/attacks/admin_panel_finder/__init__.py b/lib/attacks/admin_panel_finder/__init__.py index 5cdd389..55b25bc 100644 --- a/lib/attacks/admin_panel_finder/__init__.py +++ b/lib/attacks/admin_panel_finder/__init__.py @@ -167,10 +167,10 @@ def check_for_admin_page(url, exts, protocol="http://", **kwargs): "did not find any possible connections to {}'s " "admin page".format(url), level=50 )) - lib.core.settings.logger.warning(lib.core.settings.set_color( - "only writing successful connections to log file...", level=30 - )) if len(connections) > 0: + lib.core.settings.logger.warning(lib.core.settings.set_color( + "only writing successful connections to log file...", level=30 + )) lib.core.common.write_to_log_file( list(connections), lib.core.settings.ADMIN_PAGE_FILE_PATH, lib.core.settings.ADMIN_PAGE_FILE_PATH.format( lib.core.settings.replace_http(url) @@ -194,44 +194,49 @@ def main(url, show=False, verbose=False, **kwargs): do_threading = kwargs.get("do_threading", False) proc_num = kwargs.get("proc_num", 5) batch = kwargs.get("batch", False) - lib.core.settings.logger.info(lib.core.settings.set_color( - "parsing robots.txt..." - )) - results = check_for_externals(url, robots=True, batch=batch) - if not results: - lib.core.settings.logger.warning(lib.core.settings.set_color( - "seems like this page is either blocking access to robots.txt or it does not exist...", level=30 + + try: + lib.core.settings.logger.info(lib.core.settings.set_color( + "parsing robots.txt..." )) - lib.core.settings.logger.info(lib.core.settings.set_color( - "checking for a sitemap..." - )) - check_for_externals(url, sitemap=True) - lib.core.settings.logger.info(lib.core.settings.set_color( - "loading extensions..." - )) - extensions = __load_extensions() - if verbose: - lib.core.settings.logger.debug(lib.core.settings.set_color( - "loaded a total of {} extensions...".format(len(extensions)), level=10 + results = check_for_externals(url, robots=True, batch=batch) + if not results: + lib.core.settings.logger.warning(lib.core.settings.set_color( + "seems like this page is either blocking access to robots.txt or it does not exist...", level=30 + )) + lib.core.settings.logger.info(lib.core.settings.set_color( + "checking for a sitemap..." )) - lib.core.settings.logger.info(lib.core.settings.set_color( - "attempting to bruteforce admin panel..." - )) - if do_threading: - lib.core.settings.logger.warning(lib.core.settings.set_color( - "starting {} threads, you will not be able to end the process until " - "it is completed...".format(proc_num), level=30 + check_for_externals(url, sitemap=True) + lib.core.settings.logger.info(lib.core.settings.set_color( + "loading extensions..." )) - tasks = [] - for _ in range(0, proc_num): - t = threading.Thread(target=check_for_admin_page, args=(url, extensions), kwargs={ - "verbose": verbose, - "show_possibles": show - }) - t.daemon = True - tasks.append(t) - for thread in tasks: - thread.start() - thread.join() - else: - check_for_admin_page(url, extensions, show_possibles=show, verbose=verbose) \ No newline at end of file + extensions = __load_extensions() + if verbose: + lib.core.settings.logger.debug(lib.core.settings.set_color( + "loaded a total of {} extensions...".format(len(extensions)), level=10 + )) + lib.core.settings.logger.info(lib.core.settings.set_color( + "attempting to bruteforce admin panel..." + )) + if do_threading: + lib.core.settings.logger.warning(lib.core.settings.set_color( + "starting {} threads, you will not be able to end the process until " + "it is completed...".format(proc_num), level=30 + )) + tasks = [] + for _ in range(0, proc_num): + t = threading.Thread(target=check_for_admin_page, args=(url, extensions), kwargs={ + "verbose": verbose, + "show_possibles": show + }) + t.daemon = True + tasks.append(t) + for thread in tasks: + thread.start() + thread.join() + else: + check_for_admin_page(url, extensions, show_possibles=show, verbose=verbose) + except KeyboardInterrupt: + if not lib.core.common.pause(): + lib.core.common.shutdown() \ No newline at end of file diff --git a/lib/attacks/clickjacking_scan/__init__.py b/lib/attacks/clickjacking_scan/__init__.py index 3db75f3..e2de4cb 100644 --- a/lib/attacks/clickjacking_scan/__init__.py +++ b/lib/attacks/clickjacking_scan/__init__.py @@ -105,7 +105,9 @@ def clickjacking_main(url, **kwargs): url ), level=40 )) - + except KeyboardInterrupt: + if not lib.core.common.pause(): + lib.core.common.shutdown() except Exception as e: # until I figure out the errors, we'll just make issues about them lib.core.settings.logger.exception(lib.core.settings.set_color( "Zeus failed to process the clickjacking test and received " diff --git a/lib/attacks/gist_lookup/__init__.py b/lib/attacks/gist_lookup/__init__.py index 34256d4..98b7fb0 100644 --- a/lib/attacks/gist_lookup/__init__.py +++ b/lib/attacks/gist_lookup/__init__.py @@ -54,11 +54,13 @@ def get_raw_data(page_set, proxy=None, agent=None, verbose=False): gist_file = item["files"] gist_filename = gist_file.keys() try: - if verbose: - lib.core.settings.logger.debug(lib.core.settings.set_color( - "found filename '{}'...".format(''.join(gist_filename)), level=10 - )) - retval.add(gist_file[''.join(gist_filename)]["raw_url"]) + skip_schema = ("-", " ", "") + if not any(s == gist_filename for s in skip_schema): + if verbose: + lib.core.settings.logger.debug(lib.core.settings.set_color( + "found filename '{}'...".format(''.join(gist_filename)), level=10 + )) + retval.add(gist_file[''.join(gist_filename)]["raw_url"]) # sometimes the URL doesn't like being pulled, so we'll just skip those ones except Exception: pass @@ -110,44 +112,48 @@ def github_gist_search_main(query, **kwargs): agent = kwargs.get("agent", None) verbose = kwargs.get("verbose", False) thread = kwargs.get("do_threading", False) - proc_num = kwargs.get("proc_num", 5) + proc_num = kwargs.get("proc_num", 5) # TODO:/ page_set = kwargs.get("page_set", (1, 2, 3, 4, 5)) total_found = 0 - if verbose: - lib.core.settings.logger.debug(lib.core.settings.set_color( - "checking if you have exceeded your search limit...", level=10 - )) - __check_remaining_rate_limit() - lib.core.settings.logger.info(lib.core.settings.set_color( - "searching Github Gists for '{}'...".format(query) - )) - gathered_links = get_raw_data(page_set, proxy=proxy, agent=agent, verbose=verbose) - lib.core.settings.logger.info(lib.core.settings.set_color( - "pulled a total of {} URL's to search...".format(len(gathered_links)), level=25 - )) - if not thread: + try: + if verbose: + lib.core.settings.logger.debug(lib.core.settings.set_color( + "checking if you have exceeded your search limit...", level=10 + )) + __check_remaining_rate_limit() lib.core.settings.logger.info(lib.core.settings.set_color( - "performing Github Gist search, this will probably take awhile..." - )) - for url in gathered_links: - total = check_files_for_information(url, query) - total_found += total - else: - lib.core.settings.logger.warning(lib.core.settings.set_color( - "multi-threading is not implemented yet...", level=35 + "searching Github Gists for '{}'...".format(query) )) + gathered_links = get_raw_data(page_set, proxy=proxy, agent=agent, verbose=verbose) lib.core.settings.logger.info(lib.core.settings.set_color( - "performing Github Gist search, this will probably take awhile..." - )) - for url in gathered_links: - total = check_files_for_information(url, query) - total_found += total - if total_found > 0: - lib.core.settings.logger.info(lib.core.settings.set_color( - "found a total of {} interesting Gists...".format(total_found) - )) - else: - lib.core.settings.logger.warning(lib.core.settings.set_color( - "did not find any interesting Gists...", level=30 + "pulled a total of {} URL's to search...".format(len(gathered_links)), level=25 )) + if not thread: + lib.core.settings.logger.info(lib.core.settings.set_color( + "performing Github Gist search, this will probably take awhile..." + )) + for url in gathered_links: + total = check_files_for_information(url, query) + total_found += total + else: + lib.core.settings.logger.warning(lib.core.settings.set_color( + "multi-threading is not implemented yet...", level=35 + )) + lib.core.settings.logger.info(lib.core.settings.set_color( + "performing Github Gist search, this will probably take awhile..." + )) + for url in gathered_links: + total = check_files_for_information(url, query) + total_found += total + if total_found > 0: + lib.core.settings.logger.info(lib.core.settings.set_color( + "found a total of {} interesting Gists...".format(total_found) + )) + else: + lib.core.settings.logger.warning(lib.core.settings.set_color( + "did not find any interesting Gists...", level=30 + )) + except KeyboardInterrupt: + if not lib.core.common.pause(): + lib.core.common.shutdown() \ No newline at end of file diff --git a/lib/attacks/nmap_scan/__init__.py b/lib/attacks/nmap_scan/__init__.py index 55b1e15..2678bee 100644 --- a/lib/attacks/nmap_scan/__init__.py +++ b/lib/attacks/nmap_scan/__init__.py @@ -1,5 +1,4 @@ import json -import os import socket import shlex import subprocess @@ -19,20 +18,15 @@ class NmapHook(object): NM = nmap.PortScanner() - def __init__(self, ip, verbose=False, pretty=True, - dirname="{}/log/scanner-log".format(os.getcwd()), filename="nmap_scan-results-{}.json", - opts=None): + def __init__(self, ip, **kwargs): self.ip = ip - self.verbose = verbose - self.pretty = pretty - self.dir = dirname - self.file = filename - if opts is None: - self.opts = "" - else: - self.opts = " ".join(opts) + self.verbose = kwargs.get("verbose", False) + self.pretty = kwargs.get("pretty", True) + self.dir = lib.core.settings.PORT_SCAN_LOG_PATH + self.file = lib.core.settings.NMAP_FILENAME + self.opts = kwargs.get("opts", "") - def _get_all_info(self): + def get_all_info(self): """ get all the information from the scan """ @@ -89,10 +83,13 @@ def find_nmap(item_name="nmap"): return lib.core.settings.find_application(item_name) -def perform_port_scan(url, scanner=NmapHook, verbose=False, opts=None, **kwargs): +def perform_port_scan(url, scanner=NmapHook, **kwargs): """ main function that will initalize the port scanning """ + verbose = kwargs.get("verbose", False) + opts = kwargs.get("opts", None) + url = url.strip() lib.core.settings.logger.info(lib.core.settings.set_color( "attempting to find IP address for hostname '{}'...".format(url) @@ -116,7 +113,7 @@ def perform_port_scan(url, scanner=NmapHook, verbose=False, opts=None, **kwargs) )) try: data = scanner(found_ip_address, opts=opts) - json_data = data._get_all_info() + json_data = data.get_all_info() data.show_open_ports(json_data) file_path = data.send_to_file(json_data) lib.core.settings.logger.info(lib.core.settings.set_color( @@ -128,6 +125,9 @@ def perform_port_scan(url, scanner=NmapHook, verbose=False, opts=None, **kwargs) url, found_ip_address ), level=50 )) + except KeyboardInterrupt: + if not lib.core.common.pause(): + lib.core.common.shutdown() except Exception as e: lib.core.settings.logger.exception(lib.core.settings.set_color( "ran into exception '{}', cannot continue quitting...".format(e), level=50 diff --git a/lib/attacks/sqlmap_scan/__init__.py b/lib/attacks/sqlmap_scan/__init__.py index c85194c..15bc02f 100644 --- a/lib/attacks/sqlmap_scan/__init__.py +++ b/lib/attacks/sqlmap_scan/__init__.py @@ -227,6 +227,9 @@ def sqlmap_scan_main(url, port=None, verbose=None, opts=None, auto_start=False): "the server port and try again...".format(e), level=50 )) pass + except KeyboardInterrupt: + if not lib.core.common.pause(): + lib.core.common.shutdown() except Exception as e: if "HTTPConnectionPool(host='127.0.0.1'" in str(e): lib.core.settings.logger.error(lib.core.settings.set_color( diff --git a/lib/attacks/whois_lookup/whois.py b/lib/attacks/whois_lookup/whois.py index 05bf6ae..774e5c9 100644 --- a/lib/attacks/whois_lookup/whois.py +++ b/lib/attacks/whois_lookup/whois.py @@ -71,35 +71,40 @@ def whois_lookup_main(domain, **kwargs): verbose = kwargs.get("verbose", False) timeout = kwargs.get("timeout", None) domain = lib.core.settings.replace_http(domain) - lib.core.settings.logger.info(lib.core.settings.set_color( - "performing WhoIs lookup on given domain '{}'...".format(domain) - )) - if timeout is not None: - time.sleep(timeout) + try: - raw_information = gather_raw_whois_info(domain) - except Exception as e: - lib.core.settings.logger.exception(str(e)) - lib.core.settings.logger.error(lib.core.settings.set_color( - "unable to produce information from WhoIs lookup...", level=40 + lib.core.settings.logger.info(lib.core.settings.set_color( + "performing WhoIs lookup on given domain '{}'...".format(domain) )) - return None - lib.core.settings.logger.info(lib.core.settings.set_color( - "discovered raw information...", level=25 - )) - lib.core.settings.logger.info(lib.core.settings.set_color( - "gathering interesting information..." - )) - interesting_data = get_interesting(raw_information) - if verbose: + if timeout is not None: + time.sleep(timeout) try: - human_readable_display(domain, interesting_data) - except (ValueError, Exception): + raw_information = gather_raw_whois_info(domain) + except Exception as e: + lib.core.settings.logger.exception(str(e)) lib.core.settings.logger.error(lib.core.settings.set_color( - "unable to display any information from WhoIs lookup on domain '{}'...".format(domain), level=50 + "unable to produce information from WhoIs lookup...", level=40 )) return None - lib.core.common.write_to_log_file( - raw_information, lib.core.settings.WHOIS_RESULTS_LOG_PATH, - lib.core.settings.WHOIS_LOOKUP_FILENAME.format(domain) - ) \ No newline at end of file + lib.core.settings.logger.info(lib.core.settings.set_color( + "discovered raw information...", level=25 + )) + lib.core.settings.logger.info(lib.core.settings.set_color( + "gathering interesting information..." + )) + interesting_data = get_interesting(raw_information) + if verbose: + try: + human_readable_display(domain, interesting_data) + except (ValueError, Exception): + lib.core.settings.logger.error(lib.core.settings.set_color( + "unable to display any information from WhoIs lookup on domain '{}'...".format(domain), level=50 + )) + return None + lib.core.common.write_to_log_file( + raw_information, lib.core.settings.WHOIS_RESULTS_LOG_PATH, + lib.core.settings.WHOIS_LOOKUP_FILENAME.format(domain) + ) + except KeyboardInterrupt: + if not lib.core.common.pause(): + lib.core.common.shutdown() \ No newline at end of file diff --git a/lib/attacks/xss_scan/__init__.py b/lib/attacks/xss_scan/__init__.py index 366405a..241664a 100644 --- a/lib/attacks/xss_scan/__init__.py +++ b/lib/attacks/xss_scan/__init__.py @@ -136,89 +136,93 @@ def main_xss(start_url, proxy=None, agent=None, **kwargs): verbose = kwargs.get("verbose", False) batch = kwargs.get("batch", False) - if tamper: + try: + if tamper: + lib.core.settings.logger.info(lib.core.settings.set_color( + "tampering payloads with '{}'...".format(tamper) + )) + find_xss_script(start_url) lib.core.settings.logger.info(lib.core.settings.set_color( - "tampering payloads with '{}'...".format(tamper) + "loading payloads..." )) - find_xss_script(start_url) - lib.core.settings.logger.info(lib.core.settings.set_color( - "loading payloads..." - )) - payloads = __load_payloads() - if verbose: - lib.core.settings.logger.debug(lib.core.settings.set_color( - "a total of {} payloads loaded...".format(len(payloads)), level=10 - )) - lib.core.settings.logger.info(lib.core.settings.set_color( - "payloads will be written to a temporary file and read from there..." - )) - filename = create_urls(start_url, payloads, tamper=tamper, verbose=verbose) - lib.core.settings.logger.info(lib.core.settings.set_color( - "loaded URL's have been saved to '{}'...".format(filename), level=25 - )) - lib.core.settings.logger.info(lib.core.settings.set_color( - "testing for XSS vulnerabilities on host '{}'...".format(start_url) - )) - if proxy is not None: + payloads = __load_payloads() + if verbose: + lib.core.settings.logger.debug(lib.core.settings.set_color( + "a total of {} payloads loaded...".format(len(payloads)), level=10 + )) lib.core.settings.logger.info(lib.core.settings.set_color( - "using proxy '{}'...".format(proxy) + "payloads will be written to a temporary file and read from there..." )) - success = set() - with open(filename) as urls: - for i, url in enumerate(urls.readlines(), start=1): - url = url.strip() - result = scan_xss(url, proxy=proxy, agent=agent) - payload = find_xss_script(url) - if verbose: - lib.core.settings.logger.info(lib.core.settings.set_color( - "trying payload '{}'...".format(payload) - )) - if result[0] != "sqli" and result[0] is True: - success.add(url) + filename = create_urls(start_url, payloads, tamper=tamper, verbose=verbose) + lib.core.settings.logger.info(lib.core.settings.set_color( + "loaded URL's have been saved to '{}'...".format(filename), level=25 + )) + lib.core.settings.logger.info(lib.core.settings.set_color( + "testing for XSS vulnerabilities on host '{}'...".format(start_url) + )) + if proxy is not None: + lib.core.settings.logger.info(lib.core.settings.set_color( + "using proxy '{}'...".format(proxy) + )) + success = set() + with open(filename) as urls: + for i, url in enumerate(urls.readlines(), start=1): + url = url.strip() + result = scan_xss(url, proxy=proxy, agent=agent) + payload = find_xss_script(url) if verbose: - lib.core.settings.logger.debug(lib.core.settings.set_color( - "payload '{}' appears to be usable...".format(payload), level=10 - )) - elif result[0] is "sqli": - if i <= 1: - lib.core.settings.logger.error(lib.core.settings.set_color( - "loaded URL '{}' threw a DBMS error and appears to be injectable, test for SQL injection, " - "backend DBMS appears to be '{}'...".format( - url, result[1] - ), level=40 + lib.core.settings.logger.info(lib.core.settings.set_color( + "trying payload '{}'...".format(payload) )) + if result[0] != "sqli" and result[0] is True: + success.add(url) + if verbose: + lib.core.settings.logger.debug(lib.core.settings.set_color( + "payload '{}' appears to be usable...".format(payload), level=15 + )) + elif result[0] is "sqli": + if i <= 1: + lib.core.settings.logger.error(lib.core.settings.set_color( + "loaded URL '{}' threw a DBMS error and appears to be injectable, test for SQL injection, " + "backend DBMS appears to be '{}'...".format( + url, result[1] + ), level=40 + )) + else: + if verbose: + lib.core.settings.logger.error(lib.core.settings.set_color( + "SQL error discovered...", level=40 + )) else: if verbose: - lib.core.settings.logger.error(lib.core.settings.set_color( - "SQL error discovered...", level=40 + lib.core.settings.logger.debug(lib.core.settings.set_color( + "host '{}' does not appear to be vulnerable to XSS attacks with payload '{}'...".format( + start_url, payload + ), level=10 )) - else: - if verbose: - lib.core.settings.logger.debug(lib.core.settings.set_color( - "host '{}' does not appear to be vulnerable to XSS attacks with payload '{}'...".format( - start_url, payload - ), level=10 - )) - if len(success) != 0: - lib.core.settings.logger.info(lib.core.settings.set_color( - "possible XSS scripts to be used:", level=25 - )) - lib.core.settings.create_tree(start_url, list(success)) - else: - lib.core.settings.logger.error(lib.core.settings.set_color( - "host '{}' does not appear to be vulnerable to XSS attacks...".format(start_url) - )) - question_msg = "would you like to keep the URL's saved for further testing" - if not batch: - save = lib.core.common.prompt( - question_msg, opts="yN" - ) - else: - save = lib.core.common.prompt( - question_msg, opts="yN", default="n" - ) + if len(success) != 0: + lib.core.settings.logger.info(lib.core.settings.set_color( + "possible XSS scripts to be used:", level=25 + )) + lib.core.settings.create_tree(start_url, list(success)) + else: + lib.core.settings.logger.error(lib.core.settings.set_color( + "host '{}' does not appear to be vulnerable to XSS attacks...".format(start_url) + )) + question_msg = "would you like to keep the URL's saved for further testing" + if not batch: + save = lib.core.common.prompt( + question_msg, opts="yN" + ) + else: + save = lib.core.common.prompt( + question_msg, opts="yN", default="n" + ) - if save.lower().startswith("n"): - os.remove(filename) - else: - os.remove(filename) \ No newline at end of file + if save.lower().startswith("n"): + os.remove(filename) + else: + os.remove(filename) + except KeyboardInterrupt: + if not lib.core.common.pause(): + lib.core.common.shutdown() \ No newline at end of file