diff --git a/extra/kerberos/client.py b/extra/kerberos/client.py
index f8fe44e7f06..3157fce1ced 100644
--- a/extra/kerberos/client.py
+++ b/extra/kerberos/client.py
@@ -309,6 +309,15 @@ def _asReq(realm, username, etypes, nonce, padata=None):
     parts.append(der.tagged(4, reqBody))
     return der.application(AS_REQ, der.sequence(*parts))
 
+def _selectEtype(etypes, hints):
+    """
+    The etype getTGT commits to after a preauth-required hint: OUR first offered etype that is also
+    KDC-hinted and supported, falling back to our top preference. The unauthenticated hint can only
+    reorder WITHIN what we offered - it can never pull us onto an etype we did not offer (anti-downgrade).
+    """
+
+    return next((_ for _ in etypes if _ in hints and _ in ENCTYPES), etypes[0])
+
 def getTGT(realm, username, password, kdcHost, kdcPort=88, etypes=DEFAULT_ETYPES, salt=None):
     """Run the AS exchange and return the TGT and its session key.
 
@@ -343,10 +352,7 @@ def getTGT(realm, username, password, kdcHost, kdcPort=88, etypes=DEFAULT_ETYPES
         # the hint is unauthenticated, so it may only choose among the etypes we actually offered, and
         # in *our* order of preference rather than the KDC's (otherwise it could force a downgrade)
         hints = _preauthHints(errorFields)
-        for offered in etypes:
-            if offered in hints and offered in ENCTYPES:
-                etype = offered
-                break
+        etype = _selectEtype(etypes, hints)
         chosenSalt, iterations = _hintFor(hints, etype, salt, chosenSalt)
 
     enc = _enctype(etype)
diff --git a/lib/core/common.py b/lib/core/common.py
index ef3b2936449..a4be0827218 100644
--- a/lib/core/common.py
+++ b/lib/core/common.py
@@ -183,6 +183,7 @@
 from lib.core.settings import STRUCTURAL_TAG_REGEX
 from lib.core.settings import SUPPORTED_DBMS
 from lib.core.settings import TEXT_TAG_REGEX
+from lib.core.settings import TIME_OUTLIER_MAD_COEFF
 from lib.core.settings import TIME_STDEV_COEFF
 from lib.core.settings import UNICODE_ENCODING
 from lib.core.settings import UNKNOWN_DBMS_VERSION
@@ -2009,7 +2010,10 @@ def parseUnionPage(page):
                 entry = entry.split(kb.chars.start)[-1]
 
             if kb.unionDuplicates:
-                key = entry.lower()
+                # Note: de-dup on the EXACT entry, not entry.lower() - the doubled emission repeats each
+                # row verbatim, so case-folding here would silently drop genuinely case-distinct rows
+                # (e.g. 'Admin' vs 'admin'). Force-uppercased pages are already lower-cased before parsing.
+                key = entry
                 if key not in keys:
                     keys.add(key)
                 else:
@@ -2877,6 +2881,38 @@ def wasLastResponseHTTPError():
     threadData = getCurrentThreadData()
     return threadData.lastHTTPError and threadData.lastHTTPError[0] == threadData.lastRequestUID
 
+def stripTimeOutliers(values):
+    """
+    Returns L{values} with high (spike) outliers removed, using a robust median/MAD cutoff.
+
+    A single network spike that lands in the time-response model would otherwise inflate both the
+    average and the standard deviation, exploding the delay threshold (avg + 7*stdev) so that genuine
+    time-delays are no longer recognized. MAD is robust to a minority of outliers, so the cutoff is
+    computed from the clean bulk even when the sample already contains a spike. On a clean model no
+    value exceeds median + 10*MAD, so it is returned unchanged (identical avg/stdev/threshold).
+
+    >>> len(stripTimeOutliers([0.1, 0.12] * 8 + [9.0]))   # a lone 9s spike is dropped (17->16)
+    16
+    >>> len(stripTimeOutliers([0.1, 0.12] * 8))           # a clean model is left intact (no-op)
+    16
+    """
+
+    if not values or len(values) < MIN_TIME_RESPONSES // 2:
+        return values
+
+    ordered = sorted(values)
+    median = ordered[len(ordered) // 2]
+    mad = sorted(abs(_ - median) for _ in values)[len(values) // 2]
+
+    if mad <= 0:  # degenerate (near-constant model) - nothing robust to trim on
+        return values
+
+    cutoff = median + TIME_OUTLIER_MAD_COEFF * 1.4826 * mad
+    retVal = [_ for _ in values if _ <= cutoff]
+
+    # never trim away the bulk (guards a genuinely wide/bimodal model from being gutted)
+    return retVal if len(retVal) >= max(MIN_TIME_RESPONSES // 2, len(values) // 2) else values
+
 def wasLastResponseDelayed():
     """
     Returns True if the last web request resulted in a time-delay
@@ -2886,7 +2922,10 @@ def wasLastResponseDelayed():
     # response times should be inside +-7*stdev([normal response times])
     # Math reference: http://www.answers.com/topic/standard-deviation
 
