Fix lightfuzz false positives: crypto jitter stability, XSS context verification, SQLi WAF detection

liquidsec · liquidsec · commit 749a695decd2 · 2026-04-03T15:45:33.000-04:00
- Add endpoint stability pre-check to padding oracle and CBC bitflip tests
- Verify XSS probe matches appear in the correct HTML context
- Suppress SQLi findings when single-quote probe triggers WAF 403
- Blacklist PKCE and Akamai Bot Manager parameters
diff --git a/bbot/defaults.yml b/bbot/defaults.yml
@@ -272,6 +272,14 @@ parameter_blacklist:
   # ASP.NET session/identity cookies
   - .ASPXANONYMOUS
   - .ASPXAUTH
+  # PKCE (Proof Key for Code Exchange)
+  - code_verifier
+  - code_challenge
+  # Akamai Bot Manager
+  - _abck
+  - bm_sz
+  - bm_sv
+  - ak_bmsc
 
 parameter_blacklist_prefixes:
   - TS01
diff --git a/bbot/modules/lightfuzz/submodules/crypto.py b/bbot/modules/lightfuzz/submodules/crypto.py
@@ -262,6 +262,9 @@ async def cbc_bitflip(self, probe_value, cookies):
         data, encoding = self.format_agnostic_decode(probe_value)
         if encoding == "unknown":
             return
+        # Stability pre-check: verify the endpoint returns consistent responses
+        if not await self._check_endpoint_stability(probe_value, encoding, cookies):
+            return
         for block_size in self.possible_block_sizes(len(data)):
             num_blocks = len(data) // block_size
             if num_blocks < 2:
@@ -389,12 +392,42 @@ async def padding_oracle_execute(self, original_data, encoding, block_size, cook
             return None
         return False
 
+    async def _check_endpoint_stability(self, probe_value, encoding, cookies):
+        """Send the same probe value multiple times and verify the endpoint returns consistent responses.
+        Returns True if stable, False if responses vary for identical inputs (jitter)."""
+        data, _ = self.format_agnostic_decode(probe_value)
+        if encoding == "unknown":
+            return True
+        # Build a fixed probe to test stability
+        stability_value = self.format_agnostic_encode(b"\x00" * 16 + data[-16:], encoding)
+        stability_hashes = []
+        for _ in range(3):
+            r = await self.standard_probe(self.event.data["type"], cookies, stability_value)
+            if r:
+                body = r.text
+                for encoded in [stability_value, stability_value.replace("+", " "), quote(stability_value)]:
+                    body = body.replace(encoded, "")
+                stability_hashes.append(hash(body))
+        if len(set(stability_hashes)) > 1:
+            self.debug(
+                f"Endpoint produces unstable responses for identical inputs "
+                f"({len(set(stability_hashes))}/{len(stability_hashes)} unique), "
+                f"skipping differential analysis"
+            )
+            return False
+        return True
+
     async def padding_oracle(self, probe_value, cookies):
         data, encoding = self.format_agnostic_decode(probe_value)
         possible_block_sizes = self.possible_block_sizes(
             len(data)
         )  # determine possible block sizes for the ciphertext
 
+        # Stability pre-check: verify the endpoint returns consistent responses
+        # for identical inputs before attempting differential analysis
+        if not await self._check_endpoint_stability(probe_value, encoding, cookies):
+            return
+
         for block_size in possible_block_sizes:
             padding_oracle_result = await self.padding_oracle_execute(data, encoding, block_size, cookies)
             # if we get a negative result first, theres a 1/255 change it's a false negative. To rule that out, we must retry again with possible_first_byte set to false
diff --git a/bbot/modules/lightfuzz/submodules/sqli.py b/bbot/modules/lightfuzz/submodules/sqli.py
@@ -1,5 +1,6 @@
 from .base import BaseLightfuzz
 from bbot.errors import HttpCompareError
+from bbot.core.helpers.misc import get_waf_strings
 
 import statistics
 
@@ -119,14 +120,32 @@ async def fuzz(self):
                     if "code" in single_quote[1] and (
                         single_quote[3].status_code != double_single_quote[3].status_code
                     ):
-                        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 ({http_compare.baseline.status_code}->{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
+                        if single_quote[3].status_code == 403:
+                            waf_detected = any(ws in single_quote[3].text for ws in get_waf_strings())
+                            if waf_detected:
+                                self.debug(
+                                    "Single quote probe returned 403 with WAF signature, "
+                                    "suppressing SQL injection finding"
+                                )
+                            else:
+                                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 ({http_compare.baseline.status_code}->{single_quote[3].status_code}->{double_single_quote[3].status_code})]",
+                                    }
+                                )
+                        else:
+                            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 ({http_compare.baseline.status_code}->{single_quote[3].status_code}->{double_single_quote[3].status_code})]",
+                                }
+                            )
             else:
                 self.debug("Failed to get responses for both single_quote and double_single_quote")
         except HttpCompareError as e:
