[pull] master from sqlmapproject:master

.github/workflows/tests.yml

@@ -122,6 +122,11 @@ jobs:
       - name: Smoke test
         run: python sqlmap.py --smoke-test
 
+      - name: Payload lint
+        # offline: emulates blind + UNION enumeration across all DBMSes and checks
+        # every payload agent.py builds with lib/utils/sqllint (structural sanity)
+        run: python sqlmap.py --payload-lint
+
       - name: Vuln test
         run: python sqlmap.py --vuln-test
 

lib/core/optiondict.py

@@ -288,6 +288,7 @@
         "murphyRate": "integer",
         "smokeTest": "boolean",
         "fpTest": "boolean",
+        "payloadLint": "boolean",
         "apiTest": "boolean",
     },
 

lib/core/settings.py

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

lib/core/testing.py

@@ -51,6 +51,8 @@ def vulnTest(tests=None, label="vuln"):
         ("--dependencies", ("sqlmap requires", "third-party library")),
         ("-u <url> --data=\"reflect=1\" --flush-session --wizard --disable-coloring", ("Please choose:", "back-end DBMS: SQLite", "current user is DBA: True", "banner: '3.")),
         ("-u <url> --data=\"code=1\" --code=200 --technique=B --banner --no-cast --flush-session", ("back-end DBMS: SQLite", "banner: '3.", "~COALESCE(CAST(")),
+        ("-u <url> --flush-session --technique=B --banner", ("Type: boolean-based blind", "banner: '3.")),   # session resume (1/2): detect + STORE the injection into the (serialized) session
+        ("-u <url> --technique=B --banner", ("sqlmap resumed the following injection point(s) from stored session", "Type: boolean-based blind", "banner: '3.")),   # session resume (2/2): NO --flush-session, so the injection must round-trip back OUT of the serialized session and still work (guards session serialization/deserialization end-to-end)
         (u"-c <config> --flush-session --output-dir=\"<tmpdir>\" --smart --roles --statements --hostname --privileges --sql-query=\"SELECT '\u0161u\u0107uraj'\" --technique=U", (u": '\u0161u\u0107uraj'", "on SQLite it is not possible", "as the output directory")),
         (u"-u <url> --flush-session --sql-query=\"SELECT '\u0161u\u0107uraj'\" --titles --technique=B --no-escape --string=luther --unstable", (u": '\u0161u\u0107uraj'", "~with --string",)),
         ("-m <multiple> --flush-session --technique=B --banner", ("/3] URL:", "back-end DBMS: SQLite", "banner: '3.")),
@@ -545,3 +547,239 @@ def _(node):
         logger.error("smoke test final result: FAILED")
 
     return retVal
