feat(sql): schema-qualified table denylist + information_schema/lo_* defaults

UPDATING.md

@@ -24,6 +24,17 @@ assists people when migrating to a new version.
 
 ## Next
 
+### SQL Lab denies large-object and information_schema access by default
+
+`DISALLOWED_SQL_FUNCTIONS` and `DISALLOWED_SQL_TABLES` now ship with additional default entries, so SQL Lab and chart-data queries that reference them are rejected where they were previously allowed:
+
+- PostgreSQL large-object routines (`lo_from_bytea`, `lo_export`, `lo_import`, `lo_put`, `lo_create`, `lo_creat`, `lowrite`, `lo_get`, `loread`, `lo_unlink`), which read and write bytes on the database server's filesystem.
+- The SQL-standard `information_schema` views (`tables`, `columns`, `routines`, `views`, the privilege/grant views, etc.), which expose table, column, privilege, and view-definition metadata across the whole database.
+
+Deployments that legitimately query these (for example tooling that introspects `information_schema`) can restore the previous behavior by overriding `DISALLOWED_SQL_FUNCTIONS` / `DISALLOWED_SQL_TABLES` in `superset_config.py` to drop the entries they need.
+
+Because the denylist now resolves the effective schema through the query-aware path, PostgreSQL queries that change the `search_path` (e.g. `SET search_path = ...`) are rejected on the SQL Lab execution and cost-estimate paths whenever any `DISALLOWED_SQL_TABLES` entry is configured (the default for PostgreSQL), matching the behavior previously applied only when `RLS_IN_SQLLAB` was enabled.
+
 ### SQL parser input length cap (SQL_MAX_PARSE_LENGTH)
 
 The SQL parser now rejects scripts whose UTF-8 byte length exceeds the new

superset/commands/sql_lab/estimate.py

@@ -34,7 +34,6 @@
 )
 from superset.jinja_context import get_template_processor
 from superset.models.core import Database
-from superset.models.sql_lab import Query
 from superset.sql.parse import SQLScript
 from superset.utils import core as utils
 from superset.utils.rls import apply_rls
@@ -102,57 +101,43 @@ 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)
+        rls_enabled = is_feature_enabled("RLS_IN_SQLLAB")
+
+        # Resolve the effective per-query schema once, the same way the execution
+        # path does (``sql_lab.execute_sql_statements``), but only when a control
+        # below actually needs it. Going through ``get_default_schema_for_query``
+        # rather than the static ``get_default_schema`` runs engine-specific
+        # per-query security gates too — e.g. ``PostgresEngineSpec`` rejects a
+        # query that sets ``search_path`` — and resolves unqualified references to
+        # the schema the engine uses at runtime, so both the denylist check and
+        # RLS injection match the execution path exactly.
+        catalog: str | None = None
+        effective_schema = ""
+        if disallowed_tables or rls_enabled:
+            catalog = self._catalog or self._database.get_default_catalog()
+            resolved_schema = self._database.resolve_query_default_schema(
+                self._sql, self._schema, catalog, self._template_params
+            )
+            # An explicit schema still wins for matching/RLS targeting; otherwise
+            # fall back to the runtime-resolved default.
+            effective_schema = self._schema or resolved_schema or ""
+
+        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, effective_schema
+            )
+            if found_tables:
+                raise SupersetDisallowedSQLTableException(found_tables)
 
         if parsed_script.has_mutation() and not self._database.allow_dml:
             raise SupersetDMLNotAllowedException()
 
-        if is_feature_enabled("RLS_IN_SQLLAB"):
-            # Resolve the default catalog/schema the same way the execution path
-            # does (``sql_lab.execute_sql_statements``) before injecting RLS.
-            # Crucially this goes through ``get_default_schema_for_query`` rather
-            # than the plain ``get_default_schema``, so engine-specific per-query
-            # security gates run too — e.g. ``PostgresEngineSpec`` rejects a query
-            # that sets ``search_path``. Resolving against the static default
-            # schema instead would both skip that gate and let unqualified tables
-            # dodge the RLS predicates the real query enforces, defeating the
-            # security parity this command exists to provide.
-            catalog = self._catalog or self._database.get_default_catalog()
-            # Build a transient (unsaved) Query so the engine spec can resolve the
-            # effective per-query schema exactly as the executor does. Mirror the
-            # probe built in ``SupersetSecurityManager.raise_for_access``: set a
-            # ``client_id`` (the column is ``nullable=False``) and expunge it, so
-            # the ``database`` backref's ``cascade="all, delete-orphan"`` cannot
-            # autoflush this incomplete row into the session when ``apply_rls``
-            # issues its own ``db.session`` query below.
-            probe_query = Query(
-                database=self._database,
-                sql=self._sql,
-                schema=self._schema or None,
-                catalog=catalog,
-                client_id=utils.shortid()[:10],
-                user_id=utils.get_user_id(),
-            )
-            db.session.expunge(probe_query)
-            # Always resolve through ``get_default_schema_for_query`` — even when
-            # the caller pinned a schema — so the engine's per-query security gate
-            # runs (e.g. ``PostgresEngineSpec`` rejects a query that sets
-            # ``search_path``), exactly as the executor does unconditionally. Only
-            # the resulting value falls back to the resolved default; an explicit
-            # schema still wins for the RLS predicate target.
-            resolved_schema = self._database.get_default_schema_for_query(
-                probe_query, self._template_params
-            )
-            schema = self._schema or resolved_schema or ""
+        if rls_enabled:
             for statement in parsed_script.statements:
-                apply_rls(self._database, catalog, schema, statement)
+                apply_rls(self._database, catalog, effective_schema, statement)
             return parsed_script.format()
 
         return sql

superset/config.py

@@ -1958,6 +1958,24 @@ 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_creat, lo_import), readers can pull bytes back
+        # out (lo_get, loread), lo_truncate/lo_truncate64 shrink existing
+        # large objects, and lo_unlink deletes them outright. Defense-in-depth
+        # on top of is_mutating()'s function-name check.
+        "lo_from_bytea",
+        "lo_export",
+        "lo_import",
+        "lo_put",
+        "lo_create",
+        "lo_creat",
+        "lowrite",
+        "lo_get",
+        "loread",
+        "lo_truncate",
+        "lo_truncate64",
+        "lo_unlink",
         # XML functions that can execute SQL
         "database_to_xml",
         "database_to_xmlschema",
@@ -2048,6 +2066,30 @@ 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
+        # reference resolves to `information_schema.<view>` -- either written
+        # explicitly or as an unqualified name under an `information_schema`
+        # search_path -- 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/core.py

@@ -698,6 +698,46 @@ def get_default_schema_for_query(
             self, query, template_params
         )
 
+    def resolve_query_default_schema(
+        self,
+        sql: str,
+        schema: str | None,
+        catalog: str | None,
+        template_params: Optional[dict[str, Any]] = None,
+    ) -> str | None:
+        """
+        Resolve the effective per-query default schema for a SQL string through
+        the query-aware :meth:`get_default_schema_for_query`.
+
+        Builds a transient (unsaved) ``Query`` probe so the engine spec resolves
+        the schema exactly as execution does -- running engine-specific per-query
+        security gates too -- then expunges it so the ``database`` backref's
+        ``cascade="all, delete-orphan"`` cannot autoflush this incomplete row
+        into the session. Centralizes the probe construction shared by the SQL
+        Lab executor, the cost-estimate command, and datasource denylist checks
+        so the schema-resolution behavior stays in sync across these
+        security-sensitive paths.
+
+        :param sql: Original (pre-render) SQL the query will execute
+        :param schema: Explicit per-query schema, if any
+        :param catalog: Resolved catalog
+        :param template_params: Jinja template parameters, if any
+        :returns: The runtime-resolved default schema, or None
+        """
+        from superset.models.sql_lab import Query
+
+        probe_query = Query(
+            database=self,
+            sql=sql,
+            schema=schema or None,
+            catalog=catalog,
+            client_id=utils.shortid()[:10],
+            user_id=utils.get_user_id(),
+        )
+        if probe_query in db.session:
+            db.session.expunge(probe_query)
+        return self.get_default_schema_for_query(probe_query, template_params)
+
     @staticmethod
     def post_process_df(df: pd.DataFrame) -> pd.DataFrame:
         def column_needs_conversion(df_series: pd.Series) -> bool:

superset/models/helpers.py

@@ -1187,6 +1187,31 @@ def get_sqla_row_level_filters(
         # for datasources of type query
         return []
 
+    def _resolve_denylist_schema(self, sql: str) -> Optional[str]:
+        """
+        Resolve the effective default schema for ``DISALLOWED_SQL_TABLES``
+        checks, memoized per datasource instance.
+
+        An explicit datasource schema always wins. Otherwise the schema is
+        resolved through the query-aware ``get_default_schema_for_query`` (not
+        the static ``get_default_schema``) so the value matches what the engine
+        uses at runtime -- honoring dynamic-schema engines and URI/connect-arg
+        schema rules -- exactly like the SQL Lab / executor denylist gate.
+
+        The result is cached on the instance so adhoc-expression validation,
+        which runs once per selected column/metric/order-by, resolves the schema
+        at most once instead of issuing a fallback inspector round-trip per
+        expression.
+        """
+        if self.schema:
+            return self.schema
+        if not hasattr(self, "_denylist_default_schema"):
+            catalog = self.catalog or self.database.get_default_catalog()
+            self._denylist_default_schema = self.database.resolve_query_default_schema(
+                sql, None, catalog
+            )
+        return self._denylist_default_schema
+
     def _process_sql_expression(  # pylint: disable=too-many-arguments
         self,
         expression: Optional[str],
@@ -1236,24 +1261,23 @@ 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``) and resolves unqualified
+                    # references against the same runtime schema the SQL Lab /
+                    # executor gate resolves them to (via the query-aware
+                    # resolver, not the static inspector default), so both
+                    # surfaces enforce the denylist consistently.
+                    effective_schema = self._resolve_denylist_schema(sql_to_check)
+                    found_tables = parsed.get_disallowed_tables(
+                        disallowed_tables, effective_schema
                     )
+                    if found_tables:
+                        raise SupersetDisallowedSQLTableException(found_tables)
         return expression
 
     def _process_select_expression(
@@ -1480,19 +1504,21 @@ 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``) and resolves
+            # unqualified references against the same runtime schema the SQL Lab
+            # / executor gate resolves them to (via the query-aware resolver,
+            # not the static inspector default), so both surfaces enforce the
+            # denylist consistently.
+            effective_schema = self._resolve_denylist_schema(sql)
+            found_tables = parsed_script.get_disallowed_tables(
+                disallowed_tables, effective_schema
+            )
+            if found_tables:
+                raise SupersetDisallowedSQLTableException(found_tables)
 
     def query(self, query_obj: QueryObjectDict) -> QueryResult:
         """