diff --git a/bbot/modules/lightfuzz/submodules/xss.py b/bbot/modules/lightfuzz/submodules/xss.py
@@ -84,21 +84,62 @@ def is_balanced(section, target_index, quote_char):
         # If we have no matches, the target string is most likely not within quotes
         return "outside"
 
+    def _verify_match_context(self, html, match, context):
+        """Verify the match appears in the correct HTML context, not just anywhere in the response.
+        When the same parameter is reflected in multiple contexts with different encoding,
+        a match found in the wrong context can cause false positives."""
+        if "Tag Attribute" in context:
+            # Verify match is inside a tag (between < and >), not in text content
+            pos = html.find(match)
+            while pos != -1:
+                preceding = html[:pos]
+                last_open = preceding.rfind("<")
+                last_close = preceding.rfind(">")
+                if last_open > last_close:
+                    return True
+                pos = html.find(match, pos + 1)
+            return False
+        elif "Between Tags" in context:
+            pos = html.find(match)
+            while pos != -1:
+                preceding = html[:pos]
+                last_open = preceding.rfind("<")
+                last_close = preceding.rfind(">")
+                if last_close > last_open:
+                    return True
+                pos = html.find(match, pos + 1)
+            return False
+        elif "In Javascript" in context:
+            in_js_regex = re.compile(
+                rf"<script\b[^>]*>[^<]*(?:<(?!\/script>)[^<]*)*{re.escape(match)}"
+                rf"[^<]*(?:<(?!\/script>)[^<]*)*<\/script>"
+            )
+            return bool(in_js_regex.search(html))
+        return True
+
     async def check_probe(self, cookies, probe, match, context):
         # Send the defined probe and look for the expected match value in the response
         probe_result = await self.standard_probe(self.event.data["type"], cookies, probe)
