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

superset/config.py

@@ -1775,6 +1775,19 @@ 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) and readers can pull bytes back out
+        # (lo_get, loread). 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",
         # XML functions that can execute SQL
         "database_to_xml",
         "database_to_xmlschema",
@@ -1865,6 +1878,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/sql/parse.py

@@ -562,6 +562,51 @@ class SQLStatement(BaseSQLStatement[exp.Expression]):
     This class is used for all engines with dialects that can be parsed using sqlglot.
     """
 
+    # Function names that mutate server-side state but appear in the AST as
+    # plain function calls inside a non-mutating wrapper. Used by
+    # ``is_mutating()`` to classify e.g. PostgreSQL large-object writers.
+    # Names are uppercased for comparison.
+    _MUTATING_FUNCTION_NAMES: frozenset[str] = frozenset(
+        {
+            "LO_FROM_BYTEA",
+            "LO_EXPORT",
+            "LO_IMPORT",
+            "LO_PUT",
+            "LO_CREATE",
+            "LOWRITE",
+            # PostgreSQL sequence mutators. `SELECT setval('seq', N)` looks
+            # like a read but changes sequence state for every subsequent
+            # `nextval` caller.
+            "SETVAL",
+        }
+    )
+
+    # PostgreSQL constructs that sqlglot represents as an opaque ``exp.Command``
+    # (no structured AST). Each can mutate server state or wrap a DML body that
+    # would otherwise be detected by node-type matching. Used by
+    # ``is_mutating()``.
+    _POSTGRES_MUTATING_COMMAND_NAMES: frozenset[str] = frozenset(
+        {
+            "DO",  # PL/pgSQL anonymous block
+            "PREPARE",  # PREPARE u AS UPDATE ... ; EXECUTE u
+            "EXECUTE",  # body is the prepared DML
+            "CALL",  # procedure body may mutate
+            "COPY",  # server-side file ingest into a table
+            "GRANT",
+            "REVOKE",
+            "SET",  # SET ROLE / SET SESSION AUTHORIZATION change effective user
+            "REFRESH",  # REFRESH MATERIALIZED VIEW
+            "REINDEX",
+            "VACUUM",
+            # SHOW reads server configuration (version, hba_file, ssl state,
+            # search_path, etc.). It does not mutate, but in a read-only
+            # context (`allow_dml=False`) it is information-disclosure
+            # equivalent to the read-side entries in DISALLOWED_SQL_FUNCTIONS
+            # which are also blocked. Treat as gated alongside the writers.
+            "SHOW",
+        }
+    )
+
     def __init__(
         self,
         statement: str | None = None,
@@ -690,25 +735,52 @@ def is_mutating(self) -> bool:
             exp.Drop,
             exp.TruncateTable,
             exp.Alter,
+            # sqlglot has structured nodes for these DML/DCL forms in
+            # PostgreSQL and other dialects; without them an opaque exp.Command
+            # check would still miss the structured-parse path.
+            exp.Copy,  # COPY <table> FROM/TO (server-side file ingest)
+            exp.Grant,
+            exp.Revoke,
         )
 
         for node_type in mutating_nodes:
             if self._parsed.find(node_type):
                 return True
 
+        # `SELECT ... INTO new_table FROM ...` parses as `exp.Select` with an
+        # `into` arg (Postgres-style CTAS variant). It creates a new table and
+        # therefore mutates schema.
+        if isinstance(self._parsed, exp.Select) and self._parsed.args.get("into"):
+            return True
+
+        # Function calls that mutate server-side state without an enclosing
+        # mutating AST node. Notable example: PostgreSQL large-object writers
+        # (`lo_export` writes to the server filesystem, `lo_from_bytea`/
+        # `lo_create`/`lo_put`/`lo_import`/`lowrite` mutate the pg_largeobject
+        # catalog). These appear as plain function calls inside an `exp.Select`
+        # and would otherwise pass the read-only gate.
+        for function in self._parsed.find_all(exp.Func):
+            name = (
+                function.name.upper()
+                if function.sql_name() == "ANONYMOUS"
+                else function.sql_name().upper()
+            )
+            if name in self._MUTATING_FUNCTION_NAMES:
+                return True
+
         # depending on the dialect (Oracle, MS SQL) the `ALTER` is parsed as a
         # command, not an expression - check at root level
         if isinstance(self._parsed, exp.Command) and self._parsed.name == "ALTER":
             return True  # pragma: no cover
 
+        # PostgreSQL constructs that sqlglot represents as an opaque
+        # `exp.Command` rather than a structured AST. Each of these can mutate
+        # state or wrap a DML body that would otherwise be detected.
         if (
             self._dialect == Dialects.POSTGRES
             and isinstance(self._parsed, exp.Command)
-            and self._parsed.name == "DO"
+            and self._parsed.name in self._POSTGRES_MUTATING_COMMAND_NAMES
         ):
-            # anonymous blocks can be written in many different languages (the default
-            # is PL/pgSQL), so parsing them it out of scope of this class; we just
-            # assume the anonymous block is mutating
             return True
 
         # Postgres runs DMLs prefixed by `EXPLAIN ANALYZE`, see
@@ -829,17 +901,45 @@ def check_functions_present(self, functions: set[str]) -> bool:
             else:
                 present.add(function.name.upper())
 
+        # MySQL `@@<name>` syntax (also Oracle/SQL-Server `@@name`) parses as
+        # `exp.SessionParameter`, which is *not* a subclass of `exp.Func`, so
+        # the walk above misses it. Include those names so denylist entries
+        # like `version` or `hostname` match `SELECT @@version`.
+        for param in self._parsed.find_all(exp.SessionParameter):
+            present.add(param.name.upper())
+
         return any(function.upper() in present for function in functions)
 
     def check_tables_present(self, tables: set[str]) -> bool:
         """
         Check if any of the given tables are present in the statement.
 
+        Denylist entries may be bare (``pg_stat_activity``) or
+        schema-qualified (``information_schema.tables``). Bare entries
+        match by table name regardless of schema; qualified entries
+        require the schema to match too. This lets us block all access
+        to ``information_schema`` without also blocking any
+        user-authored table that happens to be named ``tables``.
+
         :param tables: Set of table names to check for (case-insensitive)
-        :return: True if any of the tables are present
-        """
-        present = {table.table.lower() for table in self.tables}
-        return any(table.lower() in present for table in tables)
+        :return: True if any of the given tables is referenced
+        """
+        present_bare: set[str] = set()
+        present_qualified: set[str] = set()
+        for t in self.tables:
+            bare = t.table.lower()
+            present_bare.add(bare)
+            if t.schema:
+                present_qualified.add(f"{t.schema.lower()}.{bare}")
+        for entry in tables:
+            needle = entry.lower()
+            if "." in needle:
+                if needle in present_qualified:
+                    return True
+            else:
+                if needle in present_bare:
+                    return True
+        return False
 
     def get_limit_value(self) -> int | None:
         """