diff --git a/lib/core/agent.py b/lib/core/agent.py index ab28d2bf4..455e44f35 100644 --- a/lib/core/agent.py +++ b/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 diff --git a/lib/core/settings.py b/lib/core/settings.py index 5e0accc99..4ca26c663 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.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) diff --git a/plugins/generic/custom.py b/plugins/generic/custom.py index cdab69237..4d3ce0b70 100644 --- a/plugins/generic/custom.py +++ b/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)) diff --git a/tests/test_agent.py b/tests/test_agent.py index 1a6f6882e..bfbec6dbf 100644 --- a/tests/test_agent.py +++ b/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) diff --git a/tests/test_generic_takeover.py b/tests/test_generic_takeover.py index 95dda4ddb..fc8889730 100644 --- a/tests/test_generic_takeover.py +++ b/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 "), []) # --------------------------------------------------------------------------- #