Merge origin/lightfuzz-fp-improvements-6-26 (#3357) into bleeding-edge

# Conflicts:
#	bbot/core/helpers/diff.py

Author: liquidsec

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
@@ -1966,7 +1971,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 = []
@@ -1979,6 +1986,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/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)
@@ -250,13 +273,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(
@@ -317,6 +340,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:

bbot/modules/lightfuzz/submodules/sqli.py

@@ -1,5 +1,8 @@
+from urllib.parse import quote
+
 from .base import BaseLightfuzz
 from bbot.errors import HttpCompareError
+from bbot.core.helpers.diff import parse_body
 
 
 class sqli(BaseLightfuzz):
@@ -13,6 +16,10 @@ class sqli(BaseLightfuzz):
        - Tests quote escape sequence variations
        - Matches against known SQL error patterns
 
+    * Code-change Detection:
+       - Compares the status code of a single-quote probe against a doubled-quote probe
+       - Requires a positive boolean (TRUE/FALSE) content differential to confirm
+
     * Time-based Blind Detection:
        - Uses vendor-specific time delay payloads
        - Confirms delays with statistical analysis
@@ -41,6 +48,14 @@ class sqli(BaseLightfuzz):
         "string not properly terminated",
     ]
 
+    # TRUE/FALSE payload pairs used to confirm the value reaches a SQL query. Both halves of a
+    # pair are the same length and differ by a single character, so a reflected payload can be
+    # stripped cleanly and any surviving body difference comes from the query result set.
+    BOOLEAN_PROBE_PAIRS = [
+        ("' AND '1'='1", "' AND '1'='2"),
+        (" AND 1=1", " AND 1=2"),
+    ]
+
     DELAY_PROBE_TEMPLATES = [
         "'||pg_sleep({d})--",
         "' OR (SELECT TRUE FROM pg_sleep({d})) LIMIT 1-- -",
@@ -115,6 +130,106 @@ async def _confirm_code_change(self, probe_value, cookies, initial_status_codes,
 
         return True
 
+    @staticmethod
+    def _strip_payload(text, payload):
+        """Remove reflected copies of a payload from a response body, in raw and encoded form."""
+        for variant in (payload, quote(payload), payload.replace(" ", "+")):
+            text = text.replace(variant, "")
+        return text
+
+    async def _probe_body(self, http_compare, payload, cookies):
+        """Send ``payload`` and return its parsed response body, reflections of the payload
+        stripped, ready for ``http_compare.compare_body()``.
+
+        Returns None when the probe fails, or when the response is 403/429 (WAF or rate limit,
+        which tells us nothing about the query behind the parameter).
+        """
+        try:
+            probe = await self.compare_probe(
+                http_compare,
+                self.event.data["type"],
+                payload,
+                cookies,
+                additional_params_populate_empty=True,
+            )
+        except HttpCompareError as e:
+            self.debug(f"Boolean probe [{payload}] failed: {e}")
+            return None
+        if not probe[3]:
+            return None
+        if probe[3].status_code in (403, 429):
+            self.debug(f"Boolean probe [{payload}] returned {probe[3].status_code}, cannot confirm")
+            return None
+        return parse_body(self._strip_payload(probe[3].text, payload))
+
+    async def confirm_boolean_differential(self, http_compare, probe_value, cookies):
+        """Require positive SQL-logic evidence before asserting injection from a status change.
+
+        A bare status flip is not SQL: a WAF signature match, or an envelope whose structural
+        validity changes when the payload is repacked, both produce one. Only a TRUE/FALSE pair
+        that changes the response *content* shows the value is reaching a query.
+
+        Returns the confirming ``(true_payload, false_payload)`` pair, or None.
+        """
+        for true_suffix, false_suffix in self.BOOLEAN_PROBE_PAIRS:
+            true_payload = f"{probe_value}{true_suffix}"
+            false_payload = f"{probe_value}{false_suffix}"
+
+            true_body = await self._probe_body(http_compare, true_payload, cookies)
+            if true_body is None:
+                continue
+            false_body = await self._probe_body(http_compare, false_payload, cookies)
+            if false_body is None:
+                continue
+
+            if http_compare.compare_body(true_body, false_body) is not False:
+                self.debug(f"No boolean differential for [{true_suffix}] / [{false_suffix}]")
+                continue
+
+            # A page that renders differently on every request produces a differential on its
+            # own. Re-send the TRUE payload; the body must reproduce for the pair to mean anything.
+            repeat_body = await self._probe_body(http_compare, true_payload, cookies)
+            if repeat_body is None:
+                continue
+            if http_compare.compare_body(true_body, repeat_body) is False:
+                self.debug("Response body is not deterministic, discarding boolean differential")
+                continue
+
+            self.verbose(f"Boolean differential confirmed for {self.event.url}: [{true_suffix}] vs [{false_suffix}]")
+            return true_payload, false_payload
+        return None
+
+    async def is_quote_specific(self, http_compare, probe_value, cookies, status_codes):
+        """Verify the status flip tracks the quote characters and not the payload's shape.
+
+        Appending one vs. two benign characters mirrors the `'`/`''` pair in length while
+        carrying no SQL meaning. If that benign pair reproduces the same status triplet, the
+        flip tracks value length or envelope validity rather than quoting.
+        """
+        control_codes = []
+        for suffix in ("a", "aa"):
+            try:
+                control = await self.compare_probe(
+                    http_compare,
+                    self.event.data["type"],
+                    f"{probe_value}{suffix}",
+                    cookies,
+                    additional_params_populate_empty=True,
+                )
+            except HttpCompareError as e:
+                self.debug(f"Quote-specificity control probe failed: {e}")
+                return True
+            if not control[3]:
+                return True
+            control_codes.append(control[3].status_code)
+
+        if (status_codes[0], *control_codes) == status_codes:
+            self.debug(
+                f"Benign control pair reproduced the status triplet {status_codes}, the change is not quote-specific"
+            )
+            return False
+        return True
+
     async def fuzz(self):
         cookies = self.event.data.get("assigned_cookies", {})
         probe_value = self.incoming_probe_value(populate_empty=True)
@@ -165,19 +280,14 @@ async def fuzz(self):
                     if "code" in single_quote[1] and (
                         single_quote[3].status_code != double_single_quote[3].status_code
                     ):
-                        # Check if the status code change is due to a WAF, not SQL injection
-                        is_waf = False
-                        if single_quote[3].status_code == 403:
-                            waf_matches = await self.lightfuzz.helpers.yara.match(
-                                self.lightfuzz.waf_yara_rules, single_quote[3].text
+                        # A transition into 403 is access control (usually a WAF matching its managed
+                        # SQLi signature on the bare quote), not a code change driven by the query.
+                        if 403 in (single_quote[3].status_code, double_single_quote[3].status_code):
+                            self.debug(
+                                "Quote probe transitioned into 403 (access control/WAF), "
+                                "suppressing SQL injection finding"
                             )
-                            if waf_matches:
-                                self.debug(
-                                    "Single quote probe returned 403 with WAF signature, "
-                                    "suppressing SQL injection finding"
-                                )
-                                is_waf = True
-                        if not is_waf:
+                        else:
                             # Confirmation loop: require 2 additional rounds with fresh baselines
                             # to confirm the status-code triplet is stable and not a transient CDN/server flap.
                             # TODO: apply this same confirmation pattern to other submodules that use compare_probe-based detection.
@@ -187,15 +297,28 @@ async def fuzz(self):
                                 double_single_quote[3].status_code,
                             )
                             confirmed = await self._confirm_code_change(probe_value, cookies, initial_status_codes)
-                            if confirmed:
+                            quote_specific = confirmed and await self.is_quote_specific(
+                                http_compare, probe_value, cookies, initial_status_codes
+                            )
+                            boolean_pair = (
+                                await self.confirm_boolean_differential(http_compare, probe_value, cookies)
+                                if quote_specific
+                                else None
+                            )
+                            if boolean_pair:
                                 self.results.append(
                                     {
                                         "name": "Possible SQL Injection",
                                         "severity": "HIGH",
                                         "confidence": "MEDIUM",
-                                        "description": f"Possible SQL Injection. {self.metadata()} Detection Method: [Single Quote/Two Single Quote, Code Change ({initial_status_codes[0]}->{initial_status_codes[1]}->{initial_status_codes[2]})]",
+                                        "description": f"Possible SQL Injection. {self.metadata()} Detection Method: [Single Quote/Two Single Quote, Code Change ({initial_status_codes[0]}->{initial_status_codes[1]}->{initial_status_codes[2]})] Boolean Confirmation: [{boolean_pair[0]}] vs [{boolean_pair[1]}]",
                                     }
                                 )
+                            elif confirmed and quote_specific:
+                                self.verbose(
+                                    f"Discarding code change {initial_status_codes} for {self.event.url}: "
+                                    "no boolean differential, the value does not reach a query"
+                                )
             else:
                 self.debug("Failed to get responses for both single_quote and double_single_quote")
         except HttpCompareError as e: