From 306738539e2b01f8f819de22b3602f31988c5167 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Wed, 29 Jul 2026 01:08:47 +0200 Subject: [PATCH 1/2] Making boolean inference more robust against jitter --- lib/core/settings.py | 2 +- lib/techniques/blind/inference.py | 19 ++- tests/test_boolean_jitter.py | 258 ++++++++++++++++++++++++++++++ 3 files changed, 273 insertions(+), 6 deletions(-) create mode 100644 tests/test_boolean_jitter.py diff --git a/lib/core/settings.py b/lib/core/settings.py index 8428bcaabb2..a0a6fdd13b6 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from thirdparty import six # sqlmap version (...) -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) diff --git a/lib/techniques/blind/inference.py b/lib/techniques/blind/inference.py index 5a66f4d943f..fcfaad35c5f 100644 --- a/lib/techniques/blind/inference.py +++ b/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()) diff --git a/tests/test_boolean_jitter.py b/tests/test_boolean_jitter.py new file mode 100644 index 00000000000..b05292444d4 --- /dev/null +++ b/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 = "welcome %s, here is your dashboard with 12 private items" % _STRING +_FALSE_BODY = "invalid credentials, no such record, please retry" +_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 "

502 Bad Gateway

", 502, "text/html" +def _gw504(intended, rng): return "

504 Gateway Time-out