-        if probe_result and match in probe_result.text:
-            self.results.append(
-                {
-                    "name": "Possible Reflected XSS",
-                    "severity": "MEDIUM",
-                    "confidence": "MEDIUM",
-                    "type": "FINDING",
-                    "description": f"Possible Reflected XSS. Parameter: [{self.event.data['name']}] Context: [{context}] Parameter Type: [{self.event.data['type']}]{self.conversion_note()}",
-                }
+        if not probe_result or match not in probe_result.text:
+            return False
+
+        if not self._verify_match_context(probe_result.text, match, context):
+            self.debug(
+                f"Probe match found in response but not in the expected context [{context}]. "
+                f"Likely reflected in a different context with different encoding. Suppressing."
             )
-            return True
-        return False
+            return False
+
+        self.results.append(
+            {
+                "name": "Possible Reflected XSS",
+                "severity": "MEDIUM",
+                "confidence": "MEDIUM",
+                "type": "FINDING",
+                "description": f"Possible Reflected XSS. Parameter: [{self.event.data['name']}] Context: [{context}] Parameter Type: [{self.event.data['type']}]{self.conversion_note()}",
+            }
+        )
+        return True
 
     async def fuzz(self):
         lightfuzz_event = self.event.parent
diff --git a/bbot/test/test_step_2/module_tests/test_module_lightfuzz.py b/bbot/test/test_step_2/module_tests/test_module_lightfuzz.py
@@ -2853,3 +2853,164 @@ def check(self, module_test, events):
         assert sqli_postparam_converted_finding_emitted, (
             "SQLi POSTPARAM (converted from GETPARAM) FINDING not emitted (try_get_as_post failed)"
         )
+
+
+# Padding Oracle Jitter Stability Pre-Check
+class Test_Lightfuzz_PaddingOracleDetection_JitterStability(Test_Lightfuzz_PaddingOracleDetection):
+    """Padding oracle negative test: the endpoint produces different response bodies for identical inputs
+    (e.g. ADFS with embedded timestamps/nonces). The stability pre-check should detect this and skip."""
+
+    jitter_counter = 0
+
+    def request_handler(self, request):
+        encrypted_value = quote(
+            "dplyorsu8VUriMW/8DqVDU6kRwL/FDk3Q+4GXVGZbo0CTh9YX1YvzZZJrYe4cHxvAICyliYtp1im4fWoOa54Zg=="
+        )
+        default_html_response = f"""
+        <html>
+            <body>
+                <form action="/decrypt" method="post">
+                    <input type="hidden" name="encrypted_data" value="{encrypted_value}" />
+                    <button type="submit">Decrypt</button>
+                </form>
+            </body>
+        </html>
+        """
+
+        if "/decrypt" in request.url and request.method == "POST":
+            # Every response is unique, simulating ADFS-style dynamic content
+            Test_Lightfuzz_PaddingOracleDetection_JitterStability.jitter_counter += 1
+            response_content = f"Error correlation_id={Test_Lightfuzz_PaddingOracleDetection_JitterStability.jitter_counter} nonce=abc{Test_Lightfuzz_PaddingOracleDetection_JitterStability.jitter_counter}"
+            return Response(response_content, status=200)
+        else:
+            return Response(default_html_response, status=200)
+
+    def check(self, module_test, events):
+        web_parameter_extracted = False
+        padding_oracle_detected = False
+        for e in events:
+            if e.type == "WEB_PARAMETER":
+                if "HTTP Extracted Parameter [encrypted_data] (POST Form" in e.data["description"]:
+                    web_parameter_extracted = True
+            if e.type == "FINDING":
+                if "Padding Oracle" in e.data["description"]:
+                    padding_oracle_detected = True
+
+        assert web_parameter_extracted, "Web parameter was not extracted"
+        assert not padding_oracle_detected, (
+            "Padding oracle should NOT be detected when endpoint has jittery responses (stability pre-check should abort)"
+        )
+
+
+# XSS Multi-Context Reflection False Positive
+class Test_Lightfuzz_xss_multicontext(Test_Lightfuzz_xss):
+    """XSS negative test: parameter reflected in multiple contexts with different encoding.
+    Quote survives in text content but is encoded in tag attribute. Should NOT report Tag Attribute XSS."""
+
+    def request_handler(self, request):
+        qs = str(request.query_string.decode())
+
+        parameter_block = """
+        <html>
+            <form action="/" method="GET">
+                <input type="text" name="path" value="default">
+                <button type="submit">Submit</button>
+            </form>
+        </html>
+        """
+        if "path=" in qs:
+            value = qs.split("path=")[1]
+            if "&" in value:
+                value = value.split("&")[0]
+            decoded = unquote(value)
+            # Tag attribute context: quotes are URL-encoded (safe)
+            attr_value = decoded.replace('"', "%22")
+            # Text content: raw reflection (quotes survive but harmless here)
+            text_value = decoded
+            # JS context: everything URL-encoded (safe)
+            js_value = value
+
+            html = f"""
+<html>
+<form action="/page?path={attr_value}" method="GET">
+    <input type="text" name="path">
+</form>
+<h1>{text_value}</h1>
+<script>if('{js_value}') {{ }}</script>
+</html>
+            """
+            return Response(html, status=200)
+        return Response(parameter_block, status=200)
+
+    async def setup_after_prep(self, module_test):
+        module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA"
+        expect_args = re.compile("/")
+        module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler)
+
+    def check(self, module_test, events):
+        web_parameter_emitted = False
+        tag_attribute_xss_emitted = False
+        for e in events:
+            if e.type == "WEB_PARAMETER":
+                if "HTTP Extracted Parameter [path]" in e.data["description"]:
+                    web_parameter_emitted = True
+            if e.type == "FINDING":
+                if "Possible Reflected XSS" in e.data["description"] and "Tag Attribute" in e.data["description"]:
+                    tag_attribute_xss_emitted = True
+
+        assert web_parameter_emitted, "WEB_PARAMETER was not emitted"
+        assert not tag_attribute_xss_emitted, (
+            "Tag Attribute XSS should NOT be reported when the quote only survives in text content, not in tag attributes"
+        )
+
+
+# SQLi WAF False Positive (Akamai-style 403)
+class Test_Lightfuzz_sqli_waf(Test_Lightfuzz_sqli):
+    """SQLi negative test: endpoint returns 403 with WAF signature when single quote is sent.
+    Should NOT report SQL injection."""
+
+    def request_handler(self, request):
+        qs = str(request.query_string.decode())
+        parameter_block = """
+        <section class=search>
+            <form action=/ method=GET>
+                <input type=text placeholder='Search the blog...' name=search>
+                <button type=submit class=button>Search</button>
+            </form>
+        </section>
+        """
+        if "search=" in qs:
+            value = qs.split("=")[1]
+            if "&" in value:
+                value = value.split("&")[0]
+
+            if value.endswith("'") and not value.endswith("''"):
+                # WAF blocks the request with a known WAF string
+                waf_response = """
+                <html>
+                <head><title>Access Denied</title></head>
+                <body>
+                <h1>Access Denied</h1>
+                <p>The requested URL was rejected. Please consult with your administrator.</p>
+                </body>
+                </html>
+                """
+                return Response(waf_response, status=403)
+            return Response(parameter_block, status=200)
+        return Response(parameter_block, status=200)
+
+    def check(self, module_test, events):
+        web_parameter_emitted = False
+        sqli_finding_emitted = False
+        for e in events:
+            if e.type == "WEB_PARAMETER":
+                if "HTTP Extracted Parameter [search]" in e.data["description"]:
+                    web_parameter_emitted = True
+            if e.type == "FINDING":
+                if "Possible SQL Injection" in e.data["description"] and "Code Change" in e.data["description"]:
+                    sqli_finding_emitted = True
+
+        assert web_parameter_emitted, "WEB_PARAMETER was not emitted"
+        assert not sqli_finding_emitted, (
+            "SQLi should NOT be reported when single quote probe triggers a WAF 403 response"
+        )
