Merge pull request #2910 from blacklanternsecurity/lightfuzz-bugfix

Fix lightfuzz envelope cross-contamination between submodules

Author: liquidsec

bbot/core/helpers/web/envelopes.py

@@ -166,6 +166,36 @@ def get_subparam(self, key=None, recursive=True):
                 data = data[segment]
         return data
 
+    def pack_value(self, value, key=None):
+        """
+        Pack a value through the envelope chain WITHOUT modifying internal state.
+        """
+        if key is None:
+            key = self.selected_subparam
+
+        inner = self.unpacked_data(recursive=False)
+
+        if hasattr(inner, "pack_value"):
+            # Inner is another envelope - delegate down the chain
+            data = inner.pack_value(value, key)
+        elif self.singleton:
+            # At the leaf singleton - use the new value directly
+            data = value
+        else:
+            # At the leaf non-singleton (JSON/XML) - copy the data and substitute
+            import copy
+
+            if key is None:
+                raise ValueError("No subparam selected for non-singleton envelope")
+            data = copy.deepcopy(inner)
+            # In the loop: Traverse all the way down to the parent of the target value (all segments except the last),
+            target = data
+            for segment in key[:-1]:
+                target = target[segment]
+            # Use the final segment to actually assign the value.
+            target[key[-1]] = value
+        return self._pack(data)
+
     def set_subparam(self, key=None, value=None, recursive=True):
         envelope = self
         if recursive:

bbot/modules/lightfuzz/submodules/base.py

@@ -265,13 +265,16 @@ def incoming_probe_value(self, populate_empty=True):
 
     def outgoing_probe_value(self, outgoing_probe_value):
         """
-        Transparently modifies the outgoing probe value (fuzz probe being sent to the target), given any envelopes that may have been identified, so that fuzzing within the envelopes can occur.
+        Transparently packs the outgoing probe value (fuzz probe being sent to the target) through
+        any envelopes that may have been identified, so that fuzzing within the envelopes can occur.
+
+        Uses pack_value() to avoid mutating the envelope's internal state, preventing cross-contamination
+        between submodules that share the same event/envelope object.
         """
         self.debug(f"outgoing_probe_value (before packing): {outgoing_probe_value} / {self.event}")
         envelopes = getattr(self.event, "envelopes", None)
         if envelopes is not None:
-            envelopes.set_subparam(value=outgoing_probe_value)
-            outgoing_probe_value = envelopes.pack()
+            outgoing_probe_value = envelopes.pack_value(outgoing_probe_value)
             self.debug(
                 f"outgoing_probe_value (after packing): {outgoing_probe_value} with envelopes [{envelopes}] / {self.event}"
             )

bbot/test/test_step_1/test_web_envelopes.py

@@ -341,3 +341,77 @@ async def test_web_envelopes():
 
     tiny_base64 = BaseEnvelope.detect("YWJi")
     assert isinstance(tiny_base64, TextEnvelope)
