Offload wayback URL parsing into separate process

bbot/core/helpers/validators.py

@@ -221,6 +221,7 @@ def collapse_urls(urls, threshold=10):
         ["http://evilcorp.com/user/11111/info"]
 
     """
+    log.verbose(f"Collapsing {len(urls):,} URLs")
     url_hashes = {}
     for url in urls:
         try:

bbot/modules/wayback.py

@@ -1,3 +1,5 @@
+from datetime import datetime
+
 from bbot.modules.templates.subdomain_enum import subdomain_enum
 
 
@@ -49,8 +51,19 @@ async def query(self, query):
             except KeyError:
                 continue
 
+        self.verbose(f"Found {len(urls):,} URLs for {query}")
+
         dns_names = set()
-        for parsed_url in self.helpers.validators.collapse_urls(urls, threshold=self.garbage_threshold):
+        collapsed_urls = 0
+        start_time = datetime.now()
+        parsed_urls = await self.scan.run_in_executor_mp(
+            self.execute_callback,
+            self.helpers.validators.collapse_urls,
+            urls,
+            threshold=self.garbage_threshold,
+        )
+        for parsed_url in parsed_urls:
+            collapsed_urls += 1
             if not self.urls:
                 dns_name = parsed_url.hostname
                 h = hash(dns_name)
@@ -59,4 +72,14 @@ async def query(self, query):
                     results.add((dns_name, "DNS_NAME"))
             else:
                 results.add((parsed_url.geturl(), "URL_UNVERIFIED"))
+        end_time = datetime.now()
+        duration = self.helpers.human_timedelta(end_time - start_time)
+        self.verbose(f"Collapsed {len(urls):,} -> {collapsed_urls:,} URLs in {duration}")
         return results
+
+    @staticmethod
+    def execute_callback(callback, *args, **kwargs):
+        """
+        This exists so that we can run our URL parsing logic in a separate process.
+        """
+        return list(callback(*args, **kwargs))

bbot/scanner/scanner.py

@@ -255,7 +255,7 @@ def __init__(
             mp.set_start_method("spawn")
         except Exception:
             self.warning(f"Failed to set multiprocessing spawn method. This may negatively affect performance.")
-        self.process_pool = ProcessPoolExecutor()
+        self.process_pool = ProcessPoolExecutor(max_tasks_per_child=10)
 
         self._stopping = False