fix: Dashboard aware RBAC dataset permission (#24789)

superset/security/manager.py

@@ -24,7 +24,6 @@ from typing import Any, Callable, cast, NamedTuple, Optional, TYPE_CHECKING, Uni
 
 from flask import current_app, Flask, g, Request
 from flask_appbuilder import Model
-from flask_appbuilder.models.sqla.interface import SQLAInterface
 from flask_appbuilder.security.sqla.manager import SecurityManager
 from flask_appbuilder.security.sqla.models import (
     assoc_permissionview_role,
@@ -1781,7 +1780,7 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
 
         # pylint: disable=import-outside-toplevel
         from superset.connectors.sqla.models import SqlaTable
-        from superset.extensions import feature_flag_manager
+        from superset.daos.dashboard import DashboardDAO
         from superset.sql_parse import Table
 
         if database and table or query:
@@ -1827,25 +1826,26 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
                 )
 
         if datasource or query_context or viz:
+            form_data = None
+
             if query_context:
                 datasource = query_context.datasource
+                form_data = query_context.form_data
             elif viz:
                 datasource = viz.datasource
+                form_data = viz.form_data
 
             assert datasource
 
-            should_check_dashboard_access = (
-                feature_flag_manager.is_feature_enabled("DASHBOARD_RBAC")
-                or self.is_guest_user()
-            )
-
             if not (
                 self.can_access_schema(datasource)
                 or self.can_access("datasource_access", datasource.perm or "")
                 or self.is_owner(datasource)
                 or (
-                    should_check_dashboard_access
-                    and self.can_access_based_on_dashboard(datasource)
+                    form_data
+                    and (dashboard_id := form_data.get("dashboardId"))
+                    and (dashboard := DashboardDAO.find_by_id(dashboard_id))
+                    and self.can_access_dashboard(dashboard)
                 )
             ):
                 raise SupersetSecurityException(
@@ -2036,31 +2036,6 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
 
         raise DashboardAccessDeniedError()
 
-    @staticmethod
-    def can_access_based_on_dashboard(datasource: "BaseDatasource") -> bool:
-        # pylint: disable=import-outside-toplevel
-        from superset import db
-        from superset.dashboards.filters import DashboardAccessFilter
-        from superset.models.dashboard import Dashboard
-        from superset.models.slice import Slice
-
-        dashboard_ids = db.session.query(Dashboard.id)
-        dashboard_ids = DashboardAccessFilter(
-            "id", SQLAInterface(Dashboard, db.session)
-        ).apply(dashboard_ids, None)
-
-        datasource_class = type(datasource)
-        query = (
-            db.session.query(Dashboard.id)
-            .join(Slice, Dashboard.slices)
-            .join(Slice.table)
-            .filter(datasource_class.id == datasource.id)
-            .filter(Dashboard.id.in_(dashboard_ids))
-        )
-
-        exists = db.session.query(query.exists()).scalar()
-        return exists
-
     @staticmethod
     def _get_current_epoch_time() -> float:
         """This is used so the tests can mock time"""

tests/integration_tests/dashboards/security/security_rbac_tests.py

@@ -169,7 +169,6 @@ class TestDashboardRoleBasedSecurity(BaseTestDashboardSecurity):
             return
 
         # arrange
-
         username = random_str()
         new_role = f"role_{random_str()}"
         self.create_user_with_roles(username, [new_role], should_create_roles=True)
@@ -191,7 +190,7 @@ class TestDashboardRoleBasedSecurity(BaseTestDashboardSecurity):
 
         request_payload = get_query_context("birth_names")
         rv = self.post_assert_metric(CHART_DATA_URI, request_payload, "data")
-        self.assertEqual(rv.status_code, 200)
+        self.assertEqual(rv.status_code, 403)
 
         # post
         revoke_access_to_dashboard(dashboard_to_access, new_role)

tests/integration_tests/security/guest_token_security_tests.py

@@ -159,41 +159,6 @@ class TestGuestUserDashboardAccess(SupersetTestCase):
         has_guest_access = security_manager.has_guest_access(self.dash)
         self.assertFalse(has_guest_access)
 
-    def test_chart_raise_for_access_as_guest(self):
-        chart = self.dash.slices[0]
-        g.user = self.authorized_guest
-
-        security_manager.raise_for_access(viz=chart)
-
-    def test_chart_raise_for_access_as_unauthorized_guest(self):
-        chart = self.dash.slices[0]
-        g.user = self.unauthorized_guest
-
-        with self.assertRaises(SupersetSecurityException):
-            security_manager.raise_for_access(viz=chart)
-
-    def test_dataset_raise_for_access_as_guest(self):
-        dataset = self.dash.slices[0].datasource
-        g.user = self.authorized_guest
-
-        security_manager.raise_for_access(datasource=dataset)
-
-    def test_dataset_raise_for_access_as_unauthorized_guest(self):
-        dataset = self.dash.slices[0].datasource
-        g.user = self.unauthorized_guest
-
-        with self.assertRaises(SupersetSecurityException):
-            security_manager.raise_for_access(datasource=dataset)
-
-    def test_guest_token_does_not_grant_access_to_underlying_table(self):
-        sqla_table = self.dash.slices[0].table
-        table = Table(table=sqla_table.table_name)
-
-        g.user = self.authorized_guest
-
-        with self.assertRaises(Exception):
-            security_manager.raise_for_access(table=table, database=sqla_table.database)
-
     def test_raise_for_dashboard_access_as_guest(self):
         g.user = self.authorized_guest
 

tests/integration_tests/security_tests.py

@@ -47,6 +47,7 @@ from superset.utils.database import get_example_database
 from superset.utils.urls import get_url_host
 
 from .base_tests import SupersetTestCase
+from tests.integration_tests.conftest import with_feature_flags
 from tests.integration_tests.fixtures.public_role import (
     public_role_like_gamma,
     public_role_like_test_role,
@@ -1643,17 +1644,19 @@ class TestSecurityManager(SupersetTestCase):
         with self.assertRaises(SupersetSecurityException):
             security_manager.raise_for_access(query=query)
 
+    @patch("superset.security.manager.g")
     @patch("superset.security.SupersetSecurityManager.is_owner")
     @patch("superset.security.SupersetSecurityManager.can_access")
     @patch("superset.security.SupersetSecurityManager.can_access_schema")
     def test_raise_for_access_query_context(
-        self, mock_can_access_schema, mock_can_access, mock_is_owner
+        self, mock_can_access_schema, mock_can_access, mock_is_owner, mock_g
     ):
         query_context = Mock(datasource=self.get_datasource_mock())
 
         mock_can_access_schema.return_value = True
         security_manager.raise_for_access(query_context=query_context)
 
+        mock_g.user = security_manager.find_user("gamma")
         mock_can_access.return_value = False
         mock_can_access_schema.return_value = False
         mock_is_owner.return_value = False
@@ -1674,17 +1677,19 @@ class TestSecurityManager(SupersetTestCase):
         with self.assertRaises(SupersetSecurityException):
             security_manager.raise_for_access(database=database, table=table)
 
+    @patch("superset.security.manager.g")
     @patch("superset.security.SupersetSecurityManager.is_owner")
     @patch("superset.security.SupersetSecurityManager.can_access")
     @patch("superset.security.SupersetSecurityManager.can_access_schema")
     def test_raise_for_access_viz(
-        self, mock_can_access_schema, mock_can_access, mock_is_owner
+        self, mock_can_access_schema, mock_can_access, mock_is_owner, mock_g
     ):
         test_viz = viz.TimeTableViz(self.get_datasource_mock(), form_data={})
 
         mock_can_access_schema.return_value = True
         security_manager.raise_for_access(viz=test_viz)
 
+        mock_g.user = security_manager.find_user("gamma")
         mock_can_access.return_value = False
         mock_can_access_schema.return_value = False
         mock_is_owner.return_value = False
@@ -1692,6 +1697,44 @@ class TestSecurityManager(SupersetTestCase):
         with self.assertRaises(SupersetSecurityException):
             security_manager.raise_for_access(viz=test_viz)
 
+    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
+    @with_feature_flags(DASHBOARD_RBAC=True)
+    @patch("superset.security.manager.g")
+    @patch("superset.security.SupersetSecurityManager.is_owner")
+    @patch("superset.security.SupersetSecurityManager.can_access")
+    @patch("superset.security.SupersetSecurityManager.can_access_schema")
+    def test_raise_for_access_rbac(
+        self,
+        mock_can_access_schema,
+        mock_can_access,
+        mock_is_owner,
+        mock_g,
+    ):
+        dashboard = self.get_dash_by_slug("births")
+
+        obj = Mock(
+            datasource=self.get_datasource_mock(),
+            form_data={"dashboardId": dashboard.id},
+        )
+
+        mock_g.user = security_manager.find_user("gamma")
+        mock_is_owner.return_value = False
+        mock_can_access.return_value = False
+        mock_can_access_schema.return_value = False
+
+        for kwarg in ["query_context", "viz"]:
+            dashboard.roles = []
+            db.session.flush()
+
+            with self.assertRaises(SupersetSecurityException):
+                security_manager.raise_for_access(**{kwarg: obj})
+
+            dashboard.roles = [self.get_role("Gamma")]
+            db.session.flush()
+            security_manager.raise_for_access(**{kwarg: obj})
+
+        db.session.rollback()
+
     @patch("superset.security.manager.g")
     def test_get_user_roles(self, mock_g):
         admin = security_manager.find_user("admin")