fix(security): keep chart perm in sync when a renamed datasource is already loaded

superset/security/manager.py

@@ -79,7 +79,8 @@
     Table as SQLATable,
 )
 from sqlalchemy.engine.base import Connection
-from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import joinedload, object_session
+from sqlalchemy.orm.attributes import set_committed_value
 from sqlalchemy.orm.exc import MultipleResultsFound
 from sqlalchemy.orm.mapper import Mapper
 from sqlalchemy.orm.query import Query as SqlaQuery
@@ -3874,10 +3875,73 @@ def database_after_update(
         self._update_vm_datasources_access(
             mapper, connection, old_database_name, target
         )
+        self._sync_loaded_dataset_perms(connection, target)
         # Note schema permissions are updated at the API level
         # (database.commands.update). Since we need to fetch all existing schemas from
         # the db
 
+    @staticmethod
+    def _set_committed_perms(instance: Any, values: dict[str, Optional[str]]) -> None:
+        """
+        Record perm strings that a Core UPDATE wrote for an instance loaded in
+        the session.
+
+        The permission event hooks rewrite ``perm``, ``schema_perm`` and
+        ``catalog_perm`` through the connection, so a loaded instance keeps
+        the previous strings until it is expired. ``set_related_perm`` copies
+        a chart's perm from its datasource instance, so a chart flushed later
+        in the same transaction would write the previous strings back.
+
+        An attribute that is not loaded reads the new value from the database
+        anyway, and one with a pending change keeps that change.
+
+        :param instance: The loaded datasource instance
+        :param values: The written value of each perm column
+        """
+        state = inspect(instance)
+        for key, value in values.items():
+            if key in state.unloaded or state.attrs[key].history.has_changes():
+                continue
+            set_committed_value(instance, key, value)
+
+    def _sync_loaded_dataset_perms(
+        self,
+        connection: Connection,
+        target: "Database",
+    ) -> None:
+        """
+        Refresh the ``perm`` of the database's datasets loaded in the session
+        after a database rename rewrote it through the connection.
+
+        :param connection: The SQLA connection
+        :param target: The renamed database
+        """
+        from superset.connectors.sqla.models import (  # pylint: disable=import-outside-toplevel
+            SqlaTable,
+        )
+
+        if not (session := object_session(target)):
+            return
+        loaded_ids = [
+            key[1][0]
+            for key in session.identity_map.keys()
+            if issubclass(key[0], SqlaTable)
+        ]
+        if not loaded_ids:
+            return
+        sqlatable_table = SqlaTable.__table__  # pylint: disable=no-member
+        rows = connection.execute(
+            select(sqlatable_table.c.id, sqlatable_table.c.perm).where(
+                sqlatable_table.c.database_id == target.id,
+                sqlatable_table.c.id.in_(loaded_ids),
+            )
+        )
+        for row in rows:
+            if dataset := session.identity_map.get(
+                session.identity_key(SqlaTable, row.id)
+            ):
+                self._set_committed_perms(dataset, {"perm": row.perm})
+
     def _delete_vm_database_access(
         self,
         mapper: Mapper,
@@ -4312,6 +4376,13 @@ def _update_dataset_catalog_schema_perm(
                 schema_perm=schema_permission_name,
             )
         )
+        self._set_committed_perms(
+            target,
+            {
+                "catalog_perm": catalog_permission_name,
+                "schema_perm": schema_permission_name,
+            },
+        )
 
         # Update charts (Slice schema_perm field)
         connection.execute(
@@ -4390,6 +4461,7 @@ def _update_dataset_perm(  # pylint: disable=too-many-arguments
             )
             .values(perm=new_permission_name)
         )
+        self._set_committed_perms(target, {"perm": new_permission_name})
         # Update charts (Slice perm field)
         connection.execute(
             chart_table.update()
@@ -4483,6 +4555,7 @@ def semantic_layer_before_update(
             views = connection.execute(
                 sv_table.select().where(sv_table.c.semantic_layer_uuid == target.uuid)
             ).fetchall()
+            session = object_session(target)
             for view_row in views:
                 new_view_perm = f"[{target.name}].[{view_row.name}](id:{view_row.id})"
                 old_view_perm = view_row.perm
@@ -4499,6 +4572,12 @@ def semantic_layer_before_update(
                         .where(sv_table.c.id == view_row.id)
                         .values(perm=new_view_perm)
                     )
+                    if session and (
+                        view := session.identity_map.get(
+                            session.identity_key(SemanticView, view_row.id)
+                        )
+                    ):
+                        self._set_committed_perms(view, {"perm": new_view_perm})
 
                     # Update dependent charts so their denormalized perm stays
                     # in sync with the view: chart-list access filters match

tests/integration_tests/commands_test.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 import copy
+from typing import Any
 from unittest.mock import patch
 from uuid import uuid4
 
@@ -25,7 +26,9 @@
 from superset.commands.exceptions import CommandInvalidError
 from superset.commands.importers.v1.assets import ImportAssetsCommand
 from superset.commands.importers.v1.utils import is_valid_config
+from superset.connectors.sqla.models import SqlaTable
 from superset.models.annotations import Annotation, AnnotationLayer
+from superset.models.core import Database
 from superset.models.dashboard import Dashboard
 from superset.models.slice import Slice
 from superset.utils import json
@@ -122,6 +125,57 @@ def test_import_assets_resolves_native_annotation_layer(self, mock_add_permissio
             db.session.delete(database)
             db.session.commit()
 
+    @patch("superset.commands.database.importers.v1.utils.add_permissions")
+    def test_import_assets_dataset_change_keeps_chart_perms(self, mock_add_permissions):
+        """
+        Re-importing a bundle whose dataset has a new table name and schema
+        gives its chart the dataset's new perm strings. The dataset update
+        rewrites them through the connection before the chart is written in
+        the same transaction.
+        """
+
+        def import_bundle(dataset: dict[str, Any]) -> None:
+            ImportAssetsCommand(
+                {
+                    "metadata.yaml": yaml.safe_dump(metadata_config),
+                    "databases/imported_database.yaml": yaml.safe_dump(database_config),
+                    "datasets/imported_dataset.yaml": yaml.safe_dump(dataset),
+                    "charts/imported_chart.yaml": yaml.safe_dump(chart_config),
+                }
+            ).run()
+
+        try:
+            import_bundle(dataset_config)
+            import_bundle(
+                {
+                    **dataset_config,
+                    "table_name": "imported_dataset_renamed",
+                    "schema": "renamed_schema",
+                }
+            )
+
+            db.session.expire_all()
+            chart = db.session.query(Slice).filter_by(uuid=chart_config["uuid"]).one()
+            dataset = chart.table
+            assert dataset.perm == (
+                f"[imported_database].[imported_dataset_renamed](id:{dataset.id})"
+            )
+            assert dataset.schema_perm == (
+                "[imported_database].[default].[renamed_schema]"
+            )
+            assert chart.perm == dataset.perm
+            assert chart.schema_perm == dataset.schema_perm
+        finally:
+            db.session.rollback()
+            for model, uuid in (
+                (Slice, chart_config["uuid"]),
+                (SqlaTable, dataset_config["uuid"]),
+                (Database, database_config["uuid"]),
+            ):
+                if obj := db.session.query(model).filter_by(uuid=uuid).one_or_none():
+                    db.session.delete(obj)
+            db.session.commit()
+
     @patch("superset.commands.database.importers.v1.utils.add_permissions")
     def test_import_assets(self, mock_add_permissions):
         """Test that we can import multiple assets"""