From da1943cbc2a0a1ba5ab9a7bf733608ed19b06196 Mon Sep 17 00:00:00 2001 From: ajmeese7 Date: Sat, 11 Feb 2023 15:27:42 -0500 Subject: [PATCH] style: misc code improvements --- lib/core/common.py | 24 ++++++------- lib/core/settings.py | 11 +++--- var/auto_issue/github.py | 16 ++++----- var/search/pgp_search.py | 32 +++++++++--------- var/search/selenium_search.py | 64 +++++++++++++++++------------------ 5 files changed, 74 insertions(+), 73 deletions(-) diff --git a/lib/core/common.py b/lib/core/common.py index da50408..cb47588 100644 --- a/lib/core/common.py +++ b/lib/core/common.py @@ -86,7 +86,7 @@ class URLParser(object): def extract_webcache_url(self, splitter="+"): """ - extract the URL from Google's webcache URL + Extract the URL from Google's webcache URL """ url = self.url data = self.webcache_regex.split(url) @@ -98,7 +98,7 @@ class URLParser(object): def extract_ip_ban_url(self): """ - extract the true URL from Google's IP ban URL + Extract the true URL from Google's IP ban URL """ url = unquote(self.url) to_use_separator = None @@ -124,7 +124,7 @@ class URLParser(object): def write_to_log_file(data_to_write, path, filename, blacklist=False): """ - write all found data to a log file + Write all found data to a log file """ lib.core.settings.create_dir(path.format(os.getcwd())) full_file_path = "{}/{}".format( @@ -173,14 +173,14 @@ def write_to_log_file(data_to_write, path, filename, blacklist=False): else: log.write(data_to_write + "\n") lib.core.settings.logger.info(lib.core.settings.set_color( - "successfully wrote found items to '{}'".format(full_file_path) + "Successfully wrote found items to '{}'".format(full_file_path) )) return full_file_path def start_up(): """ - start the program and display the time it was started + Start the program and display the time it was started """ print( "\n\n[*] starting up at {}..\n\n".format(time.strftime("%H:%M:%S")) @@ -189,17 +189,17 @@ def start_up(): def shutdown(): """ - shut down the program and the time it stopped + Shut down the program and the time it stopped """ print( - "\n\n[*] shutting down at {}..\n\n".format(time.strftime("%H:%M:%S")) + "\n\n[*] shutting down at {}.\n\n".format(time.strftime("%H:%M:%S")) ) exit(0) def prompt(question, opts=None, default=None, paused=False): """ - ask a question + Ask a question """ if opts is not None and default is None: options = '/'.join(opts) @@ -247,10 +247,10 @@ def prompt(question, opts=None, default=None, paused=False): def pause(): """ - interactive pause function, as of now you are only able to skip and exit + Interactive pause function, as of now you are only able to skip and exit from this function """ - message = "program has been paused, how do you want to proceed?" + message = "Program has been paused, how do you want to proceed?" return prompt( message, paused=True ) @@ -258,7 +258,7 @@ def pause(): def run_fix(message, command, fail_message, exit_process=False): """ - run the fix script for the program + Run the fix script for the program """ do_fix = prompt( message, opts="yN" @@ -268,7 +268,7 @@ def run_fix(message, command, fail_message, exit_process=False): subprocess.call(cmd) if exit_process: lib.core.settings.logger.info(lib.core.settings.set_color( - "command completed successfully, should be safe to re-run Zeus" + "Command completed successfully, should be safe to re-run Zeus" )) else: lib.core.settings.logger.fatal(lib.core.settings.set_color( diff --git a/lib/core/settings.py b/lib/core/settings.py index a86e23d..2270049 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -61,11 +61,11 @@ DISCLAIMER = ( # sexy banner BANNER = """\033[36m - __ __________ __ - / / \____ /____ __ __ ______ \ \ - / / ______ / // __ \| | \/ ___/ ______ \ \ - \ \ /_____/ / /\ ___/| | /\___ \ /_____/ / / - \_\ /_______ \___ >____//____ > /_/ + __ __________ __ + / / \____ /____ __ __ ______ \ \\ + / / ______ / // __ \| | \/ ___/ ______ \ \\ + \ \ /_____/ / /\ ___/| | /\___ \ /_____/ / / + \_\ /_______ \___ >____//____ > /_/ \/ \/ \/ {} \t{}\n\t\t{}\033[0m\n\n\n{}""".format(VERSION_STRING, CLONE, SAYING, DISCLAIMER) @@ -620,6 +620,7 @@ def get_browser_version(output=True): return output except Exception as e: logger.error(set_color( + # TODO: Fix this - 'cannot use a string pattern on a bytes-like object' "Received an exception from Firefox: '{}'".format(str(e), level=40) )) return str(e) diff --git a/var/auto_issue/github.py b/var/auto_issue/github.py index 5277622..d35e14f 100644 --- a/var/auto_issue/github.py +++ b/var/auto_issue/github.py @@ -13,7 +13,7 @@ import lib.core.settings def find_url(params, search="https://github.com/ekultek/zeus-scanner/issues"): """ - get the URL that your issue is created at + Get the URL that your issue is created at """ retval = "https://github.com{}" href = None @@ -36,13 +36,13 @@ def find_url(params, search="https://github.com/ekultek/zeus-scanner/issues"): def request_issue_creation(): if not lib.core.settings.get_md5sum(): lib.core.settings.logger.fatal(lib.core.settings.set_color( - "it appears that your checksums did not match, therefore it is assumed " + "It appears that your checksums did not match, therefore it is assumed " "that you have edited some of the code, issue request denied", level=50 )) lib.core.common.shutdown() question = lib.core.common.prompt( - "would you like to create an anonymous issue and post it to Zeus's Github", opts="yN" + "Would you like to create an anonymous issue and post it to Zeus's Github? >", opts="yN" ) if question.lower().startswith("n"): lib.core.settings.logger.error(lib.core.settings.set_color( @@ -57,7 +57,7 @@ def request_issue_creation(): def __extract_stacktrace(file_data): lib.core.settings.logger.info(lib.core.settings.set_color( - "extracting traceback from log file" + "Extracting traceback from log file" )) retval, buff_mode, _buffer = [], False, "" with open(file_data, "r+") as log: @@ -75,7 +75,7 @@ def request_issue_creation(): return "".join(retval) lib.core.settings.logger.info(lib.core.settings.set_color( - "getting authorization" + "Getting authorization..." )) token = lib.core.settings.get_token(lib.core.settings.GITHUB_AUTH_PATH) @@ -117,13 +117,13 @@ def request_issue_creation(): ) urllib2.urlopen(req, timeout=10).read() lib.core.settings.logger.info(lib.core.settings.set_color( - "issue has been created successfully with the following name '{}', your unique identifier " + "Issue has been created successfully with the name '{}', your unique identifier " "for this issue is '{}' and the URL to your issue is '{}'".format( issue_title, identifier, find_url(identifier) ) )) except Exception as e: lib.core.settings.logger.exception(lib.core.settings.set_color( - "failed to auto create the issue, got exception '{}', " - "you may manually create an issue".format(e), level=50 + "Failed to auto create the issue, got exception: '{}'. " + "You may manually create an issue".format(e), level=50 )) diff --git a/var/search/pgp_search.py b/var/search/pgp_search.py index 9cb1b1e..664ea78 100644 --- a/var/search/pgp_search.py +++ b/var/search/pgp_search.py @@ -10,7 +10,7 @@ import lib.core.settings def __create_url(ext): """ - create the URL with the identifier, usually a hash + Create the URL with the identifier (usually a hash) """ url = lib.core.settings.AUTHORIZED_SEARCH_ENGINES["pgp"] items = url.split("/") @@ -25,7 +25,7 @@ def __create_url(ext): def __set_headers(**kwargs): """ - set the HTTP headers + Set the HTTP headers """ agent = kwargs.get("agent", None) xforward = kwargs.get("xforward", False) @@ -52,7 +52,7 @@ def __set_headers(**kwargs): def obtain_html(url, query, **kwargs): """ - obtain the HTML containing the URL redirects to the public PGP keys + Obtain the HTML containing the URL redirects to the public PGP keys """ agent = kwargs.get("agent", None) xforward = kwargs.get("xforwad", False) @@ -78,7 +78,7 @@ def obtain_html(url, query, **kwargs): def gather_urls(html, attribute="a", descriptor="href"): """ - get the URLs within the HTML + Get the URLs within the HTML """ redirects, retval = set(), set() soup = BeautifulSoup(html, "html.parser") @@ -95,7 +95,7 @@ def gather_urls(html, attribute="a", descriptor="href"): def get_pgp_keys(url_list, query, attribute="pre", **kwargs): """ - get the PGP keys by connecting to the URLs and pulling the information from the HTML + Get the PGP keys by connecting to the URLs and pulling the information from the HTML """ agent = kwargs.get("agent", None) proxy = kwargs.get("proxy", None) @@ -109,18 +109,18 @@ def get_pgp_keys(url_list, query, attribute="pre", **kwargs): identity_matcher = re.compile(r"\bbegin.pgp.public.key.block", re.I) amount_left = len(url_list) lib.core.settings.logger.info(lib.core.settings.set_color( - "checking a maximum of {} PGP keys".format(amount_to_search) + "Checking a maximum of {} PGP keys".format(amount_to_search) )) for i, url in enumerate(url_list, start=1): if i >= amount_to_search: break if verbose: lib.core.settings.logger.debug(lib.core.settings.set_color( - "checking '{}'".format(url), level=10 + "Checking '{}'".format(url), level=10 )) if i % 25 == 0: lib.core.settings.logger.info(lib.core.settings.set_color( - "currently checking PGP key #{}, {} left to check ({} total found)".format( + "Currently checking PGP key #{}, {} left to check ({} total found)".format( i, amount_to_search - i, amount_left ) )) @@ -147,7 +147,7 @@ def get_pgp_keys(url_list, query, attribute="pre", **kwargs): pgp_key = pgp_key[1].split("".format(attribute))[0] # split it again by the end tag if verbose: lib.core.settings.logger.debug(lib.core.settings.set_color( - "found PGP:", level=10 + "Found PGP:", level=10 )) # output the found PGP key if you run in verbose print("{}\n{}\n{}".format(data_sep, pgp_key, data_sep)) @@ -164,7 +164,7 @@ def pgp_main(query, verbose=False): except Exception: query = query lib.core.settings.logger.info(lib.core.settings.set_color( - "searching public PGP files with given query '{}'".format(query) + "Searching public PGP files with given query '{}'".format(query) )) try: html = obtain_html( @@ -172,26 +172,26 @@ def pgp_main(query, verbose=False): ) except (Exception, ReadTimeout): lib.core.settings.logger.warning(lib.core.settings.set_color( - "connection failed, assuming no PGP keys", level=30 + "Connection failed, assuming no PGP keys", level=30 )) html = None if html is not None: urls = gather_urls(html) lib.core.settings.logger.info(lib.core.settings.set_color( - "found a total of {} URLs".format(len(urls)) + "Found a total of {} URLs".format(len(urls)) )) if verbose: lib.core.settings.logger.debug(lib.core.settings.set_color( - "found a '{}'".format(urls), level=10 + "Found a '{}'".format(urls), level=10 )) lib.core.settings.logger.info(lib.core.settings.set_color( - "gathering PGP key(s) and writing to a file", level=25 + "Gathering PGP key(s) and writing to a file", level=25 )) return get_pgp_keys(urls, query, verbose=verbose) else: lib.core.settings.logger.warning(lib.core.settings.set_color( - "did not find anything using query '{}'".format(query), level=30 + "Did not find anything using query '{}'".format(query), level=30 )) except KeyboardInterrupt: if not lib.core.common.pause(): - lib.core.common.shutdown() \ No newline at end of file + lib.core.common.shutdown() diff --git a/var/search/selenium_search.py b/var/search/selenium_search.py index 6490573..836f441 100644 --- a/var/search/selenium_search.py +++ b/var/search/selenium_search.py @@ -252,7 +252,7 @@ def parse_search_results(query, url_to_search, verbose=False, **kwargs): HTTP_HEADER.USER_AGENT: user_agent } logger.info(set_color( - "attempting to gather query URL" + "Attempting to gather query URL" )) try: query_url = get_urls( @@ -262,60 +262,60 @@ def parse_search_results(query, url_to_search, verbose=False, **kwargs): except Exception as e: if "'/usr/lib/firefoxdriver/webdriver.xpi'" in str(e): logger.fatal(set_color( - "firefox was not found in the default location on your system, " - "check your installation and make sure it is in /usr/lib, if you " + "Firefox was not found in the default location on your system, " + "check your installation and make sure it is in /usr/lib. If you " "find it there, restart your system and try again", level=50 )) elif "connection refused" in str(e).lower(): logger.fatal(set_color( - "there are to many sessions of firefox opened and selenium cannot " + "There are to many sessions of Firefox open, Selenium cannot " "create a new one", level=50 )) run_fix( - "would you like to attempt to auto clean the open sessions", + "Would you like to attempt to auto clean the open sessions? >", "sudo sh {}".format(CLEANUP_TOOL_PATH), - "kill off the open sessions of firefox and re-run Zeus", + "Kill off the open sessions of firefox and re-run Zeus", exit_process=True ) elif "Program install error!" in str(e): logger.error(set_color( - "seems the program is having some trouble installing would you like " + "Seems the program is having some trouble installing; would you like " "to try and automatically fix this issue", level=40 )) run_fix( - "would you like to attempt to fix this issue automatically", + "Would you like to attempt to fix this issue automatically? >", "sudo sh {}".format(FIX_PROGRAM_INSTALL_PATH), "you can manually try and re-install Xvfb to fix the problem", exit_process=True ) elif "Message: Reached error page:" in str(e): logger.fatal(set_color( - "geckodriver has hit an error that usually means it needs to be reinstalled", level=50 + "geckodriver has hit an error; that usually means it needs to be reinstalled", level=50 )) question = prompt( - "would you like to attempt a reinstallation of the geckodriver", opts="yN" + "Would you like to attempt a reinstallation of the geckodriver? >", opts="yN" ) if question.lower().startswith("y"): logger.warning(set_color( - "rewriting all executed information, path information, and removing geckodriver", level=30 + "Rewriting all executed information, path information, and removing geckodriver", level=30 )) rewrite_all_paths() logger.info(set_color( - "all paths rewritten, you will be forced to re-install everything next run of Zeus" + "All paths rewritten, you will be forced to re-install everything next run of Zeus" )) else: logger.fatal(set_color( - "you will need to remove the geckodriver from /usr/bin and reinstall it", level=50 + "You will need to remove the geckodriver from /usr/bin and reinstall it", level=50 )) shutdown() elif "Unable to find a matching set of capabilities" in str(e): logger.fatal(set_color( - "it appears that firefox, selenium, and geckodriver are not playing nice with one another", level=50 + "It appears that Firefox, Selenium, and geckodriver are not playing nice with one another", level=50 )) run_fix( - "would you like to attempt to resolve this issue automatically", + "Would you like to attempt to resolve this issue automatically? >", "sudo sh {}".format(REINSTALL_TOOL), - ("you will need to reinstall firefox to a later version, update selenium, and reinstall the " + ("You will need to reinstall firefox to a later version, update selenium, and reinstall the " "geckodriver to continue using Zeus"), exit_process=True ) @@ -337,7 +337,7 @@ def parse_search_results(query, url_to_search, verbose=False, **kwargs): req = requests.get(query_url, proxies=proxy_string, params=headers) except ConnectionError: logger.warning(set_color( - "target machine refused connection, delaying and trying again", level=30 + "Target machine refused connection, delaying and trying again", level=30 )) time.sleep(3) req = requests.get(query_url, proxies=proxy_string, params=headers) @@ -380,12 +380,12 @@ def parse_search_results(query, url_to_search, verbose=False, **kwargs): if parse_webcache: if "webcache" in url: logger.info(set_color( - "found a webcache URL, extracting" + "Found a webcache URL, extracting..." )) url = URLParser(url).extract_webcache_url() if verbose: logger.debug(set_color( - "found '{}'".format(url), level=15 + "Found '{}'".format(url), level=15 )) true_retval.add(url) else: @@ -399,23 +399,23 @@ def parse_search_results(query, url_to_search, verbose=False, **kwargs): amount_of_urls = len(open(file_path).readlines()) success_rate = calculate_success(amount_of_urls) logger.info(set_color( - "provided query has a {} success rate".format(success_rate) + "Provided query has a {} success rate".format(success_rate) )) else: logger.warning(set_color( - "did not find any URLs with given query '{}' writing query to blacklist".format(query), level=50 + "Did not find any URLs with given query '{}' writing query to blacklist".format(query), level=50 )) write_to_log_file(query, BLACKLIST_FILE_PATH, BLACKLIST_FILENAME, blacklist=True) logger.info(set_color( - "found a total of {} URLs with given query '{}'".format(len(true_retval), query) + "Found a total of {} URLs with given query '{}'".format(len(true_retval), query) )) def search_multiple_pages(query, link_amount, verbose=False, **kwargs): """ - search multiple pages for a lot of links, this will not be done via Google + Search multiple pages for a lot of links, this will not be done via Google """ proxy = kwargs.get("proxy", None) agent = kwargs.get("agent", None) @@ -427,7 +427,7 @@ def search_multiple_pages(query, link_amount, verbose=False, **kwargs): search_engine = AUTHORIZED_SEARCH_ENGINES["search-results"] logger.warning(set_color( - "searching multiple pages will not be done on Google".format(search_engine), level=30 + "Searching multiple pages will not be done on Google".format(search_engine), level=30 )) if not parse_blacklist(query, BLACKLIST_FILE_PATH, batch=batch): @@ -451,11 +451,11 @@ def search_multiple_pages(query, link_amount, verbose=False, **kwargs): while len(retval) <= link_amount: if verbose: logger.debug(set_color( - "searching page number {}".format(page_number), level=10 + "Searching page number {}".format(page_number), level=10 )) if page_number % 10 == 0: logger.info(set_color( - "currently on page {} of search results".format( + "Currently on page {} of search results".format( page_number ) )) @@ -482,16 +482,16 @@ def search_multiple_pages(query, link_amount, verbose=False, **kwargs): break else: logger.warning(set_color( - "no more results found for given query '{}'".format(query), level=30 + "No more results found for given query '{}'".format(query), level=30 )) break except KeyboardInterrupt: logger.error(set_color( - "user aborted, dumping already found URL(s)", level=40 + "User aborted, dumping already found URL(s)", level=40 )) write_to_log_file(retval, URL_LOG_PATH, URL_FILENAME) logger.info(set_color( - "found a total of {} URL(s)".format(len(retval)), level=25 + "Found a total of {} URL(s)".format(len(retval)), level=25 )) shutdown() except Exception as e: @@ -503,18 +503,18 @@ def search_multiple_pages(query, link_amount, verbose=False, **kwargs): if len(retval) > 0: logger.info(set_color( - "a total of {} URL(s) found out of the requested {}".format(len(retval), link_amount), level=25 + "A total of {} URL(s) found out of the requested {}".format(len(retval), link_amount), level=25 )) file_path = write_to_log_file(retval, URL_LOG_PATH, URL_FILENAME) if show_success: amount_of_urls = len(open(file_path).readlines()) success_rate = calculate_success(amount_of_urls) logger.info(set_color( - "provided query has a {} success rate".format(success_rate) + "Provided query has a {} success rate".format(success_rate) )) return list(retval) else: logger.warning(set_color( - "did not find any links with given query '{}' writing to blacklist".format(query), level=30 + "Did not find any links with given query '{}' writing to blacklist".format(query), level=30 )) write_to_log_file(query, BLACKLIST_FILE_PATH, BLACKLIST_FILENAME)