Implementing automatic auto-recover on charset mismatch (E/U)

Author: stamparm

lib/core/settings.py

@@ -20,7 +20,7 @@
 from thirdparty import six
 
 # sqlmap version (<major>.<minor>.<month>.<monthly commit>)
-VERSION = "1.10.7.224"
+VERSION = "1.10.7.225"
 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

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 "

tests/test_techniques.py

@@ -1778,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)