lot's of patch, great optimization, better comments

Author: ugomeguerditchian

api/global_parser.py

@@ -1,47 +1,48 @@
+import importlib
+import pkgutil
 import lib.custom_logger as custom_logger
 from lib import generics as gen
 from lib.configuration import configuration
-from api import *
-import os
+import api
 
 logger = custom_logger.logger
 
 
-def main(domain: str, config: configuration):
-    # get all the lib in the api folder and call the main function
+def main(domain: str, config: configuration) -> list:
+    """
+    This function is the entry point of the program. It takes a domain name and a configuration object as parameters.
+    :param domain: The domain name to search for subdomains.
+    :type domain: str
+    :param config: The configuration object.
+    :type config: configuration
+    :return: A list of subdomains.
+    :rtype: list
+    """
     results = []
-    for os_file in os.listdir(os.path.dirname(__file__)):
-        if os_file.endswith(".py") and os_file not in [
-            "__init__.py",
-            "global_parser.py",
-        ]:
-            try:
-                try:
-                    data = config.config["API"]["mapper"][os_file.split(".")[0]]
-                    if not data["activate"]:
-                        logger.info(f"[*] {os_file} is deactivated")
-                        continue
-                    key = data["api_key"]
-                    if key == "":
-                        key == None
-                except:
-                    logger.info(
-                        "[*] This API seems to not appear in the config file or no key is provided"
-                    )
-                    logger.info("[*] Continuing without API key")
-                    key = None
-                logger.info(f"[*] Getting subdomains of {domain} from {os_file}")
-                results += eval(
-                    os_file.split(".")[0] + ".main(domain, config.handler, key)"
-                )
-                if len(results) == 0:
-                    warn = ""
-                    logger.info(f"[*] No subdomains of {domain} found from {os_file}")
-                else:
-                    # remove duplicates
-                    results = list(dict.fromkeys(results))
-                    logger.info(f"[*] {len(results)} subdomains of {domain} found")
-            except Exception as e:
-                logger.error(f"Impossible to get subdomains of {domain} from {os_file}")
+    # Get a list of all the modules in the api package
+    for _, module_name, _ in pkgutil.walk_packages(api.__path__):
+        if module_name in ["global_parser", "__init__"]:
+            continue
+        # Dynamically import the module
+        module = importlib.import_module(f"api.{module_name}")
+        try:
+            # Call the main() function of the module
+            data = config.config["API"]["mapper"][module_name]
+            if not data["activate"]:
+                logger.info(f"[*] {module_name} is deactivated")
                 continue
+            key = data["api_key"] or None
+            logger.info(f"[*] Getting subdomains of {domain} from {module_name}")
+            results += module.main(domain, config.handler, key)
+            if len(results) == 0:
+                logger.info(f"[*] No subdomains of {domain} found from {module_name}")
+            else:
+                # Remove duplicates and invalid subdomains
+                results = list(set(results))
+                logger.info(f"[*] {len(results)} subdomains of {domain} found")
+        except Exception as e:
+            logger.error(
+                f"Impossible to get subdomains of {domain} from {module_name}: {e}"
+            )
+            continue
     return results

api/hackertarget.py

@@ -5,27 +5,33 @@
 logger = custom_logger.logger
 
 
-def main(domain, handler: handler, key: str):
-    # get all the subdomain of the domain from hackertarget
-    # the url is https://api.hackertarget.com/hostsearch/?q={domain}
-    # key is the api key
+def main(domain: str, handler: handler, key: str = "") -> list:
+    """
+    Get all the subdomains of the domain from hackertarget.
+    The url is https://api.hackertarget.com/hostsearch/?q={domain}.
+    :param domain: The domain to search for subdomains.
+    :param handler: The handler to use for the request.
+    :param key: The API key to use for the request.
+    :return: A list of subdomains.
+    """
+    # Construct the URL for the API request.
     if key:
-        url = "https://api.hackertarget.com/hostsearch/?q=" + domain + "&apikey=" + key
+        url = f"https://api.hackertarget.com/hostsearch/?q={domain}&apikey={key}"
     else:
-        url = "https://api.hackertarget.com/hostsearch/?q=" + domain
+        url = f"https://api.hackertarget.com/hostsearch/?q={domain}"
     try:
+        # Make the API request and decode the response.
         response = handler.get(url, until_ok=True)._body.decode("utf-8")
         if response == "API count exceeded - Increase Quota with Membership":
             raise Exception("API")
-        # split the response in linesr
+        # Split the response into lines.
         lines = response.split("\n")
