GeezerVison is showing github.com — sqlmap/tests/test_audit.py at master · sqlmapproject/sqlmap

Expand file tree

test_audit.py

Copy path

More file actions

Latest commit

History

244 lines (217 loc) · 13 KB

test_audit.py

Copy path

File metadata and controls

244 lines (217 loc) · 13 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

#!/usr/bin/env python

"""

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)