+
+
+async def test_web_envelope_pack_value():
+    """
+    Test pack_value() - encodes a value through the envelope chain without modifying internal state.
+    """
+    import base64
+    import json
+
+    from bbot.core.helpers.web.envelopes import BaseEnvelope
+
+    # Text envelope (singleton, transparent)
+    text_envelope = BaseEnvelope.detect("original_text")
+    assert text_envelope.pack_value("new_text") == "new_text"
+    assert text_envelope.get_subparam() == "original_text"
+
+    # Hex envelope (singleton chain: hex -> text)
+    hex_envelope = BaseEnvelope.detect("706172616d")  # "param" in hex
+    packed = hex_envelope.pack_value("modified")
+    assert packed == "modified".encode().hex()
+    assert hex_envelope.get_subparam() == "param"
+
+    # Base64 envelope (singleton chain: base64 -> text)
+    b64_envelope = BaseEnvelope.detect("cGFyYW0=")  # "param" in base64
+    packed = b64_envelope.pack_value("modified")
+    assert packed == base64.b64encode(b"modified").decode()
+    assert b64_envelope.get_subparam() == "param"
+
+    # Nested hex -> base64 -> text chain
+    nested_envelope = BaseEnvelope.detect("634746795957303d")  # hex(base64("param"))
+    packed = nested_envelope.pack_value("modified")
+    expected = base64.b64encode(b"modified").decode().encode().hex()
+    assert packed == expected
+    assert nested_envelope.get_subparam() == "param"
+
+    # URL envelope (singleton chain: url -> text)
+    url_envelope = BaseEnvelope.detect("a%20b%20c")
+    packed = url_envelope.pack_value("x y z")
+    assert packed == "x%20y%20z"
+    assert url_envelope.get_subparam() == "a b c"
+
+    # JSON inside base64 (non-singleton: base64 -> json) - only the selected subparam is substituted in the output
+    b64_json = BaseEnvelope.detect("eyJwYXJhbTEiOiAidmFsMSIsICJwYXJhbTIiOiB7InBhcmFtMyI6ICJ2YWwzIn19")
+    b64_json.selected_subparam = ["param2", "param3"]
+    packed = b64_json.pack_value("new_val3")
+    decoded_json = json.loads(base64.b64decode(packed).decode())
+    assert decoded_json["param1"] == "val1"
+    assert decoded_json["param2"]["param3"] == "new_val3"
+    assert b64_json.get_subparam() == "val3"
+    assert b64_json.get_subparam(["param1"]) == "val1"
+
+    # Repeated calls do not accumulate - each starts from the original state
+    hex_envelope = BaseEnvelope.detect("706172616d")
+    hex_envelope.pack_value("first_modification")
+    hex_envelope.pack_value("second_modification")
+    hex_envelope.pack_value("third_modification")
+    assert hex_envelope.get_subparam() == "param"
+
+    # Multiple callers sharing the same envelope each produce correct output independently
+    shared_envelope = BaseEnvelope.detect("706172616d")  # "param" in hex
+
+    probe_a = shared_envelope.pack_value("param' OR 1=1--")
+    assert probe_a == "param' OR 1=1--".encode().hex()
+    assert shared_envelope.get_subparam() == "param"
+
+    probe_b = shared_envelope.pack_value("param| echo 1234 |")
+    assert probe_b == "param| echo 1234 |".encode().hex()
+    assert shared_envelope.get_subparam() == "param"
+
+    probe_c = shared_envelope.pack_value("../../etc/passwd")
+    assert probe_c == "../../etc/passwd".encode().hex()
+
+    assert shared_envelope.get_subparam() == "param"
+    assert shared_envelope.pack() == "706172616d"

bbot/test/test_step_2/module_tests/test_module_lightfuzz.py

@@ -1885,3 +1885,29 @@ def check(self, module_test, events):
 
         assert web_parameter_emitted, "WEB_PARAMETER was not emitted"
         assert esi_finding_emitted, "ESI FINDING not emitted"
+
+
+# Envelope state isolation: crypto error detection with all submodules enabled.
+# Crypto runs after sqli/cmdi/xss/path/ssti. Each prior submodule calls outgoing_probe_value()
+# which must not corrupt the envelope state that crypto reads via incoming_probe_value().
+class Test_Lightfuzz_envelope_isolation_crypto(Test_Lightfuzz_crypto_error):
+    config_overrides = {
+        "interactsh_disable": True,
+        "modules": {
+            "lightfuzz": {
+                "enabled_submodules": ["sqli", "cmdi", "xss", "path", "ssti", "crypto", "serial", "esi"],
+            }
+        },
+    }
+
+
+# Envelope state isolation: padding oracle detection with all submodules enabled.
+class Test_Lightfuzz_envelope_isolation_paddingoracle(Test_Lightfuzz_PaddingOracleDetection):
+    config_overrides = {
+        "interactsh_disable": True,
+        "modules": {
+            "lightfuzz": {
+                "enabled_submodules": ["sqli", "cmdi", "xss", "path", "ssti", "crypto", "serial", "esi"],
+            }
+        },
+    }