Making boolean inference more robust against jitter

Author: stamparm

lib/core/settings.py

@@ -20,7 +20,7 @@
 from thirdparty import six
 
 # sqlmap version (<major>.<minor>.<month>.<monthly commit>)
-VERSION = "1.10.7.240"
+VERSION = "1.10.7.241"
 TYPE = "dev" if VERSION.count('.') > 2 and VERSION.split('.')[-1] != '0' else "stable"
 TYPE_COLORS = {"dev": 33, "stable": 90, "pip": 34}
 VERSION_STRING = "sqlmap/%s#%s" % ('.'.join(VERSION.split('.')[:-1]) if VERSION.count('.') > 2 and VERSION.split('.')[-1] == '0' else VERSION, TYPE)

lib/techniques/blind/inference.py

@@ -513,11 +513,20 @@ def validateChar(idx, value):
 
             result = not Request.queryPage(forgedPayload, timeBasedCompare=timeBasedCompare, raise404=False)
 
-            if result and timeBasedCompare and getTechniqueData().trueCode:
-                result = threadData.lastCode == getTechniqueData().trueCode
-                if not result:
-                    warnMsg = "detected HTTP code '%s' in validation phase is differing from expected '%s'" % (threadData.lastCode, getTechniqueData().trueCode)
-                    singleTimeWarnMessage(warnMsg)
+            if result and getTechniqueData() is not None:
+                trueCode, falseCode = getTechniqueData().trueCode, getTechniqueData().falseCode
+                if timeBasedCompare:
+                    if trueCode:
+                        result = threadData.lastCode == trueCode
+                        if not result:
+                            warnMsg = "detected HTTP code '%s' in validation phase is differing from expected '%s'" % (threadData.lastCode, trueCode)
+                            singleTimeWarnMessage(warnMsg)
+                # A boolean validation confirmed under an UNEXPECTED HTTP code (a transient 5xx/403/429/..
+                # landing on the validation request itself) is not trustworthy - fail it so the character is
+                # re-extracted, riding out the blip. On a clean target every code is true/false -> no-op.
+                elif threadData.lastCode is not None and any((trueCode, falseCode)) and threadData.lastCode not in (trueCode, falseCode):
+                    result = False
+                    singleTimeWarnMessage("unexpected HTTP code '%s' during validation phase; will re-extract" % threadData.lastCode)
 
             incrementCounter(getTechnique())
 

tests/test_boolean_jitter.py

