Merge pull request #2433 from blacklanternsecurity/lightfuzz-fp-tweaks

Excavate unhandled error / Lightfuzz FP Fix

Author: liquidsec

bbot/modules/internal/excavate.py

@@ -5,7 +5,7 @@
 import inspect
 import regex as re
 from pathlib import Path
-from bbot.errors import ExcavateError
+from bbot.errors import ExcavateError, ValidationError
 import bbot.core.helpers.regexes as bbot_regexes
 from bbot.modules.base import BaseInterceptModule
 from bbot.modules.internal.base import BaseInternalModule
@@ -622,14 +622,15 @@ async def process(self, yara_results, event, yara_rule_settings, discovery_conte
                                 base_url += f"?{event.parsed_url.query}"
                             url = urljoin(base_url, endpoint)
 
+                        try:
+                            # Validate the URL before using it
+                            parsed_url = self.excavate.helpers.validators.validate_url_parsed(url)
+                        except (ValidationError, ValueError) as e:
+                            self.excavate.debug(f"Invalid URL [{url}]: {e}")
+                            continue
+
                         if self.excavate.helpers.validate_parameter(parameter_name, parameter_type):
                             if self.excavate.in_bl(parameter_name) is False:
-                                parsed_url = urlparse(url)
-                                if not parsed_url.hostname:
-                                    self.excavate.warning(
-                                        f"Error Parsing reconstructed URL [{url}] during parameter extraction, missing hostname"
-                                    )
-                                    continue
                                 description = f"HTTP Extracted Parameter [{parameter_name}] ({parameterExtractorSubModule.name} Submodule)"
                                 data = {
                                     "host": parsed_url.hostname,
@@ -848,45 +849,51 @@ async def process(self, yara_results, event, yara_rule_settings, discovery_conte
                 urls_found = 0
                 final_url = ""
                 for url_str in results:
-                    if identifier == "url_full":
-                        if not await self.helpers.re.search(self.full_url_regex, url_str):
+                    try:
+                        if identifier == "url_full":
+                            if not await self.helpers.re.search(self.full_url_regex, url_str):
+                                self.excavate.debug(
+                                    f"Rejecting potential full URL [{url_str}] as did not match full_url_regex"
+                                )
+                                continue
+                            final_url = url_str
+                            self.excavate.debug(f"Discovered Full URL [{final_url}]")
+
+                        elif identifier == "url_attr" and hasattr(event, "parsed_url"):
+                            m = await self.helpers.re.search(self.tag_attribute_regex, url_str)
+                            if not m:
+                                self.excavate.debug(
+                                    f"Rejecting potential attribute URL [{url_str}] as did not match tag_attribute_regex"
+                                )
+                                continue
+                            unescaped_url = html.unescape(m.group(1))
+                            source_url = event.parsed_url.geturl()
+                            final_url = urldefrag(urljoin(source_url, unescaped_url)).url
+                            if not await self.helpers.re.search(self.full_url_regex_strict, final_url):
+                                self.excavate.debug(
+                                    f"Rejecting reconstructed URL [{final_url}] as did not match full_url_regex_strict"
+                                )
+                                continue
                             self.excavate.debug(
-                                f"Rejecting potential full URL [{url_str}] as did not match full_url_regex"
+                                f"Reconstructed Full URL [{final_url}] from extracted relative URL [{unescaped_url}] "
                             )
-                            continue
-                        final_url = url_str
 
-                        self.excavate.debug(f"Discovered Full URL [{final_url}]")
-                    elif identifier == "url_attr" and hasattr(event, "parsed_url"):
-                        m = await self.helpers.re.search(self.tag_attribute_regex, url_str)
-                        if not m:
-                            self.excavate.debug(
-                                f"Rejecting potential attribute URL [{url_str}] as did not match tag_attribute_regex"
+                        if final_url:
+                            # Validate the URL before using it
+                            self.excavate.helpers.validators.validate_url_parsed(final_url)
+                            if self.excavate.scan.in_scope(final_url):
+                                urls_found += 1
+                            await self.report(
+                                final_url,
+                                event,
+                                yara_rule_settings,
+                                discovery_context,
+                                event_type="URL_UNVERIFIED",
+                                urls_found=urls_found,
                             )
-                            continue
-                        unescaped_url = html.unescape(m.group(1))
-                        source_url = event.parsed_url.geturl()
-                        final_url = urldefrag(urljoin(source_url, unescaped_url)).url
-                        if not await self.helpers.re.search(self.full_url_regex_strict, final_url):
-                            self.excavate.debug(
-                                f"Rejecting reconstructed URL [{final_url}] as did not match full_url_regex_strict"
-                            )
-                            continue
-                        self.excavate.debug(
-                            f"Reconstructed Full URL [{final_url}] from extracted relative URL [{unescaped_url}] "
-                        )
-
-                    if final_url:
-                        if self.excavate.scan.in_scope(final_url):
-                            urls_found += 1
-                        await self.report(
-                            final_url,
-                            event,
-                            yara_rule_settings,
-                            discovery_context,
-                            event_type="URL_UNVERIFIED",
-                            urls_found=urls_found,
-                        )
+                    except (ValidationError, ValueError) as e:
+                        self.excavate.debug(f"Invalid URL [{url_str if not final_url else final_url}]: {e}")
+                        continue
 
         async def report_prep(self, event_data, event_type, event, tags, **kwargs):
             event_draft = self.excavate.make_event(event_data, event_type, parent=event)
@@ -1114,7 +1121,10 @@ async def search(self, data, event, content_type, discovery_context="HTTP respon
 
                 # Check if rule processing function exists
                 if rule_name in self.yara_preprocess_dict:
-                    await self.yara_preprocess_dict[rule_name](result, event, discovery_context)
+                    try:
+                        await self.yara_preprocess_dict[rule_name](result, event, discovery_context)
+                    except ValidationError as e:
+                        self.debug(f"ValidationError in rule {rule_name} for result {result}: {e}")
                 else:
                     self.hugewarning(f"YARA Rule {rule_name} not found in pre-compiled rules")
 

bbot/modules/lightfuzz/submodules/serial.py

@@ -22,7 +22,7 @@ class serial(BaseLightfuzz):
     CONTROL_PAYLOAD_PHP_RAW = "z:0:{}"
 
     BASE64_SERIALIZATION_PAYLOADS = {
-        "php_base64": "YTowOnt9",
+        "php_base64": "YToxOntpOjA7aToxO30=",
         "java_base64": "rO0ABXNyABFqYXZhLmxhbmcuQm9vbGVhbs0gcoDVnPruAgABWgAFdmFsdWV4cAA=",
         "java_base64_string_error": "rO0ABXQABHRlc3Q=",
         "java_base64_OptionalDataException": "rO0ABXcEAAAAAAEAAAABc3IAEGphdmEudXRpbC5IYXNoTWFwAAAAAAAAAAECAAJMAARrZXkxYgABAAAAAAAAAAJ4cHcBAAAAB3QABHRlc3Q=",

bbot/test/test_step_2/module_tests/test_module_excavate.py

@@ -1418,3 +1418,17 @@ def check(self, module_test, events):
 
         url_events = [e for e in events if e.type == "URL_UNVERIFIED"]
         assert sorted([e.data for e in url_events]) == sorted(["https://ssl/", "http://127.0.0.1:8888/"])
+
+
+class TestExcavateURL_InvalidPort(TestExcavate):
+    modules_overrides = ["excavate", "httpx", "hunt"]
+
+    async def setup_before_prep(self, module_test):
+        # Test URL with invalid port (greater than 65535)
+        module_test.httpserver.expect_request("/").respond_with_data(
+            '<div><img loading="lazy" src="https://asdffoo.test.notreal:9212952841/whatever.jpg" width="576" height="382" alt="...." /></div>'
+        )
+
+    def check(self, module_test, events):
+        # Verify we got the hostname
+        assert any(e.data == "asdffoo.test.notreal" for e in events)