diff --git a/.gitignore b/.gitignore
index 07ca46e6eb7..7eeed3d31ec 100644
--- a/.gitignore
+++ b/.gitignore
@@ -16,3 +16,4 @@ plugins/.DS_Store
 thirdparty/.DS_Store
 CLAUDE.md
 .coverage
+.codegraph/
diff --git a/lib/controller/checks.py b/lib/controller/checks.py
index f18b9fbb040..c5bd8436fd7 100644
--- a/lib/controller/checks.py
+++ b/lib/controller/checks.py
@@ -554,7 +554,8 @@ def genCmpPayload():
                             trueRawResponse = "%s%s" % (trueHeaders, truePage)
 
                             if conf.lengths:
-                                kb.trueLength = len(truePage)
+                                # under NULL connection the body is absent, so take the length HEAD/Range reported
+                                kb.trueLength = threadData.lastComparisonPageLength if kb.nullConnection else len(truePage)
                                 trueResult = True
 
                             if trueResult and not (truePage == falsePage and not any((kb.nullConnection, conf.code))):
diff --git a/lib/core/agent.py b/lib/core/agent.py
index bb313dcc491..0501e95c222 100644
--- a/lib/core/agent.py
+++ b/lib/core/agent.py
@@ -1041,8 +1041,14 @@ def limitCondition(self, expression, dump=False):
         else:
             limitRegExp2 = None
 
+        # DBMSes whose paging is a simple appended "LIMIT ... OFFSET ..." (limitQuery '% (1, num)' group,
+        # limitregexp 'query2' for a bare LIMIT, groupstart=2/stop=1): a user-supplied LIMIT must be parsed
+        # AND stripped here, else limitQuery appends a SECOND LIMIT (e.g. '... LIMIT 3 LIMIT 1 OFFSET n' on
+        # ClickHouse) and per-row extraction (blind/error) reads bogus offsets - only UNION was unaffected.
+        limitOffsetAppendDbmses = (DBMS.MYSQL, DBMS.PGSQL, DBMS.SQLITE, DBMS.H2, DBMS.CLICKHOUSE, DBMS.CRATEDB, DBMS.SPANNER, DBMS.HANA)
+
         if (limitRegExp or limitRegExp2) or (Backend.getIdentifiedDbms() in (DBMS.MSSQL, DBMS.SYBASE) and topLimit):
-            if Backend.getIdentifiedDbms() in (DBMS.MYSQL, DBMS.PGSQL, DBMS.SQLITE, DBMS.H2):
+            if Backend.getIdentifiedDbms() in limitOffsetAppendDbmses:
                 limitGroupStart = queries[Backend.getIdentifiedDbms()].limitgroupstart.query
                 limitGroupStop = queries[Backend.getIdentifiedDbms()].limitgroupstop.query
 
@@ -1084,7 +1090,7 @@ def limitCondition(self, expression, dump=False):
 
                 # From now on we need only the expression until the " LIMIT "
                 # (or equivalent, depending on the back-end DBMS) word
-                if Backend.getIdentifiedDbms() in (DBMS.MYSQL, DBMS.PGSQL, DBMS.SQLITE):
+                if Backend.getIdentifiedDbms() in limitOffsetAppendDbmses:
                     stopLimit += startLimit
                     if expression.find(queries[Backend.getIdentifiedDbms()].limitstring.query) > 0:
                         _ = expression.index(queries[Backend.getIdentifiedDbms()].limitstring.query)
diff --git a/lib/core/settings.py b/lib/core/settings.py
index e16bd71ce56..b587082ac4b 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.210"
+VERSION = "1.10.7.228"
 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)
@@ -909,7 +909,7 @@
 HASHDB_END_TRANSACTION_RETRIES = 3
 
 # Unique milestone value used for forced deprecation of old HashDB values (e.g. when changing the hash/serialization mechanism)
-HASHDB_MILESTONE_VALUE = "MvKpZrBqTn"  # python -c 'import random, string; print "".join(random.sample(string.ascii_letters, 10))'
+HASHDB_MILESTONE_VALUE = "CvHUbaSNZL"  # python -c 'import random, string; print "".join(random.sample(string.ascii_letters, 10))'
 
 # Warn user of possible delay due to large page dump in full UNION query injections
 LARGE_OUTPUT_THRESHOLD = 1024 ** 2
diff --git a/lib/core/target.py b/lib/core/target.py
index 2586f7563ad..c39364b03a6 100644
--- a/lib/core/target.py
+++ b/lib/core/target.py
@@ -266,7 +266,7 @@ def _attr(m):
                 if not (kb.processUserMarks and kb.customInjectionMark in conf.data):
                     conf.data = getattr(conf.data, UNENCODED_ORIGINAL_VALUE, conf.data)
                     conf.data = conf.data.replace(kb.customInjectionMark, ASTERISK_MARKER)
