GithubStar/sqlmap
1
0
Fork 0
mirror of https://github.com/sqlmapproject/sqlmap.git synced 2026-10-28 20:12:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

Implements --links (#21)

Browse Source
This commit is contained in:
Miroslav Štampar
2026-10-19 10:19:34 +02:00
parent 559d6f4543
commit e90078184d
13 changed files with 179 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
data/xml/queries.xml
+8
View File
@@ -173,6 +173,10 @@
</dbms>
<dbms value="Microsoft SQL Server">
<links>
<inband query="SELECT CAST(name AS NVARCHAR(MAX)) COLLATE DATABASE_DEFAULT+N' -> '+ISNULL(data_source COLLATE DATABASE_DEFAULT,N'&lt;unknown&gt;') FROM master.sys.servers WHERE is_linked=1 ORDER BY server_id"/>
<blind query="SELECT link_name FROM (SELECT CAST(name AS NVARCHAR(MAX)) COLLATE DATABASE_DEFAULT+N' -> '+ISNULL(data_source COLLATE DATABASE_DEFAULT,N'&lt;unknown&gt;') AS link_name,ROW_NUMBER() OVER (ORDER BY server_id) AS rn FROM master.sys.servers WHERE is_linked=1) AS links WHERE rn=%d+1" count="SELECT COUNT(*) FROM master.sys.servers WHERE is_linked=1"/>
</links>
<cast query="CAST(%s AS NVARCHAR(MAX))"/>
<length query="LTRIM(STR(LEN(%s+'.')-1))"/>
<isnull query="ISNULL(%s,' ')"/>
@@ -253,6 +257,10 @@
</dbms>
<dbms value="Oracle">
<links>
<inband query="SELECT OWNER||'.'||DB_LINK||' -> '||NVL(HOST,'&lt;unknown&gt;') FROM ALL_DB_LINKS ORDER BY OWNER,DB_LINK"/>
<blind query="SELECT link_name FROM (SELECT OWNER||'.'||DB_LINK||' -> '||NVL(HOST,'&lt;unknown&gt;') AS link_name,ROW_NUMBER() OVER (ORDER BY OWNER,DB_LINK) AS rn FROM ALL_DB_LINKS) WHERE rn=%d+1" count="SELECT COUNT(*) FROM ALL_DB_LINKS"/>
</links>
<cast query="CAST(%s AS VARCHAR(4000))"/>
<length query="LENGTH(%s)"/>
<isnull query="NVL(%s,' ')"/>
lib/controller/action.py
+6
View File
@@ -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)
lib/core/dump.py
+3
View File
@@ -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()
lib/core/enums.py
+1
View File
@@ -429,6 +429,7 @@ class CONTENT_TYPE(object):
REG_READ = 25
STATEMENTS = 26
PROCEDURES = 27
LINKS = 28
class CONTENT_STATUS(object):
IN_PROGRESS = 0
lib/core/optiondict.py
+1
View File
@@ -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"),
lib/core/settings.py
+1 -1
View File
@@ -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.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)
lib/core/testing.py
+1
View File
@@ -113,6 +113,7 @@ def vulnTest(tests=None, label="vuln"):
("-u \"<url>&query=*\" --flush-session --technique=Q --banner", ("Title: SQLite inline queries", "banner: '3.")),
("-d \"<direct>\" --flush-session --dump -T creds --dump-format=SQLITE --binary-fields=password_hash --where \"user_id=5\"", ("3137396164343563366365326362393763663130323965323132303436653831", "dumped to SQLITE database")),
("-d \"<direct>\" --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 \"<direct>\" --links", ("back-end DBMS: SQLite", "enumeration of database links is not supported on SQLite", "~unhandled exception")),
("-u <base>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")),
)
lib/parse/cmdline.py
+3
View File
@@ -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")
plugins/generic/databases.py
+36
View File
@@ -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)
sqlmap.conf
+4
View File
@@ -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 =
tests/test_configfile.py
+3
View File
@@ -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()
tests/test_databases_enum.py
+107 -1
View File
@@ -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 -> <unknown>"]]
dbmod.inject.getValue = gv
expected = ["remote -> db.example", "other -> <unknown>"]
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 -> <unknown>", "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")
tests/test_dump_format.py
+5
View File
@@ -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")