fix: chart import validation

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

@@ -55,6 +55,12 @@ def import_chart(
     can_write = ignore_permissions or security_manager.can_access("can_write", "Chart")
     existing = session.query(Slice).filter_by(uuid=config["uuid"]).first()
     if existing:
+        if overwrite and can_write and hasattr(g, "user") and g.user:
+            if not security_manager.can_access_chart(existing):
+                raise ImportFailedError(
+                    "A chart already exists and user doesn't "
+                    "have permissions to overwrite it"
+                )
         if not overwrite or not can_write:
             return existing
         config["id"] = existing.id

superset/errors.py

@@ -66,6 +66,7 @@ class SupersetErrorType(StrEnum):
     MISSING_OWNERSHIP_ERROR = "MISSING_OWNERSHIP_ERROR"
     USER_ACTIVITY_SECURITY_ACCESS_ERROR = "USER_ACTIVITY_SECURITY_ACCESS_ERROR"
     DASHBOARD_SECURITY_ACCESS_ERROR = "DASHBOARD_SECURITY_ACCESS_ERROR"
+    CHART_SECURITY_ACCESS_ERROR = "CHART_SECURITY_ACCESS_ERROR"
 
     # Other errors
     BACKEND_TIMEOUT_ERROR = "BACKEND_TIMEOUT_ERROR"

superset/security/manager.py

@@ -85,6 +85,7 @@
     )
     from superset.models.core import Database
     from superset.models.dashboard import Dashboard
+    from superset.models.slice import Slice
     from superset.models.sql_lab import Query
     from superset.sql_parse import Table
     from superset.viz import BaseViz
@@ -422,6 +423,19 @@
 
         return True
 
+    def can_access_chart(self, chart: "Slice") -> bool:
+        """
+        Return True if the user can access the specified chart, False otherwise.
+        :param chart: The chart
+        :return: Whether the user can access the chart
+        """
+        try:
+            self.raise_for_access(chart=chart)
+        except SupersetSecurityException:
+            return False
+
+        return True
+
     def get_dashboard_access_error_object(  # pylint: disable=invalid-name
         self,
         dashboard: "Dashboard",  # pylint: disable=unused-argument
@@ -439,6 +453,23 @@
             level=ErrorLevel.ERROR,
         )
 
+    def get_chart_access_error_object(  # pylint: disable=invalid-name
+        self,
+        dashboard: "Dashboard",  # pylint: disable=unused-argument
+    ) -> SupersetError:
+        """
+        Return the error object for the denied Superset dashboard.
+
+        :param dashboard: The denied Superset dashboard
+        :returns: The error object
+        """
+
+        return SupersetError(
+            error_type=SupersetErrorType.CHART_SECURITY_ACCESS_ERROR,
+            message="You don't have access to this chart.",
+            level=ErrorLevel.ERROR,
+        )
+
     @staticmethod
     def get_datasource_access_error_msg(datasource: "BaseDatasource") -> str:
         """
@@ -1822,6 +1853,7 @@
         # pylint: disable=too-many-arguments,too-many-branches,too-many-locals,too-many-statements
         self,
         dashboard: Optional["Dashboard"] = None,
+        chart: Optional["Slice"] = None,
         database: Optional["Database"] = None,
         datasource: Optional["BaseDatasource"] = None,
         query: Optional["Query"] = None,
@@ -2044,6 +2076,15 @@
                 self.get_dashboard_access_error_object(dashboard)
             )
 
+        if chart:
+            if self.is_admin() or self.is_owner(chart):
+                return
+
+            if chart.datasource and self.can_access_datasource(chart.datasource):
+                return
+
+            raise SupersetSecurityException(self.get_chart_access_error_object(chart))
+
     def get_user_by_username(self, username: str) -> Optional[User]:
         """
         Retrieves a user by it's username case sensitive. Optional session parameter

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

