Merge origin/http-response-url-extension (#3406) into bleeding-edge

Author: liquidsec

bbot/core/event/base.py

@@ -142,6 +142,9 @@ class BaseEvent:
     _discovery_context_regex = re.compile(r"\{(?:event|module)[^}]*\}")
     # Stats class for the status line — override in subclasses for custom formatting
     _stats_class = None
+    # Whether this event is subject to `url_extension_special` distribution filtering.
+    # False for events representing already-retrieved content, whose consumers analyze the body.
+    _url_special_filterable = True
 
     # using __slots__ dramatically reduces memory usage in large scans
     __slots__ = [
@@ -1164,18 +1167,24 @@ def sanitize_data(self, data):
 class DictEvent(BaseEvent):
     __slots__ = ["url_extension"]
 
+    def _set_url_extension(self):
+        """Extract the file extension from self.parsed_url and record it as an attribute and a tag."""
+        parsed_url = getattr(self, "parsed_url", None)
+        if parsed_url is None:
+            return
+        url_path = parsed_url.path
+        if not url_path:
+            return
+        extension = get_file_extension(str(url_path).lower())
+        if extension:
+            self.url_extension = extension
+            self.add_tag(f"extension-{extension}")
+
     def sanitize_data(self, data):
         url = data.get("url", "")
         if url:
             self.parsed_url = self.validators.validate_url_parsed(url)
-            # extract url_extension from any dict event with a URL
-            url_path = self.parsed_url.path
-            if url_path:
-                parsed_path_lower = str(url_path).lower()
-                extension = get_file_extension(parsed_path_lower)
-                if extension:
-                    self.url_extension = extension
-                    self.add_tag(f"extension-{extension}")
+            self._set_url_extension()
         return data
 
     def _data_load(self, data):
@@ -1446,15 +1455,7 @@ def sanitize_data(self, data):
         self.parsed_url = self.validators.validate_url_parsed(url)
         data["url"] = self.parsed_url.geturl()
 
-        # special handling of URL extensions
-        if self.parsed_url is not None:
-            url_path = self.parsed_url.path
-            if url_path:
-                parsed_path_lower = str(url_path).lower()
-                extension = get_file_extension(parsed_path_lower)
-                if extension:
-                    self.url_extension = extension
-                    self.add_tag(f"extension-{extension}")
+        self._set_url_extension()
 
         # tag as dir or endpoint
         if str(self.parsed_url.path).endswith("/"):
@@ -1716,6 +1717,10 @@ def _words(self):
 
 
 class HTTP_RESPONSE(URL_UNVERIFIED):
+    # the response body has already been retrieved, so content-analysis modules
+    # should still receive it even for special extensions like .js
+    _url_special_filterable = False
+
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         # count number of consecutive redirects
@@ -1776,6 +1781,7 @@ def sanitize_data(self, data):
         url = data.get("url", "")
         self.parsed_url = self.validators.validate_url_parsed(url)
         data["url"] = self.parsed_url.geturl()
+        self._set_url_extension()
 
         if not "raw_header" in data:
             raise ValueError("raw_header is required for HTTP_RESPONSE events")

bbot/modules/base.py

@@ -871,7 +871,7 @@ def _event_precheck(self, event):
             return False, "it did not meet target_only filter criteria"
 
         # limit events with special URL extensions (e.g. .js) to modules that opt in
-        if not self.accept_url_special:
+        if not self.accept_url_special and event._url_special_filterable:
             extension = getattr(event, "url_extension", "")
             if extension in self.scan.url_extension_special:
                 return (

bbot/test/test_step_1/test_events.py

@@ -175,6 +175,28 @@ async def test_events(events, helpers):
     )
     assert getattr(wp_no_ext, "url_extension", "NOT_SET") == "NOT_SET"
 
+    # url_extension: HTTP_RESPONSE events
+    # modules that filter on url_extension (e.g. paramminer, lightfuzz) rely on this being set
+    def _http_response(url):
+        return scan.make_event(
+            {"url": url, "raw_header": "HTTP/1.1 200 OK\r\n\r\n"},
+            "HTTP_RESPONSE",
+            dummy=True,
+        )
+
+    hr_pdf = _http_response("https://evilcorp.com/files/document.pdf?foo=bar")
+    assert getattr(hr_pdf, "url_extension", "") == "pdf"
+    assert "extension-pdf" in hr_pdf.tags
+    hr_no_ext = _http_response("https://evilcorp.com/search")
+    assert getattr(hr_no_ext, "url_extension", "NOT_SET") == "NOT_SET"
+
+    # special extensions (.js) must still reach modules that don't opt in to special URLs,
+    # since the response body has already been retrieved
+    hr_js = _http_response("https://evilcorp.com/app.js")
+    assert getattr(hr_js, "url_extension", "") == "js"
+    assert hr_js._url_special_filterable is False
+    assert scan.make_event("https://evilcorp.com/app.js", dummy=True)._url_special_filterable is True
+
     # http response
     assert events.http_response.host == "example.com"
     assert events.http_response.port == 80

bbot/test/test_step_2/module_tests/test_module_paramminer_getparams.py

@@ -288,3 +288,11 @@ def check(self, module_test, events):
         assert excavate_extracted_param, "Excavate should still extract the parameter from the HTML link"
         assert not paramminer_recycled_to_php, "Paramminer should not recycle words from static-URL WEB_PARAMETERs"
         assert not paramminer_bruted_pdf, "Paramminer should not brute-force parameters on static URLs"
+
+        # paramminer's static filter also applies to HTTP_RESPONSE events, which requires
+        # url_extension to be populated on them
+        pdf_responses = [e for e in events if e.type == "HTTP_RESPONSE" and e.data["url"].endswith("test2.pdf")]
+        assert pdf_responses, "expected an HTTP_RESPONSE for the .pdf URL"
+        assert getattr(pdf_responses[0], "url_extension", None) == "pdf", (
+            "HTTP_RESPONSE is missing url_extension, so paramminer's static filter is bypassed"
+        )