Merge pull request #3021 from blacklanternsecurity/blasthttp-integration-sharedloop

blasthttp 0.2.0: shared event loop (eliminate executor bridge)

Author: liquidsec

bbot/core/helpers/helper.py

@@ -208,10 +208,8 @@ def loop(self):
         """
         if self._loop is None:
             self._loop = get_event_loop()
-            # increase default thread pool size to prevent executor starvation
-            # during heavy scans where YARA, regex, DNS, and HTTP all compete for threads
-            thread_pool_size = max(32, (os.cpu_count() or 1) * 4)
-            self._io_executor = ThreadPoolExecutor(max_workers=thread_pool_size)
+            # only current caller is wafw00f (sync requests library)
+            self._io_executor = ThreadPoolExecutor(max_workers=max(8, (os.cpu_count() or 1) + 4))
             self._cpu_executor = ThreadPoolExecutor(max_workers=max(8, os.cpu_count() or 4))
             self._loop.set_default_executor(self._io_executor)
         return self._loop

bbot/core/helpers/web/web.py

@@ -266,8 +266,8 @@ async def request(self, *args, **kwargs):
             if self.http_debug:
                 log.trace(f"blasthttp request: {method} {url}")
 
-            # Run in executor — blasthttp blocks the thread (tokio runtime)
-            blast_response = await self.parent_helper.run_in_executor_io(self.client.request, url, **blast_kwargs)
+            # blasthttp returns a native coroutine via pyo3-async-runtimes
+            blast_response = await self.client.request(url, **blast_kwargs)
 
             response = BlasthttpResponse(blast_response, request_url=url, method=method)
 

bbot/modules/http.py

@@ -212,8 +212,8 @@ async def handle_batch(self, *events):
             )
             configs.append(config)
 
-        # Run batch in executor to avoid blocking the event loop
-        results = await self.helpers.run_in_executor_io(self.client.request_batch, configs, self.threads)
+        # blasthttp batch returns a native coroutine via pyo3-async-runtimes
+        results = await self.client.request_batch(configs, self.threads)
 
         # Index results by URL for the dedup check
         results_by_url = {r.url: r for r in results}

bbot/modules/web_brute.py

@@ -180,9 +180,7 @@ async def baseline_fuzz(self, url, exts=None, prefix="", suffix=""):
                 )
 
             canary_results = []
-            results = await self.helpers.run_in_executor_io(
-                self.blast_client.request_batch, canary_configs, 4, rate_limit=self.rate
-            )
+            results = await self.blast_client.request_batch(canary_configs, 4, rate_limit=self.rate)
             for result in results:
                 if result.success:
                     canary_results.append(self._batch_response_metrics(result.response))
@@ -306,9 +304,7 @@ async def execute_fuzz(
             self.debug(f"Fuzzing {len(configs)} URLs for ext [{ext}]")
 
             # Fire all requests via native blasthttp batch (Rust concurrency)
-            results = await self.helpers.run_in_executor_io(
-                self.blast_client.request_batch, configs, self.concurrency, rate_limit=self.rate
-            )
+            results = await self.blast_client.request_batch(configs, self.concurrency, rate_limit=self.rate)
 
             # Index results by URL for ordered processing
             results_by_url = {}
@@ -367,9 +363,7 @@ async def execute_fuzz(
                             proxy=proxy,
                         )
                     ]
-                    canary_batch = await self.helpers.run_in_executor_io(
-                        self.blast_client.request_batch, canary_configs, 1, rate_limit=self.rate
-                    )
+                    canary_batch = await self.blast_client.request_batch(canary_configs, 1, rate_limit=self.rate)
                     if canary_batch and canary_batch[0].success:
                         canary_metrics = self._batch_response_metrics(canary_batch[0].response)
                         if not self._is_baseline_match(canary_metrics, ext_filter):

bbot/scanner/scanner.py

@@ -768,15 +768,13 @@ def modules_status(self, _log=False, detailed=False):
                 )
 
             num_queued_events = self.num_queued_events
-            io_backlog = self.helpers._io_executor._work_queue.qsize()
-            cpu_backlog = self.helpers._cpu_executor._work_queue.qsize()
             if num_queued_events:
                 self.info(
-                    f"{self.name}: {num_queued_events:,} events in queue ({self.stats.speedometer.speed:,} processed in the past {self.status_frequency} seconds) | Thread pool backlog: I/O: {io_backlog:,}, CPU: {cpu_backlog:,}"
+                    f"{self.name}: {num_queued_events:,} events in queue ({self.stats.speedometer.speed:,} processed in the past {self.status_frequency} seconds)"
                 )
             else:
                 self.info(
-                    f"{self.name}: No events in queue ({self.stats.speedometer.speed:,} processed in the past {self.status_frequency} seconds) | Thread pool backlog: I/O: {io_backlog:,}, CPU: {cpu_backlog:,}"
+                    f"{self.name}: No events in queue ({self.stats.speedometer.speed:,} processed in the past {self.status_frequency} seconds)"
                 )
 
             if detailed or self.log_level <= logging.DEBUG:

pyproject.toml

@@ -47,7 +47,7 @@ dependencies = [
     "ansible-core>=2.17,<3",
     "tldextract>=5.3.0,<6",
     "cloudcheck>=9.2.0,<10",
-    "blasthttp>=0.1.4",
+    "blasthttp>=0.2.0",
 ]
 
 [project.urls]