@@ -17,104 +17,130 @@
 # pylint: disable=unused-argument, import-outside-toplevel, unused-import, invalid-name
 
 import copy
+from collections.abc import Generator
 
 import pytest
+from flask_appbuilder.security.sqla.models import Role, User
 from pytest_mock import MockFixture
 from sqlalchemy.orm.session import Session
 
+from superset import security_manager
+from superset.commands.chart.importers.v1.utils import import_chart
 from superset.commands.exceptions import ImportFailedError
+from superset.connectors.sqla.models import Database, SqlaTable
+from superset.models.slice import Slice
+from superset.utils.core import override_user
+from tests.integration_tests.fixtures.importexport import chart_config
 
 
-def test_import_chart(mocker: MockFixture, session: Session) -> None:
+@pytest.fixture
+def session_with_data(session: Session) -> Generator[Session, None, None]:
+    engine = session.get_bind()
+    SqlaTable.metadata.create_all(engine)  # pylint: disable=no-member
+
+    dataset = SqlaTable(
+        table_name="test_table",
+        metrics=[],
+        main_dttm_col=None,
+        database=Database(database_name="my_database", sqlalchemy_uri="sqlite://"),
+    )
+    session.add(dataset)
+    session.flush()
+    slice = Slice(
+        id=1,
+        datasource_id=dataset.id,
+        datasource_type="table",
+        datasource_name="tmp_perm_table",
+        slice_name="slice_name",
+        uuid=chart_config["uuid"],
+    )
+    session.add(slice)
+    session.flush()
+
+    yield session
+    session.rollback()
+
+
+@pytest.fixture
+def session_with_schema(session: Session) -> Generator[Session, None, None]:
+    from superset.connectors.sqla.models import SqlaTable
+
+    engine = session.get_bind()
+    SqlaTable.metadata.create_all(engine)  # pylint: disable=no-member
+
+    yield session
+
+
+def test_import_chart(mocker: MockFixture, session_with_schema: Session) -> None:
     """
     Test importing a chart.
     """
-    from superset import security_manager
-    from superset.commands.chart.importers.v1.utils import import_chart
-    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 chart_config
 
     mocker.patch.object(security_manager, "can_access", return_value=True)
 
-    engine = session.get_bind()
-    Slice.metadata.create_all(engine)  # pylint: disable=no-member
-
     config = copy.deepcopy(chart_config)
     config["datasource_id"] = 1
     config["datasource_type"] = "table"
 
-    chart = import_chart(session, config)
+    chart = import_chart(session_with_schema, config)
     assert chart.slice_name == "Deck Path"
     assert chart.viz_type == "deck_path"
     assert chart.is_managed_externally is False
     assert chart.external_url is None
 
+    # Assert that the can write to chart was checked
+    security_manager.can_access.assert_called_once_with("can_write", "Chart")
 
-def test_import_chart_managed_externally(mocker: MockFixture, session: Session) -> None:
+
+def test_import_chart_managed_externally(
+    mocker: MockFixture, session_with_schema: Session
+) -> None:
     """
     Test importing a chart that is managed externally.
     """
-    from superset import security_manager
-    from superset.commands.chart.importers.v1.utils import import_chart
-    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 chart_config
-
     mocker.patch.object(security_manager, "can_access", return_value=True)
 
-    engine = session.get_bind()
-    Slice.metadata.create_all(engine)  # pylint: disable=no-member
-
     config = copy.deepcopy(chart_config)
     config["datasource_id"] = 1
     config["datasource_type"] = "table"
     config["is_managed_externally"] = True
     config["external_url"] = "https://example.org/my_chart"
 
-    chart = import_chart(session, config)
+    chart = import_chart(session_with_schema, config)
     assert chart.is_managed_externally is True
     assert chart.external_url == "https://example.org/my_chart"
 
