fix: dashboard import validation (#26887)

superset/commands/dashboard/importers/v1/utils.py

@@ -157,7 +157,13 @@ def import_dashboard(
     )
     existing = session.query(Dashboard).filter_by(uuid=config["uuid"]).first()
     if existing:
-        if not overwrite or not can_write:
+        if overwrite and can_write and hasattr(g, "user") and g.user:
+            if not security_manager.can_access_dashboard(existing):
+                raise ImportFailedError(
+                    "A dashboard already exists and user doesn't "
+                    "have permissions to overwrite it"
+                )
+        elif not overwrite or not can_write:
             return existing
         config["id"] = existing.id
     elif not can_write:

tests/unit_tests/dashboards/commands/importers/v1/import_test.py

@@ -23,6 +23,7 @@
 from sqlalchemy.orm.session import Session
 
 from superset.commands.exceptions import ImportFailedError
+from superset.utils.core import override_user
 
 
 def test_import_dashboard(mocker: MockFixture, session: Session) -> None:
@@ -31,8 +32,6 @@ def test_import_dashboard(mocker: MockFixture, session: Session) -> None:
     """
     from superset import security_manager
     from superset.commands.dashboard.importers.v1.utils import import_dashboard
-    from superset.connectors.sqla.models import SqlaTable
-    from superset.models.core import Database
     from superset.models.slice import Slice
     from tests.integration_tests.fixtures.importexport import dashboard_config
 
@@ -48,6 +47,8 @@ def test_import_dashboard(mocker: MockFixture, session: Session) -> None:
     assert dashboard.description is None
     assert dashboard.is_managed_externally is False
     assert dashboard.external_url is None
+    # Assert that the can write to dashboard was checked
+    security_manager.can_access.assert_called_once_with("can_write", "Dashboard")
 
 
 def test_import_dashboard_managed_externally(
@@ -59,8 +60,6 @@ def test_import_dashboard_managed_externally(
     """
     from superset import security_manager
     from superset.commands.dashboard.importers.v1.utils import import_dashboard
-    from superset.connectors.sqla.models import SqlaTable
-    from superset.models.core import Database
     from superset.models.slice import Slice
     from tests.integration_tests.fixtures.importexport import dashboard_config
 
@@ -77,6 +76,9 @@ def test_import_dashboard_managed_externally(
     assert dashboard.is_managed_externally is True
     assert dashboard.external_url == "https://example.org/my_dashboard"
 
+    # Assert that the can write to dashboard was checked
+    security_manager.can_access.assert_called_once_with("can_write", "Dashboard")
+
 
 def test_import_dashboard_without_permission(
     mocker: MockFixture,
@@ -87,8 +89,6 @@ def test_import_dashboard_without_permission(
     """
     from superset import security_manager
     from superset.commands.dashboard.importers.v1.utils import import_dashboard
-    from superset.connectors.sqla.models import SqlaTable
-    from superset.models.core import Database
     from superset.models.slice import Slice
     from tests.integration_tests.fixtures.importexport import dashboard_config
 
@@ -105,3 +105,101 @@ def test_import_dashboard_without_permission(
         str(excinfo.value)
         == "Dashboard doesn't exist and user doesn't have permission to create dashboards"
     )
+
+    # Assert that the can write to dashboard was checked
+    security_manager.can_access.assert_called_once_with("can_write", "Dashboard")
+
+
+def test_import_existing_dashboard_without_permission(
+    mocker: MockFixture,
+    session: Session,
+) -> None:
+    """
+    Test importing a dashboard when a user doesn't have permissions to create.
+    """
+    from superset import security_manager
+    from superset.commands.dashboard.importers.v1.utils import g, import_dashboard
+    from superset.models.dashboard import Dashboard
+    from superset.models.slice import Slice
+    from tests.integration_tests.fixtures.importexport import dashboard_config
+
+    mocker.patch.object(security_manager, "can_access", return_value=True)
+    mocker.patch.object(security_manager, "can_access_dashboard", return_value=False)
+    mock_g = mocker.patch(
+        "superset.commands.dashboard.importers.v1.utils.g"
+    )  # Replace with the actual path to g
+    mock_g.user = mocker.MagicMock(return_value=True)
+
+    engine = session.get_bind()
+    Slice.metadata.create_all(engine)  # pylint: disable=no-member
+    Dashboard.metadata.create_all(engine)  # pylint: disable=no-member
+
+    dashboard_obj = Dashboard(
+        id=100,
+        dashboard_title="Test dash",
+        slug=None,
+        slices=[],
+        published=True,
+        uuid="c4b28c4e-a1fe-4cf8-a5ac-d6f11d6fdd51",
+    )
+    session.add(dashboard_obj)
+    session.flush()
+    config = copy.deepcopy(dashboard_config)
+
+    with pytest.raises(ImportFailedError) as excinfo:
+        import_dashboard(session, config, overwrite=True)
+    assert (
+        str(excinfo.value)
+        == "A dashboard already exists and user doesn't have permissions to overwrite it"
+    )
+
+    # Assert that the can write to dashboard was checked
+    security_manager.can_access.assert_called_once_with("can_write", "Dashboard")
+    security_manager.can_access_dashboard.assert_called_once_with(dashboard_obj)
+
+
+def test_import_existing_dashboard_with_permission(
+    mocker: MockFixture,
+    session: Session,
+) -> None:
+    """
+    Test importing a dashboard when a user doesn't have permissions to create.
+    """
+    from flask_appbuilder.security.sqla.models import Role, User
+
+    from superset import security_manager
+    from superset.commands.dashboard.importers.v1.utils import import_dashboard
+    from superset.models.dashboard import Dashboard
+    from tests.integration_tests.fixtures.importexport import dashboard_config
+
+    mocker.patch.object(security_manager, "can_access", return_value=True)
+    mocker.patch.object(security_manager, "can_access_dashboard", return_value=True)
+
+    engine = session.get_bind()
+    Dashboard.metadata.create_all(engine)  # pylint: disable=no-member
+
+    admin = User(
+        first_name="Alice",
+        last_name="Doe",
+        email="[email protected]",
+        username="admin",
+        roles=[Role(name="Admin")],
+    )
+
+    dashboard_obj = Dashboard(
+        id=100,
+        dashboard_title="Test dash",
+        slug=None,
+        slices=[],
+        published=True,
+        uuid="c4b28c4e-a1fe-4cf8-a5ac-d6f11d6fdd51",
+    )
+    session.add(dashboard_obj)
+    session.flush()
+    config = copy.deepcopy(dashboard_config)
+
+    with override_user(admin):
+        import_dashboard(session, config, overwrite=True)
+    # Assert that the can write to dashboard was checked
+    security_manager.can_access.assert_called_once_with("can_write", "Dashboard")
+    security_manager.can_access_dashboard.assert_called_once_with(dashboard_obj)