+
+def payloadLintTest():
+    """
+    Offline payload-sanity coverage test.
+
+    For every supported back-end DBMS an emulation oracle drives the blind
+    enumeration handlers (no live target), so agent.py builds the full range of
+    inference payloads it would emit while dumping a schema, and each one is
+    checked with lib.utils.sqllint. A planted-malformation self-check first
+    proves the pipeline actually catches a defect, so the gate can never pass
+    vacuously.
+    """
+
+    import lib.core.common as common_module
+
+    from lib.controller.handler import setHandler
+    from lib.core.agent import agent
+    from lib.core.common import Backend
+    from lib.core.datatype import AttribDict
+    from lib.core.datatype import InjectionDict
+    from lib.core.enums import PAYLOAD
+    from lib.core.enums import PLACE
+    from lib.core.threads import getCurrentThreadData
+    from lib.request.connect import Connect
+    from lib.utils.sqllint import checkSanity
+
+    unisonRandom()
+
+    collected = []
+    guard = {"count": 0}
+    CAP_PER_METHOD = 600
+
+    # A "consistent liar": parse the injected char comparison in whatever form the
+    # dialect uses (bare int / CHAR(n) / quoted 'x') and answer so counts->'2',
+    # lengths->small, names->'a'. This keeps every dialect's enumeration walking a
+    # few shallow levels, exercising the payload builders without a real backend.
+    def _oracle(value=None, **kwargs):
+        if value is None:
+            return None
+        sql = agent.removePayloadDelimiters(agent.adjustLateValues(value))
+        collected.append(sql)
+        guard["count"] += 1
+        if guard["count"] > CAP_PER_METHOD:
+            raise KeyboardInterrupt
+        # UNION/inband path reads the response body: hand back a page carrying the
+        # delimited marker value so extraction "succeeds" and keeps producing payloads
+        if kwargs.get("content"):
+            start, stop, delim = kb.chars.start, kb.chars.stop, kb.chars.delimiter
+            return ("x%s%s%s%s%sx" % (start, "1", delim, "1", stop), None, 200)
+        upper = sql.upper()
+        match = re.search(r">\s*'(.?)'", sql)
+        if match:
+            return True if match.group(1) == "" else (50 if "COUNT(" in upper else 97) > ord(match.group(1))
+        match = re.search(r">\s*(?:N?CHAR|CHR|NCHR)\s*\(\s*(\d+)", sql, re.I)
+        if match:
+            return (50 if "COUNT(" in upper else 97) > int(match.group(1))
+        match = re.search(r">\s*(\d+)", sql)
+        if match:
+            value_ = int(match.group(1))
+            target = 1 if "COUNT(" in upper else (3 if any(_ in upper for _ in ("LENGTH(", "LEN(", "DATALENGTH(")) else 2)
+            return target > value_
+        match = re.search(r"(\d+)\s*(=|<|>=|<=|<>|!=)\s*(\d+)", sql)
+        if match:
+            left, operator, right = int(match.group(1)), match.group(2), int(match.group(3))
+            return {"=": left == right, "<": left < right, ">=": left >= right, "<=": left <= right, "<>": left != right, "!=": left != right}[operator]
+        return False
+
+    def _injection(dbms):
+        injection = InjectionDict()
+        injection.place = PLACE.GET
+        injection.parameter = "id"
+        injection.ptype = 1
+        injection.prefix = ""
+        injection.suffix = ""
+        injection.clause = [1]
+        injection.dbms = dbms
+        boolean = AttribDict()
+        boolean.title = "AND boolean-based blind"
+        boolean.vector = "AND [INFERENCE]"
+        boolean.comment = ""
+        boolean.payload = "x"
+        boolean.where = 1
+        boolean.templatePayload = None
+        boolean.matchRatio = None
+        boolean.trueCode = None
+        boolean.falseCode = None
+        injection.data = AttribDict()
+        injection.data[PAYLOAD.TECHNIQUE.BOOLEAN] = boolean
+        return injection
+
+    def _unionInjection(dbms):
+        injection = InjectionDict()
+        injection.place = PLACE.GET
+        injection.parameter = "id"
+        injection.ptype = 1
+        injection.prefix = ""
+        injection.suffix = ""
+        injection.clause = [1]
+        injection.dbms = dbms
+        union = AttribDict()
+        union.title = "Generic UNION query"
+        # vector = (position, count, comment, prefix, suffix, char, where,
+        #           unionDuplicates, forcePartial, tableFrom, unionTemplate)
+        union.vector = (0, 3, "-- -", "", "", "NULL", PAYLOAD.WHERE.ORIGINAL, False, False, None, None)
+        union.comment = "-- -"
+        union.prefix = ""
+        union.suffix = ""
+        union.where = PAYLOAD.WHERE.ORIGINAL
+        union.templatePayload = None
+        union.matchRatio = None
+        union.trueCode = None
+        union.falseCode = None
+        injection.data = AttribDict()
+        injection.data[PAYLOAD.TECHNIQUE.UNION] = union
+        return injection
+
+    methods = ("getBanner", "getCurrentDb", "getCurrentUser", "getHostname", "isDba",
+               "getDbs", "getTables", "getColumns", "getSchema", "getUsers",
+               "getPasswordHashes", "getPrivileges", "getRoles", "getStatements",
+               "getComments", "getProcedures")
+
+    _stdout = sys.stdout
+
+    def _progress(dbms, count, bad):
+        # write to the real stdout (sqlmap's is redirected to a sink during the walk)
+        _stdout.write("[payload-lint] %-26s %7d payloads  %s\n" % (dbms, count, ("%d FLAGGED" % bad) if bad else "ok"))
+        _stdout.flush()
+
+    class _Sink(object):
+        def write(self, *args, **kwargs): pass
+        def flush(self, *args, **kwargs): pass
+
+    _queryPage = Connect.queryPage
+    _getPageTemplate = common_module.getPageTemplate
+    _level = logger.level
+    _threads = conf.threads
+    threadData = getCurrentThreadData()
+
+    retVal = True
+    total = walked = 0
+    flagged = []
+    allDbms = sorted(queries.keys())
+
+    try:
+        conf.batch = True
+        conf.threads = 1                    # deterministic, single-threaded (clean output)
+        conf.disableHashing = True
+        if not conf.base64Parameter:
+            conf.base64Parameter = []
+        common_module.getPageTemplate = lambda *args, **kwargs: ("<html>x</html>", False)
+        Connect.queryPage = staticmethod(_oracle)
+        logger.setLevel(logging.CRITICAL)   # mute the emulated walk's "unable to retrieve" noise
+        threadData.disableStdOut = True     # mute dataToStdout at its gate
+        sys.stdout = _Sink()                # and mute everything else (readInput prompts, tables, ...)
+
+        def _runMethods():
+            found = set()
+            for name in methods:
+                method = getattr(conf.dbmsHandler, name, None)
+                if method is None:
+                    continue
+                del collected[:]
+                guard["count"] = 0
+                try:
+                    method()
+                except (KeyboardInterrupt, Exception):
+                    pass
+                found.update(collected)
+            return found
+
+        for dbms in allDbms:
+            try:
+                Backend.flushForcedDbms(force=True)
+                kb.stickyDBMS = False
+                Backend.forceDbms(dbms)
+                conf.forceDbms = conf.dbms = dbms
+                conf.parameters = {PLACE.GET: "id=1"}
+                conf.paramDict = {PLACE.GET: {"id": "1"}}
+                # a user/column so user-filtered and column-specific query variants
+                # (WHERE grantee='...', per-column extraction) are exercised too
+                conf.db, conf.tbl, conf.col, conf.user = "testdb", "testtbl", "testcol", "testuser"
+                setHandler()
+            except Exception:
+                _progress(dbms, 0, 0)
+                continue
+
+            seen = set()
+            # (a) boolean-based blind pass, then (b) UNION-query (inband) pass -
+            # two techniques exercise two distinct payload-builder families in agent.py
+            for technique, injector in ((PAYLOAD.TECHNIQUE.BOOLEAN, _injection),
+                                        (PAYLOAD.TECHNIQUE.UNION, _unionInjection)):
+                for key in list(kb.data.keys()):
+                    if key.startswith("cached"):
+                        kb.data[key] = None
+                kb.jsonAggMode = False
+                kb.injection = injector(dbms)
+                kb.injections = [kb.injection]
+                kb.technique = technique
+                seen.update(_runMethods())
+
+            bad = [(dbms, p, checkSanity(p)) for p in seen if checkSanity(p)]
+            flagged.extend(bad)
+            total += len(seen)
+            if seen:
+                walked += 1
+            _progress(dbms, len(seen), len(bad))
+    finally:
+        sys.stdout = _stdout
+        Connect.queryPage = _queryPage
+        common_module.getPageTemplate = _getPageTemplate
+        Backend.flushForcedDbms(force=True)
+        logger.setLevel(_level)
+        conf.threads = _threads
+        threadData.disableStdOut = False
+
+    # self-check: the linter must flag a known-malformed payload, and the walk
+    # must have actually produced payloads (guards against a vacuous pass)
+    if not checkSanity("1 AND (SELECT id 1 FROM users)"):
+        logger.error("payload-lint self-check failed: a known-malformed payload was not flagged")
+        retVal = False
+    if total < 1000:
+        logger.error("payload-lint self-check failed: only %d payloads generated (walk did not run)" % total)
+        retVal = False
+
+    for dbms, payload, issues in flagged[:20]:
+        retVal = False
+        logger.error("[%s] malformed payload: %s -> %s" % (dbms, payload, "; ".join(issues)))
+
+    clearConsoleLine()
+    logger.info("payload-lint: %d payloads across %d/%d DBMSes checked, %d malformed" % (total, walked, len(allDbms), len(flagged)))
+    if retVal:
+        logger.info("payload-lint final result: PASSED")
+    else:
+        logger.error("payload-lint final result: FAILED")
+
+    return retVal

lib/core/threads.py

@@ -52,6 +52,8 @@ def reset(self):
         self.lastComparisonCode = None
         self.lastComparisonRatio = None
         self.lastPageTemplateCleaned = None
+        self.lastPageTemplateJsonMinimized = None
+        self.lastPageTemplateStructural = None
         self.lastPageTemplate = None
         self.lastErrorPage = tuple()
         self.lastHTTPError = None

lib/parse/cmdline.py

@@ -930,6 +930,9 @@ def cmdLineParser(argv=None):
         parser.add_argument("--fp-test", dest="fpTest", action="store_true",
             help=SUPPRESS)
 
+        parser.add_argument("--payload-lint", dest="payloadLint", action="store_true",
+            help=SUPPRESS)
+
         parser.add_argument("--api-test", dest="apiTest", action="store_true",
             help=SUPPRESS)
 
@@ -1187,7 +1190,7 @@ def _format_action_invocation(self, action):
         else:
             args.stdinPipe = None
 
-        if not any((args.direct, args.url, args.logFile, args.bulkFile, args.googleDork, args.configFile, args.requestFile, args.openApiFile, args.updateAll, args.smokeTest, args.vulnTest, args.fpTest, args.apiTest, args.wizard, args.dependencies, args.purge, args.listTampers, args.hashFile, args.stdinPipe)):
+        if not any((args.direct, args.url, args.logFile, args.bulkFile, args.googleDork, args.configFile, args.requestFile, args.openApiFile, args.updateAll, args.smokeTest, args.vulnTest, args.fpTest, args.payloadLint, args.apiTest, args.wizard, args.dependencies, args.purge, args.listTampers, args.hashFile, args.stdinPipe)):
             errMsg = "missing a mandatory option (-d, -u, -l, -m, -r, -g, -c, --wizard, --shell, --update, --purge, --list-tampers or --dependencies). "
             errMsg += "Use -h for basic and -hh for advanced help\n"
             parser.error(errMsg)

