mirror of
https://github.com/sqlmapproject/sqlmap.git
synced 2026-11-07 17:54:40 -10:00
245 lines
13 KiB
Python
245 lines
13 KiB
Python
#!/usr/bin/env python
|
|
|
|
"""
|
|
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
|
|
See the file 'LICENSE' for copying permission
|
|
|
|
Audit acquisition preserves NULL/empty/unknown states and interpretation boundaries.
|
|
Real DBMS/SQLi coverage is opt-in under tests/integration/audit/.
|
|
"""
|
|
|
|
import importlib
|
|
import os
|
|
import subprocess
|
|
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.common import getSQLSnippet
|
|
from lib.core.data import conf, kb, paths, queries
|
|
from lib.core.enums import DBMS, FORK, EXPECTED, PAYLOAD
|
|
from lib.core.settings import DBMS_DIRECTORY_DICT
|
|
from lib.request import inject
|
|
from plugins.generic.audit import Audit
|
|
from plugins.generic.databases import Databases
|
|
|
|
tearDownModule = reset_dbms
|
|
|
|
|
|
class TestAudit(unittest.TestCase):
|
|
def setUp(self):
|
|
self._saved = dict(conf)
|
|
self._cache = kb.data.get('cachedAudit')
|
|
self._getValue = inject.getValue
|
|
self._forced = kb.get('forcedDbms')
|
|
self._injection = kb.injection.data
|
|
self._gadget = kb.get("gadget")
|
|
kb.injection.data = {}
|
|
kb.gadget = None
|
|
conf.direct = False
|
|
kb.data.cachedAudit = []
|
|
set_dbms(DBMS.MYSQL)
|
|
|
|
def tearDown(self):
|
|
kb.injection.data = self._injection
|
|
kb.gadget = self._gadget
|
|
inject.getValue = self._getValue
|
|
kb.data.cachedAudit = self._cache
|
|
kb.forcedDbms = self._forced
|
|
conf.clear()
|
|
conf.update(self._saved)
|
|
|
|
def test_preserves_null_empty_and_marker_like_values(self):
|
|
inject.getValue = lambda *a, **k: [['null', 'N'], ['empty', 'V'], ['text', 'VNULL'], ['prefix', 'VVvalue'], ['padded ', 'Vkeep '], ['trailing\n', 'Vvalue']]
|
|
audit = Audit()
|
|
audit.dbms = DBMS.MYSQL
|
|
self.assertEqual(audit._auditFetch('audit_identity'), {'null': None, 'empty': '', 'text': 'NULL', 'prefix': 'Vvalue', 'padded': 'keep ', 'trailing\n': 'value'})
|
|
|
|
def test_single_row_and_failed_results(self):
|
|
audit = Audit()
|
|
audit.dbms = DBMS.MYSQL
|
|
for value, expected in ((['name', 'Vvalue'], {'name': 'value'}), (None, None), ([], None), (['truncated'], None), ([['key', 'invalid']], None)):
|
|
inject.getValue = lambda *a, **k: value
|
|
self.assertEqual(audit._auditFetch('audit_identity'), expected)
|
|
|
|
def test_major_dbmses_dispatch_without_xml_audit_blocks(self):
|
|
inject.getValue = lambda *a, **k: [['effective account', 'Vtest']]
|
|
for dbms in (DBMS.MYSQL, DBMS.PGSQL, DBMS.MSSQL, DBMS.ORACLE):
|
|
set_dbms(dbms)
|
|
d = Databases()
|
|
self.assertNotIn('audit', queries[dbms])
|
|
result = d.getAudit()
|
|
self.assertTrue(any('test' in item for item in result))
|
|
self.assertFalse(any('OS boundary:' in item for item in result))
|
|
|
|
def test_unsupported_dbmses_do_not_query(self):
|
|
inject.getValue = lambda *a, **k: self.fail('unsupported DBMS must not query')
|
|
for dbms in queries:
|
|
if dbms not in (DBMS.MYSQL, DBMS.PGSQL, DBMS.MSSQL, DBMS.ORACLE):
|
|
set_dbms(dbms)
|
|
self.assertEqual(Databases().getAudit(), [])
|
|
|
|
def test_failures_are_not_reported_as_disabled(self):
|
|
inject.getValue = lambda *a, **k: None
|
|
for dbms in (DBMS.MYSQL, DBMS.PGSQL, DBMS.MSSQL, DBMS.ORACLE):
|
|
set_dbms(dbms)
|
|
result = Databases().getAudit()
|
|
self.assertTrue(any('retrieval failed' in item for item in result))
|
|
self.assertFalse(any(item.endswith(': disabled') for item in result))
|
|
|
|
def test_cache_and_bulk_queries_use_existing_extraction(self):
|
|
calls = []
|
|
def getValue(query, **kwargs):
|
|
self.assertTrue(kwargs.get('fromUser'))
|
|
self.assertTrue(kwargs.get('batch'))
|
|
self.assertNotIn('\n', query)
|
|
self.assertNotIn('\r', query)
|
|
calls.append(query)
|
|
return [['effective account', 'Vtest']]
|
|
inject.getValue = getValue
|
|
d = Databases()
|
|
result = d.getAudit()
|
|
count = len(calls)
|
|
self.assertEqual(count, 4)
|
|
self.assertEqual(d.getAudit(), result)
|
|
self.assertEqual(len(calls), count)
|
|
|
|
def test_mysql_directory_null_and_empty_are_different(self):
|
|
from plugins.dbms.mysql.audit import Audit as MySQLAudit
|
|
audit = MySQLAudit()
|
|
cases = ((None, 'blocks'), ('NULL', 'blocks'), ('', 'no directory restriction'), ('/var/lib/mysql-files/', 'must be inside'))
|
|
for value, expected in cases:
|
|
result = audit.explain({'file access': {'secure_file_priv': value}})
|
|
self.assertTrue(any(expected in item for item in result))
|
|
self.assertFalse(any('must be inside' in item for item in audit.explain({})))
|
|
|
|
def test_mariadb_selects_fork_specific_configuration(self):
|
|
from lib.core.common import Backend
|
|
from plugins.dbms.mysql.audit import Audit as MySQLAudit
|
|
original = Backend.isFork
|
|
Backend.isFork = staticmethod(lambda fork: fork == FORK.MARIADB)
|
|
try:
|
|
self.assertEqual(MySQLAudit().queries[1][1][0], 'audit_configuration_mariadb')
|
|
self.assertFalse(any('blocks' in item for item in MySQLAudit().explain({'file access': {'secure_file_priv': None}})))
|
|
finally:
|
|
Backend.isFork = staticmethod(original)
|
|
|
|
def test_mssql_explicit_permission_denial(self):
|
|
from plugins.dbms.mssqlserver.audit import Audit as MSSQLAudit
|
|
audit = MSSQLAudit()
|
|
groups = {'permissions': {'read configuration metadata': '0'}}
|
|
self.assertEqual(audit.failure('configuration', groups), 'configuration: SELECT permission missing on master.sys.configurations')
|
|
self.assertEqual(audit.failure('linked servers', groups), 'linked servers: retrieval failed')
|
|
|
|
def test_guidance_requires_retrieved_capabilities(self):
|
|
from plugins.dbms.mysql.audit import Audit as MySQLAudit
|
|
from plugins.dbms.postgresql.audit import Audit as PostgreSQLAudit
|
|
from plugins.dbms.mssqlserver.audit import Audit as MSSQLAudit
|
|
from plugins.dbms.oracle.audit import Audit as OracleAudit
|
|
for audit in (MySQLAudit(), PostgreSQLAudit(), MSSQLAudit(), OracleAudit()):
|
|
self.assertEqual(audit.explain({}), [])
|
|
self.assertEqual(MySQLAudit().explain({'privileges': {'direct FILE grant': '0'}}), [])
|
|
result = MySQLAudit().explain({'file access': {'secure_file_priv': '/allowed/', 'max_allowed_packet': '1024'}, 'privileges': {'direct FILE grant': '1'}})
|
|
self.assertTrue(any('must be inside' in item for item in result))
|
|
self.assertTrue(any('file size < 1024 bytes' in item for item in result))
|
|
result = MySQLAudit().explain({'file access': {'secure_file_priv': None}, 'privileges': {'direct FILE grant': '1'}})
|
|
self.assertFalse(any('FILE granted; use' in item for item in result))
|
|
result = PostgreSQLAudit().explain({'privileges': {'superuser': '0', 'pg_read_server_files': '1', 'pg_execute_server_program': '0'}})
|
|
self.assertTrue(any('--file-read' in item for item in result))
|
|
self.assertFalse(any('--os-cmd' in item for item in result))
|
|
result = MSSQLAudit().explain({'configuration': {'xp_cmdshell': '1'}, 'permissions': {'execute xp_cmdshell': '0'}})
|
|
self.assertTrue(any('EXECUTE permission missing' in item for item in result))
|
|
self.assertFalse(any('--os-cmd' in item for item in result))
|
|
conf.direct = True
|
|
result = MSSQLAudit().explain({'configuration': {'xp_cmdshell': '0'}, 'identity': {'sysadmin': '1'}})
|
|
self.assertTrue(any('can enable xp_cmdshell' in item for item in result))
|
|
result = MSSQLAudit().explain({'configuration': {'xp_cmdshell': '1'}, 'permissions': {'execute xp_cmdshell': '1'}})
|
|
self.assertTrue(any('--os-cmd' in item for item in result))
|
|
result = PostgreSQLAudit().explain({'privileges': {'pg_execute_server_program': '1'}})
|
|
self.assertTrue(any('--os-cmd' in item for item in result))
|
|
|
|
def test_execution_advice_uses_known_routes_without_probing(self):
|
|
from plugins.dbms.postgresql.audit import Audit as PostgreSQLAudit
|
|
from plugins.dbms.mssqlserver.audit import Audit as MSSQLAudit
|
|
inject.getValue = lambda *a, **k: self.fail('guidance must not send probes')
|
|
pg = {'privileges': {'pg_execute_server_program': '1'}}
|
|
ms = {'configuration': {'xp_cmdshell': '1'}, 'permissions': {'execute xp_cmdshell': '1'}}
|
|
for audit, groups in ((PostgreSQLAudit(), pg), (MSSQLAudit(), ms)):
|
|
for route in ('none', 'stacked', 'gadget', 'direct'):
|
|
kb.injection.data = {PAYLOAD.TECHNIQUE.STACKED: {'title': 'stacked queries'}} if route == 'stacked' else {}
|
|
kb.gadget = {'command': 'known gadget'} if route == 'gadget' else None
|
|
conf.direct = route == 'direct'
|
|
result = audit.explain(groups)
|
|
self.assertEqual(any('--os-cmd' in item for item in result), route != 'none')
|
|
self.assertEqual(any('no statement execution route identified' in item for item in result), route == 'none')
|
|
|
|
def test_rls_setting_does_not_grant_bypass(self):
|
|
from plugins.dbms.postgresql.audit import Audit as PostgreSQLAudit
|
|
result = PostgreSQLAudit().explain({'configuration': {'row_security': 'off'}, 'privileges': {'superuser': '0', 'bypass RLS': '0'}})
|
|
self.assertFalse(any('bypasses' in item for item in result))
|
|
result = PostgreSQLAudit().explain({'privileges': {'bypass RLS': '1'}})
|
|
self.assertTrue(any('bypasses' in item for item in result))
|
|
|
|
def test_empty_result_requires_confirmed_zero_count(self):
|
|
audit = Audit()
|
|
audit.dbms = DBMS.MYSQL
|
|
for count, expected in ((0, {}), ('0', {}), ([[0]], {}), (None, None), ('1', None)):
|
|
calls = []
|
|
def getValue(query, **kwargs):
|
|
calls.append(query)
|
|
return count if kwargs.get('expected') == EXPECTED.INT else None
|
|
inject.getValue = getValue
|
|
self.assertEqual(audit._auditFetch('audit_identity'), expected)
|
|
self.assertEqual(len(calls), 2)
|
|
audit.queries = (('empty scope', ('audit_identity',)),)
|
|
inject.getValue = lambda query, **kwargs: 0 if kwargs.get('expected') == EXPECTED.INT else None
|
|
self.assertEqual(audit.getAudit(), ['empty scope: no entries'])
|
|
|
|
def test_control_characters_are_escaped_in_facts_and_guidance(self):
|
|
from plugins.dbms.mysql.audit import Audit as MySQLAudit
|
|
value = "/allowed/\n[*] forged\r\x1b[2J"
|
|
inject.getValue = lambda query, **kwargs: [['secure_file_priv', 'V' + value]] if '@@GLOBAL.secure_file_priv' in query else [['name\nforged', 'Vsafe']]
|
|
result = MySQLAudit().getAudit()
|
|
self.assertTrue(any('\\n[*] forged' in item for item in result))
|
|
self.assertFalse(any(any(char in item for char in ('\n', '\r', '\x1b')) for item in result))
|
|
|
|
def test_listing_summary_reports_displayed_and_total(self):
|
|
from plugins.dbms.mssqlserver.audit import Audit as MSSQLAudit
|
|
from plugins.dbms.oracle.audit import Audit as OracleAudit
|
|
links = {'visible linked servers': '200', 'link one': 'destination'}
|
|
self.assertIn('Linked servers: showing 1 of 200; use --links for the full list', MSSQLAudit().explain({'linked servers': links}))
|
|
grants = {'visible directory grants': '200', 'ONE [READ]': '/one'}
|
|
self.assertEqual(OracleAudit().explain({'directory grants': grants}), ['Directory grants: showing 1 of 200'])
|
|
|
|
def test_only_observed_settings_are_listed_once(self):
|
|
from plugins.dbms.mysql.audit import Audit as MySQLAudit
|
|
def getValue(query, **kwargs):
|
|
if '@@GLOBAL.secure_file_priv' in query:
|
|
return [['secure_file_priv', 'V/allowed/']]
|
|
return [['local_infile', 'VOFF']]
|
|
inject.getValue = getValue
|
|
result = MySQLAudit().getAudit()
|
|
self.assertFalse(any('super_read_only' in item for item in result))
|
|
self.assertEqual(sum('/ secure_file_priv:' in item for item in result), 1)
|
|
self.assertFalse(any('not established' in item or 'does not establish' in item for item in result))
|
|
|
|
def test_snippets_are_single_read_only_statements(self):
|
|
for dbms in (DBMS.MYSQL, DBMS.PGSQL, DBMS.MSSQL, DBMS.ORACLE):
|
|
set_dbms(dbms)
|
|
audit = importlib.import_module('plugins.dbms.%s.audit' % DBMS_DIRECTORY_DICT[dbms]).Audit()
|
|
for scope, snippets in audit.queries:
|
|
for snippet in snippets:
|
|
sql = getSQLSnippet(dbms, snippet, LIMIT="20")
|
|
self.assertTrue(sql.startswith('SELECT '))
|
|
self.assertNotIn(';', sql.replace('; RPC out: ', ''))
|
|
self.assertIn('audit_value', sql)
|
|
|
|
def test_help_is_one_line(self):
|
|
output = subprocess.check_output([sys.executable, os.path.join(paths.SQLMAP_ROOT_PATH, 'sqlmap.py'), '-hh'])
|
|
lines = [line for line in output.decode('utf-8').splitlines() if '--audit' in line]
|
|
self.assertEqual(len(lines), 1)
|
|
self.assertLessEqual(len(lines[0]), 78)
|