fix(security): handle a missing dataset permission when renaming a database

superset/security/manager.py

@@ -3959,6 +3959,8 @@ def _update_vm_database_access(
         new_db_view_menu = self._find_view_menu_on_sqla_event(
             connection, new_view_menu_name
         )
+        if not new_db_view_menu:
+            return None
 
         self.on_view_menu_after_update(mapper, connection, new_db_view_menu)
         return new_db_view_menu
@@ -4041,6 +4043,20 @@ def _update_vm_datasources_access(  # pylint: disable=too-many-locals
                     connection,
                     new_dataset_vm_name,
                 )
+                if not new_dataset_view_menu:
+                    # The rename above matched no row: the dataset had no
+                    # permission under the old database name, so create one.
+                    logger.warning(
+                        "Could not find previous dataset permission %s",
+                        old_dataset_vm_name,
+                    )
+                    self._insert_pvm_on_sqla_event(
+                        mapper,
+                        connection,
+                        "datasource_access",
+                        new_dataset_vm_name,
+                    )
+                    continue
                 self.on_view_menu_after_update(
                     mapper,
                     connection,
@@ -4859,7 +4875,7 @@ def _delete_pvm_on_sqla_event(  # pylint: disable=too-many-arguments
 
     def _find_permission_on_sqla_event(
         self, connection: Connection, name: str
-    ) -> Permission:
+    ) -> Optional[Permission]:
         """
         Find a FAB Permission using a SQLA connection.
 
@@ -4868,13 +4884,15 @@ def _find_permission_on_sqla_event(
 
         :param connection: SQLAlchemy connection
         :param name: The permission name (it's unique)
-        :return: Permission
+        :return: Permission, or None if no row has that name
         """
         permission_table = self.permission_model.__table__  # pylint: disable=no-member
 
         permission_ = connection.execute(
             permission_table.select().where(permission_table.c.name == name)
         ).fetchone()
+        if permission_ is None:
+            return None
         permission = Permission()
         # ensures this object is never persisted
         permission.metadata = None
@@ -4884,7 +4902,7 @@ def _find_permission_on_sqla_event(
 
     def _find_view_menu_on_sqla_event(
         self, connection: Connection, name: str
-    ) -> ViewMenu:
+    ) -> Optional[ViewMenu]:
         """
         Find a FAB ViewMenu using a SQLA connection.
 
@@ -4893,13 +4911,15 @@ def _find_view_menu_on_sqla_event(
 
         :param connection: SQLAlchemy connection
         :param name: The ViewMenu name (it's unique)
-        :return: ViewMenu
+        :return: ViewMenu, or None if no row has that name
         """
         view_menu_table = self.viewmenu_model.__table__  # pylint: disable=no-member
 
         view_menu_ = connection.execute(
             view_menu_table.select().where(view_menu_table.c.name == name)
         ).fetchone()
+        if view_menu_ is None:
+            return None
         view_menu = ViewMenu()
         # ensures this object is never persisted
         view_menu.metadata = None
@@ -4943,10 +4963,20 @@ def _insert_pvm_on_sqla_event(
             permission = self._find_permission_on_sqla_event(
                 connection, permission_name
             )
+            if not permission:
+                logger.warning(
+                    "Could not read back inserted permission %s", permission_name
+                )
+                return
             self.on_permission_after_insert(mapper, connection, permission)
         if not view_menu:
             _ = connection.execute(view_menu_table.insert().values(name=view_menu_name))
             view_menu = self._find_view_menu_on_sqla_event(connection, view_menu_name)
+            if not view_menu:
+                logger.warning(
+                    "Could not read back inserted view menu %s", view_menu_name
+                )
+                return
             self.on_view_menu_after_insert(mapper, connection, view_menu)
         connection.execute(
             permission_view_table.insert().values(
@@ -4959,6 +4989,13 @@ def _insert_pvm_on_sqla_event(
                 permission_view_table.c.view_menu_id == view_menu.id,
             )
         ).fetchone()
+        if permission_view is None:
+            logger.warning(
+                "Could not read back inserted permission view %s on %s",
+                permission_name,
+                view_menu_name,
+            )
+            return
         permission_view_model = PermissionView()
         permission_view_model.metadata = None
         permission_view_model.id = permission_view.id

tests/unit_tests/security/database_rename_permissions_test.py

@@ -0,0 +1,148 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Database rename when the old database or dataset permission is missing."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+from sqlalchemy import inspect
+from sqlalchemy.orm.session import Session
+
+from superset import db, security_manager
+
+if TYPE_CHECKING:
+    from pytest_mock import MockerFixture
+
+    from superset.connectors.sqla.models import SqlaTable
+    from superset.models.core import Database
+
+
+def _create_database_with_dataset() -> tuple[Database, SqlaTable]:
+    from superset.connectors.sqla.models import SqlaTable
+    from superset.models.core import Database
+
+    SqlaTable.metadata.create_all(db.session.get_bind())  # pylint: disable=no-member
+
+    database = Database(database_name="examples", sqlalchemy_uri="sqlite://")
+    db.session.add(database)
+    db.session.flush()
+    dataset = SqlaTable(table_name="my_table", database=database)
+    db.session.add(dataset)
+    db.session.flush()
+    return database, dataset
+
+
+def _delete_view_menu(name: str) -> None:
+    """Delete a view menu and its permission views without touching the ORM."""
+    view_menu = security_manager.find_view_menu(name)
+    assert view_menu is not None
+    view_menu_table = security_manager.viewmenu_model.__table__
+    permission_view_table = security_manager.permissionview_model.__table__
+    db.session.execute(
+        permission_view_table.delete().where(
+            permission_view_table.c.view_menu_id == view_menu.id
+        )
+    )
+    db.session.execute(
+        view_menu_table.delete().where(view_menu_table.c.id == view_menu.id)
+    )
+
+
+@pytest.mark.parametrize(
+    "missing",
+    [
+        pytest.param({"database"}, id="database"),
+        pytest.param({"dataset"}, id="dataset"),
+        pytest.param({"database", "dataset"}, id="database-and-dataset"),
+    ],
+)
+def test_rename_database_with_missing_permission(
+    session: Session, missing: set[str]
+) -> None:
+    """
+    Renaming a database whose old database or dataset permission is missing
+    recreates the permission under the new name instead of failing with
+    ``'NoneType' object has no attribute 'id'``.
+    """
+    from superset.connectors.sqla.models import SqlaTable
+    from superset.models.slice import Slice
+
+    database, dataset = _create_database_with_dataset()
+    dataset_id = dataset.id
+    assert dataset.schema is None
+    chart = Slice(
+        slice_name="rename_chart",
+        datasource_id=dataset_id,
+        datasource_type="table",
+        viz_type="table",
+    )
+    db.session.add(chart)
+    db.session.flush()
+    chart_id = chart.id
+    assert chart.perm == f"[examples].[my_table](id:{dataset_id})"
+    if "database" in missing:
+        _delete_view_menu(f"[examples].(id:{database.id})")
+    if "dataset" in missing:
+        _delete_view_menu(f"[examples].[my_table](id:{dataset_id})")
+
+    database.database_name = "examples_renamed"
+    db.session.flush()
+
+    assert security_manager.find_permission_view_menu(
+        "database_access",
+        f"[examples_renamed].(id:{database.id})",
+    )
+    new_dataset_perm = f"[examples_renamed].[my_table](id:{dataset_id})"
+    assert security_manager.find_permission_view_menu(
+        "datasource_access", new_dataset_perm
+    )
+    assert (
+        security_manager.find_view_menu(f"[examples].[my_table](id:{dataset_id})")
+        is None
+    )
+    db.session.expire_all()
+    assert db.session.get(SqlaTable, dataset_id).perm == new_dataset_perm
+    assert db.session.get(Slice, chart_id).perm == new_dataset_perm
+
+
+def test_database_permission_update_skips_hook_when_view_menu_missing(
+    session: Session, mocker: MockerFixture
+) -> None:
+    """Do not invoke the update hook with a missing database view-menu readback."""
+    from superset.models.core import Database
+
+    database, _ = _create_database_with_dataset()
+    target = Database(id=database.id, database_name="examples_renamed")
+    connection = session.connection()
+    # A successful UPDATE is visible on the same connection; inject a missing
+    # readback to exercise the defensive guard independently of database behavior.
+    find_view_menu = mocker.patch.object(
+        security_manager, "_find_view_menu_on_sqla_event", return_value=None
+    )
+    update_hook = mocker.patch.object(security_manager, "on_view_menu_after_update")
+
+    result = security_manager._update_vm_database_access(
+        inspect(Database), connection, "examples", target
+    )
+
+    assert result is None
+    find_view_menu.assert_called_once_with(
+        connection, f"[examples_renamed].(id:{database.id})"
+    )
+    update_hook.assert_not_called()