Fixes #4708

lib/core/agent.py

@@ -658,7 +658,7 @@ class Agent(object):
         fieldsSelectCase = re.search(r"\ASELECT%s\s+(\(CASE WHEN\s+.+\s+END\))" % prefixRegex, query, re.I)
         fieldsSelectFrom = re.search(r"\ASELECT%s\s+(.+?)\s+FROM " % prefixRegex, query, re.I)
         fieldsExists = re.search(r"EXISTS\(([^)]*)\)\Z", query, re.I)
-        fieldsSelect = re.search(r"\ASELECT%s\s+(.*)" % prefixRegex, query, re.I)
+        fieldsSelect = re.search(r"\ASELECT%s\s+(.*)" % prefixRegex, query, re.I | re.S)
         fieldsSubstr = re.search(r"\A(SUBSTR|MID\()", query, re.I)
         fieldsMinMaxstr = re.search(r"(?:MIN|MAX)\(([^\(\)]+)\)", query, re.I)
         fieldsNoSelect = query

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.10.4"
+VERSION = "1.10.10.5"
 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)

plugins/generic/custom.py

@@ -12,10 +12,12 @@ import sys
 
 from lib.core.common import Backend
 from lib.core.common import dataToStdout
-from lib.core.common import getSQLSnippet
 from lib.core.common import isListLike
 from lib.core.common import isStackingAvailable
 from lib.core.common import joinValue
+from lib.core.common import readCachedFileContent
+from lib.core.common import splitFields
+from lib.core.common import zeroDepthSearch
 from lib.core.compat import xrange
 from lib.core.convert import getUnicode
 from lib.core.data import conf
@@ -148,12 +150,12 @@ class Custom(object):
             if not filename:
                 continue
 
-            snippet = getSQLSnippet(Backend.getDbms(), filename)
+            # User SQL is not a bundled procedure template: preserve literals and percent tokens.
+            snippet = readCachedFileContent(filename)
+            queries = [_.strip() for _ in splitFields(snippet, delimiter=';' if zeroDepthSearch(snippet, ';') else '\n') if _.strip()]
 
-            if snippet and all(query.strip().upper().startswith("SELECT") for query in (_ for _ in snippet.split(';' if ';' in snippet else '\n') if _)):
-                for query in (_ for _ in snippet.split(';' if ';' in snippet else '\n') if _):
-                    query = query.strip()
-                    if query:
-                        conf.dumper.sqlQuery(query, self.sqlQuery(query))
-            else:
+            if queries and all(query.upper().startswith("SELECT") for query in queries):
+                for query in queries:
+                    conf.dumper.sqlQuery(query, self.sqlQuery(query))
+            elif queries:
                 conf.dumper.sqlQuery(snippet, self.sqlQuery(snippet))

tests/test_agent.py

@@ -186,6 +186,12 @@ class TestPrefixSuffix(unittest.TestCase):
 
 
 class TestGetFields(unittest.TestCase):
+    def test_multiline_scalar_literal(self):
+        field = "HEX('a;\n  b')"
+        result = agent.getFields("SELECT %s" % field)
+        self.assertEqual(result[5], [field])
+        self.assertEqual(result[6], field)
+
     def test_extracts_select_list(self):
         # getFields(query) returns an 8-tuple; the fields-bearing slots are:
         #   [0],[1] = regex match objects for the SELECT/expression (must be found, not None)

tests/test_generic_takeover.py

@@ -21,8 +21,10 @@ prompts non-interactive, and conf.dumper is a recording stub. Every test restore
 all touched conf.* / kb.* / patched module attributes in tearDown so nothing leaks.
 """
 
+import io
 import os
 import sys
+import tempfile
 import unittest
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
@@ -39,6 +41,7 @@ import plugins.generic.entries as emod
 import plugins.generic.custom as cmod
 import plugins.generic.misc as mmod
 import plugins.generic.takeover as tmod
+import lib.core.common as common
 from plugins.generic.custom import Custom
 from plugins.generic.misc import Miscellaneous
 
@@ -250,38 +253,65 @@ class TestCustomSqlQuery(_GenericBase):
 
 
 class TestCustomSqlFile(_GenericBase):
-    def test_sqlfile_select_snippets(self):
-        set_dbms("MySQL")
-        c = Custom()
-        cmod.inject.getValue = lambda query, **k: "r"
+    def _runSqlFile(self, content, suffix=".sql"):
+        handle, filename = tempfile.mkstemp(suffix=suffix)
+        os.close(handle)
+        self.addCleanup(os.unlink, filename)
+        with io.open(filename, "w", encoding="utf-8") as f:
+            f.write(content)
 
-        # getSQLSnippet reads from disk; patch it to return inline SQL.
-        saved = cmod.getSQLSnippet
-        try:
-            cmod.getSQLSnippet = lambda dbms, filename, **kw: "SELECT 1;SELECT 2"
-            conf.sqlFile = "dummy.sql"
-            c.sqlFile()
-            # two SELECT statements => two recorded dumper.sqlQuery calls
-            self.assertEqual(len(conf.dumper.sqlQueries), 2)
-        finally:
-            cmod.getSQLSnippet = saved
+        set_dbms("MySQL")
+        cmod.inject.getValue = lambda query, **k: "r"
+        conf.sqlFile = filename
+        Custom().sqlFile()
+        return [query for query, output in conf.dumper.sqlQueries]
+
+    def test_sqlfile_select_snippets(self):
+        self.assertEqual(self._runSqlFile(u"SELECT 1; SELECT 2;\n"), ["SELECT 1", "SELECT 2"])
 
     def test_sqlfile_nonselect_snippet(self):
-        set_dbms("MySQL")
-        conf.direct = True
-        c = Custom()
         cmod.inject.goStacked = lambda *a, **k: None
+        self.assertEqual(self._runSqlFile(u"DROP TABLE x"), ["DROP TABLE x"])
 
-        saved = cmod.getSQLSnippet
+    def test_sqlfile_preserves_percent_tokens_without_prompting(self):
+        saved = common.readInput
         try:
-            cmod.getSQLSnippet = lambda dbms, filename, **kw: "DROP TABLE x"
-            conf.sqlFile = "dummy.sql"
-            c.sqlFile()
-            # non-SELECT => single recorded call with the whole snippet
-            self.assertEqual(len(conf.dumper.sqlQueries), 1)
-            self.assertEqual(conf.dumper.sqlQueries[0][0], "DROP TABLE x")
+            common.readInput = lambda *a, **k: self.fail("user SQL must not request template substitutions")
+            query = u"SELECT DATE_FORMAT(NOW(), '%m%m'), '%RANDSTR1%', '%RANDINT1%'"
+            self.assertEqual(self._runSqlFile(query), [query])
         finally:
-            cmod.getSQLSnippet = saved
+            common.readInput = saved
+
+    def test_sqlfile_preserves_hash_and_quoted_delimiters(self):
+        first = u"SELECT '#keep;  value', 'it''s;  quoted', \"semi;  colon\""
+        self.assertEqual(self._runSqlFile(first + u"; SELECT 2;"), [first, "SELECT 2"])
+
+    def test_sqlfile_preserves_multiline_literal(self):
+        query = u"SELECT 'a;\n  b'"
+        self.assertEqual(self._runSqlFile(query), [query])
+
+    def test_sqlfile_newline_separated_selects(self):
+        self.assertEqual(self._runSqlFile(u"SELECT 'a;b'\nSELECT 2\n"), ["SELECT 'a;b'", "SELECT 2"])
+
+    def test_sqlfile_multiline_select(self):
+        query = u"SELECT name\nFROM users"
+        self.assertEqual(self._runSqlFile(query), [query])
+
+    def test_sqlfile_preserves_mixed_batch(self):
+        batch = u"UPDATE users SET name='#keep;  value';\nSELECT name FROM users;"
+        executed = []
+        cmod.inject.goStacked = lambda query, **k: executed.append(query)
+        self.assertEqual(self._runSqlFile(batch), [batch])
+        self.assertEqual(executed, [batch.rstrip(';')])
+
+    def test_sqlfile_arbitrary_extension(self):
+        self.assertEqual(self._runSqlFile(u"SELECT 1", suffix=".txt"), ["SELECT 1"])
+
+    def test_sqlfile_without_extension(self):
+        self.assertEqual(self._runSqlFile(u"SELECT 1", suffix=""), ["SELECT 1"])
+
+    def test_sqlfile_empty(self):
+        self.assertEqual(self._runSqlFile(u" \n;\n "), [])
 
 
 # --------------------------------------------------------------------------- #