Blasthttp mock migration

bbot/core/helpers/web/web.py

@@ -10,9 +10,10 @@
 from bs4 import MarkupResemblesLocatorWarning
 from bs4.builder import XMLParsedAsHTMLWarning
 
+from blasthttp import HTTPStatusError
+
 from bbot.core.helpers.misc import truncate_filename, bytes_to_human, get_exception_chain
 from bbot.errors import WordlistError, WebError
-from .blast_response import BlasthttpResponse, BlasthttpHTTPError
 
 warnings.filterwarnings("ignore", category=XMLParsedAsHTMLWarning)
 warnings.filterwarnings("ignore", category=MarkupResemblesLocatorWarning)
@@ -284,9 +285,7 @@ async def request(self, *args, **kwargs):
                 log.trace(f"blasthttp request: {method} {url}")
 
             # 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)
+            response = await self.client.request(url, **blast_kwargs)
 
             if self.http_debug:
                 log.trace(
@@ -389,10 +388,7 @@ async def request_batch_stream(self, urls, threads=10, **kwargs):
             trackers_by_url.setdefault(config.url, deque()).append(tracker)
 
         async for br in iter_batch_results(self.client.request_batch_stream(configs, concurrency=threads)):
-            if br.response is not None:
-                response = BlasthttpResponse(br.response, request_url=br.url, method="GET")
-            else:
-                response = None
+            response = br.response  # blasthttp.Response or None
             if has_tracker:
                 queue = trackers_by_url.get(br.url)
                 tracker = queue.popleft() if queue else None
@@ -455,7 +451,7 @@ async def download(self, url, **kwargs):
                 response = await self.request(url, **kwargs)
 
                 if response is None:
-                    raise BlasthttpHTTPError(f"No response from {url}")
+                    raise HTTPStatusError(f"No response from {url}")
 
                 log.debug(f"Download result: HTTP {response.status_code}")
                 response.raise_for_status()
@@ -473,7 +469,7 @@ async def download(self, url, **kwargs):
                     f.write(content)
                 success = True
 
-            except (BlasthttpHTTPError, WebError, RuntimeError) as e:
+            except (HTTPStatusError, WebError, RuntimeError) as e:
                 log_fn = log.verbose
                 if warn:
                     log_fn = log.warning

bbot/modules/http.py

@@ -109,14 +109,15 @@ def _response_to_json(self, url_input, response):
         parsed = urlparse(response.url)
         path = parsed.path or "/"
 
-        # Build raw_header string (required by HTTP_RESPONSE validation)
+        # Build raw_header string (required by HTTP_RESPONSE validation).
+        # blasthttp already builds the canonical "Name: Value\r\n..." form
+        # — reuse it instead of rebuilding.
         status_line = f"HTTP/1.1 {response.status} \r\n"
-        header_lines = "\r\n".join(f"{k}: {v}" for k, v in response.headers)
-        raw_header = f"{status_line}{header_lines}\r\n\r\n"
+        raw_header = f"{status_line}{response.raw_headers}\r\n\r\n"
 
         # Build header dict (lowercase keys, comma-joined for dupes)
         header_dict = {}
-        for k, v in response.headers:
+        for k, v in response.headers.items():
             key = k.lower().replace("-", "_")
             if key in header_dict:
                 header_dict[key] += f", {v}"

bbot/modules/telerik.py

@@ -156,7 +156,7 @@ class telerik(BaseModule):
     options = {"exploit_RAU_crypto": False, "include_subdirs": False}
     options_desc = {
         "exploit_RAU_crypto": "Attempt to confirm any RAU AXD detections are vulnerable",
-        "include_subdirs": "Include subdirectories in the scan (off by default)",  # will create many finding events if used in conjunction with web spider or web_brute
+        "include_subdirs": "Include subdirectories in the scan (off by default)",  # will create many finding events if used in conjunction with web spider or webbrute
     }
 
     in_scope_only = True

bbot/modules/url_manipulation.py

@@ -68,7 +68,7 @@ async def handle_event(self, event):
                 self.debug(f"Encountered HttpCompareError: [{e}] for URL [{event.url}]")
 
             if subject_response:
-                subject_content = "".join([str(x) for x in subject_response.headers])
+                subject_content = subject_response.raw_headers
                 if subject_response.text is not None:
                     subject_content += subject_response.text
 

bbot/modules/web_brute.py → bbot/modules/webbrute.py

@@ -7,7 +7,7 @@
 from bbot.modules.base import BaseModule
 
 
-class web_brute(BaseModule):
+class webbrute(BaseModule):
     watched_events = ["URL"]
     produced_events = ["URL_UNVERIFIED"]
     flags = ["active", "loud"]
@@ -124,16 +124,6 @@ def _response_metrics(self, response):
             "lines": text.count("\n") + 1,
         }
 
-    def _batch_response_metrics(self, response):
-        """Extract metrics from a raw blasthttp batch response."""
-        body = response.body or ""
-        return {
-            "status": response.status,
-            "length": len(response.body_bytes),
-            "words": len(body.split()),
-            "lines": body.count("\n") + 1,
-        }
-
     def _is_baseline_match(self, metrics, baseline_filter):
         """Return True if the response matches the baseline (i.e. should be filtered OUT)."""
         if baseline_filter.get("abort"):
@@ -187,7 +177,7 @@ async def baseline_fuzz(self, url, exts=None, prefix="", suffix=""):
                 self.blast_client.request_batch_stream(canary_configs, 4, rate_limit=self.rate)
             ):
                 if result.success:
-                    canary_results.append(self._batch_response_metrics(result.response))
+                    canary_results.append(self._response_metrics(result.response))
                     if await self.helpers.yara.match(self.waf_yara_rules, result.response.body):
                         canary_waf_count += 1
 
@@ -331,7 +321,7 @@ async def execute_fuzz(
                     continue
 
                 response = result.response
-                metrics = self._batch_response_metrics(response)
+                metrics = self._response_metrics(response)
 
                 # Check if this matches the baseline (should be filtered out)
                 if ext_filter and self._is_baseline_match(metrics, ext_filter):
@@ -353,7 +343,7 @@ async def execute_fuzz(
                 # not real findings (e.g. mod_userdir sending ~user to /)
                 if 300 <= response.status < 400:
                     location = ""
-                    for hdr_name, hdr_val in response.headers:
+                    for hdr_name, hdr_val in response.headers.items():
                         if hdr_name.lower() == "location":
                             location = hdr_val
                             break
@@ -373,12 +363,15 @@ async def execute_fuzz(
                 self.debug("Found canary in results, all hits are likely false positives — aborting")
                 return
 
-            # Mid-scan validation: one canary check per extension
+            # Mid-scan validation: one canary check per extension.
+            # Single request — use client.request() directly instead of a
+            # 1-config request_batch_stream loop (the streaming API only
+            # earns its keep with multiple in-flight requests).
             if hits and not baseline and ext_filter:
                 canary_word = "".join(random.choice(string.ascii_lowercase) for _ in range(4))
                 canary_url = f"{url}{prefix}{canary_word}{suffix}{ext}"
-                canary_configs = [
-                    blasthttp.BatchConfig(
+                try:
+                    canary_response = await self.blast_client.request(
                         canary_url,
                         headers=headers,
                         timeout=self.scan.http_timeout,
@@ -387,14 +380,11 @@ async def execute_fuzz(
                         follow_redirects=False,
                         proxy=proxy,
                     )
-                ]
-                canary_result = None
-                async for r in iter_batch_results(
-                    self.blast_client.request_batch_stream(canary_configs, 1, rate_limit=self.rate)
-                ):
-                    canary_result = r
-                if canary_result is not None and canary_result.success:
-                    canary_metrics = self._batch_response_metrics(canary_result.response)
+                except Exception as e:
+                    self.debug(f"Mid-scan canary request failed: {e}")
+                    canary_response = None
+                if canary_response is not None:
+                    canary_metrics = self._response_metrics(canary_response)
                     if not self._is_baseline_match(canary_metrics, ext_filter):
                         self.verbose(
                             f"Would have reported {len(hits)} hit(s), but mid-scan baseline check failed. "

bbot/modules/web_brute_shortnames.py → bbot/modules/webbrute_shortnames.py

@@ -3,10 +3,10 @@
 import random
 import string
 
-from bbot.modules.web_brute import web_brute
+from bbot.modules.webbrute import webbrute
 
 
-class web_brute_shortnames(web_brute):
+class webbrute_shortnames(webbrute):
     watched_events = ["URL_HINT"]
     produced_events = ["URL_UNVERIFIED"]
     flags = ["loud", "active", "iis-shortnames", "web-heavy"]
@@ -183,7 +183,7 @@ def find_delimiter(self, hint):
 
     async def filter_event(self, event):
         if "iis-magic-url" in event.tags:
-            return False, "iis-magic-url URL_HINTs are not solvable by web_brute_shortnames"
+            return False, "iis-magic-url URL_HINTs are not solvable by webbrute_shortnames"
         if event.parent.type != "URL":
             return False, "its parent event is not of type URL"
         return True
@@ -204,7 +204,7 @@ async def handle_event(self, event):
         elif "shortname-directory" in event.tags:
             shortname_type = "directory"
         else:
-            self.error("web_brute_shortnames received URL_HINT without proper 'shortname-' tag")
+            self.error("webbrute_shortnames received URL_HINT without proper 'shortname-' tag")
             return
 
         host = f"{event.parent.parsed_url.scheme}://{event.parent.parsed_url.netloc}/"
@@ -329,7 +329,7 @@ async def finish(self):
                             elif "shortname-directory" in self.shortname_to_event[hint].tags:
                                 shortname_type = "directory"
                             else:
-                                self.error("web_brute_shortnames received URL_HINT without proper 'shortname-' tag")
+                                self.error("webbrute_shortnames received URL_HINT without proper 'shortname-' tag")
                                 continue
 
                             partial_hint = hint[len(prefix) :]