-    deviation = stdev(kb.responseTimes.get(kb.responseTimeMode, []))
+    # spike outliers (e.g. a GC pause / retransmit during baseline sampling) are stripped first, so a
+    # single lagging response can't inflate the model and hide every genuine delay behind it
+    sample = stripTimeOutliers(kb.responseTimes.get(kb.responseTimeMode, []))
+    deviation = stdev(sample)
     threadData = getCurrentThreadData()
 
     if deviation and not conf.direct and not conf.disableStats:
@@ -2895,7 +2934,7 @@ def wasLastResponseDelayed():
             warnMsg += "with less than %d response times" % MIN_TIME_RESPONSES
             logger.warning(warnMsg)
 
-        lowerStdLimit = average(kb.responseTimes[kb.responseTimeMode]) + TIME_STDEV_COEFF * deviation
+        lowerStdLimit = average(sample) + TIME_STDEV_COEFF * deviation
         retVal = (threadData.lastQueryDuration >= max(MIN_VALID_DELAYED_RESPONSE, lowerStdLimit))
 
         if not kb.testMode and retVal:
@@ -3859,7 +3898,7 @@ def joinValue(value, delimiter=','):
     """
 
     if isListLike(value):
-        retVal = delimiter.join(getText(_ if _ is not None else "None") for _ in value)
+        retVal = delimiter.join(getText(getUnicode(_) if _ is not None else "None") for _ in value)
     else:
         retVal = value
 
@@ -4264,9 +4303,14 @@ def decodeStringEscape(value):
     retVal = value
 
     if value and '\\' in value:
-        charset = "\\%s" % string.whitespace.replace(" ", "")
-        for _ in charset:
+        # Note: shield an escaped backslash ('\\\\') behind a marker BEFORE decoding the whitespace
+        # escapes, then restore it - otherwise decoding '\\\\' -> '\\' first turns a literal '\\n'
+        # into a newline (i.e. the round-trip with encodeStringEscape was not lossless)
+        _marker = "\x00"
+        retVal = retVal.replace("\\\\", _marker)
+        for _ in string.whitespace.replace(" ", ""):
             retVal = retVal.replace(repr(_).strip("'"), _)
+        retVal = retVal.replace(_marker, "\\")
 
     return retVal
 
diff --git a/lib/core/settings.py b/lib/core/settings.py
index b587082ac4b..8428bcaabb2 100644
--- a/lib/core/settings.py
+++ b/lib/core/settings.py
@@ -20,7 +20,7 @@
 from thirdparty import six
 
 # sqlmap version (<major>.<minor>.<month>.<monthly commit>)
-VERSION = "1.10.7.228"
+VERSION = "1.10.7.240"
 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)
@@ -232,6 +232,12 @@
 # Coefficient used for a time-based query delay checking (must be >= 7)
 TIME_STDEV_COEFF = 7
 
+# Robust (median/MAD) cutoff for discarding spike outliers from the time-response model before
+# computing avg/stdev - a single network spike landing in the baseline would otherwise inflate the
+# delay threshold and miss genuine delays. Deliberately wide (~10 robust sigmas) so a clean model is
+# left untouched (identical threshold) and only true outliers are dropped.
+TIME_OUTLIER_MAD_COEFF = 10
+
 # Minimum response time that can be even considered as delayed (not a complete requirement)
 MIN_VALID_DELAYED_RESPONSE = 0.5
 
diff --git a/lib/parse/cmdline.py b/lib/parse/cmdline.py
index 03bddb531be..a4ceb781940 100644
--- a/lib/parse/cmdline.py
+++ b/lib/parse/cmdline.py
@@ -1171,7 +1171,7 @@ def _format_action_invocation(self, action):
                 dataToStdout("[!] detected usage of long-option without a starting hyphen ('%s')\n" % argv[i])
                 raise SystemExit
 
-        for verbosity in (_ for _ in argv if re.search(r"\A\-v+\Z", _)):
+        for verbosity in [_ for _ in argv if re.search(r"\A\-v+\Z", _)]:  # Note: list (not a generator) - the loop mutates argv (del), which would desync a live generator and leave a stray '-v' for argparse to choke on
             try:
                 if argv.index(verbosity) == len(argv) - 1 or not argv[argv.index(verbosity) + 1].isdigit():
                     conf.verbose = verbosity.count('v')
diff --git a/lib/parse/configfile.py b/lib/parse/configfile.py
index 88f91ce7828..9e104923e2a 100644
--- a/lib/parse/configfile.py
+++ b/lib/parse/configfile.py
@@ -84,14 +84,14 @@ def configFileParser(configFile):
 
     mandatory = False
 
-    for option in ("direct", "url", "logFile", "bulkFile", "googleDork", "requestFile", "wizard"):
+    for option in ("direct", "url", "logFile", "bulkFile", "googleDork", "requestFile", "openApiFile", "wizard"):
         if config.has_option("Target", option) and config.get("Target", option) or cmdLineOptions.get(option):
             mandatory = True
             break
 
     if not mandatory:
         errMsg = "missing a mandatory option in the configuration file "
-        errMsg += "(direct, url, logFile, bulkFile, googleDork, requestFile or wizard)"
+        errMsg += "(direct, url, logFile, bulkFile, googleDork, requestFile, openApiFile or wizard)"
         raise SqlmapMissingMandatoryOptionException(errMsg)
 
     for family, optionData in optDict.items():
diff --git a/lib/utils/har.py b/lib/utils/har.py
index ebdef3bfde3..c9a17b25a11 100644
--- a/lib/utils/har.py
+++ b/lib/utils/har.py
@@ -116,7 +116,7 @@ def toDict(self):
             "httpVersion": self.httpVersion,
             "method": self.method,
             "url": self.url,
-            "headers": [dict(name=key.capitalize(), value=value) for key, value in self.headers.items()],
+            "headers": [dict(name="-".join(_.capitalize() for _ in key.split("-")), value=value) for key, value in self.headers.items()],
             "cookies": [],
             "queryString": [],
             "headersSize": -1,
@@ -202,7 +202,7 @@ def toDict(self):
             "httpVersion": self.httpVersion,
             "status": self.status,
             "statusText": self.statusText,
-            "headers": [dict(name=key.capitalize(), value=value) for key, value in self.headers.items() if key.lower() != "uri"],
+            "headers": [dict(name="-".join(_.capitalize() for _ in key.split("-")), value=value) for key, value in self.headers.items() if key.lower() != "uri"],
             "cookies": [],
             "content": content,
             "headersSize": -1,
diff --git a/lib/utils/safe2bin.py b/lib/utils/safe2bin.py
index d6004ef7a57..1b7aa605d61 100644
--- a/lib/utils/safe2bin.py
+++ b/lib/utils/safe2bin.py
@@ -55,6 +55,9 @@ def safecharencode(value):
 
     if isinstance(value, string_types):
         if any(_ not in SAFE_CHARS for _ in value):
+            # NOTE (checked twice, do NOT "fix" by deleting): this marker keeps an already-`\x`-carrying
+            # value (incl. getUnicode's 'reversible' \xNN) from double-escaping to `\\x` in console/CSV.
+            # Dropping it only swaps that for the rare literal-`\x`-in-data round-trip - a bad trade.
             retVal = retVal.replace(HEX_ENCODED_PREFIX, HEX_ENCODED_PREFIX_MARKER)
             retVal = retVal.replace('\\', SLASH_MARKER)
 
diff --git a/plugins/generic/databases.py b/plugins/generic/databases.py
index f01d4095bd1..becad0332ae 100644
--- a/plugins/generic/databases.py
+++ b/plugins/generic/databases.py
@@ -418,9 +418,9 @@ def getTables(self, bruteForce=None):
                     else:
                         for index in indexRange:
                             if Backend.isDbms(DBMS.SYBASE):
-                                query = _query % (db, (kb.data.cachedTables[-1] if kb.data.cachedTables else " "))
+                                query = _query % (db, (tables[-1] if tables else " "))
                             elif Backend.getIdentifiedDbms() in (DBMS.MAXDB, DBMS.ACCESS, DBMS.MCKOI, DBMS.EXTREMEDB):
-                                query = _query % (kb.data.cachedTables[-1] if kb.data.cachedTables else " ")
+                                query = _query % (tables[-1] if tables else " ")
                             elif Backend.getIdentifiedDbms() in (DBMS.SQLITE, DBMS.FIREBIRD):
                                 query = _query % index
                             elif Backend.getIdentifiedDbms() in (DBMS.HSQLDB, DBMS.INFORMIX, DBMS.FRONTBASE, DBMS.VIRTUOSO):
diff --git a/plugins/generic/users.py b/plugins/generic/users.py
index e22c96d5aff..7bd070510af 100644
--- a/plugins/generic/users.py
+++ b/plugins/generic/users.py
@@ -648,13 +648,16 @@ def getPrivileges(self, query2=False):
                         break
 
                 if privileges:
-                    kb.data.cachedUsersPrivileges[user] = list(privileges)
+                    # Note: 'user' may be a LIKE-wrapped form (e.g. '%root%') built for the MySQL
+                    # query above; key/record under the real name so the output isn't mislabelled
+                    # (and the retrievedUsers de-dup check, which compares the unwrapped name, works)
+                    kb.data.cachedUsersPrivileges[outuser] = list(privileges)
                 else:
                     warnMsg = "unable to retrieve the privileges "
                     warnMsg += "for user '%s'" % outuser
                     logger.warning(warnMsg)
 
-                retrievedUsers.add(user)
+                retrievedUsers.add(outuser)
 
         if not kb.data.cachedUsersPrivileges:
             errMsg = "unable to retrieve the privileges "
diff --git a/tamper/charunicodeescape.py b/tamper/charunicodeescape.py
index 0bc2624aec5..30e1e433726 100644
--- a/tamper/charunicodeescape.py
+++ b/tamper/charunicodeescape.py
@@ -33,7 +33,13 @@ def tamper(payload, **kwargs):
                 retVal += "\\u00%s" % payload[i + 1:i + 3]
                 i += 3
             else:
-                retVal += '\\u%.4X' % ord(payload[i])
+                ordinal = ord(payload[i])
+                if ordinal > 0xFFFF:
+                    # non-BMP: emit a UTF-16 surrogate pair (a bare 5-hex '\uXXXXX' is invalid)
+                    ordinal -= 0x10000
+                    retVal += "\\u%04X\\u%04X" % (0xD800 + (ordinal >> 10), 0xDC00 + (ordinal & 0x3FF))
+                else:
+                    retVal += "\\u%04X" % ordinal
                 i += 1
 
     return retVal
diff --git a/tamper/if2case.py b/tamper/if2case.py
index f3c01ddb1c4..029c130b9cd 100644
--- a/tamper/if2case.py
+++ b/tamper/if2case.py
@@ -14,6 +14,34 @@
 def dependencies():
     pass
 
+def _unwrap(expr):
+    """
+    Strips only FULLY-wrapping outer parentheses (e.g. '(1=1)' -> '1=1'), leaving a bare function
+    call such as 'SLEEP(5)' intact - unlike str.strip('()') which would drop its trailing ')'
+    """
+
+    expr = expr.strip()
+
+    while len(expr) > 1 and expr[0] == '(' and expr[-1] == ')':
+        depth = 0
+        wrapper = True
+
+        for i in xrange(len(expr)):
+            if expr[i] == '(':
+                depth += 1
+            elif expr[i] == ')':
+                depth -= 1
+                if depth == 0 and i != len(expr) - 1:  # the opening '(' closes before the end
+                    wrapper = False
+                    break
+
+        if not wrapper:
+            break
+
+        expr = expr[1:-1].strip()
+
+    return expr
+
 def tamper(payload, **kwargs):
     """
     Replaces instances like 'IF(A, B, C)' with 'CASE WHEN (A) THEN (B) ELSE (C) END' counterpart