", 504, "text/html" +def _rate429(intended, rng): return "{\"error\":\"rate limited\"}", 429, "application/json" +def _waf403(intended, rng): return "Request blocked by security policy #%d" % rng.randint(1, 9), 403, "text/html" +def _cf(intended, rng): return "Just a moment...Checking your browser (Cloudflare)", 200, "text/html" +def _captcha(intended, rng): return "Please complete the CAPTCHA to continue", 200, "text/html" +def _maintenance(intended, rng):return "We'll be back shortly - scheduled maintenance", 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 "bienvenue, voici votre tableau de bord", 200, "text/html" +def _dynamic(intended, rng): return intended.replace("", "%d%d views" % (rng.getrandbits(32), rng.randint(1, 999))), 200, "text/html" +def _coincidence(intended, rng):return "system message from %s: degraded, retry later" % _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 "

503 Service Unavailable

", 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() From 39d22d26c3b0495ae7c285c65e0df971a93595f1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Wed, 29 Jul 2026 02:13:39 +0200 Subject: [PATCH 2/2] Making boolean inference some more robust against jitter --- lib/controller/checks.py | 6 +++++ lib/core/option.py | 4 +++ lib/core/settings.py | 7 ++++- lib/techniques/blind/inference.py | 38 +++++++++++++++++++++++++- tests/test_boolean_jitter.py | 44 ++++++++++++++++++++++++++++--- 5 files changed, 94 insertions(+), 5 deletions(-) diff --git a/lib/controller/checks.py b/lib/controller/checks.py index c5bd8436fd7..03e7abb5cd5 100644 --- a/lib/controller/checks.py +++ b/lib/controller/checks.py @@ -818,6 +818,12 @@ def genCmpPayload(): injection.data[stype].trueCode = trueCode injection.data[stype].falseCode = falseCode + # reference bodies for inference.py's "resembles neither TRUE nor FALSE model" + # anomaly guard (runtime-only; lets a transient same-HTTP-code junk response + # trigger a validateChar re-check during boolean extraction) + if method == PAYLOAD.METHOD.COMPARISON: + kb.trueTemplate, kb.falseTemplate = truePage, falsePage + injection.conf.textOnly = conf.textOnly injection.conf.titles = conf.titles injection.conf.code = conf.code diff --git a/lib/core/option.py b/lib/core/option.py index b1aa4e4bd83..babcf675a68 100644 --- a/lib/core/option.py +++ b/lib/core/option.py @@ -2305,6 +2305,10 @@ def _setKnowledgeBaseAttributes(flushAll=True): kb.pageTemplate = None kb.pageTemplates = dict() kb.pageEncoding = DEFAULT_PAGE_ENCODING + + # calibrated TRUE/FALSE reference bodies for the boolean same-HTTP-code anomaly guard (inference.py) + kb.trueTemplate = None + kb.falseTemplate = None kb.pageStable = None kb.pageStructurallyStable = None kb.partRun = None diff --git a/lib/core/settings.py b/lib/core/settings.py index a0a6fdd13b6..4b571449561 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from thirdparty import six # sqlmap version (...) -VERSION = "1.10.7.241" +VERSION = "1.10.7.242" 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) @@ -103,6 +103,11 @@ LOWER_RATIO_BOUND = 0.02 UPPER_RATIO_BOUND = 0.98 +# Minimum similarity at which a boolean extraction response is judged to "resemble" the calibrated +# TRUE or FALSE model. A response resembling NEITHER (a transient same-HTTP-code junk page: WAF/CDN +# interstitial, captcha, maintenance, empty/truncated body) triggers an extra validateChar re-check. +BOOLEAN_MODEL_MATCH_RATIO = 0.9 + # Number of candidate names probed per request while mining for hidden parameters ('--mine-params') PARAMETER_MINING_BUCKET_SIZE = 25 diff --git a/lib/techniques/blind/inference.py b/lib/techniques/blind/inference.py index fcfaad35c5f..21a93eba7a7 100644 --- a/lib/techniques/blind/inference.py +++ b/lib/techniques/blind/inference.py @@ -7,6 +7,7 @@ from __future__ import division +import difflib import heapq import re import time @@ -26,6 +27,7 @@ from lib.core.common import getTechniqueData from lib.core.common import getText from lib.core.common import predictValue +from lib.core.common import removeDynamicContent from lib.core.common import hashDBRetrieve from lib.core.common import hashDBWrite from lib.core.common import incrementCounter @@ -45,6 +47,7 @@ from lib.core.exception import SqlmapThreadException from lib.core.exception import SqlmapUnsupportedFeatureException from lib.core.wordlist import Wordlist +from lib.core.settings import BOOLEAN_MODEL_MATCH_RATIO from lib.core.settings import CHAR_INFERENCE_MARK from lib.core.settings import HUFFMAN_PROBE_LIMIT from lib.core.settings import HUFFMAN_PRIOR_WEIGHTS @@ -256,6 +259,31 @@ def oracleReliabilityLitmus(expressionUnescaped, value, timeBasedCompare): return bool(mustBeTrue) and not bool(mustBeFalse) +def _resemblesNeitherModel(page): + """ + Returns True when a boolean extraction response resembles NEITHER the calibrated TRUE nor FALSE + model (kb.trueTemplate / kb.falseTemplate). A transient response that keeps the expected HTTP code + but swaps the body for junk (WAF/CDN interstitial, captcha, maintenance banner, empty/truncated + page) is invisible to the HTTP-code check, so it is flagged here for an extra validateChar re-check. + + This only ever ADDS a re-validation - it never changes a decided bit - so it is a safe no-op on a + clean target (every response resembles its own model) and when no models were recorded (a resumed + session, or a non-boolean technique). + """ + + refs = [_ for _ in (kb.trueTemplate, kb.falseTemplate) if _] + if not refs: + return False + if page is None: + return True + + cleaned = removeDynamicContent(page) + for ref in refs: + if difflib.SequenceMatcher(None, removeDynamicContent(ref), cleaned).quick_ratio() >= BOOLEAN_MODEL_MATCH_RATIO: + return False + + return True + def bisection(payload, expression, length=None, charsetType=None, firstChar=None, lastChar=None, dump=False): """ Bisection algorithm that can be used to perform blind SQL injection @@ -714,6 +742,7 @@ def getChar(idx, charTbl=None, continuousOrder=True, expand=charsetType is None, firstCheck = False lastCheck = False unexpectedCode = False + unexpectedResponse = False if continuousOrder: while len(charTbl) > 1: @@ -787,6 +816,13 @@ def getChar(idx, charTbl=None, continuousOrder=True, expand=charsetType is None, singleTimeWarnMessage(warnMsg) + # same-HTTP-code body anomaly (WAF/CDN interstitial, captcha, maintenance, empty + # or truncated body) - invisible to the code check above, so re-validate when the + # response resembles neither calibrated model + elif not unexpectedResponse and not kb.nullConnection and _resemblesNeitherModel(threadData.lastPage): + unexpectedResponse = True + singleTimeWarnMessage("unexpected response content detected. Will use (extra) validation step in similar cases") + if result: minValue = posValue @@ -826,7 +862,7 @@ def getChar(idx, charTbl=None, continuousOrder=True, expand=charsetType is None, retVal = minValue + 1 if retVal in originalTbl or (retVal == ord('\n') and CHAR_INFERENCE_MARK in payload): - if (timeBasedCompare or unexpectedCode) and kb.get("timeless") is None and not validateChar(idx, retVal): + if (timeBasedCompare or unexpectedCode or unexpectedResponse) and kb.get("timeless") is None and not validateChar(idx, retVal): if restricted: # the character fell outside this column's observed range - re-extract # over the full charset (not timing noise, so no delay increase / retry count) diff --git a/tests/test_boolean_jitter.py b/tests/test_boolean_jitter.py index b05292444d4..8b539f7fc57 100644 --- a/tests/test_boolean_jitter.py +++ b/tests/test_boolean_jitter.py @@ -51,8 +51,14 @@ _PARSE = re.compile(r"IDX=(\d+) CMP(!=|=|>)(\d+)") _SECRET = "Str0ng!" _STRING = "luther" -_TRUE_BODY = "welcome %s, here is your dashboard with 12 private items" % _STRING -_FALSE_BODY = "invalid credentials, no such record, please retry" +# realistic-size bodies (shared nav/footer boilerplate) so the "resembles neither model" anomaly guard +# behaves as on a real page: benign dynamic noise is proportionally tiny (stays a match), while a junk +# interstitial/maintenance/empty body clearly matches neither +_BOILER = "Acme Portal
" * 8 +_FOOT = "
(c) Acme Corp - all rights reserved - support@acme.example - v4.2
" * 8 +_TRUE_BODY = _BOILER + "welcome %s, dashboard: orders profile settings billing (12 items)" % _STRING + _FOOT +_FALSE_BODY = _BOILER + "invalid credentials, no such record found, please retry" + _FOOT +_INTERSTITIAL = "Just a moment... checking your browser before access (DDoS protection)" _STRESS = os.environ.get("SQLMAP_JITTER_STRESS") @@ -97,7 +103,7 @@ class _BooleanJitterBase(unittest.TestCase): _KB = ("negativeLogic", "nullConnection", "errorIsNone", "pageTemplate", "matchRatio", "heavilyDynamic", "pageStructurallyStable", "skipSeqMatcher", "pageEncoding", "partRun", "safeCharEncode", "bruteMode", "fileReadMode", "disableShiftTable", "prependFlag", "timeless", "counters", - "originalCode", "originalPage") + "originalCode", "originalPage", "trueTemplate", "falseTemplate", "dynamicMarkings") def setUp(self): self._saved_conf = {k: conf.get(k) for k in self._CONF} @@ -129,6 +135,8 @@ def _configure(self): 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 + # calibrated reference bodies for the same-code anomaly guard (Fix B); no learned dynamic markings + kb.trueTemplate = _TRUE_BODY; kb.falseTemplate = _FALSE_BODY; kb.dynamicMarkings = [] kb.injection.data = {PAYLOAD.TECHNIQUE.BOOLEAN: _vector()} setTechnique(PAYLOAD.TECHNIQUE.BOOLEAN) kb.data.processChar = None @@ -224,6 +232,36 @@ def respond(payload, cond): self._configure() self.assertEqual(self._extract(respond), _SECRET) + def test_anomaly_classifier_flags_only_junk(self): + # Fix B core: a response resembling NEITHER calibrated model is flagged; the models themselves + # and a benign dynamic variant are not. Deterministic, no network. + self._configure() + self.assertFalse(inf._resemblesNeitherModel(_TRUE_BODY)) + self.assertFalse(inf._resemblesNeitherModel(_FALSE_BODY)) + self.assertFalse(inf._resemblesNeitherModel(_TRUE_BODY.replace("dashboard", "dashboard 7 new tok=abc123"))) + for junk in (_INTERSTITIAL, "", "

502 Bad Gateway

", _TRUE_BODY[:60]): + self.assertTrue(inf._resemblesNeitherModel(junk), msg="must flag junk %r" % junk[:40]) + + def test_same_code_body_jitter_is_ridden_out(self): + # Fix B guard: a transient same-HTTP-code junk page that resembles NEITHER model makes a + # character mis-resolve to a wrong (valid) value; the anomaly guard triggers validateChar to + # re-extract it. The junk here carries the --string token (so it reads True and pushes the char + # HIGH -> a wrong valid char, the case validateChar covers), and is unlike both models -> flagged. + junk = "notice: %s service temporarily degraded, retry" % _STRING + poisoned = {"n": 0} + + def respond(payload, cond): + m = _PARSE.search(payload) + idx = int(m.group(1)) if m else 0 + if idx == 4 and "!=" not in payload and poisoned["n"] < 3: + poisoned["n"] += 1 + return junk, 200, "text/html" + return (_TRUE_BODY if cond else _FALSE_BODY), 200, "text/html" + + self._configure() + self.assertTrue(inf._resemblesNeitherModel(junk)) # precondition: the junk IS anomalous + self.assertEqual(self._extract(respond), _SECRET) + @unittest.skipUnless(_STRESS, "creative boolean-jitter sweep is opt-in (set SQLMAP_JITTER_STRESS=1)") class TestBooleanJitterSweep(_BooleanJitterBase):