Merge pull request #1832 from blacklanternsecurity/lightfuzz-envelopes

Adding lightfuzz envelopes system

bbot/core/helpers/misc.py

@@ -11,6 +11,7 @@
 import ahocorasick
 import regex as re
 import subprocess as sp
+
 from pathlib import Path
 from contextlib import suppress
 from unidecode import unidecode  # noqa F401

bbot/core/helpers/regexes.py

@@ -116,8 +116,12 @@
 
 # For use with excavate paramaters extractor
 input_tag_regex = re.compile(
-    r"<input[^>]+?name=[\"\']?([\.$\w]+)[\"\']?(?:[^>]*?value=[\"\']([=+\/\w]*)[\"\'])?[^>]*>"
+    r"<input[^>]*?name=[\"\']?([\._=+\/\w]+)[\"\']?[^>]*?value=[\"\']?([\._=+\/\w]*)[\"\']?[^>]*?>"
 )
+input_tag_regex2 = re.compile(
+    r"<input[^>]*?value=[\"\']?([\._=+\/\w]*)[\"\']?[^>]*?name=[\"\']?([\._=+\/\w]+)[\"\']?[^>]*?>"
+)
+input_tag_novalue_regex = re.compile(r"<input(?![^>]*\bvalue=)[^>]*?name=[\"\']?([\._=+\/\w]*)[\"\']?[^>]*?>")
 # jquery_get_regex = re.compile(r"url:\s?[\"\'].+?\?(\w+)=")
 # jquery_get_regex = re.compile(r"\$.get\([\'\"].+[\'\"].+\{(.+)\}")
 # jquery_post_regex = re.compile(r"\$.post\([\'\"].+[\'\"].+\{(.+)\}")

bbot/modules/internal/excavate.py

@@ -329,6 +329,8 @@ class excavateTestRule(ExcavateRule):
             "ASP.NET_SessionId",
             "JSESSIONID",
             "PHPSESSID",
+            "AWSALB",
+            "AWSALBCORS",
         ]
     )
 
@@ -432,11 +434,13 @@ def extract(self):
         class GetForm(ParameterExtractorRule):
             name = "GET Form"
             discovery_regex = r'/<form[^>]*\bmethod=["\']?get["\']?[^>]*>.*<\/form>/s nocase'
-            form_content_regexes = [
-                bbot_regexes.input_tag_regex,
-                bbot_regexes.select_tag_regex,
-                bbot_regexes.textarea_tag_regex,
-            ]
+            form_content_regexes = {
+                "input_tag_regex": bbot_regexes.input_tag_regex,
+                "input_tag_regex2": bbot_regexes.input_tag_regex2,
+                "input_tag_novalue_regex": bbot_regexes.input_tag_novalue_regex,
+                "select_tag_regex": bbot_regexes.select_tag_regex,
+                "textarea_tag_regex": bbot_regexes.textarea_tag_regex,
+            }
             extraction_regex = bbot_regexes.get_form_regex
             output_type = "GETPARAM"
 
@@ -448,16 +452,24 @@ def extract(self):
                         form_action = form_action.lstrip(".")
 
                     form_parameters = {}
-                    for form_content_regex in self.form_content_regexes:
+                    for form_content_regex_name, form_content_regex in self.form_content_regexes.items():
                         input_tags = form_content_regex.findall(form_content)
+                        if input_tags:
 
-                        for parameter_name, original_value in input_tags:
-                            form_parameters[parameter_name] = original_value.strip()
+                            if form_content_regex_name == "input_tag_novalue_regex":
+                                form_parameters[input_tags[0]] = None
 
-                        for parameter_name, original_value in form_parameters.items():
-                            yield self.output_type, parameter_name, original_value, form_action, _exclude_key(
-                                form_parameters, parameter_name
-                            )
+                            else:
+                                if form_content_regex_name == "input_tag_regex2":
+                                    input_tags = input_tags = [(b, a) for a, b in input_tags]
+
+                                for parameter_name, original_value in input_tags:
+                                    form_parameters[parameter_name] = original_value.strip()
+
+                            for parameter_name, original_value in form_parameters.items():
+                                yield self.output_type, parameter_name, original_value, form_action, _exclude_key(
+                                    form_parameters, parameter_name
+                                )
 
         class GetForm2(GetForm):
             extraction_regex = bbot_regexes.get_form_regex2