+    # Assert that the can write to chart was checked
+    security_manager.can_access.assert_called_once_with("can_write", "Chart")
+
 
 def test_import_chart_without_permission(
     mocker: MockFixture,
-    session: Session,
+    session_with_schema: Session,
 ) -> None:
     """
     Test importing a chart when a user doesn't have permissions to create.
     """
-    from superset import security_manager
-    from superset.commands.chart.importers.v1.utils import import_chart
-    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 chart_config
-
     mocker.patch.object(security_manager, "can_access", return_value=False)
 
-    engine = session.get_bind()
-    Slice.metadata.create_all(engine)  # pylint: disable=no-member
-
     config = copy.deepcopy(chart_config)
     config["datasource_id"] = 1
     config["datasource_type"] = "table"
 
     with pytest.raises(ImportFailedError) as excinfo:
-        import_chart(session, config)
+        import_chart(session_with_schema, config)
     assert (
         str(excinfo.value)
         == "Chart doesn't exist and user doesn't have permission to create charts"
     )
+    # Assert that the can write to chart was checked
+    security_manager.can_access.assert_called_once_with("can_write", "Chart")
 
 
-def test_filter_chart_annotations(mocker: MockFixture, session: Session) -> None:
+def test_filter_chart_annotations(session: Session) -> None:
     """
     Test importing a chart.
     """
-    from superset import security_manager
     from superset.commands.chart.importers.v1.utils import filter_chart_annotations
     from tests.integration_tests.fixtures.importexport import (
         chart_config_with_mixed_annotations,
@@ -127,3 +153,70 @@ def test_filter_chart_annotations(mocker: MockFixture, session: Session) -> None
 
     assert len(annotation_layers) == 1
     assert all([al["annotationType"] == "FORMULA" for al in annotation_layers])
+
+
+def test_import_existing_chart_without_permission(
+    mocker: MockFixture,
+    session_with_data: Session,
+) -> None:
+    """
+    Test importing a chart when a user doesn't have permissions to modify.
+    """
+    mocker.patch.object(security_manager, "can_access", return_value=True)
+    mocker.patch.object(security_manager, "can_access_chart", return_value=False)
+    mock_g = mocker.patch(
+        "superset.commands.chart.importers.v1.utils.g"
+    )  # Replace with the actual path to g
+    mock_g.user = mocker.MagicMock(return_value=True)
+
+    slice = (
+        session_with_data.query(Slice)
+        .filter(Slice.uuid == chart_config["uuid"])
+        .one_or_none()
+    )
+
+    with pytest.raises(ImportFailedError) as excinfo:
+        import_chart(session_with_data, chart_config, overwrite=True)
+    assert (
+        str(excinfo.value)
+        == "A chart already exists and user doesn't have permissions to overwrite it"
+    )
+
+    # Assert that the can write to chart was checked
+    security_manager.can_access.assert_called_once_with("can_write", "Chart")
+    security_manager.can_access_chart.assert_called_once_with(slice)
+
+
+def test_import_existing_chart_with_permission(
+    mocker: MockFixture,
+    session_with_data: Session,
+) -> None:
+    """
+    Test importing a chart that exists when a user has access permission to that chart.
+    """
+    mocker.patch.object(security_manager, "can_access", return_value=True)
+    mocker.patch.object(security_manager, "can_access_chart", return_value=True)
+
+    admin = User(
+        first_name="Alice",
+        last_name="Doe",
+        email="adoe@example.org",
+        username="admin",
+        roles=[Role(name="Admin")],
+    )
+
+    config = copy.deepcopy(chart_config)
+    config["datasource_id"] = 1
+    config["datasource_type"] = "table"
+
+    slice = (
+        session_with_data.query(Slice)
+        .filter(Slice.uuid == config["uuid"])
+        .one_or_none()
+    )
+
+    with override_user(admin):
+        import_chart(session_with_data, config, overwrite=True)
+    # Assert that the can write to chart was checked
+    security_manager.can_access.assert_called_once_with("can_write", "Chart")
+    security_manager.can_access_chart.assert_called_once_with(slice)