Adding cross-DBMS security configuration audit (--audit)

This commit is contained in:
Miroslav Štampar
2026-10-19 18:34:39 +02:00
parent 6530184e2c
commit d0ad9b2e34
43 changed files with 1236 additions and 3 deletions
@@ -0,0 +1,5 @@
SELECT audit_key,audit_value FROM (
SELECT name AS audit_key,CASE WHEN value_in_use IS NULL THEN N'N' ELSE N'V'+CONVERT(NVARCHAR(MAX),value_in_use) END AS audit_value
FROM master.sys.configurations
WHERE name IN ('xp_cmdshell','clr enabled','Ole Automation Procedures','Ad Hoc Distributed Queries')
) audit_rows WHERE 1=1 ORDER BY audit_key
+11
View File
@@ -0,0 +1,11 @@
SELECT audit_key,audit_value FROM (
SELECT 'effective database user' AS audit_key,CASE WHEN CURRENT_USER IS NULL THEN N'N' ELSE N'V'+CONVERT(NVARCHAR(MAX),CURRENT_USER) END AS audit_value
UNION ALL
SELECT 'effective login' AS audit_key,CASE WHEN SUSER_SNAME() IS NULL THEN N'N' ELSE N'V'+CONVERT(NVARCHAR(MAX),SUSER_SNAME()) END AS audit_value
UNION ALL
SELECT 'original login' AS audit_key,CASE WHEN ORIGINAL_LOGIN() IS NULL THEN N'N' ELSE N'V'+CONVERT(NVARCHAR(MAX),ORIGINAL_LOGIN()) END AS audit_value
UNION ALL
SELECT 'database' AS audit_key,CASE WHEN DB_NAME() IS NULL THEN N'N' ELSE N'V'+CONVERT(NVARCHAR(MAX),DB_NAME()) END AS audit_value
UNION ALL
SELECT 'sysadmin' AS audit_key,CASE WHEN IS_SRVROLEMEMBER('sysadmin') IS NULL THEN N'N' ELSE N'V'+CONVERT(NVARCHAR(MAX),IS_SRVROLEMEMBER('sysadmin')) END AS audit_value
) audit_rows WHERE 1=1 ORDER BY audit_key
+6
View File
@@ -0,0 +1,6 @@
SELECT audit_key,audit_value FROM (
SELECT N'visible linked servers' AS audit_key,N'V'+CONVERT(NVARCHAR(MAX),COUNT(*)) AS audit_value FROM master.sys.servers WHERE is_linked=1
UNION ALL
SELECT N'link '+CAST(name AS NVARCHAR(MAX)) COLLATE DATABASE_DEFAULT AS audit_key,N'V'+ISNULL(data_source COLLATE DATABASE_DEFAULT,N'<unknown>')+N' (data access: '+RTRIM(CASE is_data_access_enabled WHEN 1 THEN N'enabled' WHEN 0 THEN N'disabled' ELSE N'unavailable' END)+N'; RPC out: '+RTRIM(CASE is_rpc_out_enabled WHEN 1 THEN N'enabled' WHEN 0 THEN N'disabled' ELSE N'unavailable' END)+N')' AS audit_value
FROM master.sys.servers v WHERE is_linked=1 AND (SELECT COUNT(*) FROM master.sys.servers s WHERE s.is_linked=1 AND s.server_id<=v.server_id)<=%LIMIT%
) audit_rows WHERE 1=1 ORDER BY audit_key
@@ -0,0 +1,7 @@
SELECT audit_key,audit_value FROM (
SELECT 'execute xp_cmdshell' AS audit_key,CASE WHEN HAS_PERMS_BY_NAME('master.dbo.xp_cmdshell','OBJECT','EXECUTE') IS NULL THEN N'N' ELSE N'V'+CONVERT(NVARCHAR(MAX),HAS_PERMS_BY_NAME('master.dbo.xp_cmdshell','OBJECT','EXECUTE')) END AS audit_value
UNION ALL
SELECT 'read configuration metadata' AS audit_key,CASE WHEN HAS_PERMS_BY_NAME('master.sys.configurations','OBJECT','SELECT') IS NULL THEN N'N' ELSE N'V'+CONVERT(NVARCHAR(MAX),HAS_PERMS_BY_NAME('master.sys.configurations','OBJECT','SELECT')) END AS audit_value
UNION ALL
SELECT 'read linked-server metadata' AS audit_key,CASE WHEN HAS_PERMS_BY_NAME('master.sys.servers','OBJECT','SELECT') IS NULL THEN N'N' ELSE N'V'+CONVERT(NVARCHAR(MAX),HAS_PERMS_BY_NAME('master.sys.servers','OBJECT','SELECT')) END AS audit_value
) audit_rows WHERE 1=1 ORDER BY audit_key
+5
View File
@@ -0,0 +1,5 @@
SELECT audit_key,audit_value FROM (
SELECT LOWER(VARIABLE_NAME) AS audit_key,CASE WHEN VARIABLE_VALUE IS NULL THEN 'N' ELSE CONCAT('V',VARIABLE_VALUE) END AS audit_value
FROM performance_schema.global_variables
WHERE LOWER(VARIABLE_NAME) IN ('local_infile','require_secure_transport','read_only','super_read_only','general_log','log_bin','skip_networking','skip_grant_tables')
) audit_rows ORDER BY audit_key
@@ -0,0 +1,5 @@
SELECT audit_key,audit_value FROM (
SELECT LOWER(VARIABLE_NAME) AS audit_key,CASE WHEN VARIABLE_VALUE IS NULL THEN 'N' ELSE CONCAT('V',VARIABLE_VALUE) END AS audit_value
FROM information_schema.GLOBAL_VARIABLES
WHERE LOWER(VARIABLE_NAME) IN ('local_infile','require_secure_transport','read_only','super_read_only','general_log','log_bin','skip_networking','skip_grant_tables')
) audit_rows ORDER BY audit_key
+5
View File
@@ -0,0 +1,5 @@
SELECT audit_key,audit_value FROM (
SELECT 'secure_file_priv' AS audit_key,CASE WHEN @@GLOBAL.secure_file_priv IS NULL THEN 'N' ELSE CONCAT('V',CAST(@@GLOBAL.secure_file_priv AS CHAR)) END AS audit_value
UNION ALL
SELECT 'max_allowed_packet' AS audit_key,CASE WHEN @@GLOBAL.max_allowed_packet IS NULL THEN 'N' ELSE CONCAT('V',CAST(@@GLOBAL.max_allowed_packet AS CHAR)) END AS audit_value
) audit_rows ORDER BY audit_key
+7
View File
@@ -0,0 +1,7 @@
SELECT audit_key,audit_value FROM (
SELECT 'effective account' AS audit_key,CASE WHEN CURRENT_USER() IS NULL THEN 'N' ELSE CONCAT('V',CAST(CURRENT_USER() AS CHAR)) END AS audit_value
UNION ALL
SELECT 'login identity' AS audit_key,CASE WHEN USER() IS NULL THEN 'N' ELSE CONCAT('V',CAST(USER() AS CHAR)) END AS audit_value
UNION ALL
SELECT 'server version' AS audit_key,CASE WHEN VERSION() IS NULL THEN 'N' ELSE CONCAT('V',CAST(VERSION() AS CHAR)) END AS audit_value
) audit_rows ORDER BY audit_key
+5
View File
@@ -0,0 +1,5 @@
SELECT audit_key,audit_value FROM (
SELECT 'direct FILE grant' AS audit_key,CONCAT('V',CAST(COUNT(*) AS CHAR)) AS audit_value
FROM information_schema.USER_PRIVILEGES
WHERE PRIVILEGE_TYPE='FILE' AND GRANTEE=CONCAT(CHAR(39),LEFT(CURRENT_USER(),LENGTH(CURRENT_USER())-LENGTH(SUBSTRING_INDEX(CURRENT_USER(),'@',-1))-1),CHAR(39),'@',CHAR(39),SUBSTRING_INDEX(CURRENT_USER(),'@',-1),CHAR(39))
) audit_rows ORDER BY audit_key
@@ -0,0 +1,5 @@
SELECT audit_key,audit_value FROM (
SELECT name AS audit_key,CASE WHEN value IS NULL THEN 'N' ELSE 'V'||value END AS audit_value
FROM v$parameter
WHERE name IN ('remote_os_authent','remote_login_passwordfile','sql92_security','audit_trail','os_authent_prefix')
) audit_rows ORDER BY audit_key
+12
View File
@@ -0,0 +1,12 @@
SELECT audit_key,audit_value FROM (
SELECT 'visible directory grants' AS audit_key,'V'||TO_CHAR(COUNT(*)) AS audit_value
FROM all_directories d JOIN all_tab_privs p ON p.table_schema=d.owner AND p.table_name=d.directory_name
WHERE p.type='DIRECTORY' AND p.privilege IN ('READ','WRITE') AND (p.grantee=SYS_CONTEXT('USERENV','CURRENT_USER') OR p.grantee='PUBLIC' OR p.grantee IN (SELECT role FROM session_roles))
UNION ALL
SELECT audit_key,audit_value FROM (
SELECT d.directory_name||' ['||p.privilege||'] (via '||p.grantee||')' AS audit_key,'V'||d.directory_path AS audit_value
FROM all_directories d JOIN all_tab_privs p ON p.table_schema=d.owner AND p.table_name=d.directory_name
WHERE p.type='DIRECTORY' AND p.privilege IN ('READ','WRITE') AND (p.grantee=SYS_CONTEXT('USERENV','CURRENT_USER') OR p.grantee='PUBLIC' OR p.grantee IN (SELECT role FROM session_roles))
ORDER BY d.directory_name,p.privilege,p.grantee
) audit_details WHERE ROWNUM<=%LIMIT%
) audit_rows ORDER BY audit_key
+9
View File
@@ -0,0 +1,9 @@
SELECT audit_key,audit_value FROM (
SELECT 'effective account' AS audit_key,CASE WHEN SYS_CONTEXT('USERENV','CURRENT_USER') IS NULL THEN 'N' ELSE 'V'||TO_CHAR(SYS_CONTEXT('USERENV','CURRENT_USER')) END AS audit_value FROM DUAL
UNION ALL
SELECT 'login identity' AS audit_key,CASE WHEN SYS_CONTEXT('USERENV','SESSION_USER') IS NULL THEN 'N' ELSE 'V'||TO_CHAR(SYS_CONTEXT('USERENV','SESSION_USER')) END AS audit_value FROM DUAL
UNION ALL
SELECT 'database' AS audit_key,CASE WHEN SYS_CONTEXT('USERENV','DB_NAME') IS NULL THEN 'N' ELSE 'V'||TO_CHAR(SYS_CONTEXT('USERENV','DB_NAME')) END AS audit_value FROM DUAL
UNION ALL
SELECT 'SYSDBA session' AS audit_key,CASE WHEN SYS_CONTEXT('USERENV','ISDBA') IS NULL THEN 'N' ELSE 'V'||TO_CHAR(SYS_CONTEXT('USERENV','ISDBA')) END AS audit_value FROM DUAL
) audit_rows ORDER BY audit_key
+5
View File
@@ -0,0 +1,5 @@
SELECT audit_key,audit_value FROM (
SELECT 'DBA role active' AS audit_key,'V'||CASE WHEN COUNT(*)>0 THEN 'yes' ELSE 'no' END AS audit_value FROM session_roles WHERE role='DBA'
UNION ALL
SELECT privilege,'Vgranted' FROM session_privs WHERE privilege IN ('CREATE ANY DIRECTORY','CREATE EXTERNAL JOB','EXECUTE ANY PROCEDURE','ALTER SYSTEM','SELECT ANY DICTIONARY')
) audit_rows ORDER BY audit_key
@@ -0,0 +1,5 @@
SELECT audit_key,audit_value FROM (
SELECT name AS audit_key,CASE WHEN setting IS NULL THEN 'N' ELSE 'V'||setting END AS audit_value
FROM pg_catalog.pg_settings
WHERE name IN ('ssl','password_encryption','row_security','logging_collector','log_statement','log_connections','log_disconnections')
) audit_rows ORDER BY audit_key
+7
View File
@@ -0,0 +1,7 @@
SELECT audit_key,audit_value FROM (
SELECT 'effective account' AS audit_key,CASE WHEN CURRENT_USER IS NULL THEN 'N' ELSE 'V'||CAST(CURRENT_USER AS TEXT) END AS audit_value
UNION ALL
SELECT 'login identity' AS audit_key,CASE WHEN SESSION_USER IS NULL THEN 'N' ELSE 'V'||CAST(SESSION_USER AS TEXT) END AS audit_value
UNION ALL
SELECT 'database' AS audit_key,CASE WHEN CURRENT_DATABASE() IS NULL THEN 'N' ELSE 'V'||CAST(CURRENT_DATABASE() AS TEXT) END AS audit_value
) audit_rows ORDER BY audit_key
@@ -0,0 +1,7 @@
SELECT audit_key,audit_value FROM (
SELECT 'superuser' AS audit_key,'V'||CASE WHEN rolsuper THEN '1' ELSE '0' END AS audit_value FROM pg_catalog.pg_roles WHERE rolname=CURRENT_USER
UNION ALL
SELECT 'bypass RLS','V'||CASE WHEN rolbypassrls THEN '1' ELSE '0' END FROM pg_catalog.pg_roles WHERE rolname=CURRENT_USER
UNION ALL
SELECT rolname,'V'||CASE WHEN pg_has_role(oid,'USAGE') THEN '1' ELSE '0' END FROM pg_catalog.pg_roles WHERE rolname IN ('pg_read_server_files','pg_write_server_files','pg_execute_server_program')
) audit_rows ORDER BY audit_key
+6
View File
@@ -140,6 +140,12 @@ def action():
except SqlmapNoneDataException as ex:
logger.critical(ex)
if conf.getAudit:
try:
conf.dumper.audit(conf.dbmsHandler.getAudit())
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)
+3
View File
@@ -227,6 +227,9 @@ class Dump(object):
def links(self, links):
self.lister("database links", links, content_type=CONTENT_TYPE.LINKS)
def audit(self, audit):
self.lister("database security audit", audit, content_type=CONTENT_TYPE.AUDIT)
def userSettings(self, header, userSettings, subHeader, content_type=None):
self._areAdmins = set()
+1
View File
@@ -430,6 +430,7 @@ class CONTENT_TYPE(object):
STATEMENTS = 26
PROCEDURES = 27
LINKS = 28
AUDIT = 29
class CONTENT_STATUS(object):
IN_PROGRESS = 0
+1
View File
@@ -164,6 +164,7 @@ optDict = {
"getPrivileges": ("boolean", "Privileges"),
"getRoles": ("boolean", "Roles"),
"getLinks": "boolean",
"getAudit": "boolean",
"getDbs": ("boolean", "Databases"),
"getTables": ("boolean", "Tables"),
"getColumns": ("boolean", "Columns"),
+4 -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.11"
VERSION = "1.10.10.12"
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)
@@ -612,6 +612,9 @@ MYSQL_UNION_VALUE_CAST = "CONVERT(%s USING utf8mb4) COLLATE utf8mb4_bin"
# Row count at/above which keyset (seek) pagination is used automatically for table dumps when a usable integer-key cursor exists (smaller tables keep the plain LIMIT/OFFSET path; '--keyset' forces it regardless of size)
KEYSET_MIN_ROWS = 1000
# Maximum detailed metadata rows per variable-size security audit listing
AUDIT_MAX_ROWS = 20
# Number of Huffman (set-membership) character attempts made before their escape ratio is judged; at/above it, escapes reaching half of all attempts latch the technique off (safety against trimmed/blocked long IN() payloads)
HUFFMAN_PROBE_LIMIT = 8
+1
View File
@@ -114,6 +114,7 @@ def vulnTest(tests=None, label="vuln"):
("-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")),
("-d \"<direct>\" --audit", ("back-end DBMS: SQLite", "security configuration audit 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")),
)
+3
View File
@@ -484,6 +484,9 @@ def cmdLineParser(argv=None):
enumeration.add_argument("--hostname", dest="getHostname", action="store_true",
help="Retrieve DBMS server hostname")
enumeration.add_argument("--audit", dest="getAudit", action="store_true",
help="Audit DBMS security configuration and capabilities")
enumeration.add_argument("--is-dba", dest="isDba", action="store_true",
help="Detect if the DBMS current user is DBA")
+5 -2
View File
@@ -37,6 +37,7 @@ from lib.core.common import singleTimeDebugMessage
from lib.core.common import singleTimeWarnMessage
from lib.core.common import unArrayizeValue
from lib.core.common import wasLastResponseDBMSError
from lib.core.common import zeroDepthSearch
from lib.core.compat import xrange
from lib.core.convert import decodeBase64
from lib.core.convert import getUnicode
@@ -344,8 +345,10 @@ def unionUse(expression, unpack=True, dump=False):
kb.partRun = getPartRun(alias=False) if (conf.api or conf.reportJson) else None
if expressionFieldsList and len(expressionFieldsList) > 1 and "ORDER BY" in expression.upper():
# Removed ORDER BY clause because UNION does not play well with it
expression = re.sub(r"(?i)\s*ORDER BY\s+[\w,]+", "", expression)
# Only remove outer ordering. Nested ORDER BY clauses can select which
# rows survive a ROWNUM/TOP limit and must remain intact.
positions = set(index for keyword in set(re.findall(r"(?i)ORDER BY", expression)) for index in zeroDepthSearch(expression, keyword))
expression = re.sub(r"(?i)ORDER BY\s+[\w,]+", lambda match: "" if match.start() in positions else match.group(0), expression)
debugMsg = "stripping ORDER BY clause from statement because "
debugMsg += "it does not play well with UNION query SQL injection"
singleTimeDebugMessage(debugMsg)
+53
View File
@@ -0,0 +1,53 @@
#!/usr/bin/env python
"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission
"""
from lib.core.enums import DBMS
from lib.core.settings import AUDIT_MAX_ROWS
from plugins.generic.audit import Audit as GenericAudit
class Audit(GenericAudit):
dbms = DBMS.MSSQL
# Snippets keep an outer WHERE for the existing blind TOP/NOT IN limiter.
queries = (("identity", ("audit_identity",)), ("permissions", ("audit_permissions",)),
("configuration", ("audit_configuration",)), ("linked servers", ("audit_links",)))
flags = dict((key, ("enabled", "disabled")) for key in ("xp_cmdshell", "clr enabled", "Ole Automation Procedures", "Ad Hoc Distributed Queries"))
flags.update(dict((key, ("yes", "no")) for key in ("sysadmin", "execute xp_cmdshell", "read configuration metadata", "read linked-server metadata")))
def failure(self, scope, groups):
permissions = groups.get("permissions") or {}
for key, name, target in (("read configuration metadata", "configuration", "master.sys.configurations"),
("read linked-server metadata", "linked servers", "master.sys.servers")):
if scope == name and permissions.get(key) == "0":
return "%s: SELECT permission missing on %s" % (scope, target)
return GenericAudit.failure(self, scope, groups)
def explain(self, groups):
configuration = groups.get("configuration") or {}
permissions = groups.get("permissions") or {}
identity = groups.get("identity") or {}
result = []
if identity.get("sysadmin") == "1" and configuration.get("xp_cmdshell") == "0":
if self._auditCanExecute():
result.append("Command execution: sysadmin can enable xp_cmdshell; use --os-cmd or --os-shell")
else:
result.append("Command execution: sysadmin can enable xp_cmdshell; no statement execution route identified")
elif configuration.get("xp_cmdshell") == "1" and permissions.get("execute xp_cmdshell") == "1":
if self._auditCanExecute():
result.append("Command execution: xp_cmdshell enabled and EXECUTE granted; use --os-cmd or --os-shell")
else:
result.append("Command execution: xp_cmdshell enabled and EXECUTE granted; no statement execution route identified")
elif permissions.get("execute xp_cmdshell") == "0":
result.append("Command execution: EXECUTE permission missing on master.dbo.xp_cmdshell")
elif configuration.get("xp_cmdshell") == "0":
result.append("Command execution: xp_cmdshell is disabled")
links = groups.get("linked servers") or {}
count = links.get("visible linked servers")
if count and count.isdigit() and int(count) > AUDIT_MAX_ROWS:
shown = sum(key.startswith("link ") for key in links)
result.append("Linked servers: showing %d of %s; use --links for the full list" % (shown, count))
return result
+44
View File
@@ -0,0 +1,44 @@
#!/usr/bin/env python
"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission
"""
from lib.core.common import Backend
from lib.core.enums import DBMS, FORK
from plugins.generic.audit import Audit as GenericAudit
class Audit(GenericAudit):
dbms = DBMS.MYSQL
flags = dict((key, ("enabled", "disabled")) for key in ("local_infile", "require_secure_transport", "read_only", "super_read_only", "general_log", "log_bin", "skip_networking", "skip_grant_tables"))
flags["direct FILE grant"] = ("present", "absent")
def __init__(self):
configuration = ("audit_configuration_mariadb", "audit_configuration") if Backend.isFork(FORK.MARIADB) else ("audit_configuration", "audit_configuration_mariadb")
self.queries = (("identity", ("audit_identity",)), ("configuration", configuration),
("file access", ("audit_file_access",)), ("privileges", ("audit_privileges",)))
def explain(self, groups):
facts = groups.get("file access") or {}
privileges = groups.get("privileges") or {}
result = []
disabled = False
if "secure_file_priv" in facts:
value = facts["secure_file_priv"]
# MariaDB can expose an unrestricted value as SQL NULL; MySQL can expose
# its disabled value as the literal string "NULL". Keep the raw fact.
disabled = (value is None or value.upper() == "NULL") and not Backend.isFork(FORK.MARIADB)
if disabled:
result.append("File operations: secure_file_priv=NULL blocks --file-read and --file-write")
elif not value:
result.append("File operations: no directory restriction from secure_file_priv")
else:
result.append("File operations: --file-read and --file-dest paths must be inside '%s'" % value)
if privileges.get("direct FILE grant") == "1" and not disabled:
message = "File read: FILE granted; use --file-read with a server-side path"
if facts.get("max_allowed_packet"):
message += " (file size < %s bytes)" % facts["max_allowed_packet"]
result.append(message)
return result
+23
View File
@@ -0,0 +1,23 @@
#!/usr/bin/env python
"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission
"""
from lib.core.enums import DBMS
from lib.core.settings import AUDIT_MAX_ROWS
from plugins.generic.audit import Audit as GenericAudit
class Audit(GenericAudit):
dbms = DBMS.ORACLE
queries = (("identity", ("audit_identity",)), ("configuration", ("audit_configuration",)),
("privileges", ("audit_privileges",)), ("directory grants", ("audit_directories",)))
def explain(self, groups):
grants = groups.get("directory grants") or {}
count = grants.get("visible directory grants")
if count and count.isdigit() and int(count) > AUDIT_MAX_ROWS:
return ["Directory grants: showing %d of %s" % (len(grants) - 1, count)]
return []
+33
View File
@@ -0,0 +1,33 @@
#!/usr/bin/env python
"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission
"""
from lib.core.enums import DBMS
from plugins.generic.audit import Audit as GenericAudit
class Audit(GenericAudit):
dbms = DBMS.PGSQL
queries = (("identity", ("audit_identity",)), ("configuration", ("audit_configuration",)),
("privileges", ("audit_privileges",)))
flags = dict((key, ("yes", "no")) for key in ("superuser", "bypass RLS", "pg_read_server_files", "pg_write_server_files", "pg_execute_server_program"))
def explain(self, groups):
privileges = groups.get("privileges") or {}
superuser = privileges.get("superuser") == "1"
result = []
if superuser or privileges.get("pg_read_server_files") == "1":
result.append("File read: server-file privilege granted; use --file-read with a server-side path")
if superuser and self._auditCanExecute():
result.append("File write: superuser and statement execution available; use --file-write and --file-dest")
if superuser or privileges.get("pg_execute_server_program") == "1":
if self._auditCanExecute():
result.append("Command execution: COPY FROM PROGRAM privilege granted; use --os-cmd or --os-shell")
else:
result.append("Command execution: COPY FROM PROGRAM privilege granted; no statement execution route identified")
if superuser or privileges.get("bypass RLS") == "1":
result.append("Row security: current role bypasses row-level security")
return result
+113
View File
@@ -0,0 +1,113 @@
#!/usr/bin/env python
"""
Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
See the file 'LICENSE' for copying permission
"""
from lib.core.common import arrayizeValue
from lib.core.common import getSQLSnippet
from lib.core.common import isListLike
from lib.core.common import isNoneValue
from lib.core.common import isStackingAvailable
from lib.core.common import unArrayizeValue
from lib.core.convert import getUnicode
from lib.core.data import conf
from lib.core.data import kb
from lib.core.data import logger
from lib.core.enums import CHARSET_TYPE
from lib.core.enums import EXPECTED
from lib.core.exception import SqlmapNoneDataException
from lib.core.settings import AUDIT_MAX_ROWS
from lib.utils.safe2bin import safecharencode
from lib.request import inject
from thirdparty import six
class Audit(object):
# Each snippet returns (audit_key, audit_value). Values are tagged N (SQL NULL) or V<text>
# so extraction's normal NULL/empty-string handling cannot silently change their meaning.
dbms = None
queries = ()
flags = {}
def _auditFetch(self, snippet):
# Extraction locates fields and FROM clauses on one line.
query = getSQLSnippet(self.dbms, snippet, LIMIT=str(AUDIT_MAX_ROWS)).replace("\r", " ").replace("\n", " ")
try:
values = inject.getValue(query, fromUser=True, batch=True, safeCharEncode=False)
except SqlmapNoneDataException:
values = None
if isNoneValue(values):
# Extraction can return None for both zero rows and a failed query.
# Confirm emptiness rather than inferring a database error from it.
countQuery = "SELECT COUNT(*) FROM (%s) audit_count" % query.rsplit(" ORDER BY ", 1)[0]
try:
count = inject.getValue(countQuery, expected=EXPECTED.INT, charsetType=CHARSET_TYPE.DIGITS)
except SqlmapNoneDataException:
return None
return {} if getUnicode(unArrayizeValue(count)) == "0" else None
if isListLike(values) and len(values) == 2 and not any(isListLike(_) for _ in values):
values = [values]
facts = {}
for row in arrayizeValue(values):
if not isListLike(row) or len(row) != 2:
return None
key, value = row
if isNoneValue(key) or not isinstance(value, six.string_types):
return None
if value == "N":
value = None
elif value.startswith("V"):
value = value[1:]
else:
return None
# DBMS CHAR expressions can pad the keys; preserve value whitespace.
facts[getUnicode(key).rstrip(" ")] = value
return facts
def failure(self, scope, groups):
return "%s: retrieval failed" % scope
def _auditCanExecute(self):
# Use known execution routes; audit must not probe takeover gadgets.
return bool(conf.direct or isStackingAvailable() or kb.get("gadget"))
def explain(self, groups):
return []
def getAudit(self):
if kb.data.cachedAudit:
return kb.data.cachedAudit
logger.info("fetching security configuration and session capabilities")
groups = {}
result = []
for scope, snippets in self.queries:
facts = None
for snippet in snippets:
facts = self._auditFetch(snippet)
if facts is not None:
break
groups[scope] = facts
if facts is None:
result.append(self.failure(scope, groups))
continue
if not facts:
result.append("%s: no entries" % scope)
for key in sorted(facts):
if facts[key] is None:
value = "<NULL>"
elif facts[key] == "":
value = "<empty>"
else:
value = facts[key]
if key in self.flags:
if value.lower() in ("1", "on", "true", "yes"):
value = self.flags[key][0]
elif value.lower() in ("0", "off", "false", "no"):
value = self.flags[key][1]
result.append("%s / %s: %s" % (scope, key, value))
result.extend(self.explain(groups))
result = [safecharencode(_) for _ in result]
kb.data.cachedAudit = result
return result
+13
View File
@@ -73,6 +73,7 @@ class Databases(object):
kb.data.cachedStatements = []
kb.data.cachedProcedures = []
kb.data.cachedLinks = []
kb.data.cachedAudit = []
def getCurrentDb(self):
infoMsg = "fetching current database"
@@ -1217,6 +1218,18 @@ class Databases(object):
return kb.data.cachedLinks
def getAudit(self):
from importlib import import_module
from lib.core.settings import DBMS_DIRECTORY_DICT
dbms = Backend.getIdentifiedDbms()
if dbms not in (DBMS.MYSQL, DBMS.PGSQL, DBMS.MSSQL, DBMS.ORACLE) or Backend.isFork(FORK.DUCKDB):
logger.warning("security configuration audit is not supported on %s" % dbms)
return []
module = import_module("plugins.dbms.%s.audit" % DBMS_DIRECTORY_DICT[dbms])
return module.Audit().getAudit()
def getProcedures(self):
infoMsg = "fetching stored procedures"
logger.info(infoMsg)
+4
View File
@@ -548,6 +548,10 @@ getStatements = False
# Valid: True or False
getLinks = False
# Audit database security configuration and current-session capabilities.
# Valid: True or False
getAudit = False
# Back-end database management system database to enumerate.
db =
+5
View File
@@ -0,0 +1,5 @@
FROM python:3.12-slim
RUN pip install --no-cache-dir flask==3.1.3 pymysql==1.2.3 'psycopg[binary]==3.3.6' python-tds==1.17.1 oracledb==26.0.1
WORKDIR /lab
COPY server.py db.py seed.py probe.py /lab/
CMD ["python", "/lab/server.py"]
+42
View File
@@ -0,0 +1,42 @@
# Disposable security audit matrix
Run from the repository root on a local Docker host:
```sh
python3 tests/integration/audit/run.py
```
Requires Python 3, Docker Compose, Internet access for images and roughly 12 GB
available memory. The SQL Server image requires an x86-64 host. Oracle image
support also depends on host architecture and page size. Application drivers are
installed only inside the fixture image; sqlmap gains no dependencies.
The runner creates an isolated `sqlmap-audit-<pid>` Compose project. Database ports
are internal; deliberately injectable HTTP endpoints bind to random localhost
ports. Credentials are synthetic and fixed for these disposable databases. The
runner does not access the remote sqlmap testbed or existing database services.
It removes its own containers, networks and volumes on completion, including a
failed assertion. `--keep` retains them for investigation; remove the retained
project with `docker compose -p <project> -f tests/integration/audit/compose.yml down --volumes`.
The matrix covers MySQL 8, MariaDB 10.11, PostgreSQL 16, SQL Server 2022 and
Oracle Free 23, with administrator, application and restricted accounts. Two
additional MySQL instances exercise unrestricted and disabled file operations.
It compares native metadata results with real UNION SQLi for every account and
boolean blind SQLi for each engine's application account, using five threads.
Variable listings retain total counts and show at most 20 details. Additional
scenarios cover oversized listings, escaped control characters, confirmed empty
queries, RLS enforcement and command advice with and without stacked execution.
It also checks that audits leave configuration unchanged, that revoking grants
and changing settings changes the result, and that database root with FILE
privilege still cannot read a root-only fixture file. Filesystem probes and
configuration changes belong to the fixture provisioner; `--audit` performs
metadata reads only.
Use `--native-only` for database queries and permission/configuration/filesystem
checks without the HTTP scans. Logs and native reference results go to a printed
temporary directory. For a previously seeded retained project:
```sh
python3 tests/integration/audit/run.py --project sqlmap-audit-<id> --reuse --keep
```
+82
View File
@@ -0,0 +1,82 @@
services:
mysql:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: AuditOnly!2026
MYSQL_ROOT_HOST: '%'
MYSQL_DATABASE: auditlab
command: ['--secure-file-priv=/var/lib/mysql-files', '--local-infile=OFF']
mem_limit: 1g
mariadb:
image: mariadb:10.11
environment:
MARIADB_ROOT_PASSWORD: AuditOnly!2026
MARIADB_ROOT_HOST: '%'
MARIADB_DATABASE: auditlab
command: ['--secure-file-priv=', '--local-infile=ON']
mem_limit: 1g
postgres:
image: postgres:16-alpine
environment:
POSTGRES_PASSWORD: AuditOnly!2026
POSTGRES_DB: auditlab
mem_limit: 1g
mssql:
image: mcr.microsoft.com/mssql/server:2022-latest
environment:
ACCEPT_EULA: Y
MSSQL_PID: Developer
MSSQL_SA_PASSWORD: AuditOnly!2026
mem_limit: 2g
oracle:
image: gvenzl/oracle-free:23.26.0-slim-faststart
environment:
ORACLE_PASSWORD: AuditOnly!2026
APP_USER: audit_owner
APP_USER_PASSWORD: AuditOnly!2026
mem_limit: 4g
web-mysql:
build: .
image: sqlmap-audit-fixture:local
environment: {ENGINE: mysql}
ports: ['127.0.0.1::8080']
web-mariadb:
image: sqlmap-audit-fixture:local
environment: {ENGINE: mariadb}
ports: ['127.0.0.1::8080']
web-postgres:
image: sqlmap-audit-fixture:local
environment: {ENGINE: postgres}
ports: ['127.0.0.1::8080']
web-mssql:
image: sqlmap-audit-fixture:local
environment: {ENGINE: mssql}
ports: ['127.0.0.1::8080']
web-oracle:
image: sqlmap-audit-fixture:local
environment: {ENGINE: oracle}
ports: ['127.0.0.1::8080']
mysql-open:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: AuditOnly!2026
MYSQL_ROOT_HOST: '%'
MYSQL_DATABASE: auditlab
command: ['--secure-file-priv=', '--local-infile=ON']
mem_limit: 1g
mysql-disabled:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: AuditOnly!2026
MYSQL_ROOT_HOST: '%'
MYSQL_DATABASE: auditlab
command: ['--secure-file-priv=NULL', '--local-infile=OFF']
mem_limit: 1g
web-mysql-open:
image: sqlmap-audit-fixture:local
environment: {ENGINE: mysql, DB_HOST: mysql-open}
ports: ['127.0.0.1::8080']
web-mysql-disabled:
image: sqlmap-audit-fixture:local
environment: {ENGINE: mysql, DB_HOST: mysql-disabled}
ports: ['127.0.0.1::8080']
+27
View File
@@ -0,0 +1,27 @@
import os
import threading
engine = os.environ['ENGINE']
password = 'AuditOnly!2026'
host = os.environ.get('DB_HOST', engine)
pools = {}
poolLock = threading.Lock()
def connect(account, pooled=False):
if engine in ('mysql', 'mariadb'):
import pymysql
return pymysql.connect(host=host, user=account, password=password, database='auditlab', autocommit=True)
if engine == 'postgres':
import psycopg
return psycopg.connect(host=host, user=account, password=password, dbname='auditlab', autocommit=True)
if engine == 'mssql':
import pytds
return pytds.connect(host, user=account, password=password, database='master', autocommit=True)
import oracledb
if pooled:
with poolLock:
if account not in pools:
pools[account] = oracledb.create_pool(user=account, password=password, dsn=host+'/FREEPDB1', min=1, max=10, increment=1)
return pools[account].acquire()
return oracledb.connect(user=account, password=password, dsn=host+'/FREEPDB1', mode=oracledb.AUTH_MODE_SYSDBA if account == 'sys' else oracledb.AUTH_MODE_DEFAULT)
+12
View File
@@ -0,0 +1,12 @@
#!/usr/bin/env python
"""Execute fixture SQL over a real DB-API connection for ground-truth checks."""
import json
import sys
from db import connect
conn = connect(sys.argv[1])
try:
cursor = conn.cursor()
cursor.execute(sys.stdin.read())
print(json.dumps(cursor.fetchall() if cursor.description else [], default=str))
finally:
conn.close()
+266
View File
@@ -0,0 +1,266 @@
#!/usr/bin/env python3
"""Opt-in Docker audit matrix. Runtime sqlmap itself acquires metadata only.
This fixture provisioner deliberately changes settings, grants and synthetic files inside
its own disposable containers, then verifies the audit over native SQL and real SQLi.
"""
import argparse
import concurrent.futures
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile
ROOT = Path(__file__).resolve().parents[3]
LAB = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / 'tests'))
from _testutils import bootstrap, set_dbms
bootstrap()
from lib.core.enums import DBMS, FORK, EXPECTED
from lib.core.common import Backend, getSQLSnippet, unArrayizeValue
from lib.core.data import conf
from lib.core.settings import AUDIT_MAX_ROWS
from lib.request import inject
from plugins.generic.databases import Databases
ENGINES = {'mysql': (DBMS.MYSQL, 'root'), 'mariadb': (DBMS.MYSQL, 'root'),
'postgres': (DBMS.PGSQL, 'postgres'), 'mssql': (DBMS.MSSQL, 'sa'),
'oracle': (DBMS.ORACLE, 'audit_owner'),
'mysql-open': (DBMS.MYSQL, 'root'), 'mysql-disabled': (DBMS.MYSQL, 'root')}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--project', default='sqlmap-audit-%d' % os.getpid())
parser.add_argument('--reuse', action='store_true', help='Use an already seeded audit project')
parser.add_argument('--keep', action='store_true', help='Keep this disposable project for debugging')
parser.add_argument('--native-only', action='store_true', help='Skip the HTTP SQLi matrix')
args = parser.parse_args()
if not re.match(r'^sqlmap-audit-[a-z0-9-]+$', args.project):
parser.error('project must have the sqlmap-audit- prefix')
compose = ['docker', 'compose', '-p', args.project, '-f', str(LAB / 'compose.yml')]
logs = Path(tempfile.mkdtemp(prefix='sqlmap-audit-matrix-'))
print('Logs: %s' % logs, flush=True)
def command(argv, data=None, check=True, timeout=240):
result = subprocess.run(argv, input=data, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, timeout=timeout)
if check and result.returncode:
raise AssertionError('%s\n%s\n%s' % (' '.join(argv), result.stdout, result.stderr))
return result
def sql(engine, account, query, check=True):
result = command(compose + ['exec', '-T', 'web-' + engine, 'python', '/lab/probe.py', account], query, check)
return json.loads(result.stdout) if result.returncode == 0 else None
def native(engine, account, statement=False):
dbms = ENGINES[engine][0]
set_dbms(dbms)
# MariaDB is a MySQL plugin fork; native probes don't run the normal fingerprint.
original_fork = Backend.isFork
Backend.isFork = staticmethod(lambda fork: engine == 'mariadb' and fork == FORK.MARIADB)
original_direct = conf.direct
conf.direct = statement
original = inject.getValue
calls = []
def getValue(query, **kwargs):
assert query.startswith('SELECT ')
calls.append(query)
rows = sql(engine, account, query, check=False)
return unArrayizeValue(rows) if kwargs.get("expected") == EXPECTED.INT else rows
inject.getValue = getValue
try:
result = Databases().getAudit()
assert calls
assert not any(line.startswith('OS boundary:') for line in result)
return sorted(set(result))
finally:
conf.direct = original_direct
inject.getValue = original
Backend.isFork = staticmethod(original_fork)
def scan(engine, account, technique, expected):
address = command(compose + ['port', 'web-' + engine, '8080']).stdout.strip()
url = 'http://%s/%s/get_int?id=1' % (address, account)
label = '%s-%s-%s' % (engine, account, technique)
cmd = [sys.executable, str(ROOT / 'sqlmap.py'), '-u', url,
'--dbms=' + ENGINES[engine][0], '--batch', '--technique=' + technique,
'--threads=5', '--audit', '--skip-waf', '--disable-precon', '--flush-session',
'--output-dir=' + str(logs / label)]
result = command(cmd, check=False, timeout=900)
(logs / (label + '.log')).write_text(result.stdout + result.stderr)
assert result.returncode == 0, label
audit_output = re.search(r'^database security audit[^\n]*\n((?:\[\*\] [^\n]*\n)+)', result.stdout, re.M)
assert audit_output, label
actual = sorted(set(re.findall(r'^\[\*\] (.*)$', audit_output.group(1), re.M)))
assert actual == expected, '%s: missing=%r extra=%r' % (label, sorted(set(expected)-set(actual)), sorted(set(actual)-set(expected)))
print('SQLi %s PASS' % label, flush=True)
try:
if not args.reuse:
command(compose + ['build', 'web-mysql'])
command(compose + ['up', '-d'])
with concurrent.futures.ThreadPoolExecutor(max_workers=7) as pool:
futures = [pool.submit(command, compose + ['exec', '-T', 'web-' + engine, 'python', '/lab/seed.py']) for engine in ENGINES]
for future in futures:
future.result()
expected = {}
for engine, (dbms, admin) in ENGINES.items():
for alias, account in (('admin', admin), ('app', 'audit_app'), ('limited', 'audit_limited')):
snapshot_query = getSQLSnippet(dbms, 'audit_configuration_mariadb' if engine == 'mariadb' else 'audit_configuration', LIMIT=str(AUDIT_MAX_ROWS))
before = sql(engine, admin, snapshot_query) if alias == 'admin' else None
result = native(engine, account)
if alias == 'admin':
assert sql(engine, admin, snapshot_query) == before, 'audit changed configuration'
expected[engine, alias] = result
(logs / ('%s-%s-native.json' % (engine, alias))).write_text(json.dumps(result, indent=2))
assert any('/ effective account:' in line or '/ effective login:' in line for line in result)
if alias == 'admin':
assert not any(re.match(r'^(identity|configuration|privileges|file access): retrieval failed', line) for line in result), (engine,result)
if engine == 'postgres' and alias == 'app':
assert 'privileges / pg_read_server_files: yes' in result
assert 'privileges / pg_execute_server_program: no' in result
if engine == 'mssql' and alias == 'limited':
assert 'configuration: SELECT permission missing on master.sys.configurations' in result
assert 'linked servers: SELECT permission missing on master.sys.servers' in result
if engine == 'oracle' and alias == 'app':
assert any('AUDIT_FILES [READ]' in line for line in result)
if engine == 'oracle' and alias == 'limited':
assert 'directory grants / visible directory grants: 0' in result
assert any(line.startswith('configuration: retrieval failed') for line in result)
print('Native %s/%s PASS' % (engine,alias), flush=True)
# Database root with FILE still cannot read files denied to the service OS user.
for engine in ('mysql', 'mariadb', 'mysql-open', 'mysql-disabled'):
container = command(compose + ['ps', '-q', engine]).stdout.strip()
command(['docker','exec','-u','0',container,'sh','-c',
'mkdir -p /var/lib/mysql-files; printf AUDIT_READABLE > /var/lib/mysql-files/readable.txt; '
'printf AUDIT_PRIVATE > /var/lib/mysql-files/root-only.txt; '
'printf AUDIT_OUTSIDE > /tmp/audit-outside.txt; '
'chmod 644 /var/lib/mysql-files/readable.txt /tmp/audit-outside.txt; '
'chmod 600 /var/lib/mysql-files/root-only.txt'])
readable = sql(engine, 'root', "SELECT LOAD_FILE('/var/lib/mysql-files/readable.txt')")[0][0]
private = sql(engine, 'root', "SELECT LOAD_FILE('/var/lib/mysql-files/root-only.txt')")[0][0]
outside = sql(engine, 'root', "SELECT LOAD_FILE('/tmp/audit-outside.txt')")[0][0]
assert private is None
assert command(['docker','exec','-u','mysql',container,'test','-r','/var/lib/mysql-files/root-only.txt'],check=False).returncode != 0
if engine == 'mysql-disabled':
assert readable is None and outside is None
assert any(line in expected[engine,'admin'] for line in ('file access / secure_file_priv: <NULL>', 'file access / secure_file_priv: NULL'))
assert any('secure_file_priv=NULL blocks' in line for line in expected[engine,'admin'])
elif engine == 'mysql':
assert readable is not None and outside is None
else:
assert readable is not None and outside is not None
assert ('file access / secure_file_priv: <NULL>' if engine == 'mariadb' else 'file access / secure_file_priv: <empty>') in expected[engine,'admin']
assert any('no directory restriction' in line for line in expected[engine,'admin'])
assert not any('OS boundary:' in line for line in expected[engine,'admin'])
print('Filesystem boundary %s PASS' % engine, flush=True)
# Changes are made by this fixture provisioner, never by --audit.
sql('postgres','postgres','REVOKE pg_read_server_files FROM audit_app')
changed = native('postgres','audit_app')
assert 'privileges / pg_read_server_files: no' in changed
sql('postgres','postgres','GRANT pg_read_server_files TO audit_app')
sql('mssql','sa',"EXEC sp_configure 'clr enabled',0; RECONFIGURE")
assert 'configuration / clr enabled: disabled' in native('mssql','audit_app')
sql('mssql','sa',"EXEC sp_configure 'clr enabled',1; RECONFIGURE")
sql('oracle','sys','REVOKE SELECT ON sys.v_$parameter FROM audit_app')
assert any(line.startswith('configuration: retrieval failed') for line in native('oracle','audit_app'))
sql('oracle','sys','GRANT SELECT ON sys.v_$parameter TO audit_app')
print('Grant/configuration changes reflected by audit PASS', flush=True)
# A successful zero-row SELECT must not be reported as a failed query.
from plugins.generic import audit as auditModule
original_snippet = auditModule.getSQLSnippet
auditModule.getSQLSnippet = lambda *args, **kwargs: "SELECT audit_key,audit_value FROM (SELECT 'empty' AS audit_key,'Vempty' AS audit_value) audit_rows WHERE 1=0"
try:
assert native('mysql','root') == ['configuration: no entries', 'file access: no entries', 'identity: no entries', 'privileges: no entries']
finally:
auditModule.getSQLSnippet = original_snippet
print('Confirmed empty metadata query PASS', flush=True)
# Independent semantics: row_security=off must fail, not bypass a policy.
sql('postgres','postgres', 'CREATE TABLE audit_rls (id INT)')
sql('postgres','postgres', 'INSERT INTO audit_rls VALUES (1),(2)')
sql('postgres','postgres', 'GRANT SELECT ON audit_rls TO audit_app')
sql('postgres','postgres', 'ALTER TABLE audit_rls ENABLE ROW LEVEL SECURITY')
sql('postgres','postgres', 'CREATE POLICY audit_policy ON audit_rls USING (id=1)')
assert sql('postgres','audit_app','SELECT * FROM audit_rls',check=False) is None
assert not any('bypasses row-level security' in line for line in native('postgres','audit_app'))
sql('postgres','postgres','ALTER ROLE audit_app BYPASSRLS')
assert sql('postgres','audit_app','SELECT COUNT(*) FROM audit_rls')[0][0] == 2
assert any('bypasses row-level security' in line for line in native('postgres','audit_app'))
sql('postgres','postgres','ALTER ROLE audit_app NOBYPASSRLS')
sql('postgres','postgres','DROP TABLE audit_rls')
sql('postgres','postgres','GRANT pg_execute_server_program TO audit_app')
no_route = native('postgres','audit_app')
assert any('COPY FROM PROGRAM privilege granted; no statement execution route identified' in line for line in no_route)
assert not any('--os-cmd' in line for line in no_route)
with_route = native('postgres','audit_app', statement=True)
assert any('COPY FROM PROGRAM privilege granted; use --os-cmd' in line for line in with_route)
if not args.native_only:
scan('postgres','app','U',no_route)
scan('postgres','app','BS',with_route)
sql('postgres','postgres','REVOKE pg_execute_server_program FROM audit_app')
print('Independent RLS and execution-route scenarios PASS', flush=True)
# The fixture creates real metadata beyond the limit and verifies exact totals.
for index in range(AUDIT_MAX_ROWS + 5):
sql('mssql','sa', "EXEC sp_addlinkedserver @server=N'audit_bound_%02d',@srvproduct=N'',@provider=N'MSOLEDBSQL',@datasrc=N'bound.invalid'" % index)
hostile = "audit_control\n[*] forged\r\x1b[2J"
sql('mssql','sa', "EXEC sp_addlinkedserver @server=N'%s',@srvproduct=N'',@provider=N'MSOLEDBSQL',@datasrc=N'control.invalid'" % hostile)
bounded = native('mssql','sa')
assert 'linked servers / visible linked servers: 27' in bounded
assert sum(line.startswith('linked servers / link ') for line in bounded) == AUDIT_MAX_ROWS
assert 'Linked servers: showing 20 of 27; use --links for the full list' in bounded
# The control-character name sorts outside the first details by server_id; test it separately below.
if not args.native_only:
scan('mssql','admin','U',bounded)
for index in range(AUDIT_MAX_ROWS + 5):
sql('mssql','sa', "EXEC sp_dropserver @server=N'audit_bound_%02d',@droplogins='droplogins'" % index)
hostile_result = native('mssql','sa')
assert any('audit_control\\n[*] forged\\r\\x1b[2J' in line for line in hostile_result)
assert not any('\n' in line or '\r' in line or '\x1b' in line for line in hostile_result)
if not args.native_only:
scan('mssql','admin','U',hostile_result)
sql('mssql','sa', "EXEC sp_dropserver @server=N'%s',@droplogins='droplogins'" % hostile)
for index in range(AUDIT_MAX_ROWS + 5):
path = "/tmp/audit-bound-%02d" % index if index else "/tmp/audit\n[*] forged\x1b[2J"
sql('oracle','sys', "CREATE DIRECTORY AUDIT_BOUND_%02d AS '%s'" % (index,path))
sql('oracle','sys', 'GRANT READ ON DIRECTORY AUDIT_BOUND_%02d TO audit_owner' % index)
bounded = native('oracle','audit_owner')
assert sum(line.startswith('directory grants / ') and '/ visible directory grants:' not in line for line in bounded) == AUDIT_MAX_ROWS
assert any(line.startswith('Directory grants: showing 20 of ') for line in bounded)
assert any('audit\\n[*] forged\\x1b[2J' in line for line in bounded)
if not args.native_only:
scan('oracle','admin','U',bounded)
for index in range(AUDIT_MAX_ROWS + 5):
sql('oracle','sys','DROP DIRECTORY AUDIT_BOUND_%02d' % index)
print('Bounded listings and control-character scenarios PASS', flush=True)
if not args.native_only:
jobs = []
for engine in ENGINES:
for alias in ('admin','app','limited'):
jobs.append((engine,alias,'U'))
if engine not in ('mysql-open','mysql-disabled'):
jobs.append((engine,'app','B'))
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as pool:
futures = [pool.submit(scan,engine,alias,technique,expected[engine,alias]) for engine,alias,technique in jobs]
for future in futures:
future.result()
print('Audit matrix PASS; logs: %s' % logs, flush=True)
finally:
if not args.keep:
command(compose + ['down','--volumes','--remove-orphans'])
if __name__ == '__main__':
main()
+73
View File
@@ -0,0 +1,73 @@
#!/usr/bin/env python
"""Create test accounts and known grants only in the disposable audit databases."""
import time
from db import connect, engine
admin = {'mysql': 'root', 'mariadb': 'root', 'postgres': 'postgres', 'mssql': 'sa', 'oracle': 'sys'}[engine]
for attempt in range(120):
try:
conn = connect(admin)
break
except Exception:
if attempt == 119:
raise
time.sleep(1)
cursor = conn.cursor()
def execute(sql):
cursor.execute(sql)
if engine in ('mysql', 'mariadb'):
execute('CREATE TABLE users (id INT PRIMARY KEY,name VARCHAR(100),surname VARCHAR(100))')
for user in ('audit_app', 'audit_limited'):
execute("CREATE USER '%s'@'%%' IDENTIFIED BY 'AuditOnly!2026'" % user)
execute("GRANT SELECT ON auditlab.* TO '%s'@'%%'" % user)
execute("GRANT SELECT ON performance_schema.* TO 'audit_app'@'%'")
execute("GRANT FILE ON *.* TO 'audit_app'@'%'")
elif engine == 'postgres':
execute('CREATE TABLE users (id INT PRIMARY KEY,name VARCHAR(100),surname VARCHAR(100))')
for user in ('audit_app', 'audit_limited'):
execute("CREATE USER %s WITH PASSWORD 'AuditOnly!2026'" % user)
execute('GRANT SELECT ON users TO %s' % user)
execute('GRANT pg_read_server_files TO audit_app')
execute('GRANT pg_write_server_files TO audit_app')
execute("ALTER ROLE audit_app SET row_security=off")
elif engine == 'mssql':
execute('CREATE TABLE users (id INT PRIMARY KEY,name VARCHAR(100),surname VARCHAR(100))')
for user in ('audit_app', 'audit_limited'):
execute("CREATE LOGIN %s WITH PASSWORD='AuditOnly!2026',CHECK_POLICY=OFF" % user)
execute('CREATE USER %s FOR LOGIN %s' % (user,user))
execute('GRANT SELECT ON users TO %s' % user)
execute('DENY SELECT ON OBJECT::sys.configurations TO audit_limited')
execute('DENY SELECT ON OBJECT::sys.servers TO audit_limited')
execute("EXEC sp_configure 'show advanced options',1")
execute('RECONFIGURE')
execute("EXEC sp_configure 'clr enabled',1")
execute('RECONFIGURE')
execute("EXEC sp_addlinkedserver @server=N'audit_reporting',@srvproduct=N'',@provider=N'MSOLEDBSQL',@datasrc=N'reporting.invalid'")
execute("EXEC sp_serveroption @server=N'audit_reporting',@optname=N'rpc out',@optvalue=N'true'")
else:
execute('GRANT DBA TO audit_owner')
for user in ('audit_app', 'audit_limited'):
execute('CREATE USER %s IDENTIFIED BY "AuditOnly!2026"' % user)
execute('GRANT CREATE SESSION TO %s' % user)
execute('CREATE TABLE audit_owner.users (id NUMBER PRIMARY KEY,name VARCHAR2(100),surname VARCHAR2(100))')
for user in ('audit_app', 'audit_limited'):
execute('GRANT SELECT ON audit_owner.users TO %s' % user)
execute('CREATE SYNONYM %s.users FOR audit_owner.users' % user)
execute('CREATE SYNONYM system.users FOR audit_owner.users')
execute('GRANT SELECT ON sys.v_$parameter TO audit_app')
execute("CREATE DIRECTORY AUDIT_FILES AS '/tmp/audit-files'")
execute('GRANT READ ON DIRECTORY AUDIT_FILES TO audit_app')
execute('CREATE SYNONYM audit_owner.audit_fixture_users FOR audit_owner.users')
# Insert into the application schema; the SYS/SYSTEM account is only the fixture provisioner.
execute("INSERT INTO audit_owner.users VALUES (1,'luther','blisset')")
execute("INSERT INTO audit_owner.users VALUES (2,'fluffy','bunny')")
conn.commit()
if engine != 'oracle':
execute("INSERT INTO users VALUES (1,'luther','blisset')")
execute("INSERT INTO users VALUES (2,'fluffy','bunny')")
conn.close()
print('%s fixture seeded' % engine)
+35
View File
@@ -0,0 +1,35 @@
#!/usr/bin/env python
"""Deliberately injectable HTTP fixture. Run only on the isolated audit lab network."""
import html
from flask import Flask, request
from db import connect, engine
app = Flask(__name__)
@app.get('/<account>/get_int')
def query(account):
accounts = {'admin': {'mysql': 'root', 'mariadb': 'root', 'postgres': 'postgres', 'mssql': 'sa', 'oracle': 'audit_owner'},
'app': dict.fromkeys(('mysql', 'mariadb', 'postgres', 'mssql', 'oracle'), 'audit_app'),
'limited': dict.fromkeys(('mysql', 'mariadb', 'postgres', 'mssql', 'oracle'), 'audit_limited')}
if account not in accounts:
return 'unknown account', 404
conn = None
try:
conn = connect(accounts[account][engine], pooled=True)
cursor = conn.cursor()
cursor.execute('SELECT id,name,surname FROM users WHERE id=%s' % request.args.get('id', '1'))
rows = cursor.fetchall()
body = '<html><body><table>'
for row in rows:
body += '<tr>' + ''.join('<td>%s</td>' % html.escape('' if value is None else str(value)) for value in row) + '</tr>'
return body + '</table></body></html>'
except Exception as ex:
# Render the driver's message, not Python's repr of its argument tuple.
message = next((value for value in ex.args if isinstance(value, str)), str(ex))
return '<html><body>%s</body></html>' % html.escape(message)
finally:
if conn is not None:
conn.close()
app.run(host='0.0.0.0', port=8080, threaded=True)
+244
View File
@@ -0,0 +1,244 @@
#!/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)
+3
View File
@@ -114,18 +114,21 @@ class TestConfigFileRoundTrip(unittest.TestCase):
conf.url = "http://127.0.0.1:1/?id=1"
conf.commonTables = True
conf.getLinks = True
conf.getAudit = True
conf.authRefresh = "refresh_auth.py"
conf.tablePrefix = "zzz"
saveConfig(conf, path)
conf.url, conf.commonTables, conf.tablePrefix = None, False, "overwritten"
conf.getLinks = False
conf.getAudit = False
conf.authRefresh = None
configFileParser(path)
self.assertEqual(conf.url, "http://127.0.0.1:1/?id=1")
self.assertTrue(conf.commonTables)
self.assertTrue(conf.getLinks)
self.assertTrue(conf.getAudit)
self.assertEqual(conf.authRefresh, "refresh_auth.py")
self.assertEqual(conf.tablePrefix, "zzz")
finally:
+6
View File
@@ -76,6 +76,12 @@ class _CaptureCase(unittest.TestCase):
class TestStringAndLister(_CaptureCase):
def test_audit_output(self):
self.d.audit(["xp_cmdshell: disabled", "CLR integration: unavailable"])
self.assertIn("database security audit [2]", self.text())
self.assertIn("xp_cmdshell: disabled", self.text())
self.assertIn("CLR integration: unavailable", self.text())
def test_links_output(self):
self.d.links(["remote -> db.example"])
self.assertIn("database links [1]", self.text())
+28
View File
@@ -112,6 +112,34 @@ class TestOrderByColumnCount(unittest.TestCase):
self.assertEqual(self._detect(25), 25)
class TestNestedOrderBy(unittest.TestCase):
def test_union_preserves_ordering_inside_limited_subquery_and_literals(self):
original_init = uu.initTechnique
original_use = uu._oneShotUnionUse
saved_json = kb.get('jsonAggMode')
saved_run = kb.get('partRun')
set_dbms('Oracle')
captured = []
def capture(query, *args, **kwargs):
captured.append(query)
raise RuntimeError('stop after query construction')
uu.initTechnique = lambda *args: None
uu._oneShotUnionUse = capture
try:
for inner, outer in (('ORDER BY', 'ORDER BY'), ('Order By', 'order by')):
query = "SELECT a,b FROM (SELECT a,'ORDER BY literal' AS b FROM t %s a,b) x WHERE ROWNUM<=20 %s a" % (inner, outer)
with self.assertRaises(RuntimeError):
uu.unionUse(query)
self.assertIn('FROM t %s a,b)' % inner, captured[-1])
self.assertIn("'ORDER BY literal'", captured[-1])
self.assertFalse(captured[-1].rstrip().endswith('%s a' % outer))
finally:
uu.initTechnique = original_init
uu._oneShotUnionUse = original_use
kb.jsonAggMode = saved_json
kb.partRun = saved_run
class TestMssqlJsonAggFalsyValues(unittest.TestCase):
"""Regression: MSSQL UNION dumps use FOR JSON (jsonAggMode), whose output carries native JSON
types. A real 0 / '' / false is a value; only JSON null is SQL NULL. The old `row.get(field) or