@@ -62,9 +90,9 @@ def tamper(payload, **kwargs):
                         depth -= 1
 
             if len(commas) == 2 and end:
-                a = payload[index + len("IF("):commas[0]].strip("()")
-                b = payload[commas[0] + 1:commas[1]].lstrip().strip("()")
-                c = payload[commas[1] + 1:end].lstrip().strip("()")
+                a = _unwrap(payload[index + len("IF("):commas[0]])
+                b = _unwrap(payload[commas[0] + 1:commas[1]])
+                c = _unwrap(payload[commas[1] + 1:end])
                 newVal = "CASE WHEN (%s) THEN (%s) ELSE (%s) END" % (a, b, c)
                 payload = payload[:index] + newVal + payload[end + 1:]
             else:
diff --git a/tamper/overlongutf8.py b/tamper/overlongutf8.py
index 75bd678e775..8246ed516c1 100644
--- a/tamper/overlongutf8.py
+++ b/tamper/overlongutf8.py
@@ -38,7 +38,12 @@ def tamper(payload, **kwargs):
                 i += 3
             else:
                 if payload[i] not in (string.ascii_letters + string.digits):
-                    retVal += "%%%.2X%%%.2X" % (0xc0 + (ord(payload[i]) >> 6), 0x80 + (ord(payload[i]) & 0x3f))
+                    ordinal = ord(payload[i])
+                    if ordinal <= 0x7FF:
+                        retVal += "%%%.2X%%%.2X" % (0xc0 + (ordinal >> 6), 0x80 + (ordinal & 0x3f))
+                    else:
+                        # the 2-byte overlong form can't hold code points > U+07FF; fall back to real UTF-8
+                        retVal += "".join("%%%.2X" % _ for _ in bytearray(payload[i].encode("utf8")))
                 else:
                     retVal += payload[i]
                 i += 1
