From 6f47bdc3716a840c3d9c46d2e7a6f054865224ea Mon Sep 17 00:00:00 2001 From: liquidsec Date: Sun, 7 Jun 2026 14:31:00 -0400 Subject: [PATCH] sqli: two-stage scaling-delay confirmation for time-based blind detection --- bbot/modules/lightfuzz/submodules/sqli.py | 181 ++++++++++-------- .../module_tests/test_module_lightfuzz.py | 68 ++++++- 2 files changed, 157 insertions(+), 92 deletions(-) diff --git a/bbot/modules/lightfuzz/submodules/sqli.py b/bbot/modules/lightfuzz/submodules/sqli.py index a677f0bbe4..545d54a494 100644 --- a/bbot/modules/lightfuzz/submodules/sqli.py +++ b/bbot/modules/lightfuzz/submodules/sqli.py @@ -1,8 +1,6 @@ from .base import BaseLightfuzz from bbot.errors import HttpCompareError -import statistics - class sqli(BaseLightfuzz): """ @@ -23,8 +21,13 @@ class sqli(BaseLightfuzz): friendly_name = "SQL Injection" - expected_delay = 5 - # These are common error strings that strongly indicate SQL injection + delay_low = 3 + delay_high = 8 + delay_margin = 1.5 + delay_scale_margin = 1.75 + delay_stage1_reps = 3 + delay_stage2_reps = 3 + sqli_error_strings = [ "Unterminated string literal", "Failed to parse string literal", @@ -38,36 +41,16 @@ class sqli(BaseLightfuzz): "string not properly terminated", ] - def evaluate_delay(self, mean_baseline, measured_delay): - """ - Evaluates if a measured delay falls within an expected range, indicating potential SQL injection. - - Parameters: - - mean_baseline (float): The average baseline delay measured from non-injected requests. - - measured_delay (float): The delay measured from a potentially injected request. - - Returns: - - bool: True if the measured delay is within the expected range or exactly twice the expected delay, otherwise False. - - The function checks if the measured delay is within a margin of the expected delay or twice the expected delay, - accounting for cases where the injected statement might be executed twice. - """ - margin = 1.5 - if ( - mean_baseline + self.expected_delay - margin - <= measured_delay - <= mean_baseline + self.expected_delay + margin - ): - return True - # check for exactly twice the delay, in case the statement gets placed in the query twice (a common occurrence) - elif ( - mean_baseline + (self.expected_delay * 2) - margin - <= measured_delay - <= mean_baseline + (self.expected_delay * 2) + margin - ): - return True - else: - return False + DELAY_PROBE_TEMPLATES = [ + "'||pg_sleep({d})--", + "' OR (SELECT TRUE FROM pg_sleep({d})) LIMIT 1-- -", + "1' AND (SLEEP({d})) AND '", + "' OR SLEEP({d}) IS NOT NULL LIMIT 1-- -", + " OR SLEEP({d}) IS NOT NULL LIMIT 1-- -", + "' AND (SELECT 1 FROM DUAL WHERE DBMS_LOCK.SLEEP({d})=0) AND '1'='1", + "'; WAITFOR DELAY '00:00:{d:02d}'--", + "; WAITFOR DELAY '00:00:{d:02d}'--", + ] async def _confirm_code_change(self, probe_value, cookies, initial_status_codes, rounds=2): """Run additional confirmation rounds with fresh baselines to rule out transient server/CDN flaps. @@ -218,27 +201,6 @@ async def fuzz(self): except HttpCompareError as e: self.verbose(f"Encountered HttpCompareError Sending Compare Probe: {e}") - # Time-based blind SQLi payloads across DB families. Each probe is engineered - # to fire its delay exactly once so the measured elapsed time stays close to - # self.expected_delay regardless of table row count. Row-independent variants - # use `IS NOT NULL` (since SLEEP returns 0 — not NULL) combined with `LIMIT 1` - # to force a single match on the first row scanned. - standard_probe_strings = [ - # postgres - f"'||pg_sleep({self.expected_delay})--", - f"' OR (SELECT TRUE FROM pg_sleep({self.expected_delay})) LIMIT 1-- -", - # mysql (row-dependent; fires once when original_value matches a row) - f"1' AND (SLEEP({self.expected_delay})) AND '", - # mysql (row-independent; one SLEEP, exits on first row via LIMIT 1) - f"' OR SLEEP({self.expected_delay}) IS NOT NULL LIMIT 1-- -", - f" OR SLEEP({self.expected_delay}) IS NOT NULL LIMIT 1-- -", - # oracle (DUAL is single-row so DBMS_LOCK.SLEEP fires once) - f"' AND (SELECT 1 FROM DUAL WHERE DBMS_LOCK.SLEEP({self.expected_delay})=0) AND '1'='1", - # mssql (stacked query, fires once) - f"'; WAITFOR DELAY '00:00:{self.expected_delay}'--", - f"; WAITFOR DELAY '00:00:{self.expected_delay}'--", - ] - baseline_1 = await self.standard_probe( self.event.data["type"], cookies, probe_value, additional_params_populate_empty=True ) @@ -246,61 +208,112 @@ async def fuzz(self): self.event.data["type"], cookies, probe_value, additional_params_populate_empty=True ) - # get a baseline from two different probes. We will average them to establish a mean baseline if baseline_1 and baseline_2: baseline_1_delay = baseline_1.elapsed.total_seconds() baseline_2_delay = baseline_2.elapsed.total_seconds() - mean_baseline = statistics.mean([baseline_1_delay, baseline_2_delay]) + base_floor = min(baseline_1_delay, baseline_2_delay) - # CDN cache-miss control: junk value misses the edge cache like our SQL payloads - # would. If its delay lands in the SLEEP() window, the latency is cache-miss, not - # SQL execution — bail to avoid false positives. junk_value = f"{probe_value}{self.lightfuzz.helpers.rand_string(20, numeric_only=True)}" junk_response = await self.standard_probe( self.event.data["type"], cookies, junk_value, additional_params_populate_empty=True ) - if junk_response and self.evaluate_delay(mean_baseline, junk_response.elapsed.total_seconds()): - self.debug( - "Junk control probe matched delay window — CDN cache-miss pattern, aborting time-based tests" - ) - return + if junk_response: + junk_delta = junk_response.elapsed.total_seconds() - base_floor + if any(abs(junk_delta - k * self.delay_low) <= self.delay_margin for k in (1, 2)): + self.debug("Junk control probe matched delay window, aborting time-based tests") + return - for p in standard_probe_strings: - confirmations = 0 - for i in range(0, 3): - # send the probe 3 times, and check if the delay is within the detection threshold + for template in self.DELAY_PROBE_TEMPLATES: + # Stage 1: fast gate at delay_low + low_times = [] + stage1_failed = False + for _ in range(self.delay_stage1_reps): + payload_low = template.format(d=self.delay_low) r = await self.standard_probe( self.event.data["type"], cookies, - f"{probe_value}{p}", + f"{probe_value}{payload_low}", additional_params_populate_empty=True, timeout=20, ) if not r: - self.debug("delay measure request failed") + self.debug("Stage 1 delay probe request failed") + stage1_failed = True break + if r.status_code == 403: + self.debug("Stage 1 probe returned 403, skipping template") + stage1_failed = True + break + low_times.append(r.elapsed.total_seconds()) - d = r.elapsed.total_seconds() - self.debug(f"measured delay: {str(d)}") - if self.evaluate_delay( - mean_baseline, d - ): # decide if the delay is within the detection threshold and constitutes a successful sleep execution - confirmations += 1 - self.debug( - f"{self.event.url}:{self.event.data['name']}:{self.event.data['type']} Increasing confirmations, now: {str(confirmations)} " - ) - else: + if stage1_failed or not low_times: + continue + + f_low = min(low_times) + d_low = f_low - base_floor + + k = None + for candidate_k in (1, 2): + if abs(d_low - candidate_k * self.delay_low) <= self.delay_margin: + k = candidate_k break - if confirmations == 3: + if k is None: + self.debug(f"Stage 1 rejected: d_low={d_low:.2f}s does not match delay_low={self.delay_low}s") + continue + + self.verbose(f"Stage 1 passed {self.event.url}: d_low={d_low:.2f}s, k={k}, proceeding to Stage 2") + + # Stage 2: scaling confirmation at delay_high + high_times = [] + stage2_failed = False + for _ in range(self.delay_stage2_reps): + payload_high = template.format(d=self.delay_high) + r = await self.standard_probe( + self.event.data["type"], + cookies, + f"{probe_value}{payload_high}", + additional_params_populate_empty=True, + timeout=30, + ) + if not r: + self.debug("Stage 2 delay probe request failed") + stage2_failed = True + break + if r.status_code == 403: + self.debug("Stage 2 probe returned 403, skipping template") + stage2_failed = True + break + high_times.append(r.elapsed.total_seconds()) + + if stage2_failed or not high_times: + continue + + f_high = min(high_times) + d_high = f_high - base_floor + + absolute_ok = abs(d_high - k * self.delay_high) <= self.delay_margin + scaling_ok = abs((f_high - f_low) - k * (self.delay_high - self.delay_low)) <= self.delay_scale_margin + + if absolute_ok and scaling_ok: self.results.append( { "name": "Possible Blind SQL Injection", "severity": "HIGH", - "confidence": "LOW", - "description": f"Possible Blind SQL Injection. {self.metadata()} Detection Method: [Delay Probe ({p})]", + "confidence": "MEDIUM", + "description": ( + f"Possible Blind SQL Injection. {self.metadata()} " + f"Detection Method: [Delay Probe " + f"(k={k}, {self.delay_low}s->{d_low:.1f}s, {self.delay_high}s->{d_high:.1f}s)] " + f"Payload: [{payload_high}]" + ), } ) + else: + self.verbose( + f"Stage 2 rejected {self.event.url}: d_high={d_high:.2f}s, " + f"absolute_ok={absolute_ok}, scaling_ok={scaling_ok}" + ) else: self.debug("Could not get baseline for time-delay tests") 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 index a79f6f1eac..1457b3a643 100644 --- a/bbot/test/test_step_2/module_tests/test_module_lightfuzz.py +++ b/bbot/test/test_step_2/module_tests/test_module_lightfuzz.py @@ -1529,8 +1529,10 @@ def request_handler(self, request):
""" - if "' AND (SLEEP(5)) AND '" in unquote(value): - sleep(5) + decoded = unquote(value) + m = re.search(r"AND \(SLEEP\((\d+)\)\) AND", decoded) + if m: + sleep(int(m.group(1))) return Response(sql_block, status=200) return Response(parameter_block, status=200) @@ -1543,9 +1545,11 @@ def check(self, module_test, events): web_parameter_emitted = True if e.type == "FINDING": + desc = e.data["description"] if ( - "Possible Blind SQL Injection. Parameter: [search] Parameter Type: [GETPARAM] Detection Method: [Delay Probe (1' AND (SLEEP(5)) AND ')]" - in e.data["description"] + "Possible Blind SQL Injection" in desc + and "Delay Probe" in desc + and "1' AND (SLEEP(8)) AND '" in desc ): sqldelay_finding_emitted = True @@ -1590,8 +1594,9 @@ def request_handler(self, request): # Only the one-shot row-independent MySQL payload triggers a delay. # The original AND-based mysql probe does not fire here, simulating # a context where the injected value does not match any row. - if "OR SLEEP(5) IS NOT NULL LIMIT 1-- -" in decoded: - sleep(5) + m = re.search(r"OR SLEEP\((\d+)\) IS NOT NULL", decoded) + if m: + sleep(int(m.group(1))) return Response(sql_block, status=200) return Response(parameter_block, status=200) @@ -1604,7 +1609,11 @@ def check(self, module_test, events): web_parameter_emitted = True if e.type == "FINDING": desc = e.data["description"] - if "Possible Blind SQL Injection" in desc and "OR SLEEP(5) IS NOT NULL LIMIT 1-- -" in desc: + if ( + "Possible Blind SQL Injection" in desc + and "Delay Probe" in desc + and "OR SLEEP(8) IS NOT NULL LIMIT 1-- -" in desc + ): one_shot_delay_finding = True # Guard against regression of the missing-comma bug: Python string-literal @@ -1618,10 +1627,53 @@ def check(self, module_test, events): assert web_parameter_emitted, "WEB_PARAMETER was not emitted" assert one_shot_delay_finding, ( - "One-shot row-independent SLEEP finding not emitted — row-independent blind sqli detection regression." + "One-shot row-independent SLEEP finding not emitted - row-independent blind sqli detection regression." ) +class Test_Lightfuzz_sqli_delay_jitter_fp(Test_Lightfuzz_sqli): + """Payload-independent jitter must not produce a blind SQLi finding.""" + + _jitter_idx = 0 + + def request_handler(self, request): + from time import sleep + + qs = str(request.query_string.decode()) + parameter_block = """ +
+
+ + +
+
+ """ + if "search=" in qs: + sql_block = """ +
+

0 search results found

+
+
+ """ + jitter = [0.1, 0.4, 0.15, 0.5, 0.2, 0.35, 0.45, 0.1, 0.3, 0.25] + idx = self.__class__._jitter_idx + sleep(jitter[idx % len(jitter)]) + self.__class__._jitter_idx = idx + 1 + return Response(sql_block, status=200) + return Response(parameter_block, status=200) + + async def setup_after_prep(self, module_test): + self.__class__._jitter_idx = 0 + await super().setup_after_prep(module_test) + + def check(self, module_test, events): + for e in events: + if e.type == "FINDING" and "SQL Injection" in e.data.get("description", ""): + raise AssertionError( + f"False positive: finding emitted under jitter-only conditions: {e.data['description']}" + ) + + # Serialization Module (Error Resolution) class Test_Lightfuzz_serial_errorresolution(ModuleTestBase): targets = ["http://127.0.0.1:8888"]