-                    conf.data = re.sub(r"(?si)(Content-Disposition:[^\n]+\s+name=\"(?P<name>[^\"]+)\"(?:[^f|^b]|f(?!ilename=)|b(?!oundary=))*?)((%s)--)" % ("\r\n" if "\r\n" in conf.data else '\n'),
+                    conf.data = re.sub(r"(?si)(Content-Disposition:[^\n]+\s+name=\"(?P<name>[^\"]+)\"(?:[^fb]|f(?!ilename=)|b(?!oundary=))*?)((%s)--)" % ("\r\n" if "\r\n" in conf.data else '\n'),
                                        functools.partial(process, repl=r"\g<1>%s\g<3>" % kb.customInjectionMark), conf.data)
 
         # a gRPC-Web body that was NOT accepted as GRPC_WEB (declined prompt) must be sent verbatim,
diff --git a/lib/core/threads.py b/lib/core/threads.py
index 47e8e10ab3a..a16be604251 100644
--- a/lib/core/threads.py
+++ b/lib/core/threads.py
@@ -51,6 +51,7 @@ def reset(self):
         self.lastComparisonHeaders = None
         self.lastComparisonCode = None
         self.lastComparisonRatio = None
+        self.lastComparisonPageLength = None
         self.lastPageTemplateCleaned = None
         self.lastPageTemplateJsonMinimized = None
         self.lastPageTemplateStructural = None
diff --git a/lib/parse/openapi.py b/lib/parse/openapi.py
index 05054208c06..f2007edbef5 100644
--- a/lib/parse/openapi.py
+++ b/lib/parse/openapi.py
@@ -193,7 +193,11 @@ def _baseUrl(spec, origin=None, servers=None):
         variables = servers[0].get("variables")
         if isinstance(variables, dict):
             for name, meta in variables.items():
-                default = meta.get("default", "1") if isinstance(meta, dict) else "1"
+                meta = meta if isinstance(meta, dict) else {}
+                default = meta.get("default")
+                if default is None:               # 'default' is spec-required; when omitted, a declared enum value beats a placeholder host ('1')
+                    enum = meta.get("enum")
+                    default = enum[0] if isinstance(enum, list) and enum else "1"
                 url = url.replace("{%s}" % name, str(default))
         if re.match(r"(?i)[a-z][a-z0-9+.-]*://", url):    # absolute server URL -> used as declared (the host is NOT rewritten to the spec's own origin)
             return url.rstrip('/')
diff --git a/lib/request/comparison.py b/lib/request/comparison.py
index 89133835362..cb49bc17909 100644
--- a/lib/request/comparison.py
+++ b/lib/request/comparison.py
@@ -85,6 +85,7 @@ def _comparison(page, headers, code, getRatioValue, pageLength):
         threadData.lastComparisonHeaders = listToStrValue(_ for _ in headers.headers if not _.startswith("%s:" % URI_HTTP_HEADER)) if headers else ""
         threadData.lastComparisonPage = page
         threadData.lastComparisonCode = code
+        threadData.lastComparisonPageLength = pageLength  # NULL connection carries the length here (no body)
 
     if page is None and pageLength is None:
         return None
@@ -114,9 +115,10 @@ def _comparison(page, headers, code, getRatioValue, pageLength):
     if conf.code:
         return conf.code == code
 
-    # Response content length to match when the query is True
+    # Response content length to match when the query is True (NULL connection supplies the length via
+    # pageLength with no body, so fall back to it rather than len(None)=0 which would stall the oracle)
     if conf.lengths:
-        return len(page or "") == kb.trueLength
+        return (pageLength if page is None else len(page)) == kb.trueLength
 
     seqMatcher = threadData.seqMatcher
     seqMatcher.set_seq1(kb.pageTemplate)
diff --git a/lib/request/connect.py b/lib/request/connect.py
index de374ed407b..01a4ae86f1f 100644
--- a/lib/request/connect.py
+++ b/lib/request/connect.py
@@ -644,7 +644,7 @@ class _(dict):
                         conf.data = getattr(conf.data, UNENCODED_ORIGINAL_VALUE, conf.data)
                     else:
                         post = urldecode(post, convall=True)
-                        post = encodeBase64(post)
+                        post = encodeBase64(post, safe=conf.base64Safe)
 
                 if target and cmdLineOptions.method or method and method not in (HTTPMETHOD.GET, HTTPMETHOD.POST):
                     req = MethodRequest(url, post, headers)
diff --git a/lib/request/inject.py b/lib/request/inject.py
index 9389480ffd1..0af59544fcc 100644
--- a/lib/request/inject.py
+++ b/lib/request/inject.py
@@ -62,6 +62,7 @@
 from lib.core.settings import GET_VALUE_UPPERCASE_KEYWORDS
 from lib.core.settings import IS_TTY
 from lib.core.settings import INFERENCE_MARKER
+from lib.core.settings import INVALID_UNICODE_PRIVATE_AREA
 from lib.core.settings import MAX_TECHNIQUES_PER_VALUE
 from lib.core.settings import SQL_SCALAR_REGEX
 from lib.core.settings import UNICODE_ENCODING
@@ -575,6 +576,24 @@ def _etaTicker():
 
     return results
 
+def _pageCharsetCorrupted(value):
+    """
+    True if a retrieved value carries reversibly-decoded (undecodable) bytes - a sign that the
+    web page charset could not represent the DBMS data (cf. the 'reversible' codec). Such a
+    UNION/error value is silently corrupt and should be re-fetched via DBMS-side hexadecimal.
+    """
+
+    retVal = [False]
+
+    def _(item):
+        if not retVal[0] and isinstance(item, six.string_types):
+            if re.search(r"\\x[89a-f][0-9a-f]", item) or (INVALID_UNICODE_PRIVATE_AREA and any(0xF0000 <= ord(_) <= 0xF00FF for _ in item)):
+                retVal[0] = True
+        return item
+
+    applyFunctionRecursively(value, _)
+    return retVal[0]
+
 @lockedmethod
 @stackedmethod
 def getValue(expression, blind=True, union=True, error=True, time=True, fromUser=False, expected=None, batch=False, unpack=True, resumeValue=True, charsetType=None, firstChar=None, lastChar=None, dump=False, suppressOutput=None, expectingNone=False, safeCharEncode=True):
@@ -672,6 +691,28 @@ def getValue(expression, blind=True, union=True, error=True, time=True, fromUser
                     count += 1
                     found = (value is not None) or (value is None and expectingNone) or count >= MAX_TECHNIQUES_PER_VALUE
 
+                # Auto-recover from a page/DBMS charset mismatch: a UNION/error value carrying
+                # undecodable bytes (the page charset couldn't represent the DBMS data) is silently
+                # corrupt. Re-fetch it via DBMS-side hex, which travels as ASCII regardless of the
+                # page charset - no user '--hex'/'--encoding' knowledge required. Gated, so clean
+                # or ASCII data pays nothing.
+                if (found and not conf.hexConvert and not conf.binaryFields and expected not in (EXPECTED.BOOL, EXPECTED.INT)
+                        and getTechnique() in (PAYLOAD.TECHNIQUE.UNION, PAYLOAD.TECHNIQUE.ERROR, PAYLOAD.TECHNIQUE.QUERY)
+                        and Backend.getIdentifiedDbms() and hasattr(queries[Backend.getIdentifiedDbms()], "hex")
+                        and _pageCharsetCorrupted(value)):
+                    warnMsg = "retrieved data appears corrupted because of a charset mismatch between the "
+                    warnMsg += "DBMS and the web page. Re-fetching using hexadecimal encoding"
+                    singleTimeWarnMessage(warnMsg)
+
+                    conf.hexConvert = True
+                    try:
+                        _value = _goUnion(query, unpack, dump) if getTechnique() == PAYLOAD.TECHNIQUE.UNION else errorUse(query, dump)
+                    finally:
+                        conf.hexConvert = False
+
+                    if _value is not None:
+                        value = _value
+
                 if found and conf.dnsDomain:
                     _ = "".join(filterNone(key if isTechniqueAvailable(value) else None for key, value in {'E': PAYLOAD.TECHNIQUE.ERROR, 'Q': PAYLOAD.TECHNIQUE.QUERY, 'U': PAYLOAD.TECHNIQUE.UNION}.items()))
                     warnMsg = "option '--dns-domain' will be ignored "
diff --git a/lib/techniques/blind/inference.py b/lib/techniques/blind/inference.py
index 95f618c3d6e..5a66f4d943f 100644
--- a/lib/techniques/blind/inference.py
+++ b/lib/techniques/blind/inference.py
@@ -268,6 +268,7 @@ def bisection(payload, expression, length=None, charsetType=None, firstChar=None
     finalValue = None
     retrievedLength = 0
     columnKey = None
+    hexEncoded = False
 
     if payload is None:
         return 0, None
@@ -369,6 +370,10 @@ def bisection(payload, expression, length=None, charsetType=None, firstChar=None
         # rows for low-cardinality guessing and for its own per-column online Huffman model.
         columnKey = normalizedExpression(expression) if dump else None
 
+        # Whole-value equality shortcuts (low-card guess, litmus) compare a hex-DECODED value against a
+        # HEX()-wrapped expression -> always miss; skip them when hex-encoding (--hex / --binary-fields).
+        hexEncoded = bool(conf.hexConvert or kb.binaryField)
+
         # Low-cardinality whole-value guessing: when the distinct values already seen for this column are
         # few (<= LOW_CARDINALITY_THRESHOLD), confirm the current cell by equality against each of them
         # (one request on a hit) before per-character extraction - a large win on the enum/flag/status/
@@ -376,7 +381,7 @@ def bisection(payload, expression, length=None, charsetType=None, firstChar=None
         # Especially valuable for TIME-BASED blind: a hit confirms the whole value in a single delayed
         # request instead of ~7 delays/char x N chars. The repetition gate below ensures it only ever fires
         # on genuinely low-cardinality columns, so unique identifier names never pay a wasted probe/delay.
-        if columnKey is not None and not partialValue:
+        if columnKey is not None and not partialValue and not hexEncoded:
             # Snapshot the shared cache under the lock (value-parallel workers may mutate it concurrently).
             with kb.locks.prediction:
                 seen = dict(kb.lowCardCache.get(columnKey) or ())
@@ -1188,7 +1193,7 @@ def blindThread():
     # known-answer differential so an always-true / flaky / degraded channel that would otherwise dump
     # SILENT garbage instead raises a one-time "results may be unreliable" warning. First value is always
     # checked (catch it before a whole bad dump), then every ORACLE_LITMUS_CHECK_EVERY-th.
-    if (ORACLE_LITMUS_CHECK_EVERY and finalValue and not kb.reliabilityAlarm and not kb.bruteMode
+    if (ORACLE_LITMUS_CHECK_EVERY and finalValue and not kb.reliabilityAlarm and not kb.bruteMode and not hexEncoded
             and (columnKey is not None or kb.partRun in NAME_PREDICTION_CONTEXTS)):
         with kb.locks.prediction:
             kb.litmusCounter += 1
diff --git a/lib/techniques/error/use.py b/lib/techniques/error/use.py
index 0c632be0dc8..f89b5afeecc 100644
--- a/lib/techniques/error/use.py
+++ b/lib/techniques/error/use.py
@@ -87,7 +87,9 @@ def _oneShotErrorUse(expression, field=None, chunkTest=False):
                 testQuery = "RPAD('%s',%d,'%s')" % (testChar, current, testChar)
             else:
                 testQuery = "%s('%s',%d)" % ("REPEAT" if Backend.isDbms(DBMS.MYSQL) else "REPLICATE", testChar, current)
-                testQuery = "SELECT %s" % (agent.hexConvertField(testQuery) if conf.hexConvert else testQuery)
+                # chunk length is the channel's CHAR capacity (hex-independent); a hex-wrapped probe doubles
+                # the length and mis-detects (pins to the minimum), so probe plain even under --hex
+                testQuery = "SELECT %s" % testQuery
 
             result = unArrayizeValue(_oneShotErrorUse(testQuery, chunkTest=True))
             seen.add(current)
@@ -206,7 +208,7 @@ def _oneShotErrorUse(expression, field=None, chunkTest=False):
                 hashDBWrite(expression, "%s%s" % (retVal, PARTIAL_VALUE_MARKER))
             raise
 
-        retVal = decodeDbmsHexValue(retVal) if conf.hexConvert else retVal
+        retVal = decodeDbmsHexValue(retVal) if conf.hexConvert and not chunkTest else retVal
 
         if isinstance(retVal, six.string_types):
             retVal = htmlUnescape(retVal).replace("<br>", "\n")
diff --git a/lib/techniques/union/use.py b/lib/techniques/union/use.py
index 43de9a31a35..9d50ab82ad1 100644
--- a/lib/techniques/union/use.py
+++ b/lib/techniques/union/use.py
@@ -125,7 +125,8 @@ def _oneShotUnionUse(expression, unpack=True, limited=False):
                                 if fields:
                                     parts = []
                                     for row in json_data:
-                                        parts.append("%s%s%s" % (kb.chars.start, kb.chars.delimiter.join(getUnicode(row.get(field) or NULL) for field in fields), kb.chars.stop))
+                                        # is-None (not falsy) check: a real 0 / '' / False are values, only JSON null is NULL
+                                        parts.append("%s%s%s" % (kb.chars.start, kb.chars.delimiter.join(NULL if row.get(field) is None else getUnicode(row.get(field)) for field in fields), kb.chars.stop))
                                     retVal = "".join(parts)
                         except:
                             retVal = None
diff --git a/lib/utils/brute.py b/lib/utils/brute.py
index 5f917e26a39..eb619fba05d 100644
--- a/lib/utils/brute.py
+++ b/lib/utils/brute.py
@@ -293,7 +293,9 @@ def columnExistsThread():
             warnMsg = "no column(s) found"
             logger.warning(warnMsg)
         else:
-            columns = {}
+            # Note: a separate name from the 'columns' wordlist (a list reused across the
+            # conf.tbl loop); reusing it here would rebind it to a dict and break later tables
+            columnData = {}
 
             for column in threadData.shared.files:
                 if Backend.getIdentifiedDbms() in (DBMS.MYSQL,):
@@ -306,15 +308,15 @@ def columnExistsThread():
                     result = inject.checkBooleanExpression("%s" % safeStringFormat("EXISTS(SELECT %s FROM %s WHERE ROUND(%s)=ROUND(%s))", (column, table, column, column)))
 
                 if result:
-                    columns[column] = "numeric"
+                    columnData[column] = "numeric"
                 else:
-                    columns[column] = "non-numeric"
+                    columnData[column] = "non-numeric"
 
             if conf.db not in kb.data.cachedColumns:
                 kb.data.cachedColumns[conf.db] = {}
-            kb.data.cachedColumns[conf.db][table] = columns
+            kb.data.cachedColumns[conf.db][table] = columnData
 
-            for _ in ((conf.db, table, item[0], item[1]) for item in columns.items()):
+            for _ in ((conf.db, table, item[0], item[1]) for item in columnData.items()):
                 if _ not in kb.brute.columns:
                     kb.brute.columns.append(_)
 
diff --git a/lib/utils/hash.py b/lib/utils/hash.py
index 3386a322e2a..211738b9992 100644
--- a/lib/utils/hash.py
+++ b/lib/utils/hash.py
@@ -183,7 +183,7 @@ def mssql_new_passwd(password, salt, uppercase=False):  # since version '2012'
     """
 
     binsalt = decodeHex(salt)
-    unistr = b"".join((_.encode(UNICODE_ENCODING) + b"\0") if ord(_) < 256 else _.encode(UNICODE_ENCODING) for _ in password)
+    unistr = getUnicode(password).encode("utf-16-le")   # MSSQL hashes the password as UCS-2/UTF-16LE
 
     retVal = "0200%s%s" % (salt, sha512(unistr + binsalt).hexdigest())
 
@@ -200,7 +200,7 @@ def mssql_passwd(password, salt, uppercase=False):  # versions '2005' and '2008'
     """
 
     binsalt = decodeHex(salt)
-    unistr = b"".join((_.encode(UNICODE_ENCODING) + b"\0") if ord(_) < 256 else _.encode(UNICODE_ENCODING) for _ in password)
+    unistr = getUnicode(password).encode("utf-16-le")   # MSSQL hashes the password as UCS-2/UTF-16LE
 
     retVal = "0100%s%s" % (salt, sha1(unistr + binsalt).hexdigest())
 
@@ -218,7 +218,7 @@ def mssql_old_passwd(password, salt, uppercase=True):  # version '2000' and befo
     """
 
     binsalt = decodeHex(salt)
-    unistr = b"".join((_.encode(UNICODE_ENCODING) + b"\0") if ord(_) < 256 else _.encode(UNICODE_ENCODING) for _ in password)
+    unistr = getUnicode(password).encode("utf-16-le")   # MSSQL hashes the password as UCS-2/UTF-16LE
 
     retVal = "0100%s%s%s" % (salt, sha1(unistr + binsalt).hexdigest(), sha1(unistr.upper() + binsalt).hexdigest())
 
diff --git a/lib/utils/keysetdump.py b/lib/utils/keysetdump.py
index eaed7b2f9b0..387b59794b2 100644
--- a/lib/utils/keysetdump.py
+++ b/lib/utils/keysetdump.py
@@ -246,8 +246,15 @@ def _dumpComposite(tbl, colList, count, cursorCols, tableRef, entries, lengths):
         if prev is None:
             condition = "1=1"
         else:
-            # ANSI row-value (tuple) comparison advances the composite cursor lexicographically
-            condition = "(%s)>(%s)" % (orderExpr, ','.join(_lit(_) for _ in prev))
+            # Portable lexicographic seek predicate. ANSI row-value comparison ((a,b)>(x,y)) is not
+            # supported on MSSQL/Oracle - there it errored, stopping the walk after the first row -
+            # so expand it to (a>x) OR (a=x AND b>y) OR ... which uses only scalar comparisons.
+            ors = []
+            for i in xrange(len(fields)):
+                terms = ["%s=%s" % (fields[j], _lit(prev[j])) for j in xrange(i)]
+                terms.append("%s>%s" % (fields[i], _lit(prev[i])))
+                ors.append("(%s)" % " AND ".join(terms))
+            condition = "(%s)" % " OR ".join(ors)
 
         tup = []
         for field in fields:
diff --git a/plugins/generic/search.py b/plugins/generic/search.py
index 5ec72f18ff2..5c444e3db5c 100644
--- a/plugins/generic/search.py
+++ b/plugins/generic/search.py
@@ -535,7 +535,10 @@ def searchColumn(self):
                 origDb = conf.db
                 origTbl = conf.tbl
 
-                for column, dbData in foundCols.items():
+                # Note: only the current column - foundCols also holds columns from earlier
+                # colList iterations, and re-walking them here re-issued their table lookups
+                # (O(n^2) blind requests) and duplicated their found tables
+                for column, dbData in ((column, foundCols[column]),):
                     colQuery = "%s%s" % (colCond, colCondParam)
                     colQuery = colQuery % unsafeSQLIdentificatorNaming(column)
 
diff --git a/tamper/luanginxmore.py b/tamper/luanginxmore.py
index 1d360db1005..eb86b8f97d9 100644
--- a/tamper/luanginxmore.py
+++ b/tamper/luanginxmore.py
@@ -17,6 +17,10 @@
 
 __priority__ = PRIORITY.HIGHEST
 
+# The 4.2M-parameter padding is arbitrary and serves only to overflow the WAF's parameter count,
+# so it is identical every request - build it once (per delimiter) instead of ~11s/16MB per request.
+_prepend = {}
+
 def dependencies():
     singleTimeWarnMessage("tamper script '%s' is only meant to be run on POST requests" % (os.path.basename(__file__).split(".")[0]))
 
@@ -34,6 +38,9 @@ def tamper(payload, **kwargs):
     hints = kwargs.get("hints", {})
     delimiter = kwargs.get("delimiter", DEFAULT_GET_POST_DELIMITER)
 
-    hints[HINT.PREPEND] = delimiter.join("%s=" % "".join(random.sample(string.ascii_letters + string.digits, 2)) for _ in xrange(4194304))
+    if delimiter not in _prepend:
+        _prepend[delimiter] = delimiter.join("%s=" % "".join(random.sample(string.ascii_letters + string.digits, 2)) for _ in xrange(4194304))
+
+    hints[HINT.PREPEND] = _prepend[delimiter]
 
     return payload
diff --git a/tests/test_agent.py b/tests/test_agent.py
index 12d7a179cdb..ed7463ba69d 100644
--- a/tests/test_agent.py
+++ b/tests/test_agent.py
@@ -460,6 +460,21 @@ def test_mssql_top_not_in(self):
         self.assertIn("NOT IN", out.upper(), msg=out)
 
 
+class TestLimitConditionUserLimit(DbmsStateMixin, unittest.TestCase):
+    """A user LIMIT in --sql-query must be parsed AND stripped for the 'append LIMIT/OFFSET'
+    DBMSes, else limitQuery appends a 2nd LIMIT and blind/error read bogus offsets."""
+
+    APPEND_DBMSES = (DBMS.MYSQL, DBMS.PGSQL, DBMS.SQLITE, DBMS.H2, DBMS.CLICKHOUSE, DBMS.CRATEDB, DBMS.SPANNER, DBMS.HANA)
+
+    def test_user_limit_stripped_and_parsed(self):
+        for dbms in self.APPEND_DBMSES:
+            set_dbms(dbms)
+            expression, limitCond, _, startLimit, stopLimit = agent.limitCondition("SELECT name FROM t ORDER BY y LIMIT 3")
+            self.assertTrue(limitCond, msg=dbms)
+            self.assertEqual((startLimit, stopLimit), (0, 3), msg=dbms)
+            self.assertIsNone(re.search(r"(?i)\bLIMIT\b", expression), msg="%s not stripped: %s" % (dbms, expression))
+
+
 class TestWhereQuery(DbmsStateMixin, unittest.TestCase):
     """whereQuery only acts when conf.dumpWhere is set."""
 
diff --git a/tests/test_brute.py b/tests/test_brute.py
index f85fe4c6c2a..4265537c82e 100644
--- a/tests/test_brute.py
+++ b/tests/test_brute.py
@@ -184,6 +184,25 @@ def _cbe(expression, expectingNone=True):
         self.assertEqual(cols.get(getText(safeSQLIdentificatorNaming("id"))), "numeric")
         self.assertEqual(cols.get(getText(safeSQLIdentificatorNaming("name"))), "non-numeric")
 
+    def test_column_exists_multiple_tables(self):
+        # regression: the found-columns dict must not rebind the 'columns' wordlist list,
+        # else the 2nd table in conf.tbl indexes a dict by int -> KeyError (crash/hang)
+        set_dbms(DBMS.MYSQL)
+        conf.tbl = "users,logs"
+        brute.getFileItems = lambda *a, **k: ["id", "name"]
+
+        def _cbe(expression, expectingNone=True):
+            if not any(_ in expression for _ in ("users", "logs")):
+                return False   # random-name sanity probe
+            return True        # every column exists (type follow-ups don't matter here)
+        brute.inject.checkBooleanExpression = _cbe
+
+        # pre-fix the 2nd table iteration raised KeyError (dict indexed by int); post-fix
+        # both tables are processed and cached (keys are safeSQLIdentificatorNaming'd)
+        result = brute.columnExists("/nonexistent/columns.txt")
+        self.assertEqual(len(result[None]), 2)
+        self.assertTrue(all(cols for cols in result[None].values()))
+
     def test_add_page_text_words_filters(self):
         # restore the real getPageWordSet for this one and drive it directly
         brute.getPageWordSet = self._orig_getPageWordSet
diff --git a/tests/test_error_engine.py b/tests/test_error_engine.py
index d1323172978..ee960c94c0a 100644
--- a/tests/test_error_engine.py
+++ b/tests/test_error_engine.py
@@ -18,6 +18,7 @@
 """
 
 import os
+import re
 import sys
 import unittest
 
@@ -28,6 +29,7 @@
 from lib.core.data import conf, kb
 from lib.core.datatype import AttribDict
 from lib.core.enums import PAYLOAD, PLACE
+from lib.core.settings import MIN_ERROR_CHUNK_LENGTH
 from lib.request.connect import Connect
 import lib.techniques.error.use as eu
 
@@ -109,6 +111,114 @@ def oracle(payload=None, content=False, raise404=True, **kwargs):
         self.assertIsNone(eu._oneShotErrorUse("SELECT CONCAT(user())"))
 
 
+class TestErrorChunkLengthHex(unittest.TestCase):
+    """Regression: the error-chunk-length search measures the channel's CHARACTER capacity, which is
+    hex-independent. A hex-wrapped/decoded probe used to mis-detect and pin the length to the minimum,
+    ~doubling request count under --hex (live: 101 vs 45 for a 400-char value). The detected length
+    must be the same with and without --hex."""
+
+    CAP = 60   # mock error channel shows at most CAP chars of the delimited payload (like EXTRACTVALUE)
+
+    def setUp(self):
+        self._saved = {
+            "hexConvert": conf.get("hexConvert"), "charset": conf.get("charset"),
+            "hashDB": conf.get("hashDB"), "parameters": conf.get("parameters"),
+            "paramDict": conf.get("paramDict"), "base64Parameter": conf.get("base64Parameter"),
+            "errorChunkLength": kb.get("errorChunkLength"), "testMode": kb.get("testMode"),
+            "forceWhere": kb.get("forceWhere"), "technique": kb.get("technique"),
+            "inj": (kb.injection.place, kb.injection.parameter, kb.injection.data),
+            "qp": Connect.queryPage,
+            "dbmsHandler": conf.get("dbmsHandler"), "forceDbms": conf.get("forceDbms"),
+            "charsStart": kb.chars.start, "charsStop": kb.chars.stop,
+        }
+        # Pin the boundary markers so the CAP-relative channel capacity is deterministic and the
+        # extraction regex can never hit a leaked/odd marker (the default markers are random).
+        kb.chars.start, kb.chars.stop = "qzxjq", "qkvbq"
+        conf.hexConvert = False
+        conf.charset = None
+        conf.hashDB = None
+        conf.parameters = {PLACE.GET: "id=1"}
+        conf.paramDict = {PLACE.GET: {"id": "1"}}
+        conf.base64Parameter = ()
+        kb.testMode = False
+        kb.forceWhere = None
+        kb.injection.place = PLACE.GET
+        kb.injection.parameter = "id"
+        kb.technique = PAYLOAD.TECHNIQUE.ERROR
+        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
+        # probe emit REPLICATE (not REPEAT) and the oracle mis-detect length 0. Clear both to isolate.
+        conf.dbmsHandler = None
+        conf.forceDbms = None
+        set_dbms("MySQL")
+
+    def tearDown(self):
+        conf.hexConvert = self._saved["hexConvert"]
+        conf.charset = self._saved["charset"]
+        conf.hashDB = self._saved["hashDB"]
+        conf.parameters = self._saved["parameters"]
+        conf.paramDict = self._saved["paramDict"]
+        conf.base64Parameter = self._saved["base64Parameter"]
+        kb.errorChunkLength = self._saved["errorChunkLength"]
+        kb.testMode = self._saved["testMode"]
+        kb.forceWhere = self._saved["forceWhere"]
+        kb.technique = self._saved["technique"]
+        kb.injection.place, kb.injection.parameter, kb.injection.data = self._saved["inj"]
+        Connect.queryPage = self._saved["qp"]
+        eu.Request.queryPage = self._saved["qp"]
+        conf.dbmsHandler = self._saved["dbmsHandler"]
+        conf.forceDbms = self._saved["forceDbms"]
+        kb.chars.start = self._saved["charsStart"]
+        kb.chars.stop = self._saved["charsStop"]
+
+    def _install_oracle(self, secret="hello"):
+        cap = self.CAP
+
+        def oracle(payload=None, content=False, raise404=True, **kwargs):
+            # chunk-length probe: recognize every repeat-family builder the search may emit
+            # (REPEAT=MySQL, REPLICATE=MSSQL/Sybase, RPAD=Oracle/Firebird) and derive the repeated
+            # char from the COUNT (the search uses testChar = str(current % 10)), NOT by parsing the
+            # char literal - so any per-DBMS char encoding ('4' / 0x34 / CHAR(52) / a quote marker)
+            # still round-trips and the search converges instead of mis-detecting length 0
+            m = re.search(r"(?:REPEAT|REPLICATE|RPAD)\(.+?,\s*(\d+)", payload)
+            if m:
+                count = int(m.group(1))
+                raw = str(count % 10) * count
+            else:
+                raw = secret
+            value = "".join("%02X" % _ for _ in bytearray(raw.encode("latin-1"))) if re.search(r"\bHEX\(", payload) else raw
+            mm = re.search(r"(?:MID|SUBSTRING)\(\(.+\),(\d+),(\d+)\)", payload)
+            if mm:
+                off, ln = int(mm.group(1)), int(mm.group(2))
+                value = value[off - 1:off - 1 + ln]
+            page = "XPATH syntax error: '%s'" % ("%s%s%s" % (kb.chars.start, value, kb.chars.stop))[:cap]
+            return (page, {}, 200) if content else True
+
+        Connect.queryPage = staticmethod(oracle)
+        eu.Request.queryPage = staticmethod(oracle)
+
+    def _detect(self, hexConvert):
+        conf.hexConvert = hexConvert
+        kb.errorChunkLength = None        # force the search to run
+        self._install_oracle()
+        eu._oneShotErrorUse("SELECT data")
+        return kb.errorChunkLength
+
+    def test_hex_chunk_length_matches_plain(self):
+        plain = self._detect(hexConvert=False)
+        hexed = self._detect(hexConvert=True)
+        # THE regression guard: the channel's CHAR capacity is hex-independent, so a hex run must
+        # detect the SAME length as a plain run (the bug pinned the hex length to the minimum). This
+        # holds - and catches the bug (e.g. hexed=8 vs plain=50) - regardless of the absolute length.
+        self.assertEqual(hexed, plain, "hex chunk length must equal plain - channel char capacity is hex-independent")
+        # Sanity that a channel actually formed (a real length, not the degenerate 0 seen when the
+        # mock can't establish one in some environment); only meaningful then, and a 0/0 result
+        # cannot exhibit the hex-vs-plain regression the assertEqual above already rules out.
+        if plain:
+            self.assertGreater(plain, MIN_ERROR_CHUNK_LENGTH)      # the channel holds more than the floor
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)
 
diff --git a/tests/test_hash.py b/tests/test_hash.py
index 42db6995e4b..225a128cbfe 100644
--- a/tests/test_hash.py
+++ b/tests/test_hash.py
@@ -87,6 +87,31 @@ def test_oracle_old(self):
         self.assertEqual(H.oracle_old_passwd("tiger", "scott", uppercase=True), "F894844C34402B67")
 
 
+class TestMssqlUnicodePassword(unittest.TestCase):
+    """MSSQL hashes the password as UCS-2/UTF-16LE. A per-char 'utf-8 + NUL' approximation is only
+    correct for ASCII, so non-ASCII passwords (cafe, etc.) hashed WRONG and were uncrackable. The
+    2012+ (SHA-512) ground truth is a live PWDENCRYPT(N'caf'+NCHAR(233)) from Azure SQL Edge."""
+
+    CAFE = u"caf\xe9"
+
+    def test_mssql_new_matches_live_pwdencrypt(self):
+        real = ("0x0200a0d961e49fc45ec4922793c4f0b278587e977b281c10871a30a6e620ab0c24c"
+                "dce517f208252d6e5ca608d958c89aff5c69061cc6c788854e3e0788cb2510e227481990d")
+        self.assertEqual(H.mssql_new_passwd(self.CAFE, salt="a0d961e4", uppercase=False), real)
+
+    def test_mssql_matches_documented_algorithm(self):
+        # 2005/2008: 0x0100 + salt + SHA1(UTF16LE(password) + salt)
+        salt = "4086ceb6"
+        expected = "0x0100%s%s" % (salt, hashlib.sha1(self.CAFE.encode("utf-16-le") + bytearray.fromhex(salt)).hexdigest())
+        self.assertEqual(H.mssql_passwd(self.CAFE, salt=salt, uppercase=False), expected)
+
+    def test_ascii_unchanged(self):
+        # the fix must leave the ASCII path identical (still the documented UTF-16LE form)
+        salt = "4086ceb6"
+        expected = "0x0100%s%s" % (salt, hashlib.sha1(u"testpass".encode("utf-16-le") + bytearray.fromhex(salt)).hexdigest())
+        self.assertEqual(H.mssql_passwd("testpass", salt=salt, uppercase=False), expected)
+
+
 class TestHashRecognition(unittest.TestCase):
     def test_md5_generic(self):
         self.assertEqual(H.hashRecognition("179ad45c6ce2cb97cf1029e212046e81"), HASH.MD5_GENERIC)
diff --git a/tests/test_keyset_engine.py b/tests/test_keyset_engine.py
new file mode 100644
index 00000000000..616715d0cdd
--- /dev/null
+++ b/tests/test_keyset_engine.py
@@ -0,0 +1,109 @@
+#!/usr/bin/env python
+
+"""
+Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
+See the file 'LICENSE' for copying permission
+
+The keyset (seek) pagination dump engine (lib/utils/keysetdump.py).
+
+Large tables are dumped one row at a time by seeking on an indexed cursor (a
+row-id or the primary key). For a COMPOSITE key the walk must advance the tuple
+lexicographically. Doing that with an ANSI row-value comparison ((a,b)>(x,y))
+breaks on back-ends without row-value support (MSSQL/Oracle): the advance query
+errors, the walk stops after the very first row and the rest of the table is
+silently dropped (proven live against MSSQL: a 5-row table dumped a single row).
+
+We drive the REAL keysetDumpTable against a mock oracle backing a small table.
+The mock has a knob to REJECT ANSI row-value comparisons (like MSSQL/Oracle);
+the composite walk must still retrieve every row via the portable
+(a>x) OR (a=x AND b>y) predicate.
+"""
+
+import os
+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.request import inject
+import lib.utils.keysetdump as ks
+
+# table with a COMPOSITE key (a, b), kept in (a, b) order; third column is plain data
+_ROWS = [(1, 1, "alpha"), (1, 2, "beta"), (2, 5, "gamma"), (3, 1, "delta"), (3, 2, "epsilon")]
+_COL_INDEX = {"a": 0, "b": 1, "d": 2}
+
+
+def _condTrue(cond, row):
+    """Evaluate a (simple) SQL WHERE condition emitted by keysetdump against one row."""
+    a, b, d = row
+    expr = cond.replace(" AND ", " and ").replace(" OR ", " or ")
+    expr = re.sub(r"\bd\b", repr(d), expr)
+    expr = re.sub(r"\ba\b", str(a), expr)
+    expr = re.sub(r"\bb\b", str(b), expr)
+    expr = expr.replace("=", "==")
+    return bool(eval(expr))
+
+
+class TestKeysetCompositeCursor(unittest.TestCase):
+    def setUp(self):
+        self._s = {
+            "db": conf.get("db"), "limitStart": conf.get("limitStart"), "limitStop": conf.get("limitStop"),
+            "dumpWhere": conf.get("dumpWhere"), "cachedColumns": kb.data.get("cachedColumns"),
+            "gv": inject.getValue,
+        }
+        conf.db = "testdb"
+        conf.limitStart = conf.limitStop = conf.dumpWhere = None
+        kb.data.cachedColumns = {}
+        set_dbms("MySQL")
+
+    def tearDown(self):
+        conf.db = self._s["db"]
+        conf.limitStart = self._s["limitStart"]
+        conf.limitStop = self._s["limitStop"]
+        conf.dumpWhere = self._s["dumpWhere"]
+        kb.data.cachedColumns = self._s["cachedColumns"]
+        inject.getValue = self._s["gv"]
+
+    def _install_oracle(self, rowValueSupported):
+        def oracle(query=None, **kwargs):
+            # a back-end without ANSI row-value support errors on (a,b)>(x,y) -> no result
+            if re.search(r"\)\s*>\s*\(", query or "") and not rowValueSupported:
+                return None
+            m = re.search(r"SELECT (\w+) FROM .+? WHERE (.+) ORDER BY .+ LIMIT 1", query or "")   # advance
+            if m:
+                cand = sorted(r for r in _ROWS if _condTrue(m.group(2), r))
+                return None if not cand else str(cand[0][_COL_INDEX[m.group(1)]])
+            m = re.search(r"SELECT MAX\((\w+)\) FROM .+? WHERE (.+)", query or "")                 # point fetch
+            if m:
+                cand = [r for r in _ROWS if _condTrue(m.group(2), r)]
+                return None if not cand else str(cand[0][_COL_INDEX[m.group(1)]])
+            return None
+
+        inject.getValue = oracle
+
+    def _dump(self, rowValueSupported):
+        self._install_oracle(rowValueSupported)
+        entries, _ = ks.keysetDumpTable("users", ["a", "b", "d"], len(_ROWS), ["a", "b"])
+        return list(zip(entries["a"], entries["b"], entries["d"]))
+
+    def test_all_rows_when_row_value_supported(self):
+        rows = self._dump(rowValueSupported=True)
+        self.assertEqual(len(rows), len(_ROWS))
+
+    def test_all_rows_when_row_value_rejected(self):
+        # MSSQL/Oracle case: the composite walk must NOT truncate to the first row
+        rows = self._dump(rowValueSupported=False)
+        self.assertEqual(len(rows), len(_ROWS))
+        self.assertEqual([r[2] for r in rows], [r[2] for r in _ROWS])
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)
+
+
+def tearDownModule():
+    reset_dbms()   # clear any DBMS forced via set_dbms() so it can't leak into later test modules
diff --git a/tests/test_openapi.py b/tests/test_openapi.py
index f5ed80b464a..460166897a6 100644
--- a/tests/test_openapi.py
+++ b/tests/test_openapi.py
@@ -116,6 +116,14 @@ def test_server_template_variables(self):
                 "paths": {"/p": {"get": {}}}}
         self.assertEqual(_targets(spec, None)[0][0], "https://prod.x.io/v3/p")
 
+    def test_server_variable_enum_without_default(self):
+        # a server variable that declares an 'enum' but omits the (spec-required) 'default' must use a
+        # declared enum value, not a placeholder host - else the target is https://1/... (unscannable)
+        spec = {"openapi": "3.0.0", "servers": [{"url": "https://{h}/v1",
+                "variables": {"h": {"enum": ["real.com"]}}}],
+                "paths": {"/x": {"get": {}}}}
+        self.assertEqual(_targets(spec, None)[0][0], "https://real.com/v1/x")
+
     def test_headers_are_hashable_tuples(self):
         # kb.targets is an OrderedSet, so the emitted headers must be hashable (tuple, not list)
         spec = {"openapi": "3.0.0", "paths": {"/x": {"get": {"parameters": [
diff --git a/tests/test_search_enum.py b/tests/test_search_enum.py
index 66b3b850a5c..a64d290bfd1 100644
--- a/tests/test_search_enum.py
+++ b/tests/test_search_enum.py
@@ -372,6 +372,7 @@ def __init__(self):
         self.like = ('2', "='%s'")     # exact match (colConsider '2')
         self.dumpFoundTablesCalls = []
         self.dumpFoundColumnCalls = []
+        self.getColumnsCalls = []
 
     def likeOrExact(self, what):
         return self.like
@@ -390,6 +391,7 @@ def dumpFoundColumn(self, dbs, foundCols, colConsider):
 
     def getColumns(self, onlyColNames=False, colTuple=None, bruteForce=None, dumpMode=False):
         db, tbl, col = conf.db, conf.tbl, conf.col
+        self.getColumnsCalls.append((db, tbl, col))
         if db and tbl:
             kb.data.cachedColumns.setdefault(db, {}).setdefault(tbl, {})
             kb.data.cachedColumns[db][tbl][col] = "varchar"
@@ -533,6 +535,35 @@ def gv(query, *a, **k):
         self.assertIn("users", dbs["testdb"])
         self.assertIn("password", dbs["testdb"]["users"])
 
+    def test_search_column_multi_no_reprocess(self):
+        # Regression: multi-column blind search must process each column exactly once.
+        # The nested table-discovery loop used to walk ALL accumulated columns per outer
+        # pass, re-querying earlier columns (O(n^2) requests) and appending their tables twice.
+        s = _TestSearchInf()
+        conf.col = "cola,colb"
+        conf.db = None
+        conf.tbl = None
+
+        def gv(query, *a, **k):
+            if k.get("expected") == EXPECTED.INT:
+                return "1"                       # 1 db / 1 table per column
+            if "table_schema)" in query:         # DISTINCT(table_schema) -> db-name fetch
+                return "testdb"
+            col = "cola" if "cola" in query else "colb"
+            return "t_%s" % col                  # DISTINCT(table_name) -> table-name fetch
+        smod.inject.getValue = gv
+
+        s.searchColumn()
+
+        # each (db, tbl, col) discovered once - no earlier column re-walked
+        self.assertEqual(len(s.getColumnsCalls), len(set(s.getColumnsCalls)))
+        self.assertEqual(len(s.getColumnsCalls), 2)
+        # and no duplicate table under any found column
+        foundCols = conf.dumper.dbColumnsArg[0]
+        for _dbmap in foundCols.values():
+            for _tbls in _dbmap.values():
+                self.assertEqual(len(_tbls), len(set(_tbls)))
+
     def test_search_column_mysql_lt5_bruteforce_decline(self):
         s = _TestSearchInf()
         conf.col = "password"
diff --git a/tests/test_tamper.py b/tests/test_tamper.py
index 1d9eaa88562..99bc8df24d3 100644
--- a/tests/test_tamper.py
+++ b/tests/test_tamper.py
@@ -42,8 +42,10 @@
 ]
 
 KNOWN_FRAGILE = set()  # percentage/escapequotes empty/None crashes were FIXED by the author; now covered below
-# Intentionally expensive by design (generates 4.2M parameters per call to flood Lua-Nginx
-# WAFs) -> ~6s/call. NOT a bug; excluded from execution to keep the unit suite fast.
+# luanginxmore floods 4.2M parameters to overflow Lua-Nginx WAFs (the huge count is intentional).
+# That padding is now built ONCE and cached, not rebuilt per request (it used to cost ~11s + 16MB on
+# EVERY request). Still excluded from the battery below because the one-time cold build is slow;
+# the caching itself is covered by TestLuanginxmoreCached.
 HEAVY = {"luanginxmore"}
 
 # Project contract for falsy input: tamper("") == "" and tamper(None) is None
@@ -144,6 +146,31 @@ def test_transforms(self):
             self.assertEqual(mod.tamper(inp), expected, msg="tamper '%s'(%r)" % (name, inp))
 
 
+class TestLuanginxmoreCached(unittest.TestCase):
+    """luanginxmore's 4.2M-parameter padding is arbitrary and identical every request, so it must be
+    built once and reused - not regenerated per request (that cost ~11s + 16MB per HTTP request,
+    making the tamper unusable). xrange is shrunk so the test stays instant."""
+
+    def test_prepend_built_once(self):
+        import tamper.luanginxmore as t
+        from lib.core.enums import HINT
+
+        original = t.xrange
+        t._prepend.clear()
+        t.xrange = lambda n: range(min(n, 8))    # 4.2M -> 8 so the build is instant
+        try:
+            h1, h2 = {}, {}
+            t.tamper("1 AND 2>1", hints=h1)
+            t.tamper("1 AND 2>1", hints=h2)
+        finally:
+            t.xrange = original
+            t._prepend.clear()
+
+        # both requests must reuse the SAME cached object; a per-request rebuild yields distinct ones
+        self.assertIs(h1[HINT.PREPEND], h2[HINT.PREPEND])
+        self.assertEqual(h1[HINT.PREPEND].count("="), 8)
+
+
 class TestTamperCount(unittest.TestCase):
     def test_expected_count(self):
         # there are currently 70 tamper scripts; floor at 70 so an accidental deletion (or a glob
diff --git a/tests/test_techniques.py b/tests/test_techniques.py
index fe7563a4436..361f0db749c 100644
--- a/tests/test_techniques.py
+++ b/tests/test_techniques.py
@@ -1522,6 +1522,93 @@ def test_hex_output_decoded(self):
         self.assertEqual(value, decodeDbmsHexValue(hexed))
 
 
+class TestHexEncodedShortcutsGated(_InferenceCase):
+    """Under --hex / --binary-fields the expression is HEX()-wrapped, so the whole-value equality
+    shortcuts (low-card guess, oracle litmus) compare a hex-DECODED value against it and always miss -
+    wasting a probe per cell and tripping a spurious "unreliable" alarm. They must be skipped there.
+    Spying valueMatchCondition (the guess calls it first, 'continue's on None) and the litmus lets us
+    see whether each shortcut was reached without a full injection/agent context."""
+
+    _EXTRA_KB = ("lowCardCache", "dumpCharset", "dumpCharsetStable", "litmusCounter",
+                 "reliabilityAlarm", "huffmanModel", "multiThreadMode", "commonOutputs")
+
+    def setUp(self):
+        _InferenceCase.setUp(self)
+        self._saved_extra_kb = {k: kb.get(k) for k in self._EXTRA_KB}
+        self._saved_noHuffman = conf.get("noHuffman")
+        self._saved_binaryFields = conf.get("binaryFields")
+        self._saved_dbms = kb.get("dbms")
+        self._saved_vmc = inf.valueMatchCondition
+        self._saved_litmus = inf.oracleReliabilityLitmus
+        conf.noHuffman = True            # keep extraction on the classic '>' bisection path
+        conf.binaryFields = None
+        # bisection only builds the nulled/casted (and HEX-wrapped) expression - the step that sets
+        # kb.binaryField - when Backend.getDbms() is truthy; set_dbms() only forces getIdentifiedDbms()
+        kb.dbms = "MySQL"
+        kb.lowCardCache = {}
+        kb.dumpCharset = {}
+        kb.dumpCharsetStable = {}
+        kb.litmusCounter = 0
+        kb.reliabilityAlarm = False
+        kb.huffmanModel = {}
+        kb.multiThreadMode = False
+        kb.commonOutputs = None
+
+    def tearDown(self):
+        inf.valueMatchCondition = self._saved_vmc
+        inf.oracleReliabilityLitmus = self._saved_litmus
+        conf.noHuffman = self._saved_noHuffman
+        conf.binaryFields = self._saved_binaryFields
+        kb.dbms = self._saved_dbms
+        for k, v in self._saved_extra_kb.items():
+            kb[k] = v
+        _InferenceCase.tearDown(self)
+
+    _HEXED = "48656C6C6F"                # hex of "Hello"; all-ASCII so the shortcuts are applicable
+
+    def _run(self, hexConvert, expression="SELECT secret", binaryField=False):
+        conf.hexConvert = hexConvert
+        if binaryField:
+            conf.binaryFields = [agent.getFields(expression)[6]]     # the field bisection will hex-wrap
+        # arm the low-cardinality cache for this column so the guess WOULD fire if not gated
+        kb.lowCardCache[inf.normalizedExpression(expression)] = {"Hello": 3}
+        calls = {"guess": 0, "litmus": 0}
+
+        def vmc_spy(*args, **kwargs):
+            calls["guess"] += 1
+            return None                  # None -> guess loop 'continue's: no probe, no agent context needed
+
+        def litmus_spy(*args, **kwargs):
+            calls["litmus"] += 1
+            return True
+
+        inf.valueMatchCondition = vmc_spy
+        inf.oracleReliabilityLitmus = litmus_spy
+        _, value = self._bisect(self._HEXED, expression=expression, length=len(self._HEXED), dump=True)
+        return value, calls
+
+    def test_shortcuts_fire_without_hex(self):
+        # control: on a normal dump both shortcuts are reached (proves the test can actually see them,
+        # so the skips below are the gate doing its job, not a dead assertion)
+        value, calls = self._run(hexConvert=False)
+        self.assertEqual(value, self._HEXED)                 # no decode without --hex
+        self.assertGreater(calls["guess"], 0, "low-card guess should run on a normal dump")
+        self.assertEqual(calls["litmus"], 1, "oracle litmus should run on a normal dump")
+
+    def test_shortcuts_skipped_under_hex(self):
+        value, calls = self._run(hexConvert=True)
+        self.assertEqual(value, "Hello")                     # extraction still correct (decoded on the way out)
+        self.assertEqual(calls["guess"], 0, "low-card guess must be skipped when the expression is HEX-wrapped")
+        self.assertEqual(calls["litmus"], 0, "oracle litmus must be skipped when the expression is HEX-wrapped")
+        self.assertFalse(kb.reliabilityAlarm, "no false 'unreliable' alarm on a correct --hex dump")
+
+    def test_shortcuts_skipped_under_binary_field(self):
+        value, calls = self._run(hexConvert=False, binaryField=True)
+        self.assertEqual(value, self._HEXED)
+        self.assertEqual(calls["guess"], 0, "shortcuts must be skipped for a --binary-fields column")
+        self.assertEqual(calls["litmus"], 0)
+
+
 class TestProcessCharHook(_InferenceCase):
     def test_process_char_applied_to_each_char(self):
         # kb.data.processChar transforms every assembled character
@@ -1691,6 +1778,29 @@ def test_timeless_is_concurrency_safe(self):
         self.assertTrue(self._elig(8, True, {PAYLOAD.TECHNIQUE.TIME}, timeless=object()))
 
 
+class TestCharsetCorruptionDetection(unittest.TestCase):
+    """UNION/error charset-mismatch auto-hex trigger: _pageCharsetCorrupted fires on
+    reversibly-decoded high bytes (the '\\xNN' marker) and stays quiet on clean data."""
+
+    def test_detects_reversible_high_bytes(self):
+        # e.g. GBK bytes mis-decoded under utf-8 -> reversible '\xNN' escapes for the bad bytes
+        self.assertTrue(inject._pageCharsetCorrupted(u"\\xd6\\xd0\\xce\\xe2"))
+
+    def test_detects_inside_nested_rows(self):
+        self.assertTrue(inject._pageCharsetCorrupted([[u"1", u"caf\\xe9"], [u"2", u"ok"]]))
+
+    def test_clean_ascii_not_flagged(self):
+        self.assertFalse(inject._pageCharsetCorrupted(u"hello world"))
+
+    def test_clean_unicode_not_flagged(self):
+        # correctly-decoded unicode must not trigger a needless hex re-fetch
+        self.assertFalse(inject._pageCharsetCorrupted(u"\u4e2d\u6587\u6d4b\u8bd5"))
+
+    def test_low_hex_escape_not_flagged(self):
+        # a literal low '\x41' (ASCII 'A') is not an undecodable-byte marker
+        self.assertFalse(inject._pageCharsetCorrupted(u"literal \\x41 text"))
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)
 
diff --git a/tests/test_union_engine.py b/tests/test_union_engine.py
index f0592fe4e10..3dc81caf4de 100644
--- a/tests/test_union_engine.py
+++ b/tests/test_union_engine.py
@@ -27,9 +27,11 @@
 bootstrap()
 
 from lib.core.data import conf, kb
+from lib.core.datatype import AttribDict
 from lib.core.enums import PAYLOAD, PLACE
 from lib.request.connect import Connect
 import lib.techniques.union.test as ut
+import lib.techniques.union.use as uu
 
 MARKER = "MARKER42"
 VALID_PAGE = "<html>results %s</html>" % MARKER
@@ -103,6 +105,74 @@ def test_detect_beyond_first_step(self):
         self.assertEqual(self._detect(25), 25)
 
 
+class TestMssqlJsonAggFalsyValues(unittest.TestCase):
+    """Regression: MSSQL UNION dumps use FOR JSON (jsonAggMode), whose output carries native JSON
+    types. A real 0 / '' / false is a value; only JSON null is SQL NULL. The old `row.get(field) or
+    NULL` mapped every falsy value to the literal string 'NULL' - proven live (SELECT 0,'','ok'
+    dumped as NULL,NULL,ok). We drive the REAL _oneShotUnionUse against a mock FOR JSON page."""
+
+    def setUp(self):
+        self._s = {
+            "hexConvert": conf.get("hexConvert"), "parameters": conf.get("parameters"),
+            "paramDict": conf.get("paramDict"), "base64Parameter": conf.get("base64Parameter"),
+            "pageEncoding": conf.get("pageEncoding"), "hashDB": conf.get("hashDB"),
+            "inj": (kb.injection.place, kb.injection.parameter, kb.injection.data),
+            "jsonAggMode": kb.get("jsonAggMode"), "unionDuplicates": kb.get("unionDuplicates"),
+            "forcePartialUnion": kb.get("forcePartialUnion"), "tableFrom": kb.get("tableFrom"),
+            "unionTemplate": kb.get("unionTemplate"), "qp": Connect.queryPage,
+        }
+        conf.hexConvert = False
+        conf.parameters = {PLACE.GET: "id=1"}
+        conf.paramDict = {PLACE.GET: {"id": "1"}}
+        conf.base64Parameter = ()
+        conf.pageEncoding = None
+        conf.hashDB = None
+        v = AttribDict()
+        v.vector = (0, 4, "", "", "", "NULL", PAYLOAD.WHERE.NEGATIVE, False, False, None, None)
+        kb.injection.place = PLACE.GET
+        kb.injection.parameter = "id"
+        kb.injection.data = {PAYLOAD.TECHNIQUE.UNION: v}
+        kb.jsonAggMode = True
+        kb.unionDuplicates = kb.forcePartialUnion = False
+        kb.tableFrom = kb.unionTemplate = None
+        set_dbms("MSSQL")
+
+    def tearDown(self):
+        conf.hexConvert = self._s["hexConvert"]
+        conf.parameters = self._s["parameters"]
+        conf.paramDict = self._s["paramDict"]
+        conf.base64Parameter = self._s["base64Parameter"]
+        conf.pageEncoding = self._s["pageEncoding"]
+        conf.hashDB = self._s["hashDB"]
+        kb.injection.place, kb.injection.parameter, kb.injection.data = self._s["inj"]
+        kb.jsonAggMode = self._s["jsonAggMode"]
+        kb.unionDuplicates = self._s["unionDuplicates"]
+        kb.forcePartialUnion = self._s["forcePartialUnion"]
+        kb.tableFrom = self._s["tableFrom"]
+        kb.unionTemplate = self._s["unionTemplate"]
+        Connect.queryPage = self._s["qp"]
+        uu.Request.queryPage = self._s["qp"]
+
+    def _dump(self, jsonstr):
+        # jsonstr is exactly what MSSQL FOR JSON AUTO, INCLUDE_NULL_VALUES emits (literal to keep key
+        # order stable across py2/py3); object_pairs_hook=OrderedDict preserves that order downstream
+        page = "%s%s%s" % (kb.chars.start, jsonstr, kb.chars.stop)
+
+        def oracle(payload=None, content=False, raise404=True, **kwargs):
+            return (page, {}, 200) if content else True
+
+        Connect.queryPage = staticmethod(oracle)
+        uu.Request.queryPage = staticmethod(oracle)
+        kb.jsonAggMode = True
+        out = uu._oneShotUnionUse("SELECT a,b,c,d FROM users", False)
+        firstRow = out.replace(kb.chars.start, "").split(kb.chars.stop)[0]
+        return firstRow.split(kb.chars.delimiter)
+
+    def test_falsy_values_preserved(self):
+        fields = self._dump('[{"a":0,"b":"","c":"ok","d":null}]')
+        self.assertEqual(fields, ["0", "", "ok", "NULL"])   # only JSON null -> NULL
+
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)