Massive speed improvement of luanginxmore tamper script

lib/core/settings.py

@@ -20,7 +20,7 @@ from lib.core.enums import OS
 from thirdparty import six
 
 # sqlmap version (<major>.<minor>.<month>.<monthly commit>)
-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)

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

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

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