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 = """
+
+ """
+ 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"]