diff --git a/tamper/overlongutf8more.py b/tamper/overlongutf8more.py
index 391464f6ef7..c8ad07d313d 100644
--- a/tamper/overlongutf8more.py
+++ b/tamper/overlongutf8more.py
@@ -37,7 +37,12 @@ def tamper(payload, **kwargs):
                 retVal += payload[i:i + 3]
                 i += 3
             else:
-                retVal += "%%%.2X%%%.2X" % (0xc0 + (ord(payload[i]) >> 6), 0x80 + (ord(payload[i]) & 0x3f))
+                ordinal = ord(payload[i])
+                if ordinal <= 0x7FF:
+                    retVal += "%%%.2X%%%.2X" % (0xc0 + (ordinal >> 6), 0x80 + (ordinal & 0x3f))
+                else:
+                    # the 2-byte overlong form can't hold code points > U+07FF; fall back to real UTF-8
+                    retVal += "".join("%%%.2X" % _ for _ in bytearray(payload[i].encode("utf8")))
                 i += 1
 
     return retVal
diff --git a/tests/test_error_engine.py b/tests/test_error_engine.py
index ee960c94c0a..19f3ab9902a 100644
--- a/tests/test_error_engine.py
+++ b/tests/test_error_engine.py
@@ -26,6 +26,7 @@
 from _testutils import bootstrap, set_dbms, reset_dbms
 bootstrap()
 
+from lib.core.common import getCurrentThreadData, setTechnique
 from lib.core.data import conf, kb
 from lib.core.datatype import AttribDict
 from lib.core.enums import PAYLOAD, PLACE
@@ -52,6 +53,7 @@ def setUp(self):
             "conf.paramDict": conf.get("paramDict"), "conf.base64Parameter": conf.get("base64Parameter"),
             "kb.errorChunkLength": kb.get("errorChunkLength"), "kb.testMode": kb.get("testMode"),
             "kb.forceWhere": kb.get("forceWhere"), "kb.technique": kb.get("technique"),
