Require positive confirmation for lightfuzz code-change findings

bbot/core/event/base.py

@@ -958,6 +958,11 @@ def json(self, mode="json"):
         web_spider_distance = getattr(self, "web_spider_distance", None)
         if web_spider_distance is not None:
             j["web_spider_distance"] = web_spider_distance
+        # provenance for events whose evidence is an archived snapshot rather than the live host.
+        # kept out of `data` so it doesn't become part of the event's identity.
+        archive_url = self.archive_url
+        if archive_url:
+            j["archive_url"] = archive_url
         # scope distance
         j["scope_distance"] = self.scope_distance
         # scan
@@ -1959,7 +1964,9 @@ def _pretty_string(self):
             confidence_str = f"[\033[1m{confidence}\033[0m]"
         else:
             confidence_str = f"[{confidence}]"
-        return f"Severity: [{severity}] Confidence: {confidence_str} {description}"
+        # the evidence came out of an archive, so the severity describes a snapshot, not the live host
+        archived_str = "[ARCHIVED] " if self.archive_url else ""
+        return f"{archived_str}Severity: [{severity}] Confidence: {confidence_str} {description}"
 
     def _data_human(self):
         parts = []
@@ -1972,6 +1979,10 @@ def _data_human(self):
         cves = self.data.get("cves", [])
         if cves:
             parts.append(f"[{', '.join(cves)}]")
+        # the evidence came out of an archive, so the severity describes a snapshot, not the live host
+        archive_url = self.archive_url
+        if archive_url:
+            parts.append(f"(archived: {archive_url})")
         return " ".join(parts)
 
 

bbot/core/helpers/diff.py

