refactor module inheritance

bbot/core/helpers/modules.py

@@ -18,6 +18,9 @@ def __init__(self):
         self._configs = {}
 
     def file_filter(self, file):
+        file = file.resolve()
+        if "mixins" in file.parts:
+            return False
         return file.suffix.lower() == ".py" and file.stem not in ["base", "__init__"]
 
     def preload(self, module_dir):

bbot/modules/anubisdb.py

@@ -1,7 +1,7 @@
-from .crobat import crobat
+from bbot.modules.templates.subdomain_enum import subdomain_enum
 
 
-class anubisdb(crobat):
+class anubisdb(subdomain_enum):
     flags = ["subdomain-enum", "passive", "safe"]
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]

bbot/modules/azure_tenant.py

@@ -1,9 +1,9 @@
 import re
 
-from .viewdns import viewdns
+from bbot.modules.templates.root_domains import root_domains
 
 
-class azure_tenant(viewdns):
+class azure_tenant(root_domains):
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]
     flags = ["affiliates", "subdomain-enum", "cloud-enum", "passive", "safe"]

bbot/modules/bevigil.py

@@ -1,7 +1,7 @@
-from bbot.modules.shodan_dns import shodan_dns
+from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
 
 
-class bevigil(shodan_dns):
+class bevigil(subdomain_enum_apikey):
     """
     Retrieve OSINT data from mobile applications using BeVigil
     """

bbot/modules/binaryedge.py

@@ -1,7 +1,7 @@
-from bbot.modules.shodan_dns import shodan_dns
+from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
 
 
-class binaryedge(shodan_dns):
+class binaryedge(subdomain_enum_apikey):
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]
     flags = ["subdomain-enum", "passive", "safe"]

bbot/modules/builtwith.py

@@ -10,10 +10,10 @@
 #                                                          #
 ############################################################
 
-from .shodan_dns import shodan_dns
+from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
 
 
-class builtwith(shodan_dns):
+class builtwith(subdomain_enum_apikey):
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]
     flags = ["affiliates", "subdomain-enum", "passive", "safe"]

bbot/modules/c99.py

@@ -1,7 +1,7 @@
-from .shodan_dns import shodan_dns
+from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
 
 
-class c99(shodan_dns):
+class c99(subdomain_enum_apikey):
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]
     flags = ["subdomain-enum", "passive", "safe"]

bbot/modules/censys.py

@@ -1,7 +1,7 @@
-from bbot.modules.shodan_dns import shodan_dns
+from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
 
 
-class censys(shodan_dns):
+class censys(subdomain_enum_apikey):
     """
     thanks to https://github.com/owasp-amass/amass/blob/master/resources/scripts/cert/censys.ads
     """

bbot/modules/certspotter.py

@@ -1,7 +1,7 @@
-from bbot.modules.crobat import crobat
+from bbot.modules.templates.subdomain_enum import subdomain_enum
 
 
-class certspotter(crobat):
+class certspotter(subdomain_enum):
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]
     flags = ["subdomain-enum", "passive", "safe"]

bbot/modules/chaos.py

@@ -1,7 +1,7 @@
-from .shodan_dns import shodan_dns
+from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
 
 
-class chaos(shodan_dns):
+class chaos(subdomain_enum_apikey):
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]
     flags = ["subdomain-enum", "passive", "safe"]

bbot/modules/columbus.py

@@ -1,7 +1,7 @@
-from .crobat import crobat
+from bbot.modules.templates.subdomain_enum import subdomain_enum
 
 
-class columbus(crobat):
+class columbus(subdomain_enum):
     flags = ["subdomain-enum", "passive", "safe"]
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]

bbot/modules/crobat.py

@@ -1,145 +1,10 @@
-from bbot.modules.base import BaseModule
+from bbot.modules.templates.subdomain_enum import subdomain_enum
 
 
-class crobat(BaseModule):
-    """
-    A typical free API-based subdomain enumeration module
-    Inherited by several other modules including sublist3r, dnsdumpster, etc.
-    """
-
+class crobat(subdomain_enum):
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]
     # tag "subdomain-enum" removed 2023-02-24 because API is offline
     flags = ["passive", "safe"]
     meta = {"description": "Query Project Crobat for subdomains"}
-
     base_url = "https://sonar.omnisint.io"