+            "td.technique": getCurrentThreadData().technique,
             "kb.inj": (kb.injection.place, kb.injection.parameter, kb.injection.data),
             "qp": Connect.queryPage,
         }
@@ -67,6 +69,7 @@ def setUp(self):
         kb.injection.place = PLACE.GET
         kb.injection.parameter = "id"
         kb.technique = PAYLOAD.TECHNIQUE.ERROR
+        setTechnique(PAYLOAD.TECHNIQUE.ERROR)   # getTechnique() prefers the thread-local; set it so a leaked one can't poison us
         kb.injection.data = {PAYLOAD.TECHNIQUE.ERROR: _make_vector()}
         set_dbms("MySQL")
 
@@ -81,6 +84,7 @@ def tearDown(self):
         kb.testMode = self._saved["kb.testMode"]
         kb.forceWhere = self._saved["kb.forceWhere"]
         kb.technique = self._saved["kb.technique"]
+        setTechnique(self._saved["td.technique"])
         kb.injection.place, kb.injection.parameter, kb.injection.data = self._saved["kb.inj"]
         Connect.queryPage = self._saved["qp"]
         eu.Request.queryPage = self._saved["qp"]
@@ -126,6 +130,7 @@ def setUp(self):
             "paramDict": conf.get("paramDict"), "base64Parameter": conf.get("base64Parameter"),
             "errorChunkLength": kb.get("errorChunkLength"), "testMode": kb.get("testMode"),
             "forceWhere": kb.get("forceWhere"), "technique": kb.get("technique"),
+            "td.technique": getCurrentThreadData().technique,
             "inj": (kb.injection.place, kb.injection.parameter, kb.injection.data),
             "qp": Connect.queryPage,
             "dbmsHandler": conf.get("dbmsHandler"), "forceDbms": conf.get("forceDbms"),
@@ -145,6 +150,7 @@ def setUp(self):
         kb.injection.place = PLACE.GET
         kb.injection.parameter = "id"
         kb.technique = PAYLOAD.TECHNIQUE.ERROR
+        setTechnique(PAYLOAD.TECHNIQUE.ERROR)   # getTechnique() prefers the thread-local; set it so a leaked one can't poison us
         kb.injection.data = {PAYLOAD.TECHNIQUE.ERROR: _make_vector()}
         # With testMode=False, getIdentifiedDbms() prefers conf.dbmsHandler._dbms and conf.forceDbms
         # over the forced DBMS below; a leaked handler/option (e.g. MSSQL) would make the chunk-length
@@ -164,6 +170,7 @@ def tearDown(self):
         kb.testMode = self._saved["testMode"]
         kb.forceWhere = self._saved["forceWhere"]
         kb.technique = self._saved["technique"]
+        setTechnique(self._saved["td.technique"])
         kb.injection.place, kb.injection.parameter, kb.injection.data = self._saved["inj"]
         Connect.queryPage = self._saved["qp"]
         eu.Request.queryPage = self._saved["qp"]
diff --git a/tests/test_generic_takeover.py b/tests/test_generic_takeover.py
index 40f0f0c9d20..95dda4ddb1f 100644
--- a/tests/test_generic_takeover.py
+++ b/tests/test_generic_takeover.py
@@ -165,9 +165,9 @@ def test_select_joins_listlike_rows(self):
         c = Custom()
         cmod.inject.getValue = lambda query, **k: [["1", "alice"], ["2", "bob"]]
         out = c.sqlQuery("SELECT id, name FROM users;")
-        # SELECT + list-like rows => each row joined into a single scalar string.
-        self.assertEqual(len(out), 2)
-        self.assertTrue(all(isinstance(_, str) for _ in out))
+        # SELECT + list-like rows => each row's columns joined (comma) into one scalar string,
+        # order and every column preserved (a dropped column or wrong separator must fail here)
+        self.assertEqual(out, ["1,alice", "2,bob"])
 
     def test_select_scalar_passthrough(self):
         set_dbms("MySQL")