@@ -514,12 +526,18 @@ async def process(self, yara_results, event, yara_rule_settings, discovery_conte
                             self.excavate.debug(
                                 f"Found Parameter [{parameter_name}] in [{parameterExtractorSubModule.name}] ParameterExtractor Submodule"
                             )
-                            endpoint = event.data["url"] if not endpoint else endpoint
-                            url = (
-                                endpoint
-                                if endpoint.startswith(("http://", "https://"))
-                                else f"{event.parsed_url.scheme}://{event.parsed_url.netloc}{endpoint}"
-                            )
+                            # If we have a full URL, leave it as-is
+                            if not endpoint.startswith(("http://", "https://")):
+
+                                # The endpoint is usually a form action - we should use it if we have it. If not, defautl to URL.
+                                path = event.parsed_url.path if not endpoint else endpoint
+                                # Normalize path by remove leading slash
+                                path = path.lstrip("/")
+
+                                # Ensure the base URL has a single slash between path and endpoint
+                                url = f"{event.parsed_url.scheme}://{event.parsed_url.netloc}/{path}"
+                            else:
+                                url = endpoint
 
                             if self.excavate.helpers.validate_parameter(parameter_name, parameter_type):
 
@@ -1003,8 +1021,8 @@ async def handle_event(self, event):
                             self.debug(f"Cookie found without '=': {header_value}")
                             continue
                         else:
-                            cookie_name = header_value.split("=")[0]
-                            cookie_value = header_value.split("=")[1].split(";")[0]
+                            cookie_name, _, remainder = header_value.partition("=")
+                            cookie_value = remainder.split(";")[0]
 
                             if self.in_bl(cookie_name) == False:
                                 self.assigned_cookies[cookie_name] = cookie_value

bbot/modules/lightfuzz.py

@@ -128,6 +128,16 @@ async def run_submodule(self, submodule, event):
         if len(submodule_instance.results) > 0:
             for r in submodule_instance.results:
                 event_data = {"host": str(event.host), "url": event.data["url"], "description": r["description"]}
+
+                envelopes = getattr(event, "envelopes", None)
+                if envelopes and envelopes.envelopes:
+                    envelope_summary = f'[{"->".join(envelopes.envelopes)}]'
+                    if envelopes.end_format_type:
+                        envelope_summary += f" Format: [{envelopes.end_format_type}] with subparameter [{envelopes.end_format_subparameter}])"
+
+                    # Append the envelope summary to the description
+                    event_data["description"] += f" Envelopes: {envelope_summary}"
+
                 if r["type"] == "VULNERABILITY":
                     event_data["severity"] = r["severity"]
                 await self.emit_event(

bbot/modules/lightfuzz_submodules/base.py

@@ -19,6 +19,7 @@ def additional_params_process(self, additional_params, additional_params_populat
         return new_additional_params
 
     async def send_probe(self, probe):
+        probe = self.probe_value_outgoing(probe)
         getparams = {self.event.data["name"]: probe}
         url = self.lightfuzz.helpers.add_get_params(self.event.data["url"], getparams, encode=False).geturl()
         self.lightfuzz.debug(f"lightfuzz sending probe with URL: {url}")
@@ -29,7 +30,7 @@ async def send_probe(self, probe):
     def compare_baseline(
         self, event_type, probe, cookies, additional_params_populate_empty=False, speculative_mode="GETPARAM"
     ):
-
+        probe = self.probe_value_outgoing(probe)
         http_compare = None
 
         if event_type == "SPECULATIVE":
@@ -55,7 +56,7 @@ def compare_baseline(
                 self.event.data["url"], include_cache_buster=False, headers=headers, cookies=cookies
             )
         elif event_type == "POSTPARAM":
-            data = {self.event.data["name"]: f"{probe}"}
+            data = {self.probe_value_outgoing(self.event.data["name"]): f"{probe}"}
             if self.event.data["additional_params"] is not None:
                 data.update(
                     self.additional_params_process(
@@ -67,6 +68,21 @@ def compare_baseline(
             )
         return http_compare
 
+    async def baseline_probe(self, cookies):
+        if self.event.data.get("eventtype") == "POSTPARAM":
+            method = "POST"
+        else:
+            method = "GET"
+
+        return await self.lightfuzz.helpers.request(
+            method=method,
+            cookies=cookies,
+            url=self.event.data.get("url"),
+            allow_redirects=False,
+            retries=1,
+            timeout=10,
+        )
+
     async def compare_probe(
         self,
         http_compare,
@@ -77,6 +93,8 @@ async def compare_probe(
         additional_params_override={},
         speculative_mode="GETPARAM",
     ):
+
+        probe = self.probe_value_outgoing(probe)
         additional_params = copy.deepcopy(self.event.data.get("additional_params", {}))
         if additional_params_override:
             for k, v in additional_params_override.items():
@@ -109,34 +127,36 @@ async def standard_probe(
         self,
         event_type,
         cookies,
-        probe_value,
+        probe,
         timeout=10,
         additional_params_populate_empty=False,
         speculative_mode="GETPARAM",
     ):
 
+        probe = self.probe_value_outgoing(probe)
+
         if event_type == "SPECULATIVE":
             event_type = speculative_mode
 
         method = "GET"
         if event_type == "GETPARAM":
-            url = f"{self.event.data['url']}?{self.event.data['name']}={probe_value}"
+            url = f"{self.event.data['url']}?{self.event.data['name']}={probe}"
             if "additional_params" in self.event.data.keys() and self.event.data["additional_params"] is not None:
                 url = self.lightfuzz.helpers.add_get_params(
                     url, self.event.data["additional_params"], encode=False
                 ).geturl()
         else:
             url = self.event.data["url"]
         if event_type == "COOKIE":
-            cookies_probe = {self.event.data["name"]: probe_value}
+            cookies_probe = {self.event.data["name"]: probe}
             cookies = {**cookies, **cookies_probe}
         if event_type == "HEADER":
-            headers = {self.event.data["name"]: probe_value}
+            headers = {self.event.data["name"]: probe}
         else:
             headers = {}
         if event_type == "POSTPARAM":
             method = "POST"
-            data = {self.event.data["name"]: probe_value}
+            data = {self.event.data["name"]: probe}
             if self.event.data["additional_params"] is not None:
                 data.update(
                     self.additional_params_process(
@@ -166,10 +186,20 @@ def metadata(self):
             )
         return metadata_string
 
-    def probe_value(self, populate_empty=True):
+    def probe_value_incoming(self, populate_empty=True):
 
         probe_value = str(self.event.data.get("original_value", ""))
         if (probe_value is None or len(probe_value) == 0) and populate_empty == True:
             probe_value = self.lightfuzz.helpers.rand_string(8, numeric_only=True)
-
+        self.lightfuzz.debug(f"probe_value_incoming (before modification): {probe_value}")
+        envelopes_instance = getattr(self.event, "envelopes", None)
+        probe_value = envelopes_instance.remove_envelopes(probe_value)
+        self.lightfuzz.debug(f"probe_value_incoming (after modification): {probe_value}")
         return probe_value
+
+    def probe_value_outgoing(self, outgoing_probe_value):
+        self.lightfuzz.debug(f"probe_value_outgoing (before modification): {outgoing_probe_value}")
+        envelopes_instance = getattr(self.event, "envelopes", None)
+        outgoing_probe_value = envelopes_instance.add_envelopes(outgoing_probe_value)
+        self.lightfuzz.debug(f"probe_value_outgoing (after modification): {outgoing_probe_value}")
+        return outgoing_probe_value

bbot/modules/lightfuzz_submodules/cmdi.py

@@ -7,9 +7,8 @@
 class CmdILightfuzz(BaseLightfuzz):
 
     async def fuzz(self):
-
         cookies = self.event.data.get("assigned_cookies", {})
-        probe_value = self.probe_value()
+        probe_value = self.probe_value_incoming()
 
         canary = self.lightfuzz.helpers.rand_string(8, numeric_only=True)
         http_compare = self.compare_baseline(self.event.data["type"], probe_value, cookies)

bbot/modules/lightfuzz_submodules/crypto.py

@@ -194,7 +194,7 @@ async def padding_oracle(self, probe_value, cookies):
                 }
             )
 
-    async def error_string_search(self, text_dict):
+    async def error_string_search(self, text_dict, baseline_text):
 
         matching_techniques = set()
         matching_strings = set()
@@ -206,12 +206,18 @@ async def error_string_search(self, text_dict):
             matching_techniques.add(label)
         context = f"Lightfuzz Cryptographic Probe Submodule detected a cryptographic error after manipulating parameter: [{self.event.data['name']}]"
         if len(matching_strings) > 0:
-            self.results.append(
-                {
-                    "type": "FINDING",
-                    "description": f"Possible Cryptographic Error. {self.metadata()} Strings: [{','.join(matching_strings)}] Detection Technique(s): [{','.join(matching_techniques)}]",
-                    "context": context,
-                }
+            false_positive_check = self.lightfuzz.helpers.string_scan(self.crypto_error_strings, baseline_text)
+            false_positive_matches = set(matched_strings) & set(false_positive_check)
+            if not false_positive_matches:
+                self.results.append(
+                    {
+                        "type": "FINDING",
+                        "description": f"Possible Cryptographic Error. {self.metadata()} Strings: [{','.join(matching_strings)}] Detection Technique(s): [{','.join(matching_techniques)}]",
+                        "context": context,
+                    }
+                )
+            self.lightfuzz.debug(
+                f"Aborting cryptographic error reporting - baseline_text already contained detected string(s) ({','.join(false_positive_check)})"
             )
 
     @staticmethod
@@ -229,13 +235,20 @@ def identify_hash_function(hash_bytes):
             return hash_functions[hash_length]
 
     async def fuzz(self):
+
         cookies = self.event.data.get("assigned_cookies", {})
-        probe_value = self.probe_value(populate_empty=False)
+        probe_value = self.probe_value_incoming(populate_empty=False)
         if not probe_value:
             self.lightfuzz.debug(
                 f"The Cryptography Probe Submodule requires original value, aborting [{self.event.data['type']}] [{self.event.data['name']}]"
             )
             return
+
+        baseline_probe = await self.baseline_probe(cookies)
+        if not baseline_probe:
+            self.lightfuzz.warning(f"Couldn't get baseline_probe for url {self.event.data['url']}, aborting")
+            return
+
         try:
             truncate_probe_value = self.modify_string(probe_value, action="truncate")
             mutate_probe_value = self.modify_string(probe_value, action="mutate")
@@ -290,7 +303,7 @@ async def fuzz(self):
 
         # Cryptographic Error String Test
         await self.error_string_search(
-            {"truncate value": truncate_probe[3].text, "mutate value": mutate_probe[3].text}
+            {"truncate value": truncate_probe[3].text, "mutate value": mutate_probe[3].text}, baseline_probe.text
         )
 
         if confirmed_techniques or (

bbot/modules/lightfuzz_submodules/path.py

@@ -9,7 +9,7 @@ class PathTraversalLightfuzz(BaseLightfuzz):
 
     async def fuzz(self):
         cookies = self.event.data.get("assigned_cookies", {})
-        probe_value = self.probe_value(populate_empty=False)
+        probe_value = self.probe_value_incoming(populate_empty=False)
         if not probe_value:
             self.lightfuzz.debug(
                 f"Path Traversal detection requires original value, aborting [{self.event.data['type']}] [{self.event.data['name']}]"

bbot/modules/lightfuzz_submodules/serial.py

@@ -4,7 +4,6 @@
 
 class SerialLightfuzz(BaseLightfuzz):
     async def fuzz(self):
-
         cookies = self.event.data.get("assigned_cookies", {})
         control_payload = "DEADBEEFCAFEBABE1234567890ABCDEF"
         serialization_payloads = {
@@ -27,7 +26,7 @@ async def fuzz(self):
             "java.io.optionaldataexception",
         ]
 
-        probe_value = self.probe_value(populate_empty=False)
+        probe_value = self.probe_value_incoming(populate_empty=False)
         if probe_value:
             self.lightfuzz.debug(
                 f"The Serialization Submodule only operates when there if no original value, aborting [{self.event.data['type']}] [{self.event.data['name']}]"

bbot/modules/lightfuzz_submodules/sqli.py

@@ -1,7 +1,6 @@
 from .base import BaseLightfuzz
 from bbot.errors import HttpCompareError
 
-import urllib.parse
 import statistics
 
 
@@ -29,14 +28,7 @@ def evaluate_delay(self, mean_baseline, measured_delay):
     async def fuzz(self):
 
         cookies = self.event.data.get("assigned_cookies", {})
-
-        # custom probe_value generation
-        if "original_value" in self.event.data and self.event.data["original_value"] is not None:
-            probe_value = urllib.parse.quote(str(self.event.data["original_value"]), safe="")
-
-        else:
-            probe_value = self.lightfuzz.helpers.rand_string(8, numeric_only=True)
-
+        probe_value = self.probe_value_incoming(populate_empty=True)
         http_compare = self.compare_baseline(
             self.event.data["type"], probe_value, cookies, additional_params_populate_empty=True
         )

bbot/presets/web/lightfuzz-intense.yml

@@ -19,6 +19,6 @@ config:
   modules:
     lightfuzz:
       force_common_headers: False
-      submodules_enabled: [cmdi,crypto,path,serial,sqli,ssti,xss]
+      enabled_submodules: [cmdi,crypto,path,serial,sqli,ssti,xss]
     excavate:
       retain_querystring: True