From 0079412fd21d5bfcf020691bd34e2d85a60a736f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Tue, 28 Jul 2026 12:39:07 +0200 Subject: [PATCH] Massive speed improvement of luanginxmore tamper script --- lib/core/settings.py | 2 +- tamper/luanginxmore.py | 9 ++- tests/test_keyset_engine.py | 109 ++++++++++++++++++++++++++++++++++++ tests/test_tamper.py | 31 +++++++++- 4 files changed, 147 insertions(+), 4 deletions(-) create mode 100644 tests/test_keyset_engine.py diff --git a/lib/core/settings.py b/lib/core/settings.py index ddd02152e..3f363ee21 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from lib.core.enums import OS from thirdparty import six # sqlmap version (...) -VERSION = "1.10.7.217" +VERSION = "1.10.7.218" 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/tamper/luanginxmore.py b/tamper/luanginxmore.py index 1d360db10..eb86b8f97 100644 --- a/tamper/luanginxmore.py +++ b/tamper/luanginxmore.py @@ -17,6 +17,10 @@ from lib.core.settings import DEFAULT_GET_POST_DELIMITER __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_keyset_engine.py b/tests/test_keyset_engine.py new file mode 100644 index 000000000..616715d0c --- /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_tamper.py b/tests/test_tamper.py index 1d9eaa885..99bc8df24 100644 --- a/tests/test_tamper.py +++ b/tests/test_tamper.py @@ -42,8 +42,10 @@ PAYLOADS = [ ] 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 @@ class TestKnownTransforms(unittest.TestCase): 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