fix(security): Add table blocklist and fix MCP SQL validation bypass (#37411)

superset/sql/execution/executor.py

@@ -451,10 +451,22 @@ class SQLExecutor:
         :raises SupersetSecurityException: If security checks fail
         """
         # Check disallowed functions
-        if disallowed := self._check_disallowed_functions(script):
+        if disallowed_functions := self._check_disallowed_functions(script):
             raise SupersetSecurityException(
                 SupersetError(
-                    message=f"Disallowed SQL functions: {', '.join(disallowed)}",
+                    message=(
+                        f"Disallowed SQL functions: {', '.join(disallowed_functions)}"
+                    ),
+                    error_type=SupersetErrorType.INVALID_SQL_ERROR,
+                    level=ErrorLevel.ERROR,
+                )
+            )
+
+        # Check disallowed tables
+        if disallowed_tables := self._check_disallowed_tables(script):
+            raise SupersetSecurityException(
+                SupersetError(
+                    message=f"Disallowed SQL tables: {', '.join(disallowed_tables)}",
                     error_type=SupersetErrorType.INVALID_SQL_ERROR,
                     level=ErrorLevel.ERROR,
                 )
@@ -684,6 +696,31 @@ class SQLExecutor:
 
         return found if found else None
 
+    def _check_disallowed_tables(self, script: SQLScript) -> set[str] | None:
+        """
+        Check for disallowed SQL tables/views.
+
+        :param script: Parsed SQL script
+        :returns: Set of disallowed tables found, or None if none found
+        """
+        disallowed_config = app.config.get("DISALLOWED_SQL_TABLES", {})
+        engine_name = self.database.db_engine_spec.engine
+
+        # Get disallowed tables for this engine
+        engine_disallowed = disallowed_config.get(engine_name, set())
+        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
+
     def _apply_rls_to_script(
         self, script: SQLScript, catalog: str | None, schema: str | None
     ) -> None: