From d0ad9b2e34af7a3015e45b008d03dd501802142b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0tampar?= Date: Mon, 19 Oct 2026 18:34:39 +0200 Subject: [PATCH] Adding cross-DBMS security configuration audit (--audit) --- .../procs/mssqlserver/audit_configuration.sql | 5 + data/procs/mssqlserver/audit_identity.sql | 11 + data/procs/mssqlserver/audit_links.sql | 6 + data/procs/mssqlserver/audit_permissions.sql | 7 + data/procs/mysql/audit_configuration.sql | 5 + .../mysql/audit_configuration_mariadb.sql | 5 + data/procs/mysql/audit_file_access.sql | 5 + data/procs/mysql/audit_identity.sql | 7 + data/procs/mysql/audit_privileges.sql | 5 + data/procs/oracle/audit_configuration.sql | 5 + data/procs/oracle/audit_directories.sql | 12 + data/procs/oracle/audit_identity.sql | 9 + data/procs/oracle/audit_privileges.sql | 5 + data/procs/postgresql/audit_configuration.sql | 5 + data/procs/postgresql/audit_identity.sql | 7 + data/procs/postgresql/audit_privileges.sql | 7 + lib/controller/action.py | 6 + lib/core/dump.py | 3 + lib/core/enums.py | 1 + lib/core/optiondict.py | 1 + lib/core/settings.py | 5 +- lib/core/testing.py | 1 + lib/parse/cmdline.py | 3 + lib/techniques/union/use.py | 7 +- plugins/dbms/mssqlserver/audit.py | 53 ++++ plugins/dbms/mysql/audit.py | 44 +++ plugins/dbms/oracle/audit.py | 23 ++ plugins/dbms/postgresql/audit.py | 33 +++ plugins/generic/audit.py | 113 ++++++++ plugins/generic/databases.py | 13 + sqlmap.conf | 4 + tests/integration/audit/Dockerfile | 5 + tests/integration/audit/README.md | 42 +++ tests/integration/audit/compose.yml | 82 ++++++ tests/integration/audit/db.py | 27 ++ tests/integration/audit/probe.py | 12 + tests/integration/audit/run.py | 266 ++++++++++++++++++ tests/integration/audit/seed.py | 73 +++++ tests/integration/audit/server.py | 35 +++ tests/test_audit.py | 244 ++++++++++++++++ tests/test_configfile.py | 3 + tests/test_dump_format.py | 6 + tests/test_union_engine.py | 28 ++ 43 files changed, 1236 insertions(+), 3 deletions(-) create mode 100644 data/procs/mssqlserver/audit_configuration.sql create mode 100644 data/procs/mssqlserver/audit_identity.sql create mode 100644 data/procs/mssqlserver/audit_links.sql create mode 100644 data/procs/mssqlserver/audit_permissions.sql create mode 100644 data/procs/mysql/audit_configuration.sql create mode 100644 data/procs/mysql/audit_configuration_mariadb.sql create mode 100644 data/procs/mysql/audit_file_access.sql create mode 100644 data/procs/mysql/audit_identity.sql create mode 100644 data/procs/mysql/audit_privileges.sql create mode 100644 data/procs/oracle/audit_configuration.sql create mode 100644 data/procs/oracle/audit_directories.sql create mode 100644 data/procs/oracle/audit_identity.sql create mode 100644 data/procs/oracle/audit_privileges.sql create mode 100644 data/procs/postgresql/audit_configuration.sql create mode 100644 data/procs/postgresql/audit_identity.sql create mode 100644 data/procs/postgresql/audit_privileges.sql create mode 100644 plugins/dbms/mssqlserver/audit.py create mode 100644 plugins/dbms/mysql/audit.py create mode 100644 plugins/dbms/oracle/audit.py create mode 100644 plugins/dbms/postgresql/audit.py create mode 100644 plugins/generic/audit.py create mode 100644 tests/integration/audit/Dockerfile create mode 100644 tests/integration/audit/README.md create mode 100644 tests/integration/audit/compose.yml create mode 100644 tests/integration/audit/db.py create mode 100644 tests/integration/audit/probe.py create mode 100644 tests/integration/audit/run.py create mode 100644 tests/integration/audit/seed.py create mode 100644 tests/integration/audit/server.py create mode 100644 tests/test_audit.py diff --git a/data/procs/mssqlserver/audit_configuration.sql b/data/procs/mssqlserver/audit_configuration.sql new file mode 100644 index 000000000..297fc2913 --- /dev/null +++ b/data/procs/mssqlserver/audit_configuration.sql @@ -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 diff --git a/data/procs/mssqlserver/audit_identity.sql b/data/procs/mssqlserver/audit_identity.sql new file mode 100644 index 000000000..7cc94f1bd --- /dev/null +++ b/data/procs/mssqlserver/audit_identity.sql @@ -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 diff --git a/data/procs/mssqlserver/audit_links.sql b/data/procs/mssqlserver/audit_links.sql new file mode 100644 index 000000000..d3458f0ff --- /dev/null +++ b/data/procs/mssqlserver/audit_links.sql @@ -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'')+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 diff --git a/data/procs/mssqlserver/audit_permissions.sql b/data/procs/mssqlserver/audit_permissions.sql new file mode 100644 index 000000000..1a063a214 --- /dev/null +++ b/data/procs/mssqlserver/audit_permissions.sql @@ -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 diff --git a/data/procs/mysql/audit_configuration.sql b/data/procs/mysql/audit_configuration.sql new file mode 100644 index 000000000..2ae3afde7 --- /dev/null +++ b/data/procs/mysql/audit_configuration.sql @@ -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 diff --git a/data/procs/mysql/audit_configuration_mariadb.sql b/data/procs/mysql/audit_configuration_mariadb.sql new file mode 100644 index 000000000..43ff5d42a --- /dev/null +++ b/data/procs/mysql/audit_configuration_mariadb.sql @@ -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 diff --git a/data/procs/mysql/audit_file_access.sql b/data/procs/mysql/audit_file_access.sql new file mode 100644 index 000000000..e76214298 --- /dev/null +++ b/data/procs/mysql/audit_file_access.sql @@ -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 diff --git a/data/procs/mysql/audit_identity.sql b/data/procs/mysql/audit_identity.sql new file mode 100644 index 000000000..d2adb3bfc --- /dev/null +++ b/data/procs/mysql/audit_identity.sql @@ -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 diff --git a/data/procs/mysql/audit_privileges.sql b/data/procs/mysql/audit_privileges.sql new file mode 100644 index 000000000..117f9781a --- /dev/null +++ b/data/procs/mysql/audit_privileges.sql @@ -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 diff --git a/data/procs/oracle/audit_configuration.sql b/data/procs/oracle/audit_configuration.sql new file mode 100644 index 000000000..60e4a98b4 --- /dev/null +++ b/data/procs/oracle/audit_configuration.sql @@ -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 diff --git a/data/procs/oracle/audit_directories.sql b/data/procs/oracle/audit_directories.sql new file mode 100644 index 000000000..bf7e89640 --- /dev/null +++ b/data/procs/oracle/audit_directories.sql @@ -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 diff --git a/data/procs/oracle/audit_identity.sql b/data/procs/oracle/audit_identity.sql new file mode 100644 index 000000000..4762a74a5 --- /dev/null +++ b/data/procs/oracle/audit_identity.sql @@ -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 diff --git a/data/procs/oracle/audit_privileges.sql b/data/procs/oracle/audit_privileges.sql new file mode 100644 index 000000000..6e518eab8 --- /dev/null +++ b/data/procs/oracle/audit_privileges.sql @@ -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 diff --git a/data/procs/postgresql/audit_configuration.sql b/data/procs/postgresql/audit_configuration.sql new file mode 100644 index 000000000..37f5a375b --- /dev/null +++ b/data/procs/postgresql/audit_configuration.sql @@ -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 diff --git a/data/procs/postgresql/audit_identity.sql b/data/procs/postgresql/audit_identity.sql new file mode 100644 index 000000000..ad01c31ee --- /dev/null +++ b/data/procs/postgresql/audit_identity.sql @@ -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 diff --git a/data/procs/postgresql/audit_privileges.sql b/data/procs/postgresql/audit_privileges.sql new file mode 100644 index 000000000..bc6891dba --- /dev/null +++ b/data/procs/postgresql/audit_privileges.sql @@ -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 diff --git a/lib/controller/action.py b/lib/controller/action.py index f0b190be4..531dfa722 100644 --- a/lib/controller/action.py +++ b/lib/controller/action.py @@ -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) diff --git a/lib/core/dump.py b/lib/core/dump.py index a5f620a07..c9235f0e9 100644 --- a/lib/core/dump.py +++ b/lib/core/dump.py @@ -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() diff --git a/lib/core/enums.py b/lib/core/enums.py index 26a0e8d50..c27083724 100644 --- a/lib/core/enums.py +++ b/lib/core/enums.py @@ -430,6 +430,7 @@ class CONTENT_TYPE(object): STATEMENTS = 26 PROCEDURES = 27 LINKS = 28 + AUDIT = 29 class CONTENT_STATUS(object): IN_PROGRESS = 0 diff --git a/lib/core/optiondict.py b/lib/core/optiondict.py index afb3a65c8..835b394e1 100644 --- a/lib/core/optiondict.py +++ b/lib/core/optiondict.py @@ -164,6 +164,7 @@ optDict = { "getPrivileges": ("boolean", "Privileges"), "getRoles": ("boolean", "Roles"), "getLinks": "boolean", + "getAudit": "boolean", "getDbs": ("boolean", "Databases"), "getTables": ("boolean", "Tables"), "getColumns": ("boolean", "Columns"), diff --git a/lib/core/settings.py b/lib/core/settings.py index 7916afa27..857a1ce0b 100644 --- a/lib/core/settings.py +++ b/lib/core/settings.py @@ -20,7 +20,7 @@ from lib.core.enums import OS from thirdparty import six # sqlmap version (...) -VERSION = "1.10.10.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 diff --git a/lib/core/testing.py b/lib/core/testing.py index c074e8b0b..4b74b909b 100644 --- a/lib/core/testing.py +++ b/lib/core/testing.py @@ -114,6 +114,7 @@ def vulnTest(tests=None, label="vuln"): ("-d \"\" --flush-session --dump -T creds --dump-format=SQLITE --binary-fields=password_hash --where \"user_id=5\"", ("3137396164343563366365326362393763663130323965323132303436653831", "dumped to SQLITE database")), ("-d \"\" --flush-session --banner --schema --sql-query=\"UPDATE users SET name='foobar' WHERE id=4; SELECT * FROM users; SELECT 987654321\"", ("banner: '3.", "INTEGER", "TEXT", "id", "name", "surname", "4,foobar,nameisnull", "'987654321'",)), ("-d \"\" --links", ("back-end DBMS: SQLite", "enumeration of database links is not supported on SQLite", "~unhandled exception")), + ("-d \"\" --audit", ("back-end DBMS: SQLite", "security configuration audit is not supported on SQLite", "~unhandled exception")), ("-u csrf --data=\"id=1&csrf_token=1\" --banner --answers=\"update=y\" --flush-session --technique=B", ("back-end DBMS: SQLite", "banner: '3.")), ("--purge -v 3", ("~ERROR", "~CRITICAL", "deleting the whole directory tree")), ) diff --git a/lib/parse/cmdline.py b/lib/parse/cmdline.py index 122e0303c..cbe762c5c 100644 --- a/lib/parse/cmdline.py +++ b/lib/parse/cmdline.py @@ -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") diff --git a/lib/techniques/union/use.py b/lib/techniques/union/use.py index db434e8c8..5a83f76ab 100644 --- a/lib/techniques/union/use.py +++ b/lib/techniques/union/use.py @@ -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) diff --git a/plugins/dbms/mssqlserver/audit.py b/plugins/dbms/mssqlserver/audit.py new file mode 100644 index 000000000..db784d8e3 --- /dev/null +++ b/plugins/dbms/mssqlserver/audit.py @@ -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 diff --git a/plugins/dbms/mysql/audit.py b/plugins/dbms/mysql/audit.py new file mode 100644 index 000000000..e2f854c2c --- /dev/null +++ b/plugins/dbms/mysql/audit.py @@ -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 diff --git a/plugins/dbms/oracle/audit.py b/plugins/dbms/oracle/audit.py new file mode 100644 index 000000000..526fd207f --- /dev/null +++ b/plugins/dbms/oracle/audit.py @@ -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 [] diff --git a/plugins/dbms/postgresql/audit.py b/plugins/dbms/postgresql/audit.py new file mode 100644 index 000000000..33e1e2869 --- /dev/null +++ b/plugins/dbms/postgresql/audit.py @@ -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 diff --git a/plugins/generic/audit.py b/plugins/generic/audit.py new file mode 100644 index 000000000..58a1dbbc7 --- /dev/null +++ b/plugins/generic/audit.py @@ -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 + # 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 = "" + elif facts[key] == "": + value = "" + 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 diff --git a/plugins/generic/databases.py b/plugins/generic/databases.py index 2f9441fd7..226f19e60 100644 --- a/plugins/generic/databases.py +++ b/plugins/generic/databases.py @@ -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) diff --git a/sqlmap.conf b/sqlmap.conf index 742a9d143..b844e8598 100644 --- a/sqlmap.conf +++ b/sqlmap.conf @@ -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 = diff --git a/tests/integration/audit/Dockerfile b/tests/integration/audit/Dockerfile new file mode 100644 index 000000000..7da8a3026 --- /dev/null +++ b/tests/integration/audit/Dockerfile @@ -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"] diff --git a/tests/integration/audit/README.md b/tests/integration/audit/README.md new file mode 100644 index 000000000..28a2fba50 --- /dev/null +++ b/tests/integration/audit/README.md @@ -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-` 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 -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- --reuse --keep +``` diff --git a/tests/integration/audit/compose.yml b/tests/integration/audit/compose.yml new file mode 100644 index 000000000..39ecf881c --- /dev/null +++ b/tests/integration/audit/compose.yml @@ -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'] diff --git a/tests/integration/audit/db.py b/tests/integration/audit/db.py new file mode 100644 index 000000000..b524ab830 --- /dev/null +++ b/tests/integration/audit/db.py @@ -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) diff --git a/tests/integration/audit/probe.py b/tests/integration/audit/probe.py new file mode 100644 index 000000000..005cf808d --- /dev/null +++ b/tests/integration/audit/probe.py @@ -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() diff --git a/tests/integration/audit/run.py b/tests/integration/audit/run.py new file mode 100644 index 000000000..eb91133b1 --- /dev/null +++ b/tests/integration/audit/run.py @@ -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: ', '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: ' if engine == 'mariadb' else 'file access / secure_file_priv: ') 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() diff --git a/tests/integration/audit/seed.py b/tests/integration/audit/seed.py new file mode 100644 index 000000000..deaa726cc --- /dev/null +++ b/tests/integration/audit/seed.py @@ -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) diff --git a/tests/integration/audit/server.py b/tests/integration/audit/server.py new file mode 100644 index 000000000..90e60be89 --- /dev/null +++ b/tests/integration/audit/server.py @@ -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('//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 = '' + for row in rows: + body += '' + ''.join('' % html.escape('' if value is None else str(value)) for value in row) + '' + return body + '
%s
' + 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 '%s' % html.escape(message) + finally: + if conn is not None: + conn.close() + + +app.run(host='0.0.0.0', port=8080, threaded=True) diff --git a/tests/test_audit.py b/tests/test_audit.py new file mode 100644 index 000000000..b78b1b891 --- /dev/null +++ b/tests/test_audit.py @@ -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) diff --git a/tests/test_configfile.py b/tests/test_configfile.py index 5ea9a94fb..337d00ddc 100644 --- a/tests/test_configfile.py +++ b/tests/test_configfile.py @@ -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: diff --git a/tests/test_dump_format.py b/tests/test_dump_format.py index cb966fd0f..faef2e44b 100644 --- a/tests/test_dump_format.py +++ b/tests/test_dump_format.py @@ -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()) diff --git a/tests/test_union_engine.py b/tests/test_union_engine.py index 9aa6886ca..88181f64b 100644 --- a/tests/test_union_engine.py +++ b/tests/test_union_engine.py @@ -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