@@ -0,0 +1,258 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+
+Adversarial "shitty response" JITTER harness for BOOLEAN-based blind extraction.
+
+Boolean-blind decides each bit through the REAL comparison() oracle (--string / --not-string /
+--regexp / --code / page-ratio). In the wild a target throws transient junk between good responses -
+gateway 5xx, WAF/rate-limit pages, a Cloudflare "just a moment" interstitial, a captcha, a
+maintenance banner, a truncated or empty body, an A/B variant, even a page that COINCIDENTALLY
+contains the --string token (a direction-flipping false positive). This drives the REAL bisection() +
+REAL comparison() + REAL validateChar() re-validation against a mock oracle that injects that catalog
+(IID or in bursts) at controllable rates, with NO network, fully deterministic per seed.
+
+The template is PAYLOAD_DELIMITER-wrapped so validateChar's per-char '!=' re-check actually fires
+(the same fidelity trap the time-based harness hit), and the mock sets threadData.lastCode so the
+unexpectedCode -> validateChar defense engages exactly as in a live run.
+
+Two tiers (mirrors tests/test_jitter_stress.py):
+  * TestBooleanJitterRegression - ALWAYS runs. Deterministic, non-flaky guards: clean extraction is
+                                  perfect, benign dynamic content never corrupts, and a transient
+                                  unexpected-code response landing on a validation request is ridden out.
+  * TestBooleanJitterSweep      - OPT-IN (SQLMAP_JITTER_STRESS=1). The creative failure-surface sweep
+                                  (IID + bursty), informational + loose bounds, kept out of normal CI.
+"""
+
+import os
+import random
+import re
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _testutils import bootstrap, set_dbms, reset_dbms
+bootstrap()
+
+from lib.core.data import conf, kb
+from lib.core.common import getCurrentThreadData, setTechnique
+from lib.core.datatype import AttribDict
+from lib.core.enums import HTTP_HEADER, PAYLOAD
+from lib.core.settings import PAYLOAD_DELIMITER
+from lib.request.comparison import comparison
+from lib.request.connect import Connect
+import lib.techniques.blind.inference as inf
+
+_D = PAYLOAD_DELIMITER
+_TEMPLATE = "%sEXPR=%%s IDX=%%d CMP>%%d%s" % (_D, _D)   # delimiter-wrapped -> validateChar '!=' fires
+_PARSE = re.compile(r"IDX=(\d+) CMP(!=|=|>)(\d+)")
+_SECRET = "Str0ng!"
+_STRING = "luther"
+_TRUE_BODY = "<html><body>welcome %s, here is your dashboard with 12 private items</body></html>" % _STRING
+_FALSE_BODY = "<html><body>invalid credentials, no such record, please retry</body></html>"
+_STRESS = os.environ.get("SQLMAP_JITTER_STRESS")
+
+
+class _Headers(object):
+    def __init__(self, ct="text/html"):
+        self.headers = ["Content-Type: %s\r\n" % ct]
+        self._d = {HTTP_HEADER.CONTENT_TYPE: ct}
+
+    def get(self, key, default=None):
+        return self._d.get(key, default)
+
+
+# ---- creative jitter catalog: each maps the clean intended body -> a transient junk response --------
+def _gw502(intended, rng):      return "<html><h1>502 Bad Gateway</h1></html>", 502, "text/html"
+def _gw504(intended, rng):      return "<html><h1>504 Gateway Time-out</h1></html>", 504, "text/html"
+def _rate429(intended, rng):    return "{\"error\":\"rate limited\"}", 429, "application/json"
+def _waf403(intended, rng):     return "<html>Request blocked by security policy #%d</html>" % rng.randint(1, 9), 403, "text/html"
+def _cf(intended, rng):         return "<html><title>Just a moment...</title>Checking your browser (Cloudflare)</html>", 200, "text/html"
+def _captcha(intended, rng):    return "<html>Please complete the CAPTCHA to continue</html>", 200, "text/html"
+def _maintenance(intended, rng):return "<html>We'll be back shortly - scheduled maintenance</html>", 200, "text/html"
+def _empty(intended, rng):      return "", 200, "text/html"
+def _truncated(intended, rng):  return intended[:rng.randint(10, 30)], 200, "text/html"
+def _lang(intended, rng):       return "<html><body>bienvenue, voici votre tableau de bord</body></html>", 200, "text/html"
+def _dynamic(intended, rng):    return intended.replace("</body>", "<span id=csrf>%d</span><span>%d views</span></body>" % (rng.getrandbits(32), rng.randint(1, 999))), 200, "text/html"
+def _coincidence(intended, rng):return "<html><body>system message from %s: degraded, retry later</body></html>" % _STRING, 200, "text/html"
+
+_CODE_CHANGING = (_gw502, _gw504, _rate429, _waf403)
+_SAME_CODE = (_cf, _captcha, _maintenance, _empty, _truncated, _lang)
+
+
+def _vector():
+    d = AttribDict()
+    d.payload = _TEMPLATE; d.where = 1; d.vector = _TEMPLATE; d.comment = ""
+    d.templatePayload = None; d.matchRatio = None; d.trueCode = 200; d.falseCode = 200
+    return d
+
+
+class _BooleanJitterBase(unittest.TestCase):
+    _CONF = ("threads", "api", "verbose", "direct", "string", "notString", "regexp", "code", "lengths",
+             "titles", "textOnly", "predictOutput", "hexConvert", "charset", "firstChar", "lastChar",
+             "ignoreCode", "ignoreTimeouts")
+    _KB = ("negativeLogic", "nullConnection", "errorIsNone", "pageTemplate", "matchRatio", "heavilyDynamic",
+           "pageStructurallyStable", "skipSeqMatcher", "pageEncoding", "partRun", "safeCharEncode",
+           "bruteMode", "fileReadMode", "disableShiftTable", "prependFlag", "timeless", "counters",
+           "originalCode", "originalPage")
+
+    def setUp(self):
+        self._saved_conf = {k: conf.get(k) for k in self._CONF}
+        self._saved_kb = {k: kb.get(k) for k in self._KB}
+        self._saved_inj = kb.injection.data
+        self._saved_qp = Connect.queryPage
+        self._saved_technique = getCurrentThreadData().technique
+
+    def tearDown(self):
+        for k, v in self._saved_conf.items():
+            conf[k] = v
+        for k, v in self._saved_kb.items():
+            kb[k] = v
+        kb.injection.data = self._saved_inj
+        Connect.queryPage = self._saved_qp
+        inf.Request.queryPage = self._saved_qp
+        setTechnique(self._saved_technique)
+
+    def _configure(self):
+        set_dbms("MySQL")
+        conf.threads = 1; conf.api = False; conf.verbose = 0; conf.direct = False
+        conf.string = _STRING; conf.notString = None; conf.regexp = None; conf.code = None
+        conf.lengths = None; conf.titles = None; conf.textOnly = None; conf.predictOutput = False
+        conf.hexConvert = False; conf.charset = None; conf.firstChar = None; conf.lastChar = None
+        conf.ignoreCode = []; conf.ignoreTimeouts = False
+        kb.negativeLogic = False; kb.nullConnection = False; kb.errorIsNone = True
+        kb.pageTemplate = _FALSE_BODY; kb.matchRatio = None; kb.heavilyDynamic = False
+        kb.pageStructurallyStable = False; kb.skipSeqMatcher = False; kb.pageEncoding = None
+        kb.partRun = None; kb.safeCharEncode = False; kb.bruteMode = False; kb.fileReadMode = False
+        kb.disableShiftTable = False; kb.prependFlag = False; kb.timeless = None; kb.counters = {}
+        kb.originalCode = None; kb.originalPage = None
+        kb.injection.data = {PAYLOAD.TECHNIQUE.BOOLEAN: _vector()}
+        setTechnique(PAYLOAD.TECHNIQUE.BOOLEAN)
+        kb.data.processChar = None
+        getCurrentThreadData().validationRun = 0
+
+    def _extract(self, respond):
+        """`respond(payload, cond, rng)` returns (body, code, contentType); drives real bisection ->
+        real comparison(). Bit truth `cond` is derived from the parseable delimiter-wrapped payload."""
+        def oracle(payload=None, timeBasedCompare=False, **kwargs):
+            td = getCurrentThreadData()
+            m = _PARSE.search(payload or "")
+            if not m:
+                td.lastPage = _FALSE_BODY; td.lastCode = 200
+                return comparison(_FALSE_BODY, _Headers(), 200)
+            idx, op, thr = int(m.group(1)), m.group(2), int(m.group(3))
+            ch = ord(_SECRET[idx - 1]) if 0 <= idx - 1 < len(_SECRET) else 0
+            cond = (ch > thr) if op == ">" else (ch != thr) if op == "!=" else (ch == thr)
+            if "NOT(" in (payload or ""):
+                cond = not cond
+            body, code, ct = respond(payload or "", cond)
+            td.lastPage = body; td.lastCode = code
+            return comparison(body, _Headers(ct), code)
+
+        Connect.queryPage = staticmethod(oracle)
+        inf.Request.queryPage = staticmethod(oracle)   # staticmethod on BOTH (py2 unbound-method guard)
+        td = getCurrentThreadData()
+        td.shared.value = ""; td.shared.index = [0]; td.shared.start = 0; td.shared.count = 0
+        _, value = inf.bisection(_TEMPLATE, "SELECT secret", length=len(_SECRET), charsetType=None)
+        return value
+
+    def _rate(self, arrival, trials=40, seed0=3000):
+        ok = 0
+        for t in range(trials):
+            rng = random.Random(seed0 + t)
+
+            def respond(payload, cond, rng=rng):
+                intended = _TRUE_BODY if cond else _FALSE_BODY
+                jitter = arrival(rng)
+                return jitter(intended, rng) if jitter is not None else (intended, 200, "text/html")
+
+            self._configure()
+            try:
+                ok += (self._extract(respond) == _SECRET)
+            except Exception:
+                pass
+        return ok, trials
+
+
+def _iid(p, kinds):
+    ks = list(kinds)
+    return lambda rng: rng.choice(ks) if rng.random() < p else None
+
+
+def _burst(p_enter, mean_len, kinds):
+    ks = list(kinds); state = {"bad": 0}
+
+    def f(rng):
+        if state["bad"] > 0:
+            state["bad"] -= 1; return rng.choice(ks)
+        if rng.random() < p_enter:
+            state["bad"] = max(0, int(rng.expovariate(1.0 / mean_len))) - 1
+            return rng.choice(ks)
+        return None
+    return f
+
+
+class TestBooleanJitterRegression(_BooleanJitterBase):
+    """Always-on, deterministic, non-flaky guards for the boolean decision stack."""
+
+    def test_clean_extraction_is_perfect(self):
+        ok, n = self._rate(lambda rng: None)
+        self.assertEqual(ok, n, "clean boolean extraction must be flawless (%d/%d)" % (ok, n))
+
+    def test_benign_dynamic_content_does_not_corrupt(self):
+        # csrf tokens / view counters / timestamps churn every response body; with the --string oracle
+        # they must never flip a bit. A regression that starts trusting raw-body noise fails here.
+        ok, n = self._rate(_iid(1.0, (_dynamic,)))
+        self.assertEqual(ok, n, "benign dynamic content must not corrupt extraction (%d/%d)" % (ok, n))
+
+    def test_unexpected_code_during_validation_is_ridden_out(self):
+        # Fix A guard: a transient unexpected-code response (503) landing on validateChar's own
+        # re-check request must not confirm a bit - the char is re-extracted. Here EVERY validation
+        # ('!=') request returns 503 once, deterministically; extraction must still be exact.
+        fired = {"n": 0}
+
+        def respond(payload, cond):
+            if "!=" in payload and fired["n"] < 3:      # poison the first few validation re-checks
+                fired["n"] += 1
+                return "<html><h1>503 Service Unavailable</h1></html>", 503, "text/html"
+            body = _TRUE_BODY if cond else _FALSE_BODY
+            return body, 200, "text/html"
+
+        self._configure()
+        self.assertEqual(self._extract(respond), _SECRET)
+
+
+@unittest.skipUnless(_STRESS, "creative boolean-jitter sweep is opt-in (set SQLMAP_JITTER_STRESS=1)")
+class TestBooleanJitterSweep(_BooleanJitterBase):
+    """Opt-in creative failure-surface map (IID + bursty). Informational; asserts only the clean case."""
+
+    def _sweep(self, label, factory, kinds, rates=(0.0, 0.02, 0.05, 0.10, 0.20)):
+        print("\n[bool-jitter] %s:" % label)
+        for r in rates:
+            ok, n = self._rate(factory(r, kinds))
+            print("  rate=%.2f -> %d/%d (%3.0f%%)" % (r, ok, n, 100.0 * ok / n))
+            if r == 0.0:
+                self.assertEqual(ok, n)
+
+    def test_iid_code_changing(self):
+        self._sweep("IID code-changing (502/504/429/403)", _iid, _CODE_CHANGING)
+
+    def test_iid_same_code_body(self):
+        self._sweep("IID same-code 200 body (cf/captcha/maint/empty/trunc/lang)", _iid, _SAME_CODE)
+
+    def test_iid_string_coincidence(self):
+        self._sweep("IID string-coincidence (fake page contains --string)", _iid, (_coincidence,))
+
+    def test_burst_same_code_body(self):
+        self._sweep("BURST(mean=4) same-code 200 body", lambda r, k: _burst(r, 4, k), _SAME_CODE)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)
+
+
+def tearDownModule():
+    reset_dbms()