diff --git a/tests/test_jitter_stress.py b/tests/test_jitter_stress.py
new file mode 100644
index 00000000000..413cc9ef742
--- /dev/null
+++ b/tests/test_jitter_stress.py
@@ -0,0 +1,224 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+
+Adversarial JITTER stress harness for time-based blind extraction.
+
+Drives the REAL bisection() + REAL wasLastResponseDelayed() + REAL validateChar() re-validation
+against a mock oracle that returns a simulated RESPONSE DURATION (base + jitter + timeSec-if-condition-
+true) instead of a boolean - so the whole time-based decision stack runs under controlled network
+jitter, with NO real sleeping (thousands of extractions per second, fully deterministic per seed).
+The delimiter-wrapped template is what lets validateChar's per-char '!=' re-check actually fire (it is
+sqlmap's main defense against a single spike faking one bit); without it the harness is far too harsh.
+
+Two tiers:
+  * TestJitterRegression   - ALWAYS runs. Low/mild jitter MUST extract perfectly, and a spike in the
+                             baseline model MUST NOT hide genuine delays. Deterministic, fast, non-flaky.
+  * TestJitterStressSweep  - OPT-IN (set env SQLMAP_JITTER_STRESS=1). Adversarial sweeps (Gaussian
+                             sigma, heavy-tailed spikes) mapping where extraction finally degrades.
+                             Informational + loose bounds only; kept out of normal CI (slow/noisy).
+
+Run the sweep on demand:  SQLMAP_JITTER_STRESS=1 python -m unittest tests.test_jitter_stress -v
+"""
+
+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 ADJUST_TIME_DELAY, PAYLOAD
+from lib.core.settings import PAYLOAD_DELIMITER
+from lib.request.connect import Connect
+import lib.techniques.blind.inference as inf
+
+# The comparison must sit BETWEEN PAYLOAD_DELIMITERs: validateChar (inference.py) rewrites '>' to '!='
+# with a regex anchored on the delimiters, and without them that per-char re-validation silently
+# no-ops (defeating sqlmap's main per-request-spike defense and making this harness far too pessimistic).
+_TEMPLATE = "%sEXPR=%%s IDX=%%d CMP>%%d%s" % (PAYLOAD_DELIMITER, PAYLOAD_DELIMITER)
+_PARSE = re.compile(r"IDX=(\d+) CMP(!=|=|>)(\d+)")   # bisection '>'/'=' plus validateChar's '!='
+_TIMESEC = 5.0
+_BASE = 0.10          # base (non-delay) round-trip latency, seconds
+_STRESS = os.environ.get("SQLMAP_JITTER_STRESS")
+
+
+def _timeVector():
+    d = AttribDict()
+    d.payload = _TEMPLATE; d.where = 1; d.vector = _TEMPLATE
+    d.comment = ""; d.templatePayload = None; d.matchRatio = None
+    d.trueCode = None; d.falseCode = None
+    return d
+
+
+class _JitterBase(unittest.TestCase):
+    _CONF = ("threads", "api", "verbose", "direct", "disableStats", "timeSec", "predictOutput",
+             "hexConvert", "charset", "firstChar", "lastChar")
+    _KB = ("responseTimeMode", "adjustTimeDelay", "laggingChecked", "partRun", "safeCharEncode",
+           "bruteMode", "fileReadMode", "disableShiftTable", "prependFlag", "originalTimeDelay",
+           "counters", "responseTimes")
+
+    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)   # setTechnique() sets a thread-local; restore so it can't leak into other modules
+
+    def _configure(self, baselineJitter, rng, nBaseline=30):
+        set_dbms("MySQL")
+        conf.threads = 1; conf.api = False; conf.verbose = 0; conf.direct = False
+        conf.disableStats = False; conf.timeSec = _TIMESEC; conf.predictOutput = False
+        conf.hexConvert = False; conf.charset = None; conf.firstChar = None; conf.lastChar = None
+        kb.responseTimeMode = None
+        kb.adjustTimeDelay = ADJUST_TIME_DELAY.DISABLE   # never prompt / never mutate timeSec
+        kb.laggingChecked = True
+        kb.partRun = None; kb.safeCharEncode = False; kb.bruteMode = False
+        kb.fileReadMode = False; kb.disableShiftTable = False; kb.prependFlag = False
+        kb.originalTimeDelay = _TIMESEC; kb.counters = {}
+        kb.injection.data = {PAYLOAD.TECHNIQUE.TIME: _timeVector()}
+        setTechnique(PAYLOAD.TECHNIQUE.TIME)
+        # jitter is always ADDITIVE (network delays only slow a response, never speed it below base),
+        # so the baseline is right-skewed with a floor at base - like real kb.responseTimes, and with
+        # no fake point-mass at 0 that a clamp (max(0.0, ..)) would create and that would skew stats
+        kb.responseTimes = {None: [_BASE + abs(baselineJitter(rng)) for _ in range(nBaseline)]}
+        kb.data.processChar = None
+
+    def _extract(self, secret, jitter, rng):
+        from lib.core.common import wasLastResponseDelayed
+
+        def oracle(payload=None, timeBasedCompare=False, **kwargs):
+            td = getCurrentThreadData()
+            m = _PARSE.search(payload or "")
+            if not m:
+                td.lastQueryDuration = _BASE + abs(jitter(rng))
+                return False
+            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:
+                cond = not cond
+            td.lastQueryDuration = _BASE + abs(jitter(rng)) + (_TIMESEC if cond else 0.0)
+            return wasLastResponseDelayed() if timeBasedCompare else cond
+
+        Connect.queryPage = staticmethod(oracle)
+        inf.Request.queryPage = staticmethod(oracle)   # Note: staticmethod on BOTH (py2 makes a bare function an unbound method)
+        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, secret, jitter, trials=40, seed0=1000):
+        ok = 0
+        for t in range(trials):
+            rng = random.Random(seed0 + t)
+            self._configure(jitter, rng)
+            try:
+                ok += (self._extract(secret, jitter, rng) == secret)
+            except Exception:
+                pass
+        return ok, trials
+
+
+def _gaussian(sigma):
+    return lambda rng: rng.gauss(0, sigma)
+
+
+def _spike(sigma, p, mag):
+    def f(rng):
+        v = rng.gauss(0, sigma)
+        if rng.random() < p:
+            v += mag
+        return v
+    return f
+
+
+class TestJitterRegression(_JitterBase):
+    """Always-on, deterministic, non-flaky: under low/mild jitter (7*sigma well below timeSec and no
+    heavy tail) the time-based stack MUST reconstruct the value exactly, every seed."""
+
+    SECRET = "Str0ng!"
+
+    def test_no_jitter_is_perfect(self):
+        ok, n = self._rate(self.SECRET, _gaussian(0.0))
+        self.assertEqual(ok, n, "time-based extraction must be flawless with zero jitter (%d/%d)" % (ok, n))
+
+    def test_mild_gaussian_is_perfect(self):
+        # sigma=0.3 -> false bits at base+|N(0,0.3)| (<~1s) stay well under the threshold, << timeSec=5
+        ok, n = self._rate(self.SECRET, _gaussian(0.3))
+        self.assertEqual(ok, n, "mild gaussian jitter must not corrupt extraction (%d/%d)" % (ok, n))
+
+    def test_baseline_spike_does_not_hide_a_genuine_delay(self):
+        # A single latency spike captured in the response-time model must not raise the delay
+        # threshold (avg + 7*stdev) so high that a real timeSec delay is missed. Deterministic.
+        from lib.core.common import wasLastResponseDelayed, average, stdev
+        from lib.core.settings import TIME_STDEV_COEFF
+
+        set_dbms("MySQL")
+        conf.direct = False; conf.disableStats = False; conf.timeSec = _TIMESEC
+        kb.adjustTimeDelay = ADJUST_TIME_DELAY.DISABLE
+        kb.responseTimeMode = None
+        bulk = [0.15, 0.25] * 15                    # clean model, small non-zero stdev
+        kb.responseTimes = {None: bulk + [8.0]}     # one 8s spike poisons the baseline
+        td = getCurrentThreadData()
+        td.lastQueryDuration = _BASE + _TIMESEC      # a genuine time-based delay (~5.1s)
+
+        raw = kb.responseTimes[None]                 # the un-trimmed model WOULD miss it (fix is load-bearing)
+        self.assertLess(td.lastQueryDuration, average(raw) + TIME_STDEV_COEFF * stdev(raw))
+        self.assertTrue(wasLastResponseDelayed())    # with spike-trimming the delay is recognized
+
+
+@unittest.skipUnless(_STRESS, "adversarial jitter sweep is opt-in (set SQLMAP_JITTER_STRESS=1)")
+class TestJitterStressSweep(_JitterBase):
+    """Opt-in failure-surface map. Prints correctness vs jitter and asserts only loose, non-flaky
+    invariants (clean case perfect). Use to evaluate hardening changes."""
+
+    SECRET = "Str0ng!"
+
+    def test_gaussian_sweep(self):
+        # Continuous jitter: degrades only once sigma approaches timeSec/7 (7*stdev threshold nears the
+        # real delay). That is the FUNDAMENTAL limit of the statistic - the answer there is a larger
+        # timeSec (--time-sec), not a code change; shown here so a regression that degrades it earlier is visible.
+        print("\n[jitter] Gaussian sigma sweep (timeSec=%.0f, base=%.2f):" % (_TIMESEC, _BASE))
+        for sigma in (0.0, 0.3, 0.5, 0.7, 0.9, 1.2):
+            ok, n = self._rate(self.SECRET, _gaussian(sigma))
+            print("  sigma=%.2fs -> %d/%d (%3.0f%%)" % (sigma, ok, n, 100.0 * ok / n))
+            if sigma == 0.0:
+                self.assertEqual(ok, n)
+
+    def test_heavy_tailed_spike_sweep(self):
+        # One-off +8s spikes: baseline-trim (stripTimeOutliers) keeps the model clean and validateChar's
+        # '!=' re-check catches a spike that fakes a single bit, so extraction stays ~perfect until an
+        # absurd spike rate (a fifth of all requests). This is the payoff of both defenses together.
+        print("\n[jitter] Heavy-tailed spike sweep (base sigma=0.2, spike=+8s):")
+        for p in (0.0, 0.01, 0.03, 0.05, 0.10, 0.20):
+            ok, n = self._rate(self.SECRET, _spike(0.2, p, 8.0))
+            print("  spike_p=%.2f -> %d/%d (%3.0f%%)" % (p, ok, n, 100.0 * ok / n))
+            if p == 0.0:
+                self.assertEqual(ok, n)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)
+
+
+def tearDownModule():
+    reset_dbms()
diff --git a/tests/test_kerberos.py b/tests/test_kerberos.py
index 86971695e8f..47069d593ef 100644
--- a/tests/test_kerberos.py
+++ b/tests/test_kerberos.py
@@ -62,10 +62,6 @@ def _preauthError(entries):
     return {12: der.octetString(der.sequenceOf([paData]))}
 
 
-def _selectEtype(offered, hints):
-    """getTGT's etype choice: the client's own preference order, restricted to what it offered."""
-
-    return next((_ for _ in offered if _ in hints and _ in ENCTYPES), None)
 
 
 class TestKerberosAES(unittest.TestCase):
@@ -219,12 +215,14 @@ def test_hint_cannot_override_pinned_salt(self):
         self.assertEqual(client._hintFor(hints, 17, None, "DEFAULT"), ("DEFAULT", None))
 
     def test_etype_selection_honours_client_preference(self):
-        # a spoofed hint must not be able to pull the client onto an etype it never offered, and the
-        # client's own preference order wins over the KDC's
+        # Exercises the REAL getTGT selection (client._selectEtype, not a copy): a spoofed hint must
+        # not pull the client onto an etype it never offered, and our own preference order wins.
         hints = client._preauthHints(_preauthError([_etypeInfo2Entry(23), _etypeInfo2Entry(18)]))
-        self.assertEqual(_selectEtype((18, 17), hints), 18)                  # KDC listed rc4 first
-        self.assertEqual(_selectEtype((17, 18), hints), 18)                  # only 18 is hinted
-        self.assertIsNone(_selectEtype((18, 17), client._preauthHints(_preauthError([_etypeInfo2Entry(23)]))))
+        self.assertEqual(client._selectEtype((18, 17), hints), 18)           # 18 offered+hinted
+        self.assertEqual(client._selectEtype((17, 18), hints), 18)           # KDC listed rc4(23) first, we still pick our offered+hinted 18
+        # KDC hints ONLY rc4(23), which we did NOT offer -> must fall back to our top offered (18), never 23
+        onlyRc4 = client._preauthHints(_preauthError([_etypeInfo2Entry(23)]))
+        self.assertEqual(client._selectEtype((18, 17), onlyRc4), 18)
 
     def test_authenticator_timestamps_are_unique(self):
         # an acceptor's replay cache keys on (ctime, cusec), and a threaded scan mints one per request
diff --git a/tests/test_payload_marking.py b/tests/test_payload_marking.py
index 5b014e82999..1e2388c1b31 100644
--- a/tests/test_payload_marking.py
+++ b/tests/test_payload_marking.py
@@ -250,9 +250,11 @@ class TestHppReconstruction(unittest.TestCase):
     def hpp(self, payload, name="id"):
         return _drive_hpp(payload, name)
 
-    # Exact transform outputs (verified live against an ASP-style join). We pin the produced
-    # string rather than "reconstruct the SQL", because reconstruction depends on the SQL parser
-    # treating /* */ as a token separator (1/*,*/AND -> "1 AND"), which a string compare can't model.
+    # Expected outputs hand-derived from the HPP rule: each inter-token gap becomes the exact
+    # splitter "/*&<name>=*/". We pin the produced string rather than "reconstruct the SQL",
+    # because reconstruction depends on the SQL parser treating /* */ as a token separator
+    # (1/*,*/AND -> "1 AND"), which a string compare can't model. (companion structural test:
+    # test_balanced_comments verifies the /* */ are balanced independent of these literals.)
     CASES = [
         ("1",            "1"),
         ("1 AND 2=2",    "1/*&id=*/AND/*&id=*/2=2"),
diff --git a/tests/test_target_parsing.py b/tests/test_target_parsing.py
index c5a981f4a5a..5db31e8f181 100644
--- a/tests/test_target_parsing.py
+++ b/tests/test_target_parsing.py
@@ -19,7 +19,8 @@
 local SQLite session file. Those side-effecting paths are still pure with
 respect to the network, so they are exercised here against real temp dirs.
 
-All expected values below were probed from actual output, not assumed.
+Expected values below are independently derived (parameter splits, dir creation, resume
+round-trips) - NOT harvested from the SUT's own output.
 """
 
 import atexit
