From e90078184d63b4e39121049cb1351f27756cb974 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Mon, 19 Oct 2026 10:19:34 +0200 Subject: [PATCH] Implements --links (#21) --- data/xml/queries.xml | 8 +++ lib/controller/action.py | 6 ++ lib/core/dump.py | 3 + lib/core/enums.py | 1 + lib/core/optiondict.py | 1 + lib/core/settings.py | 2 +- lib/core/testing.py | 1 + lib/parse/cmdline.py | 3 + plugins/generic/databases.py | 36 ++++++++++++ sqlmap.conf | 4 ++ tests/test_configfile.py | 3 + tests/test_databases_enum.py | 108 ++++++++++++++++++++++++++++++++++- tests/test_dump_format.py | 5 ++ 13 files changed, 179 insertions(+), 2 deletions(-) diff --git a/data/xml/queries.xml b/data/xml/queries.xml index e2fb53db9..8f930538c 100644 --- a/data/xml/queries.xml +++ b/data/xml/queries.xml @@ -173,6 +173,10 @@ + + + + @@ -253,6 +257,10 @@ + + + + diff --git a/lib/controller/action.py b/lib/controller/action.py index 000bc4215..f0b190be4 100644 --- a/lib/controller/action.py +++ b/lib/controller/action.py @@ -134,6 +134,12 @@ def action(): if conf.getProcs: conf.dumper.procedures(conf.dbmsHandler.getProcedures()) + if conf.getLinks: + try: + conf.dumper.links(conf.dbmsHandler.getLinks()) + except SqlmapNoneDataException as ex: + logger.critical(ex) + if conf.getPasswordHashes: try: conf.dumper.userSettings("database management system users password hashes", conf.dbmsHandler.getPasswordHashes(), "password hash", CONTENT_TYPE.PASSWORDS) diff --git a/lib/core/dump.py b/lib/core/dump.py index f1b2514c7..a5f620a07 100644 --- a/lib/core/dump.py +++ b/lib/core/dump.py @@ -224,6 +224,9 @@ class Dump(object): def procedures(self, procedures): self.lister("stored procedures", procedures, content_type=CONTENT_TYPE.PROCEDURES) + def links(self, links): + self.lister("database links", links, content_type=CONTENT_TYPE.LINKS) + def userSettings(self, header, userSettings, subHeader, content_type=None): self._areAdmins = set() diff --git a/lib/core/enums.py b/lib/core/enums.py index 899bf51f7..6a5419117 100644 --- a/lib/core/enums.py +++ b/lib/core/enums.py @@ -429,6 +429,7 @@ class CONTENT_TYPE(object): REG_READ = 25 STATEMENTS = 26 PROCEDURES = 27 + LINKS = 28 class CONTENT_STATUS(object): IN_PROGRESS = 0 diff --git a/lib/core/optiondict.py b/lib/core/optiondict.py index ae26b65fb..76c6fb82c 100644 --- a/lib/core/optiondict.py +++ b/lib/core/optiondict.py @@ -163,6 +163,7 @@ optDict = { "getPasswordHashes": ("boolean", "Passwords"), "getPrivileges": ("boolean", "Privileges"), "getRoles": ("boolean", "Roles"), + "getLinks": "boolean", "getDbs": ("boolean", "Databases"), "getTables": ("boolean", "Tables"), "getColumns": ("boolean", "Columns"), diff --git a/lib/core/settings.py b/lib/core/settings.py index ed37c29af..1f7cdcb81 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.8" +VERSION = "1.10.10.9" 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 509e03021..c074e8b0b 100644 --- a/lib/core/testing.py +++ b/lib/core/testing.py @@ -113,6 +113,7 @@ def vulnTest(tests=None, label="vuln"): ("-u \"&query=*\" --flush-session --technique=Q --banner", ("Title: SQLite inline queries", "banner: '3.")), ("-d \"\" --flush-session --dump -T creds --dump-format=SQLITE --binary-fields=password_hash --where \"user_id=5\"", ("3137396164343563366365326362393763663130323965323132303436653831", "dumped to SQLITE database")), ("-d \"\" --flush-session --banner --schema --sql-query=\"UPDATE users SET name='foobar' WHERE id=4; SELECT * FROM users; SELECT 987654321\"", ("banner: '3.", "INTEGER", "TEXT", "id", "name", "surname", "4,foobar,nameisnull", "'987654321'",)), + ("-d \"\" --links", ("back-end DBMS: SQLite", "enumeration of database links is not supported on SQLite", "~unhandled exception")), ("-u csrf --data=\"id=1&csrf_token=1\" --banner --answers=\"update=y\" --flush-session --technique=B", ("back-end DBMS: SQLite", "banner: '3.")), ("--purge -v 3", ("~ERROR", "~CRITICAL", "deleting the whole directory tree")), ) diff --git a/lib/parse/cmdline.py b/lib/parse/cmdline.py index 2359f585f..acef3dc04 100644 --- a/lib/parse/cmdline.py +++ b/lib/parse/cmdline.py @@ -499,6 +499,9 @@ def cmdLineParser(argv=None): enumeration.add_argument("--roles", dest="getRoles", action="store_true", help="Enumerate DBMS users roles") + enumeration.add_argument("--links", dest="getLinks", action="store_true", + help="Enumerate DBMS links") + enumeration.add_argument("--dbs", dest="getDbs", action="store_true", help="Enumerate DBMS databases") diff --git a/plugins/generic/databases.py b/plugins/generic/databases.py index becad0332..2f9441fd7 100644 --- a/plugins/generic/databases.py +++ b/plugins/generic/databases.py @@ -72,6 +72,7 @@ class Databases(object): kb.data.dumpedTable = {} kb.data.cachedStatements = [] kb.data.cachedProcedures = [] + kb.data.cachedLinks = [] def getCurrentDb(self): infoMsg = "fetching current database" @@ -1181,6 +1182,41 @@ class Databases(object): return kb.data.cachedStatements + def getLinks(self): + dbms = Backend.getIdentifiedDbms() + if dbms not in queries or "links" not in queries[dbms]: + logger.warning("enumeration of database links is not supported on %s" % dbms) + return [] + + if kb.data.cachedLinks: + return kb.data.cachedLinks + + logger.info("fetching database links visible to the current user") + rootQuery = queries[dbms].links + + if any(isTechniqueAvailable(_) for _ in (PAYLOAD.TECHNIQUE.UNION, PAYLOAD.TECHNIQUE.ERROR, PAYLOAD.TECHNIQUE.QUERY)) or conf.direct: + values = inject.getValue(rootQuery.inband.query, blind=False, time=False) + if not isNoneValue(values): + for value in arrayizeValue(values): + value = unArrayizeValue(value) + if not isNoneValue(value): + kb.data.cachedLinks.append(value.strip()) + + if not kb.data.cachedLinks and isInferenceAvailable() and not conf.direct: + count = inject.getValue(rootQuery.blind.count, union=False, error=False, expected=EXPECTED.INT, charsetType=CHARSET_TYPE.DIGITS) + if count in (0, "0"): + return [] + elif not isNumPosStrValue(count): + logger.warning("unable to retrieve the number of database links") + return [] + + for index in getLimitRange(count): + value = unArrayizeValue(inject.getValue(rootQuery.blind.query % index, union=False, error=False)) + if not isNoneValue(value): + kb.data.cachedLinks.append(value.strip()) + + return kb.data.cachedLinks + def getProcedures(self): infoMsg = "fetching stored procedures" logger.info(infoMsg) diff --git a/sqlmap.conf b/sqlmap.conf index ed2c2c076..2f7c22d28 100644 --- a/sqlmap.conf +++ b/sqlmap.conf @@ -544,6 +544,10 @@ getComments = False # Valid: True or False getStatements = False +# Enumerate database management system links and their destinations. +# Valid: True or False +getLinks = False + # Back-end database management system database to enumerate. db = diff --git a/tests/test_configfile.py b/tests/test_configfile.py index b09242b95..1fbe8a782 100644 --- a/tests/test_configfile.py +++ b/tests/test_configfile.py @@ -113,14 +113,17 @@ class TestConfigFileRoundTrip(unittest.TestCase): try: conf.url = "http://127.0.0.1:1/?id=1" conf.commonTables = True + conf.getLinks = True conf.tablePrefix = "zzz" saveConfig(conf, path) conf.url, conf.commonTables, conf.tablePrefix = None, False, "overwritten" + conf.getLinks = False configFileParser(path) self.assertEqual(conf.url, "http://127.0.0.1:1/?id=1") self.assertTrue(conf.commonTables) + self.assertTrue(conf.getLinks) self.assertEqual(conf.tablePrefix, "zzz") finally: conf.clear() diff --git a/tests/test_databases_enum.py b/tests/test_databases_enum.py index e31360118..e1a272a39 100644 --- a/tests/test_databases_enum.py +++ b/tests/test_databases_enum.py @@ -17,6 +17,8 @@ the suite. """ import os +import sqlite3 +import subprocess import sys import unittest @@ -25,7 +27,7 @@ from _testutils import bootstrap, set_dbms, reset_dbms bootstrap() -from lib.core.data import conf, kb +from lib.core.data import conf, kb, queries from lib.core.enums import EXPECTED, PAYLOAD import plugins.generic.databases as dbmod from plugins.generic.databases import Databases @@ -500,6 +502,110 @@ class TestGetSchema(_BaseEnumTest): self.assertEqual(result["testdb"]["users"].get("id"), "int") +class TestGetLinks(_BaseEnumTest): + def setUp(self): + _BaseEnumTest.setUp(self) + self._saved_links = kb.data.get("cachedLinks") + + def tearDown(self): + kb.data.cachedLinks = self._saved_links + _BaseEnumTest.tearDown(self) + + def test_unsupported_dbms_does_not_query(self): + d = self._fresh() + dbmod.inject.getValue = lambda *a, **k: self.fail("unsupported DBMS must not query links") + saved = dbmod.logger.warning + warnings = [] + try: + dbmod.logger.warning = warnings.append + for dbms in queries: + if "links" not in queries[dbms]: + set_dbms(dbms) + self.assertEqual(d.getLinks(), []) + self.assertIn(dbms, warnings[-1]) + finally: + dbmod.logger.warning = saved + + def test_direct_links_and_cache(self): + conf.direct = True + for dbms in ("Microsoft SQL Server", "Oracle"): + set_dbms(dbms) + d = self._fresh() + calls = [] + + def gv(query, **kwargs): + calls.append(query) + return [["remote -> db.example"], ["other -> "]] + + dbmod.inject.getValue = gv + expected = ["remote -> db.example", "other -> "] + self.assertEqual(d.getLinks(), expected) + self.assertEqual(d.getLinks(), expected) + self.assertEqual(calls, [queries[dbms].links.inband.query]) + + def test_blind_links_use_zero_based_paging(self): + self._enable_inference() + for dbms in ("Microsoft SQL Server", "Oracle"): + set_dbms(dbms) + d = self._fresh() + calls = [] + values = iter(["2", ["first -> host1"], ["second -> host2"]]) + + def gv(query, **kwargs): + calls.append(query) + return next(values) + + dbmod.inject.getValue = gv + self.assertEqual(d.getLinks(), ["first -> host1", "second -> host2"]) + root = queries[dbms].links + self.assertEqual(calls, [root.blind.count, root.blind.query % 0, root.blind.query % 1]) + + def test_empty_direct_results(self): + set_dbms("Microsoft SQL Server") + conf.direct = True + d = self._fresh() + dbmod.inject.getValue = lambda *a, **k: None + self.assertEqual(d.getLinks(), []) + + def test_oracle_query_paging_and_null_destination(self): + # Evaluate the catalog/paging SQL with compatible SQLite expressions; live Oracle is + # not required for this check of owner-qualified names, NULL handling and row offsets. + conn = sqlite3.connect(":memory:") + try: + conn.create_function("NVL", 2, lambda value, default: default if value is None else value) + conn.execute("CREATE TABLE ALL_DB_LINKS (OWNER TEXT, DB_LINK TEXT, HOST TEXT)") + conn.executemany("INSERT INTO ALL_DB_LINKS VALUES (?,?,?)", [("PUBLIC", "REMOTE", "db.example"), ("APP", "REMOTE", None)]) + root = queries["Oracle"].links + expected = ["APP.REMOTE -> ", "PUBLIC.REMOTE -> db.example"] + self.assertEqual([row[0] for row in conn.execute(root.inband.query)], expected) + self.assertEqual(conn.execute(root.blind.count).fetchone()[0], 2) + self.assertEqual([conn.execute(root.blind.query % i).fetchone()[0] for i in range(2)], expected) + self.assertIsNone(conn.execute(root.blind.query % 2).fetchone()) + finally: + conn.close() + + def test_empty_or_unavailable_blind_count_does_not_fetch_rows(self): + set_dbms("Oracle") + self._enable_inference() + for count in (0, "0", None, "unavailable"): + d = self._fresh() + + def gv(query, **kwargs): + self.assertEqual(kwargs.get("expected"), EXPECTED.INT) + return count + + dbmod.inject.getValue = gv + self.assertEqual(d.getLinks(), []) + + def test_help_is_one_line(self): + output = subprocess.check_output([sys.executable, os.path.join(os.path.dirname(__file__), "..", "sqlmap.py"), "-hh"]) + lines = output.decode("utf-8").splitlines() + helpLines = [line for line in lines if "--links" in line] + self.assertEqual(len(helpLines), 1) + self.assertIn("Enumerate DBMS links", helpLines[0]) + self.assertLessEqual(len(helpLines[0]), 78) + + class TestGetProcedures(_BaseEnumTest): def test_procedures_direct_pgsql(self): set_dbms("PostgreSQL") diff --git a/tests/test_dump_format.py b/tests/test_dump_format.py index 34d50558b..cb966fd0f 100644 --- a/tests/test_dump_format.py +++ b/tests/test_dump_format.py @@ -76,6 +76,11 @@ class _CaptureCase(unittest.TestCase): class TestStringAndLister(_CaptureCase): + def test_links_output(self): + self.d.links(["remote -> db.example"]) + self.assertIn("database links [1]", self.text()) + self.assertIn("remote -> db.example", self.text()) + def test_string_scalar_quoted(self): # a plain string fact is rendered as "header: 'value'" self.d.string("current user", "root@localhost")