@@ -326,12 +326,19 @@ async def compare(
         else:
             return (False, diff_reasons, reflection, subject_response)
 
-    def _compare_sync(self, subject_response, subject):
-        """CPU-bound comparison work offloaded from the event loop."""
+    @staticmethod
+    def parse_body(text):
+        """Parse a response body into the structure compare_body() expects: a document tree
+        when the body is well-formed XML, otherwise a list of lines. Passing raw text to
+        compare_body() works but bypasses the ddiff_filters that mask dynamic content."""
         try:
-            subject_json = xmltodict.parse(subject_response.text)
+            return xmltodict.parse(text)
         except ExpatError:
-            subject_json = subject_response.text.split("\n")
+            return text.split("\n")
+
+    def _compare_sync(self, subject_response, subject):
+        """CPU-bound comparison work offloaded from the event loop."""
+        subject_json = self.parse_body(subject_response.text)
 
         diff_reasons = []
 

bbot/modules/lightfuzz/submodules/crypto.py

@@ -41,20 +41,20 @@ def _ascii_xor_score(b):
 
 def _is_structured_id_pair(bytes_a, bytes_b, xored, zero_run):
     """True when the pair looks like structured identifiers (MongoDB ObjectIds,
-    hex timestamps, sequential counters, time-ordered UUIDs) rather than
-    reused-keystream ciphertexts.
+    hex timestamps, sequential counters, record GUIDs) rather than reused-keystream
+    ciphertexts.
 
-    Structured IDs share a long byte prefix (timestamp, process ID, ...) and
-    differ only in a short counter/random suffix.  Their XOR has a long leading-
-    zero run followed by a sparse or random tail -- superficially identical to
-    keystream reuse with shared-prefix plaintexts, but distinguishable by what
-    the tail looks like.
+    Structured IDs share a byte prefix (timestamp, process ID, ...) and differ only
+    in a short counter/random suffix.  Their XOR has a leading-zero run followed by
+    a sparse or random tail -- superficially identical to keystream reuse with
+    shared-prefix plaintexts, but distinguishable by what the tail looks like.
 
     Real many-time-pad: the tail is XOR of diverging ASCII plaintexts -- dense,
     diverse bytes mostly in [0x00, 0x60].
 
-    Structured IDs: the tail is either a small counter delta (sparse zeros with
-    one nonzero byte) or random machine/process bytes (fails ASCII-XOR check).
+    Structured IDs: the tail is a small counter delta (sparse zeros with one nonzero
+    byte), random machine/process bytes (fails the ASCII-XOR check), or a fixed-field
+    template whose matching segments zero out scattered through the tail.
     """
     tail = xored[zero_run:]
 
@@ -68,6 +68,15 @@ def _is_structured_id_pair(bytes_a, bytes_b, xored, zero_run):
     if len(bytes_a) == len(bytes_b) and len(tail) <= 4:
         return True
 
+    # Zeros scattered *through* the tail mean the two values keep re-converging at
+    # fixed offsets -- a shared field template (an instance/table/timestamp segment
+    # in a record GUID). Two ciphertexts under a reused keystream zero out only where
+    # their plaintexts still coincide, which is the leading run, or a shared suffix.
+    # Trailing zeros are stripped first so shared-suffix plaintexts aren't caught here.
+    core = tail.rstrip(b"\x00")
+    if sum(1 for b in core if b == 0) >= 2:
+        return True
+
     # For longer tails: real many-time-pad XOR reveals XOR of diverging ASCII
     # plaintexts -- dense nonzero bytes mostly in [0x00, 0x60].  Random
     # suffixes (UUID v7 random bits, different-process machine bytes) and
@@ -369,11 +378,10 @@ def detect_keystream_reuse(self, probe_value):
                 # At least 2 leading zero bytes OR ≥90% of bytes in ASCII-XOR-ASCII range
                 if zero_run < 2 and ascii_score < 0.9:
                     continue
-                # A leading-zero run alone is the hallmark of structured hex
-                # identifiers (MongoDB ObjectIds, hex timestamps, sequential
-                # counters, time-ordered UUIDs) sharing a byte prefix -- not
-                # keystream reuse.  Filter those out by examining the tail.
-                if zero_run >= 2 and _is_structured_id_pair(bytes_a, bytes_b, xored, zero_run):
+                # Both entry conditions are met just as easily by structured hex identifiers
+                # as by real ciphertexts, so every candidate pair goes through the same
+                # discrimination on what the diverging region actually looks like.
+                if _is_structured_id_pair(bytes_a, bytes_b, xored, zero_run):
                     continue
                 pair_score = (zero_run, ascii_score)
                 if best is None or pair_score > (best[0], best[1]):

bbot/modules/lightfuzz/submodules/serial.py

@@ -207,6 +207,29 @@ def payload_language(payload_name):
         """Extract the language family from a payload name (e.g. 'java_base64_string_error' -> 'java')."""
         return payload_name.split("_")[0]
 
+    @staticmethod
+    def corrupt_payload(payload, encoding):
+        """Return a twin of ``payload`` with the same encoding, length and trailing bytes but a
+        scrambled magic/type header: still parses like the original, deserializes under nothing.
+        Returns None when no distinguishable twin can be built.
+        """
+        if not payload:
+            return None
+        if encoding == "php_raw":
+            # PHP serialized data leads with a single type character
+            return f"z{payload[1:]}" if payload[0] != "z" else f"q{payload[1:]}"
+        try:
+            data = bytes.fromhex(payload) if encoding == "hex" else base64.b64decode(payload)
+        except Exception:
+            return None
+        header_length = min(4, len(data))
+        if not header_length:
+            return None
+        corrupted = bytes((b + 0x55) % 256 for b in data[:header_length]) + data[header_length:]
+        if encoding == "hex":
+            return corrupted.hex().upper() if payload.isupper() else corrupted.hex()
+        return base64.b64encode(corrupted).decode()
+
     async def confirm_baseline(self, control_payload, cookies):
         """Re-send the control payload to confirm the baseline error state is stable (not transient)."""
         confirmation = await self.standard_probe(self.event.data["type"], cookies, control_payload)
@@ -248,13 +271,13 @@ async def fuzz(self):
 
         # Map each payload set to its control payload for baseline confirmation
         payload_sets = [
-            (base64_serialization_payloads, http_compare_base64, control_payload_base64),
-            (hex_serialization_payloads, http_compare_hex, control_payload_hex),
-            (php_raw_serialization_payloads, http_compare_php_raw, control_payload_php_raw),
+            (base64_serialization_payloads, http_compare_base64, control_payload_base64, "base64"),
+            (hex_serialization_payloads, http_compare_hex, control_payload_hex, "hex"),
+            (php_raw_serialization_payloads, http_compare_php_raw, control_payload_php_raw, "php_raw"),
         ]
 
         # Proceed with payload probes
-        for payload_set, payload_baseline, control_payload in payload_sets:
+        for payload_set, payload_baseline, control_payload, encoding in payload_sets:
             for payload_type, payload in payload_set.items():
                 try:
                     matches_baseline, diff_reasons, reflection, response = await self.compare_probe(
@@ -315,6 +338,22 @@ async def fuzz(self):
                         )
                         continue
 
+                    # Deserialization is a claim about the payload's content, so a same-shape twin
+                    # with a scrambled header must not resolve the error too. If it does, the value
+                    # is only being parsed (e.g. as a URL/host), not deserialized.
+                    corrupted_payload = self.corrupt_payload(payload, encoding)
+                    if corrupted_payload is not None:
+                        corrupted_response = await self.standard_probe(
+                            self.event.data["type"], cookies, corrupted_payload
+                        )
+                        corrupted_status = getattr(corrupted_response, "status_code", None)
+                        if corrupted_status == status_code:
+                            self.debug(
+                                f"Corrupted twin of {payload_type} also returned {corrupted_status}, "
+                                "outcome is independent of payload content, skipping"
+                            )
+                            continue
+
                     def get_title(text):
                         soup = self.lightfuzz.helpers.beautifulsoup(text, "html.parser")
                         if soup and soup.title and soup.title.string: