From 291abb0ee84154d99cfa73887eaf4a8ea3fed9f0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Wed, 8 Jul 2026 12:32:05 +0200 Subject: [PATCH 1/5] Minor update --- lib/core/settings.py | 2 +- lib/core/testing.py | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/lib/core/settings.py b/lib/core/settings.py index 177257d52a3..e46d16d13b6 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from thirdparty import six # sqlmap version (...) -VERSION = "1.10.7.41" +VERSION = "1.10.7.42" TYPE = "dev" if VERSION.count('.') > 2 and VERSION.split('.')[-1] != '0' else "stable" TYPE_COLORS = {"dev": 33, "stable": 90, "pip": 34} VERSION_STRING = "sqlmap/%s#%s" % ('.'.join(VERSION.split('.')[:-1]) if VERSION.count('.') > 2 and VERSION.split('.')[-1] == '0' else VERSION, TYPE) diff --git a/lib/core/testing.py b/lib/core/testing.py index a10e336213c..765ec4fabb4 100644 --- a/lib/core/testing.py +++ b/lib/core/testing.py @@ -51,6 +51,8 @@ def vulnTest(tests=None, label="vuln"): ("--dependencies", ("sqlmap requires", "third-party library")), ("-u --data=\"reflect=1\" --flush-session --wizard --disable-coloring", ("Please choose:", "back-end DBMS: SQLite", "current user is DBA: True", "banner: '3.")), ("-u --data=\"code=1\" --code=200 --technique=B --banner --no-cast --flush-session", ("back-end DBMS: SQLite", "banner: '3.", "~COALESCE(CAST(")), + ("-u --flush-session --technique=B --banner", ("Type: boolean-based blind", "banner: '3.")), # session resume (1/2): detect + STORE the injection into the (serialized) session + ("-u --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 --flush-session --output-dir=\"\" --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 --flush-session --sql-query=\"SELECT '\u0161u\u0107uraj'\" --titles --technique=B --no-escape --string=luther --unstable", (u": '\u0161u\u0107uraj'", "~with --string",)), ("-m --flush-session --technique=B --banner", ("/3] URL:", "back-end DBMS: SQLite", "banner: '3.")), From 003393442626864cc3896bad1ab2a818238fcca1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Wed, 8 Jul 2026 12:43:10 +0200 Subject: [PATCH 2/5] Minor optimization --- lib/core/settings.py | 2 +- lib/core/threads.py | 2 ++ lib/request/comparison.py | 9 +++++++-- 3 files changed, 10 insertions(+), 3 deletions(-) diff --git a/lib/core/settings.py b/lib/core/settings.py index e46d16d13b6..be28c87c64f 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from thirdparty import six # sqlmap version (...) -VERSION = "1.10.7.42" +VERSION = "1.10.7.43" TYPE = "dev" if VERSION.count('.') > 2 and VERSION.split('.')[-1] != '0' else "stable" TYPE_COLORS = {"dev": 33, "stable": 90, "pip": 34} VERSION_STRING = "sqlmap/%s#%s" % ('.'.join(VERSION.split('.')[:-1]) if VERSION.count('.') > 2 and VERSION.split('.')[-1] == '0' else VERSION, TYPE) diff --git a/lib/core/threads.py b/lib/core/threads.py index 96d64685960..47e8e10ab3a 100644 --- a/lib/core/threads.py +++ b/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 diff --git a/lib/request/comparison.py b/lib/request/comparison.py index d2e8bac0790..b800f4795c8 100644 --- a/lib/request/comparison.py +++ b/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 (§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))) From fd009027f76778c6f0d29ac452da9f711f11d0d1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Wed, 8 Jul 2026 17:51:37 +0200 Subject: [PATCH 3/5] Adding --payload-lint to CI/CD --- .github/workflows/tests.yml | 5 + lib/core/optiondict.py | 1 + lib/core/settings.py | 2 +- lib/core/testing.py | 236 ++++++++++++++++++++++++++++++++++++ lib/parse/cmdline.py | 5 +- lib/request/connect.py | 10 ++ lib/utils/sqllint.py | 165 ++++++++++++++++++++++++- sqlmap.py | 5 +- tests/test_sqllint.py | 28 ++++- 9 files changed, 449 insertions(+), 8 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index a6bedc1e686..85ae78c0ea7 100644 --- a/.github/workflows/tests.yml +++ b/.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 diff --git a/lib/core/optiondict.py b/lib/core/optiondict.py index 93e929845e3..3e7fa93c499 100644 --- a/lib/core/optiondict.py +++ b/lib/core/optiondict.py @@ -288,6 +288,7 @@ "murphyRate": "integer", "smokeTest": "boolean", "fpTest": "boolean", + "payloadLint": "boolean", "apiTest": "boolean", }, diff --git a/lib/core/settings.py b/lib/core/settings.py index be28c87c64f..0cfb735d453 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from thirdparty import six # sqlmap version (...) -VERSION = "1.10.7.43" +VERSION = "1.10.7.44" TYPE = "dev" if VERSION.count('.') > 2 and VERSION.split('.')[-1] != '0' else "stable" TYPE_COLORS = {"dev": 33, "stable": 90, "pip": 34} VERSION_STRING = "sqlmap/%s#%s" % ('.'.join(VERSION.split('.')[:-1]) if VERSION.count('.') > 2 and VERSION.split('.')[-1] == '0' else VERSION, TYPE) diff --git a/lib/core/testing.py b/lib/core/testing.py index 765ec4fabb4..758d1a8bf85 100644 --- a/lib/core/testing.py +++ b/lib/core/testing.py @@ -547,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: ("x", 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 diff --git a/lib/parse/cmdline.py b/lib/parse/cmdline.py index 64ac6319f90..817e12f6aa0 100644 --- a/lib/parse/cmdline.py +++ b/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) diff --git a/lib/request/connect.py b/lib/request/connect.py index ce39c8f2944..70d5091d062 100644 --- a/lib/request/connect.py +++ b/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 diff --git a/lib/utils/sqllint.py b/lib/utils/sqllint.py index cc220923191..867b4c7d228 100644 --- a/lib/utils/sqllint.py +++ b/lib/utils/sqllint.py @@ -42,7 +42,7 @@ | (?P<%s>'(?:''|[^'])*'|"(?:""|[^"])*") | (?P<%s>`[^`]*`|\[[^\]]*\]) | (?P<%s>0[xX][0-9A-Fa-f]+|(?:\d+\.?\d*|\.\d+)(?:[eE][+-]?\d+)?) - | (?P<%s>[A-Za-z_@$][A-Za-z0-9_@$]*) + | (?P<%s>(?:[A-Za-z_@$]|[^\x00-\x7f])(?:[A-Za-z0-9_@$]|[^\x00-\x7f])*) | (?P<%s><=|>=|<>|!=|==|<<|>>|\|\||&&|::|:=|[-+*/%%=<>!~&|^:]) | (?P<%s>,) | (?P<%s>\.) @@ -69,6 +69,39 @@ # as the SELECT/COUNT wildcard) _BINARY_SYMBOLS = frozenset(("=", "<>", "!=", "<", ">", "<=", ">=", "/", "%", "||", "&&", "|", "&", "^")) +# clause-introducing keywords that signal a dangling list item when they sit +# right after a comma ("SELECT a,b, FROM t"). GROUP/ORDER/LIMIT/OFFSET are +# excluded on purpose - they double as very common column names, so a bare +# "a,limit,b" would false-positive. +_CLAUSE_KEYWORDS = frozenset(("FROM", "WHERE", "HAVING", "INTO")) + +# sqlmap's own templating markers. If any survives into a *final* outbound payload +# a substitution failed upstream (agent.py / cleanupPayload / queries.xml) - always +# a bug. Matched on the raw payload because a marker can leak anywhere (bare, inside +# a string, or where the lexer would otherwise read it as an MSSQL [identifier]). +_LEFTOVER_MARKER = re.compile( + r"\[(?:RANDNUM\d*|RANDSTR\d*|INFERENCE|SLEEPTIME|DELAYED|DELIMITER_START|DELIMITER_STOP" + r"|ORIGVALUE|ORIGINAL|GENERIC_SQL_COMMENT|QUERY|UNION|CHAR|COLSTART|COLSTOP|DB" + r"|SINGLE_QUOTE|DOUBLE_QUOTE|AT_REPLACE|SPACE_REPLACE|DOLLAR_REPLACE|HASH_REPLACE)\]") + +# SQL words whose near-miss spelling in a structural position is almost always a +# broken payload, not a legitimate identifier (deliberately smaller than the full +# keyword list): catches payload-builder typos like UNI1ON/SEL2ECT/ORD2ER without +# flagging arbitrary application identifiers. +# only length>=5 structural keywords: short ones (ON/NOT/IN/IS/BY/OR/AND/ALL/ +# FROM/LIKE/NULL/...) are too easily near-missed by real column names (note->NOT, +# ono->ON), which the real-identifier stress test proved would false-positive. +_NEAR_KEYWORD_TARGETS = frozenset(( + "SELECT", "UNION", "DISTINCT", "GROUP", "ORDER", "HAVING", "LIMIT", + "OFFSET", "WHERE", "INNER", "RIGHT", "OUTER", "CROSS", "REGEXP", "RLIKE")) + +# single-char substitutions seen in accidental mutation/test edits +_DIGIT_KEYWORD_ALIASES = {"0": "O", "1": "I", "2": "E", "3": "E", "4": "A", "5": "S", "7": "T", "8": "B"} + +_CLAUSE_STARTERS = frozenset(( + "SELECT", "UNION", "FROM", "WHERE", "GROUP", "ORDER", "HAVING", "LIMIT", + "OFFSET", "INTO", "JOIN", "ON", "AND", "OR")) + _KEYWORDS_CACHE = None @@ -82,6 +115,96 @@ def __init__(self, type_, value, start, end): self.end = end +def _word(token): + if token is not None and token.type in (T_IDENT, T_KEYWORD): + return token.value.upper() + return None + + +def _atClauseBoundary(prev): + return prev is None or prev.type in (T_LPAREN, T_RPAREN, T_SEMI, T_COMMA) or \ + (prev.type == T_OP and prev.value not in (".",)) or \ + (prev.type == T_KEYWORD and prev.value.upper() in _CLAUSE_STARTERS) + + +def _editWithin1(a, b): + """Damerau-Levenshtein distance <= 1 (one insertion, deletion, substitution + or adjacent transposition). Catches every single-char keyword typo class.""" + la, lb = len(a), len(b) + if a == b or abs(la - lb) > 1: + return a == b + if la == lb: + diff = [i for i in range(la) if a[i] != b[i]] + if len(diff) == 1: # substitution + return True + if len(diff) == 2 and diff[1] == diff[0] + 1 and \ + a[diff[0]] == b[diff[1]] and a[diff[1]] == b[diff[0]]: # transposition + return True + return False + shorter, longer = (a, b) if la < lb else (b, a) # deletion/insertion + for i in range(len(longer)): + if shorter == longer[:i] + longer[i + 1:]: + return True + return False + + +def _nearKeywordCandidates(value): + """ + Structural SQL keywords one single-char typo away from an identifier + (Damerau distance 1; NOT generic fuzzy matching over the whole keyword file). + + >>> sorted(_nearKeywordCandidates("UNI1ON")) + ['UNION'] + >>> sorted(_nearKeywordCandidates("SEL2ECT")) + ['SELECT'] + >>> sorted(_nearKeywordCandidates("UrNION")) + ['UNION'] + >>> sorted(_nearKeywordCandidates("UNIN")) + ['UNION'] + >>> sorted(_nearKeywordCandidates("UNOIN")) + ['UNION'] + """ + upper = value.upper() + if upper in _NEAR_KEYWORD_TARGETS or len(upper) < 4: + return set() + return set(target for target in _NEAR_KEYWORD_TARGETS if _editWithin1(upper, target)) + + +def _nearKeywordIsStructural(sig, index, keyword): + """True when a near-keyword identifier sits where that keyword is expected.""" + prev = sig[index - 1] if index > 0 else None + nxt = sig[index + 1] if index + 1 < len(sig) else None + prevWord = _word(prev) + nextWord = _word(nxt) + + if keyword == "UNION": + return nextWord in ("ALL", "DISTINCT", "SELECT") and \ + prevWord not in ("SELECT", "FROM", "WHERE", "GROUP", "ORDER", "BY", "HAVING", "LIMIT", "OFFSET", "JOIN", "ON", "AS") + + if keyword == "SELECT": + return prev is None or prev.type in (T_LPAREN, T_SEMI) or prevWord in ("UNION", "ALL", "DISTINCT", "EXCEPT", "INTERSECT") + + if keyword in ("ORDER", "GROUP"): + return nextWord == "BY" + + if keyword == "BY": + return prevWord in ("ORDER", "GROUP") + + if keyword in ("AND", "OR", "LIKE", "REGEXP", "RLIKE", "IN", "IS"): + return prev is not None and nxt is not None and prev.type in _OPERANDS and nxt.type in _OPERANDS.union((T_LPAREN,)) + + if keyword in ("FROM", "WHERE", "HAVING", "LIMIT", "OFFSET", "INTO", "JOIN", "ON"): + return _atClauseBoundary(prev) or prevWord in ("SELECT", "UPDATE", "DELETE", "INSERT", "FROM", "WHERE", "HAVING") + + if keyword in ("ALL", "DISTINCT"): + return prevWord in ("UNION", "SELECT") + + if keyword in ("NULL", "NOT"): + return _atClauseBoundary(prev) + + return False + + def _keywords(): global _KEYWORDS_CACHE @@ -194,6 +317,11 @@ def checkSanity(sql, keywords=None): keywords = _keywords() issues = [] + + # -- residual templating markers (upstream substitution failed) -------- + for match in _LEFTOVER_MARKER.finditer(sql): + issues.append("leftover marker '%s' at offset %d" % (match.group(0), match.start())) + tokens = tokenize(sql, keywords) # -- edge tolerance for unterminated strings --------------------------- @@ -201,6 +329,7 @@ def checkSanity(sql, keywords=None): # that opens *inside* a group (depth > 0) has swallowed a needed ')', i.e. # an odd quote count within an owned scope (the classic "users'" abomination). depth = 0 + unterminated = False for token in tokens: if token.type == T_LPAREN: depth += 1 @@ -209,8 +338,14 @@ def checkSanity(sql, keywords=None): elif token.type == T_UNTERM: if depth > 0: issues.append("odd quote inside a parenthesized scope at offset %d" % token.start) + unterminated = True break + # unclosed '(' (a dropped ')'): well-formed payloads NEVER end paren-positive + # (leading break-out ')' only ever makes depth negative), so this is 0-FP. + if not unterminated and depth > 0: + issues.append("unbalanced parentheses (%d unclosed '(')" % depth) + sig = _significant(tokens) for i in range(len(sig)): @@ -223,9 +358,23 @@ def checkSanity(sql, keywords=None): curIsFunc = cur.type == T_KEYWORD and nxt is not None and nxt.type == T_LPAREN curBinary = _isBinary(cur) and not curIsFunc - # -- glued number/keyword boundary: '1UNION', '1AND' --------------- - if cur.type == T_NUM and nxt is not None and nxt.start == cur.end and nxt.type in (T_IDENT, T_KEYWORD): - issues.append("digit glued to a word ('%s%s') at offset %d" % (cur.value, nxt.value, cur.start)) + # -- keyword near-miss in a structural position: UNI1ON/SEL2ECT/ORD2ER + if cur.type == T_IDENT: + for keyword in sorted(_nearKeywordCandidates(cur.value)): + if _nearKeywordIsStructural(sig, i, keyword): + issues.append("keyword typo '%s' (near '%s') at offset %d" % (cur.value, keyword, cur.start)) + break + + # -- UNION must continue with SELECT/ALL/DISTINCT/'(' (catches a glued or + # corrupted continuation like 'UNION ALLSELECT' -> UNION ) + if cur.type == T_KEYWORD and cur.value.upper() == "UNION" and nxt is not None: + if not (nxt.type == T_LPAREN or (nxt.type == T_KEYWORD and nxt.value.upper() in ("SELECT", "ALL", "DISTINCT"))): + issues.append("UNION not followed by SELECT/ALL/DISTINCT at offset %d" % cur.start) + + # -- digit glued to a keyword: '1UNION', '5108AND' (a digit-started + # identifier like '4images' is legitimate and must NOT trip this) + if cur.type == T_NUM and nxt is not None and nxt.start == cur.end and nxt.type == T_KEYWORD: + issues.append("digit glued to a keyword ('%s%s') at offset %d" % (cur.value, nxt.value, cur.start)) # -- operand directly followed by a bare number: 'id 1' ------------ # a numeric literal can never be an alias, so this is always broken @@ -241,6 +390,14 @@ def checkSanity(sql, keywords=None): issues.append("comma right after '(' at offset %d" % cur.start) elif pair == (T_COMMA, T_RPAREN): issues.append("comma right before ')' at offset %d" % cur.start) + elif prev.type == T_KEYWORD and prev.value.upper() == "SELECT" and cur.type == T_COMMA: + issues.append("comma right after SELECT at offset %d" % cur.start) + elif prev.type == T_COMMA and cur.type == T_KEYWORD and cur.value.upper() in _CLAUSE_KEYWORDS \ + and nxt is not None and nxt.type not in (T_COMMA, T_RPAREN): + # a clause keyword right after a comma AND followed by real content is a + # dangling list item ("a,b, FROM t"); if it is a bare list item itself + # ("a,group,b" - a column named 'group') the next token is a comma/paren/end + issues.append("dangling comma before '%s' at offset %d" % (cur.value, cur.start)) elif pair == (T_RPAREN, T_LPAREN): issues.append("adjacent groups ')(' at offset %d" % cur.start) elif pair == (T_LPAREN, T_RPAREN) and (i < 2 or sig[i - 2].type in (T_OP, T_COMMA, T_LPAREN)): diff --git a/sqlmap.py b/sqlmap.py index a2085c4c2a2..a54f97e38af 100755 --- a/sqlmap.py +++ b/sqlmap.py @@ -195,6 +195,9 @@ def main(): elif conf.fpTest: from lib.core.testing import fpTest os._exitcode = 1 - (fpTest() or 0) + elif conf.payloadLint: + from lib.core.testing import payloadLintTest + os._exitcode = 1 - (payloadLintTest() or 0) elif conf.apiTest: from lib.core.testing import apiTest os._exitcode = 1 - (apiTest() or 0) @@ -610,7 +613,7 @@ def main(): except OSError: pass - if any((conf.vulnTest, conf.fpTest, conf.smokeTest, conf.apiTest)) or not filterNone(filepath for filepath in glob.glob(os.path.join(tempDir, '*')) if not any(filepath.endswith(_) for _ in (".lock", ".exe", ".so", '_'))): # ignore junk files + if any((conf.vulnTest, conf.fpTest, conf.smokeTest, conf.payloadLint, conf.apiTest)) or not filterNone(filepath for filepath in glob.glob(os.path.join(tempDir, '*')) if not any(filepath.endswith(_) for _ in (".lock", ".exe", ".so", '_'))): # ignore junk files try: shutil.rmtree(tempDir, ignore_errors=True) except OSError: diff --git a/tests/test_sqllint.py b/tests/test_sqllint.py index 06468690006..b5389e75612 100644 --- a/tests/test_sqllint.py +++ b/tests/test_sqllint.py @@ -79,6 +79,28 @@ "1 UNION SELECT NULL,,NULL", # empty list item "1 AND (1=1)(2=2)", # adjacent groups "SELECT count() FROM t WHERE id=)", # operator before ')' + "SELECT a,b, FROM t", # dangling comma before clause + "1 AND (SELECT [DELIMITER_START]x[DELIMITER_STOP] FROM t)", # leftover templating marker + "1 [ORIGVALUE] AND 1=1", # leftover ORIGVALUE marker + "1 UNI1ON ALL SELECT NULL,NULL", # digit-corrupted UNION + "1 UrNION ALL SELECT NULL", # char-inserted UNION + "SEL2ECT x.z FROM t", # digit-corrupted SELECT + "1 ORD2ER BY 1", # digit-corrupted ORDER + "1 UNIN ALL SELECT NULL", # deletion-typo UNION + "1 UNOIN ALL SELECT NULL", # transposition-typo UNION + "SELCT a FROM t", # deletion-typo SELECT + "1 UNION ALLSELECT NULL", # glued keyword after UNION + "1 AND ORD(MID((SELECT COUNT(x FROM t),1,1))>64", # unbalanced parentheses (dropped ')') + "1 UNION ALL SELECT ,CONCAT(0x71,a,0x71) FROM t", # comma right after SELECT +) + +# cross-dialect valid constructs the near-keyword / comma / digit-glue rules must +# NOT flag (regression guards for false positives the real-identifier stress found) +DIALECT_GOOD_HARD = ( + "SELECT 4images_users FROM t", # digit-STARTED identifier (not '1UNION') + "SELECT a,group,b FROM t", # column literally named 'group' + "SELECT a,order FROM t", # column literally named 'order' + "1 UNION SELECT orders FROM t", # 'orders' near ORDER but a real table ) # queries.xml: attributes that carry SQL (not regexes/markers) @@ -142,13 +164,17 @@ def _payload_atoms(): class TestLinterSelf(unittest.TestCase): def test_well_formed_pass(self): - for sql in GOOD + DIALECT_GOOD: + for sql in GOOD + DIALECT_GOOD + DIALECT_GOOD_HARD: self.assertEqual(checkSanity(sql), [], "false positive on valid SQL: %r" % sql) def test_malformed_flag(self): for sql in BAD: self.assertTrue(checkSanity(sql), "missed malformed SQL: %r" % sql) + def test_nonascii_identifier(self): + # a non-ASCII column name (Turkish dotless-i U+0131) must lex as an identifier, not stray + self.assertEqual(checkSanity(u"SELECT \u0131d,ad FROM users"), []) + # module-level coverage accounting (printed in tearDownModule) _COVERAGE = {"queries_dbms": 0, "queries_atoms": 0, "payload_atoms": 0} From 8db71f4d67b9d1b4416176a0cff4df3e1f3b7bcd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Wed, 8 Jul 2026 17:55:54 +0200 Subject: [PATCH 4/5] Fixes CI/CD error --- lib/core/settings.py | 2 +- lib/request/comparison.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/lib/core/settings.py b/lib/core/settings.py index 0cfb735d453..0560348ff04 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from thirdparty import six # sqlmap version (...) -VERSION = "1.10.7.44" +VERSION = "1.10.7.45" TYPE = "dev" if VERSION.count('.') > 2 and VERSION.split('.')[-1] != '0' else "stable" TYPE_COLORS = {"dev": 33, "stable": 90, "pip": 34} VERSION_STRING = "sqlmap/%s#%s" % ('.'.join(VERSION.split('.')[:-1]) if VERSION.count('.') > 2 and VERSION.split('.')[-1] == '0' else VERSION, TYPE) diff --git a/lib/request/comparison.py b/lib/request/comparison.py index b800f4795c8..30beafabc70 100644 --- a/lib/request/comparison.py +++ b/lib/request/comparison.py @@ -132,7 +132,7 @@ 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 (§below): the + # 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) From 165bf837c937cc7bd57ba4c55622af5e4fa4905c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Wed, 8 Jul 2026 18:00:42 +0200 Subject: [PATCH 5/5] Minor update --- lib/core/settings.py | 2 +- tests/test_library.py | 7 +++++++ tests/test_sqllint.py | 16 ---------------- 3 files changed, 8 insertions(+), 17 deletions(-) diff --git a/lib/core/settings.py b/lib/core/settings.py index 0560348ff04..1cb6c5ddf5d 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from thirdparty import six # sqlmap version (...) -VERSION = "1.10.7.45" +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) diff --git a/tests/test_library.py b/tests/test_library.py index 8437238e517..36a608e31a8 100644 --- a/tests/test_library.py +++ b/tests/test_library.py @@ -122,6 +122,11 @@ def test_errors_reach_the_report(self): # otherwise gate the ERROR record before it reaches the recorder (order-dependent flakiness) saved_level = logger.level logger.setLevel(logging.ERROR) + # mute the existing console handler(s) so the deliberate ERROR below does not leak to the + # unittest output; the recorder added by setupReportCollector next still captures it + muted = [(handler, handler.level) for handler in logger.handlers] + for handler, _ in muted: + handler.setLevel(logging.CRITICAL + 1) collector = setupReportCollector() try: logger.error("boom %s", "here") @@ -129,6 +134,8 @@ def test_errors_reach_the_report(self): self.assertTrue(any("boom here" in _ for _ in result["error"])) finally: logger.setLevel(saved_level) + for handler, level in muted: + handler.setLevel(level) for handler in list(logger.handlers): if isinstance(handler, ReportErrorRecorder): logger.removeHandler(handler) diff --git a/tests/test_sqllint.py b/tests/test_sqllint.py index b5389e75612..63f409ff10d 100644 --- a/tests/test_sqllint.py +++ b/tests/test_sqllint.py @@ -176,14 +176,9 @@ def test_nonascii_identifier(self): self.assertEqual(checkSanity(u"SELECT \u0131d,ad FROM users"), []) -# module-level coverage accounting (printed in tearDownModule) -_COVERAGE = {"queries_dbms": 0, "queries_atoms": 0, "payload_atoms": 0} - - class TestCatalogCoverage(unittest.TestCase): def test_queries_xml_clean(self): atoms = _queries_atoms() - _COVERAGE["queries_dbms"] = len(atoms) total = 0 for dbms, items in atoms.items(): for raw in items: @@ -191,12 +186,10 @@ def test_queries_xml_clean(self): issues = checkSanity(_materialize(raw)) self.assertEqual(issues, [], "queries.xml [%s] malformed atom: %r -> %s" % (dbms, raw, issues)) - _COVERAGE["queries_atoms"] = total self.assertGreater(total, 1000) # sanity: the catalog was actually walked def test_payloads_xml_clean(self): items = _payload_atoms() - _COVERAGE["payload_atoms"] = len(items) for name, title, raw in items: issues = checkSanity(_materialize(raw)) self.assertEqual(issues, [], @@ -204,14 +197,5 @@ def test_payloads_xml_clean(self): self.assertGreater(len(items), 500) -def tearDownModule(): - q = _COVERAGE - total = q["queries_atoms"] + q["payload_atoms"] - sys.stderr.write( - "\n[sqllint coverage] atom-level: %d SQL atoms linted clean " - "(%d queries.xml across %d/30 DBMSes + %d payloads.xml vectors)\n" - % (total, q["queries_atoms"], q["queries_dbms"], q["payload_atoms"])) - - if __name__ == "__main__": unittest.main(verbosity=2)