-        # get all the subdomains
-        subdomains = []
-        for line in lines:
-            if line != "" and "*" not in line.split(",")[0]:
-                subdomains.append(line.split(",")[0])
-        # delete all the occurences in the list
-
+        # Get all the subdomains.
+        subdomains = [
+            line.split(",")[0]
+            for line in lines
+            if line != "" and "*" not in line.split(",")[0]
+        ]
         return subdomains
     except Exception as e:
         if e.args[0] == "API":

lib/configuration.py

@@ -15,9 +15,14 @@
 
 
 class configuration:
-    """Configuration is ./congiuration.yaml"""
+    """Configuration is ./configuration.yaml"""
 
     def __init__(self):
+        """
+        Initializes the configuration object by loading the configuration.yaml file and checking its version.
+        :param self: The configuration object.
+        :return: None.
+        """
         try:
             self.config = yaml.load(
                 open("configuration.yaml", "r"), Loader=yaml.FullLoader
@@ -64,6 +69,11 @@ def __init__(self):
             exit()
 
     def load(self):
+        """
+        Loads the proxy file if it exists, otherwise it activates the proxy and loads the proxies from the links specified in the configuration.yaml file.
+        :param self: The configuration object.
+        :return: None.
+        """
         if self.config["Proxy"]["file"]:
             if os.path.isfile(self.config["Proxy"]["file"]):
                 self.proxy_file = self.config["proxy"]["file"]
@@ -106,7 +116,16 @@ def load(self):
                 # self.check_proxy()
 
     def is_in_scope(self, to_test: str, mode: str):
-        """Check if a fqdn or ip is in scope"""
+        """
+        Check if a fqdn or ip is in scope.
+
+        :param to_test: The fqdn or ip to test.
+        :type to_test: str
+        :param mode: The mode to use for testing.
+        :type mode: str
+        :return: True if the fqdn or ip is in scope, False otherwise.
+        :rtype: bool
+        """
         scope = []
         if self.config["SCOPE"][mode]["file"]:
             if os.path.isfile(self.config["SCOPE"][mode]["file"]):
@@ -121,21 +140,24 @@ def is_in_scope(self, to_test: str, mode: str):
         for i in scope:
             if i == to_test:
                 return True
-            else:
-                return False
         if self.config["SCOPE"][mode]["regex"]:
             # example : regex: ["r'tesla'"]
             for regex in self.config["SCOPE"][mode]["regex"]:
                 # convert it to r
                 regex = r"{}".format(regex)
                 if re.search(regex, to_test):
                     return True
-                else:
-                    return False
         if len(scope) == 0:
             return True
+        return False
 
     def is_there_scope(self):
+        """
+        Check if there is a scope.
+
+        :return: True if there is a scope, False otherwise.
+        :rtype: bool
+        """
         mode = ["IPs", "FQDNs"]
         there_is = False
         for i in mode:
@@ -148,11 +170,11 @@ def is_there_scope(self):
         return there_is
 
     def get_github_proxy(self, links: dict):
-        """Get a proxy from github.com
-        Format must be ip:port for each line
-        links{
-        link:type (http, https, socks)
-        }
+        """
+        Get a proxy from github.com.
+
+        :param links: The links to get the proxy from.
+        :type links: dict
         """
         for link, type in links.items():
             try:
@@ -167,13 +189,13 @@ def get_github_proxy(self, links: dict):
                             self.socks_proxy.append(line)
                 else:
                     logger.warning(
-                        "[!] Warning : {} returned status code {}".format(
+                        "[!] Warning: {} returned status code {}".format(
                             link, r.status_code
                         )
                     )
             except:
-                logger.warning("[!] Warning : {} returned an error".format(link))
-        # check if proxy are working
+                logger.warning("[!] Warning: {} returned an error".format(link))
+        # Check if proxies are working
 
     def check_proxy(self):
         # open proxy_test.txt and add all line to a list

lib/domain.py

@@ -3,70 +3,62 @@
 import lib.custom_logger as custom_logger
 import api.global_parser as global_parser
 from lib.configuration import configuration
+import re
 
 logger = custom_logger.logger
 
 
+def valid_fqdn(fqdn: str) -> bool:
+    fqdn_regex = re.compile(
+        r"^(?!-)[A-Za-z0-9-]{1,63}(?<!-)\.(?!-)[A-Za-z0-9-]{1,63}(?<!-)(?:\.(?!-)[A-Za-z0-9-]{1,63}(?<!-))*$"
+    )
+    if fqdn_regex.match(fqdn) is None:
+        return False
+    else:
+        if fqdn != "\\r" and "." in fqdn:
+            return True
+
+
 class domain:
     def __init__(self, name: str, config: configuration):
+        """
+        Initializes a domain object with a given name and configuration.
+
+        :param name: The name of the domain.
+        :param config: The configuration object to use.
+        """
         self.name = name
         self.config = config
         self.subdomains = []
-        self.ip = self.get_ip()
+        if valid_fqdn(self.name):
+            self.ip = self.get_ip()
+        else:
+            self.ip = "Invalid"
         self.handler = self.config.handler
         self.trough_proxy = config.api_trough_proxy
 
     def get_subs(self, ip_trough_proxy: bool = False):
+        """
+        Gets all the subdomains of the domain.
+
+        :param ip_trough_proxy: Whether to get the IP through a proxy.
+        :return: A list of subdomains.
+        """
         # get all the subdomains of the domain
         changed = False
         if not self.trough_proxy and self.handler.there_is_proxy():
             logger.info("[*] Disabling proxy for requesting API")
             changed = True
             olds = self.handler.remove_proxys()
         logger.info(f"[*] Getting subdomains for {self.name}")
-
-        self.subdomains += global_parser.main(self.name, self.config)
+        self.subdomains += [
+            subdomain
+            for subdomain in global_parser.main(self.name, self.config)
+            if subdomain and valid_fqdn(subdomain)
+        ]
         if changed:
             logger.info("[*] Re-enabling proxy")
             self.handler.add_proxys(olds)
-        # get the subdomains from bufferover
-        # self.subdomains += bufferover_parser(self.name)
-        # # get the subdomains from threatcrowd
-        # self.subdomains += threatcrowd_parser(self.name)
-        # # get the subdomains from threatminer
-        # self.subdomains += threatminer_parser(self.name)
-        # # get the subdomains from virustotal
-        # self.subdomains += virustotal_parser(self.name)
-        # # get the subdomains from anubis
-        # self.subdomains += anubis_parser(self.name)
-        # # get the subdomains from securitytrails
-        # self.subdomains += securitytrails_parser(self.name)
-        # # get the subdomains from certspotter
-        # self.subdomains += certspotter_parser(self.name)
-        # # get the subdomains from riddler
-        # self.subdomains += riddler_parser(self.name)
-        # # get the subdomains from sublist3r
-        # self.subdomains += sublist3r_parser(self.name)
-        # # get the subdomains from subfinder
-        # self.subdomains += subfinder_parser(self.name)
-        # # get the subdomains from amass
-        # self.subdomains += amass_parser(self.name)
-        # # get the subdomains from assetfinder
-        # self.subdomains += assetfinder_parser(self.name)
-        # # get the subdomains from findomain
-        # self.subdomains += findomain_parser(self.name)
-        # # get the subdomains from dnsdumpster
-        # self.subdomains += dnsdumpster_parser(self.name)
-        # # get the subdomains from spyse
-        # self.subdomains += spyse_parser(self.name)
-        # # get the subdomains from rapiddns
-        # self.subdomains += rapiddns_parser(self.name)
-        # # get the subdomains from threatbook
-        # self.subdomains += threatbook_parser(self.name)
-        # # get the subdomains from dnsbuffer
-        # self.subdomains += dnsbuffer_parser(self.name)
-        # # get the subdomains from threatminer
-        # self.subdomains += threatminer_parser(self.name)
         changed = False
         if not ip_trough_proxy and self.handler.there_is_proxy():
             logger.info("[*] Deactivating proxy")
@@ -75,18 +67,22 @@ def get_subs(self, ip_trough_proxy: bool = False):
         if self.config.get_fqdn_cert:
             logger.info("[*] Getting fqdns trough certificate for {}".format(self.name))
             with_cert = self.handler.get_certificate_san(self.name)
-            if with_cert:
+            if with_cert and valid_fqdn(with_cert) and with_cert not in self.subdomains:
                 self.subdomains += with_cert
             with_cert = self.handler.get_cert_fqdn(self.name)
-            if with_cert:
+            if with_cert and valid_fqdn(with_cert) and with_cert not in self.subdomains:
                 self.subdomains += [with_cert]
         if changed:
             logger.info("[*] Re-enabling proxy")
             self.handler.add_proxys(olds)
         return self.subdomains
 
     def get_ip(self):
-        # get the ip address from the domain by using dns resolver
+        """
+        Gets the IP address from the domain by using DNS resolver.
+
+        :return: The IP address of the domain.
+        """
         try:
             ip = resolver.resolve(self.name, "A")
             return ip[0].to_text()