Merge pull request blacklanternsecurity#615 from blacklanternsecurity…

…/bypass403_collapse

Bypass403 module multiple alert collapsing

bbot/modules/bypass403.py

@@ -81,12 +81,8 @@ class bypass403(BaseModule):
     meta = {"description": "Check 403 pages for common bypasses"}
     in_scope_only = True
 
-    async def handle_event(self, event):
-        try:
-            compare_helper = self.helpers.http_compare(event.data, allow_redirects=True)
-        except HttpCompareError as e:
-            self.debug(e)
-            return
+    async def do_checks(self, compare_helper, event, collapse_threshold):
+        results = set()
 
         for sig in signatures:
             sig = self.format_signature(sig, event)
@@ -113,13 +109,39 @@ async def handle_event(self, event):
                     else:
                         reported_signature = f"Modified URL: {sig[0]} {sig[1]}"
                     description = f"403 Bypass Reasons: [{','.join(reasons)}] Sig: [{reported_signature}]"
-                    self.emit_event(
-                        {"description": description, "host": str(event.host), "url": event.data},
-                        "FINDING",
-                        source=event,
-                    )
+                    results.add(description)
+                    if len(results) > collapse_threshold:
+                        return results
                 else:
                     self.debug(f"Status code changed to {str(subject_response.status_code)}, ignoring")
+        return results
+
+    async def handle_event(self, event):
+        try:
+            compare_helper = self.helpers.http_compare(event.data, allow_redirects=True)
+        except HttpCompareError as e:
+            self.debug(e)
+            return
+
+        collapse_threshold = 10
+        results = await self.do_checks(compare_helper, event, collapse_threshold)
+        if len(results) > collapse_threshold:
+            self.emit_event(
+                {
+                    "description": f"403 Bypass MULTIPLE SIGNATURES (exceeded threshold {str(collapse_threshold)})",
+                    "host": str(event.host),
+                    "url": event.data,
+                },
+                "FINDING",
+                source=event,
+            )
+        else:
+            for description in results:
+                self.emit_event(
+                    {"description": description, "host": str(event.host), "url": event.data},
+                    "FINDING",
+                    source=event,
+                )
 
     async def filter_event(self, event):
         if ("status-403" in event.tags) or ("status-401" in event.tags):

bbot/test/test_step_2/module_tests/test_module_bypass403.py

@@ -19,6 +19,57 @@ def check(self, module_test, events):
         assert "http://127.0.0.1:8888/test..;/" in finding.data["description"]
 
 
+class TestBypass403_collapsethreshold(ModuleTestBase):
+    targets = ["http://127.0.0.1:8888/test"]
+    modules_overrides = ["bypass403", "httpx"]
+
+    async def setup_after_prep(self, module_test):
+        respond_args = {"response_data": "alive"}
+
+        # some of these wont work outside of the module because of the complex logic. This doesn't matter, we just need to get more alerts than the threshold.
+
+        query_payloads = [
+            "%09",
+            "%20",
+            "%23",
+            "%2e",
+            "%2f",
+            ".",
+            "?",
+            ";",
+            "..;",
+            ";%09",
+            ";%09..",
+            ";%09..;",
+            ";%2f..",
+            "*",
+            "/*",
+            "..;/",
+            ";/",
+            "/..;/",
+            "/;/",
+            "/./",
+            "//",
+            "/.",
+            "/?anything",
+            ".php",
+            ".json",
+            ".html",
+        ]
+
+        for qp in query_payloads:
+            expect_args = {"method": "GET", "uri": f"/test{qp}"}
+            module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args)
+
+        module_test.httpserver.no_handler_status_code = 403
+
+    def check(self, module_test, events):
+        findings = [e for e in events if e.type == "FINDING"]
+        assert len(findings) == 1
+        finding = findings[0]
+        assert "403 Bypass MULTIPLE SIGNATURES (exceeded threshold" in finding.data["description"]
+
+
 class TestBypass403_aspnetcookieless(ModuleTestBase):
     targets = ["http://127.0.0.1:8888/admin.aspx"]
     modules_overrides = ["bypass403", "httpx"]