fix(sql): broaden mutating-statement detection in SQL Lab parser

superset/commands/sql_lab/estimate.py

@@ -102,14 +102,13 @@ def _apply_sql_security(self, sql: str) -> str:
             db_engine_spec.engine,
             set(),
         )
-        if disallowed_tables and parsed_script.check_tables_present(disallowed_tables):
-            found_tables = set()
-            for statement in parsed_script.statements:
-                present = {table.table.lower() for table in statement.tables}
-                for table in disallowed_tables:
-                    if table.lower() in present:
-                        found_tables.add(table)
-            raise SupersetDisallowedSQLTableException(found_tables or disallowed_tables)
+        if disallowed_tables:
+            # Honors schema-qualified denylist entries (e.g.
+            # ``information_schema.tables``) and reports only the tables
+            # actually referenced by the query.
+            found_tables = parsed_script.get_disallowed_tables(disallowed_tables)
+            if found_tables:
+                raise SupersetDisallowedSQLTableException(found_tables)
 
         if parsed_script.has_mutation() and not self._database.allow_dml:
             raise SupersetDMLNotAllowedException()

superset/config.py

@@ -1830,6 +1830,20 @@ def engine_context_manager(  # pylint: disable=unused-argument
         "pg_read_file",
         "pg_ls_dir",
         "pg_read_binary_file",
+        # PostgreSQL large-object functions: writers can plant arbitrary
+        # bytes on the server filesystem (lo_export, lo_from_bytea, lowrite,
+        # lo_put, lo_create, lo_import), readers can pull bytes back out
+        # (lo_get, loread), and lo_unlink deletes large objects outright.
+        # Defense-in-depth on top of is_mutating()'s function-name check.
+        "lo_from_bytea",
+        "lo_export",
+        "lo_import",
+        "lo_put",
+        "lo_create",
+        "lowrite",
+        "lo_get",
+        "loread",
+        "lo_unlink",
         # XML functions that can execute SQL
         "database_to_xml",
         "database_to_xmlschema",
@@ -1920,6 +1934,29 @@ def engine_context_manager(  # pylint: disable=unused-argument
         "pg_stat_replication",
         "pg_stat_wal_receiver",
         "pg_user",
+        # The SQL-standard `information_schema` views expose table /
+        # column / privilege / view-definition metadata across the entire
+        # database role the connection user can see. Entries are
+        # schema-qualified so `check_tables_present` only matches when the
+        # query actually references `information_schema.<view>`, not any
+        # user table that happens to share a name.
+        "information_schema.tables",
+        "information_schema.columns",
+        "information_schema.schemata",
+        "information_schema.views",
+        "information_schema.routines",
+        "information_schema.role_table_grants",
+        "information_schema.role_column_grants",
+        "information_schema.role_routine_grants",
+        "information_schema.table_privileges",
+        "information_schema.column_privileges",
+        "information_schema.usage_privileges",
+        "information_schema.key_column_usage",
+        "information_schema.table_constraints",
+        "information_schema.referential_constraints",
+        "information_schema.view_table_usage",
+        "information_schema.applicable_roles",
+        "information_schema.enabled_roles",
     },
     "mysql": {
         "mysql.user",

superset/models/helpers.py

@@ -1221,24 +1221,16 @@ def _process_sql_expression(  # pylint: disable=too-many-arguments
                     disallowed_functions
                 ):
                     raise SupersetDisallowedSQLFunctionException(disallowed_functions)
-                if disallowed_tables and parsed.check_tables_present(disallowed_tables):
+                if disallowed_tables:
                     # Report only the tables actually found in the expression,
                     # mirroring the canonical execution-time gate in
                     # `superset.sql_lab._validate_query` so the user-facing
-                    # error doesn't echo the operator's full denylist.
-                    present_tables = {
-                        table.table.lower()
-                        for statement in parsed.statements
-                        for table in statement.tables
-                    }
-                    found_tables = {
-                        table
-                        for table in disallowed_tables
-                        if table.lower() in present_tables
-                    }
-                    raise SupersetDisallowedSQLTableException(
-                        found_tables or disallowed_tables
-                    )
+                    # error doesn't echo the operator's full denylist. Honors
+                    # schema-qualified denylist entries (e.g.
+                    # ``information_schema.tables``).
+                    found_tables = parsed.get_disallowed_tables(disallowed_tables)
+                    if found_tables:
+                        raise SupersetDisallowedSQLTableException(found_tables)
         return expression
 
     def _process_select_expression(
@@ -1465,19 +1457,14 @@ def _raise_for_disallowed_sql(self, sql: str) -> None:
             disallowed_functions
         ):
             raise SupersetDisallowedSQLFunctionException(disallowed_functions)
-        if disallowed_tables and parsed_script.check_tables_present(disallowed_tables):
+        if disallowed_tables:
             # Report only the tables actually found in the query, mirroring the
             # canonical execution-time gate so the user-facing error doesn't
-            # echo the operator's full denylist.
-            present_tables = {
-                table.table.lower()
-                for statement in parsed_script.statements
-                for table in statement.tables
-            }
-            found_tables = {
-                table for table in disallowed_tables if table.lower() in present_tables
-            }
-            raise SupersetDisallowedSQLTableException(found_tables or disallowed_tables)
+            # echo the operator's full denylist. Honors schema-qualified
+            # denylist entries (e.g. ``information_schema.tables``).
+            found_tables = parsed_script.get_disallowed_tables(disallowed_tables)
+            if found_tables:
+                raise SupersetDisallowedSQLTableException(found_tables)
 
     def query(self, query_obj: QueryObjectDict) -> QueryResult:
         """

superset/sql/execution/executor.py

@@ -717,15 +717,9 @@ def _check_disallowed_tables(self, script: SQLScript) -> set[str] | None:
         if not engine_disallowed:
             return None
 
-        # Single-pass AST-based table detection
-        found: set[str] = set()
-        for statement in script.statements:
-            present = {table.table.lower() for table in statement.tables}
-            for table in engine_disallowed:
-                if table.lower() in present:
-                    found.add(table)
-
-        return found or None
+        # Honors schema-qualified denylist entries (e.g.
+        # ``information_schema.tables``) as well as bare names.
+        return script.get_disallowed_tables(engine_disallowed) or None
 
     def _apply_rls_to_script(
         self, script: SQLScript, catalog: str | None, schema: str | None