diff --git a/tests/test_users_enum.py b/tests/test_users_enum.py
index f20c143280a..f831108b158 100644
--- a/tests/test_users_enum.py
+++ b/tests/test_users_enum.py
@@ -160,11 +160,15 @@ def test_get_current_user(self):
 
     def test_is_dba_mysql(self):
         umod.inject.getValue = lambda query, *a, **k: "root@localhost"
-        umod.inject.checkBooleanExpression = lambda query, *a, **k: True
         users = Users()
         kb.data.currentUser = ""
+        # drive the oracle BOTH ways so a constant-True stub can't force the result
+        umod.inject.checkBooleanExpression = lambda query, *a, **k: True
         kb.data.isDba = None
         self.assertTrue(users.isDba())
+        umod.inject.checkBooleanExpression = lambda query, *a, **k: False
+        kb.data.isDba = None
+        self.assertFalse(users.isDba())
 
     def test_is_dba_postgresql_false(self):
         set_dbms("PostgreSQL")
@@ -466,12 +470,16 @@ def test_get_users_inference(self):
         self.assertEqual(sorted(res), ["guest@%", "root@localhost"])
 
     def test_is_dba_mssql(self):
-        # MSSQL isDba goes through the generic checkBooleanExpression branch.
+        # MSSQL isDba goes through the generic checkBooleanExpression branch; drive the oracle
+        # BOTH ways so a constant-True stub can't force the result
         set_dbms("Microsoft SQL Server")
-        umod.inject.checkBooleanExpression = lambda query, *a, **k: True
         users = Users()
+        umod.inject.checkBooleanExpression = lambda query, *a, **k: True
         kb.data.isDba = None
         self.assertTrue(users.isDba())
+        umod.inject.checkBooleanExpression = lambda query, *a, **k: False
+        kb.data.isDba = None
+        self.assertFalse(users.isDba())
 
 
 if __name__ == "__main__":