-    # set module error state after this many failed requests in a row
-    abort_after_failures = 5
-    # whether to reject wildcard DNS_NAMEs
-    reject_wildcards = "strict"
-    # this helps combat rate limiting by ensuring that a query doesn't execute
-    # until the queue is ready to receive its results
-    _qsize = 1
-
-    async def setup(self):
-        self.processed = set()
-        self.http_timeout = self.scan.config.get("http_timeout", 10)
-        self._failures = 0
-        return True
-
-    async def _is_wildcard(self, query):
-        if self.helpers.is_dns_name(query):
-            for domain, wildcard_rdtypes in (await self.helpers.is_wildcard_domain(query)).items():
-                if any(t in wildcard_rdtypes for t in ("A", "AAAA", "CNAME")):
-                    return True
-        return False
-
-    async def filter_event(self, event):
-        """
-        This filter_event is used across many modules
-        """
-        query = self.make_query(event)
-        # reject if already processed
-        if self.already_processed(query):
-            return False, "Event was already processed"
-        eligible, reason = await self.eligible_for_enumeration(event)
-        if eligible:
-            self.processed.add(hash(query))
-            return True, reason
-        return False, reason
-
-    async def eligible_for_enumeration(self, event):
-        query = self.make_query(event)
-        # check if wildcard
-        is_wildcard = await self._is_wildcard(query)
-        # check if cloud
-        is_cloud = False
-        if any(t.startswith("cloud-") for t in event.tags):
-            is_cloud = True
-        # reject if it's a cloud resource and not in our target
-        if is_cloud and event not in self.scan.target:
-            return False, "Event is a cloud resource and not a direct target"
-        # optionally reject events with wildcards / errors
-        if self.reject_wildcards:
-            if any(t in event.tags for t in ("a-error", "aaaa-error")):
-                return False, "Event has a DNS resolution error"
-            if self.reject_wildcards == "strict":
-                if is_wildcard:
-                    return False, "Event is a wildcard domain"
-            elif self.reject_wildcards == "cloud_only":
-                if is_wildcard and is_cloud:
-                    return False, "Event is both a cloud resource and a wildcard domain"
-        return True, ""
-
-    def already_processed(self, hostname):
-        for parent in self.helpers.domain_parents(hostname, include_self=True):
-            if hash(parent) in self.processed:
-                return True
-        return False
-
-    async def abort_if(self, event):
-        # this helps weed out unwanted results when scanning IP_RANGES and wildcard domains
-        if "in-scope" not in event.tags:
-            return True
-        if await self._is_wildcard(event.data):
-            return True
-        return False
-
-    async def handle_event(self, event):
-        query = self.make_query(event)
-        results = await self.query(query)
-        if results:
-            for hostname in set(results):
-                if hostname:
-                    try:
-                        hostname = self.helpers.validators.validate_host(hostname)
-                    except ValueError as e:
-                        self.verbose(e)
-                        continue
-                    if hostname and hostname.endswith(f".{query}") and not hostname == event.data:
-                        self.emit_event(hostname, "DNS_NAME", event, abort_if=self.abort_if)
-
-    async def request_url(self, query):
-        url = f"{self.base_url}/subdomains/{self.helpers.quote(query)}"
-        return await self.request_with_fail_count(url)
-
-    def make_query(self, event):
-        if "target" in event.tags:
-            query = str(event.data)
-        else:
-            query = self.helpers.parent_domain(event.data).lower()
-        return ".".join([s for s in query.split(".") if s != "_wildcard"])
-
-    def parse_results(self, r, query=None):
-        json = r.json()
-        if json:
-            for hostname in json:
-                yield hostname
-
-    async def query(self, query, parse_fn=None, request_fn=None):
-        if parse_fn is None:
-            parse_fn = self.parse_results
-        if request_fn is None:
-            request_fn = self.request_url
-        try:
-            response = await request_fn(query)
-            if response is None:
-                self.info(f'Query "{query}" failed (no response)')
-                return []
-            try:
-                results = list(parse_fn(response, query))
-            except Exception as e:
-                if response:
-                    self.info(
-                        f'Error parsing results for query "{query}" (status code {response.status_code})', trace=True
-                    )
-                    self.log.trace(response.text)
-                else:
-                    self.info(f'Error parsing results for "{query}": {e}', trace=True)
-                return
-            if results:
-                return results
-            self.debug(f'No results for "{query}"')
-        except Exception as e:
-            self.info(f"Error retrieving results for {query}: {e}", trace=True)

bbot/modules/crt.py

@@ -1,7 +1,7 @@
-from .crobat import crobat
+from bbot.modules.templates.subdomain_enum import subdomain_enum
 
 
-class crt(crobat):
+class crt(subdomain_enum):
     flags = ["subdomain-enum", "passive", "safe"]
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]

bbot/modules/digitorus.py

@@ -1,9 +1,9 @@
 import re
 
-from .crobat import crobat
+from bbot.modules.templates.subdomain_enum import subdomain_enum
 
 
-class digitorus(crobat):
+class digitorus(subdomain_enum):
     flags = ["subdomain-enum", "passive", "safe"]
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]

bbot/modules/dnsdumpster.py

@@ -1,10 +1,10 @@
 import re
 from bs4 import BeautifulSoup
 
-from .crobat import crobat
+from bbot.modules.templates.subdomain_enum import subdomain_enum
 
 
-class dnsdumpster(crobat):
+class dnsdumpster(subdomain_enum):
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]
     flags = ["subdomain-enum", "passive", "safe"]

bbot/modules/emailformat.py

@@ -1,7 +1,7 @@
-from .viewdns import viewdns
+from bbot.modules.templates.root_domains import root_domains
 
 
-class emailformat(viewdns):
+class emailformat(root_domains):
     watched_events = ["DNS_NAME"]
     produced_events = ["EMAIL_ADDRESS"]
     flags = ["passive", "email-enum", "safe"]

bbot/modules/fullhunt.py

@@ -1,7 +1,7 @@
-from bbot.modules.shodan_dns import shodan_dns
+from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
 
 
-class fullhunt(shodan_dns):
+class fullhunt(subdomain_enum_apikey):
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]
     flags = ["subdomain-enum", "passive", "safe"]

bbot/modules/github.py

@@ -1,7 +1,7 @@
-from bbot.modules.shodan_dns import shodan_dns
+from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
 
 
-class github(shodan_dns):
+class github(subdomain_enum_apikey):
     watched_events = ["DNS_NAME"]
     produced_events = ["URL_UNVERIFIED"]
     flags = ["passive", "subdomain-enum", "safe"]

bbot/modules/hackertarget.py

@@ -1,7 +1,7 @@
-from bbot.modules.crobat import crobat
+from bbot.modules.templates.subdomain_enum import subdomain_enum
 
 
-class hackertarget(crobat):
+class hackertarget(subdomain_enum):
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]
     flags = ["subdomain-enum", "passive", "safe"]

bbot/modules/hunterio.py

@@ -1,7 +1,7 @@
-from .shodan_dns import shodan_dns
+from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
 
 
-class hunterio(shodan_dns):
+class hunterio(subdomain_enum_apikey):
     watched_events = ["DNS_NAME"]
     produced_events = ["EMAIL_ADDRESS", "DNS_NAME", "URL_UNVERIFIED"]
     flags = ["passive", "email-enum", "subdomain-enum", "safe"]

bbot/modules/ipstack.py

@@ -1,7 +1,7 @@
-from .shodan_dns import shodan_dns
+from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
 
 
-class Ipstack(shodan_dns):
+class Ipstack(subdomain_enum_apikey):
     """
     Ipstack GeoIP
     Leverages the ipstack.com API to geolocate a host by IP address.

bbot/modules/leakix.py

@@ -1,8 +1,7 @@
-from .crobat import crobat
-from .shodan_dns import shodan_dns
+from bbot.modules.templates.subdomain_enum import subdomain_enum_apikey
 
 
-class leakix(shodan_dns):
+class leakix(subdomain_enum_apikey):
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]
     flags = ["subdomain-enum", "passive", "safe"]
@@ -14,7 +13,7 @@ class leakix(shodan_dns):
     base_url = "https://leakix.net"
 
     async def setup(self):
-        ret = await crobat.setup(self)
+        ret = await super(subdomain_enum_apikey, self).setup()
         self.headers = {"Accept": "application/json"}
         self.api_key = self.config.get("api_key", "")
         if self.api_key:

bbot/modules/massdns.py

@@ -3,10 +3,10 @@
 import random
 import subprocess
 
-from .crobat import crobat
+from bbot.modules.templates.subdomain_enum import subdomain_enum
 
 
-class massdns(crobat):
+class massdns(subdomain_enum):
     flags = ["subdomain-enum", "passive", "slow", "aggressive"]
     watched_events = ["DNS_NAME"]
     produced_events = ["DNS_NAME"]