lib/request/comparison.py

@@ -132,6 +132,11 @@ def _comparison(page, headers, code, getRatioValue, pageLength):
             page = removeDynamicContent(page)
             if threadData.lastPageTemplate != kb.pageTemplate:
                 threadData.lastPageTemplateCleaned = removeDynamicContent(kb.pageTemplate)
+                # Same template-identity memoization for the structure-aware projections (see below): the
+                # template is constant across an extraction, so it must not be re-parsed/re-tokenized on
+                # every inference request - only seq2 (from the live page) is recomputed per response
+                threadData.lastPageTemplateJsonMinimized = jsonMinimize(kb.pageTemplate)
+                threadData.lastPageTemplateStructural = "\n".join(sorted(extractStructuralTokens(kb.pageTemplate)))
                 threadData.lastPageTemplate = kb.pageTemplate
 
             seqMatcher.set_seq1(threadData.lastPageTemplateCleaned)
@@ -178,14 +183,14 @@ def _comparison(page, headers, code, getRatioValue, pageLength):
             # only on a JSON Content-Type with both bodies parseable; any doubt (or an explicit
             # --text-only/--titles) falls back to the exact text path below.
             if _isJsonResponse(headers) and not (conf.titles or conf.textOnly or kb.nullConnection):
-                seq1 = jsonMinimize(kb.pageTemplate)
+                seq1 = threadData.lastPageTemplateJsonMinimized  # memoized per template (see above)
                 seq2 = jsonMinimize(rawPage)
 
             # Structure-aware comparison for a structurally-stable (but byte-unstable) HTML page:
             # compare the value-free tag/class/id skeleton so dynamic text does not perturb the ratio
             # while a structural change (e.g. a results table appearing/disappearing) still does
             if seq1 is None and kb.pageStructurallyStable and not (conf.titles or conf.textOnly or kb.nullConnection):
-                _ = "\n".join(sorted(extractStructuralTokens(kb.pageTemplate)))
+                _ = threadData.lastPageTemplateStructural  # memoized per template (see above)
                 if _:   # only engage when the page actually exposes structure (HTML tags); tagless content falls back to text
                     seq1 = _
                     seq2 = "\n".join(sorted(extractStructuralTokens(rawPage)))

lib/request/connect.py

@@ -139,6 +139,7 @@ class WebSocketException(Exception):
 from lib.request.direct import direct
 from lib.request.methodrequest import MethodRequest
 from lib.utils.safe2bin import safecharencode
+from lib.utils.sqllint import checkSanity
 from thirdparty import six
 from collections import OrderedDict
 from thirdparty.six import unichr as _unichr
@@ -1088,6 +1089,15 @@ def queryPage(value=None, place=None, content=False, getRatioValue=False, silent
         payload = agent.extractPayload(value)
         threadData = getCurrentThreadData()
 
+        # Opt-in sanity lint of the outbound (pre-tamper) payload. Skipped during
+        # detection (kb.testMode) where deliberately-invalid probes are expected;
+        # for operational payloads a structural defect is a genuine bug worth a
+        # heads-up. Enabled via SQLMAP_LINT_PAYLOADS (e.g. CI/--vuln-test runs).
+        if payload and not kb.testMode and os.environ.get("SQLMAP_LINT_PAYLOADS"):
+            for issue in checkSanity(agent.removePayloadDelimiters(value)):
+                singleTimeWarnMessage("potentially malformed SQL payload emitted (%s): %s" % (issue, payload))
+                break
+
         if conf.httpHeaders:
             headers = OrderedDict(conf.httpHeaders)
             contentType = max(headers[_] or "" if _.upper() == HTTP_HEADER.CONTENT_TYPE.upper() else "" for _ in headers) or None