fix: revert back to use security manager authz for dashboard when get…

… by uuid (#23330)

superset/dashboards/dao.py

@@ -22,9 +22,10 @@
 from flask_appbuilder.models.sqla.interface import SQLAInterface
 from sqlalchemy.exc import SQLAlchemyError
 
+from superset import security_manager
 from superset.dao.base import BaseDAO
 from superset.dashboards.commands.exceptions import DashboardNotFoundError
-from superset.dashboards.filters import DashboardAccessFilter
+from superset.dashboards.filters import DashboardAccessFilter, is_uuid
 from superset.extensions import db
 from superset.models.core import FavStar, FavStarClassName
 from superset.models.dashboard import Dashboard, id_or_slug_filter
@@ -41,21 +42,28 @@ class DashboardDAO(BaseDAO):
 
     @classmethod
     def get_by_id_or_slug(cls, id_or_slug: Union[int, str]) -> Dashboard:
-        query = (
-            db.session.query(Dashboard)
-            .filter(id_or_slug_filter(id_or_slug))
-            .outerjoin(Slice, Dashboard.slices)
-            .outerjoin(Slice.table)
-            .outerjoin(Dashboard.owners)
-            .outerjoin(Dashboard.roles)
-        )
-        # Apply dashboard base filters
-        query = cls.base_filter("id", SQLAInterface(Dashboard, db.session)).apply(
-            query, None
-        )
-        dashboard = query.one_or_none()
+        if is_uuid(id_or_slug):
+            # just get dashboard if it's uuid
+            dashboard = Dashboard.get(id_or_slug)
+        else:
+            query = (
+                db.session.query(Dashboard)
+                .filter(id_or_slug_filter(id_or_slug))
+                .outerjoin(Slice, Dashboard.slices)
+                .outerjoin(Slice.table)
+                .outerjoin(Dashboard.owners)
+                .outerjoin(Dashboard.roles)
+            )
+            # Apply dashboard base filters
+            query = cls.base_filter("id", SQLAInterface(Dashboard, db.session)).apply(
+                query, None
+            )
+            dashboard = query.one_or_none()
         if not dashboard:
             raise DashboardNotFoundError()
+
+        # make sure we still have basic access check from security manager
+        security_manager.raise_for_dashboard_access(dashboard)
         return dashboard
 
     @staticmethod

superset/dashboards/filters.py

@@ -14,8 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-import uuid
-from typing import Any, Optional, Union
+from typing import Any, Optional
 
 from flask import g
 from flask_appbuilder.security.sqla.models import Role
@@ -26,7 +25,7 @@
 from superset import db, is_feature_enabled, security_manager
 from superset.connectors.sqla.models import SqlaTable
 from superset.models.core import Database, FavStar
-from superset.models.dashboard import Dashboard
+from superset.models.dashboard import Dashboard, is_uuid
 from superset.models.embedded_dashboard import EmbeddedDashboard
 from superset.models.slice import Slice
 from superset.security.guest_token import GuestTokenResourceType, GuestUser
@@ -89,14 +88,6 @@ class DashboardTagFilter(BaseTagFilter):  # pylint: disable=too-few-public-metho
     model = Dashboard
 
 
-def is_uuid(value: Union[str, int]) -> bool:
-    try:
-        uuid.UUID(str(value))
-        return True
-    except ValueError:
-        return False
-
-
 class DashboardAccessFilter(BaseFilter):  # pylint: disable=too-few-public-methods
     """
     List dashboards with the following criteria:

superset/dashboards/permalink/commands/create.py

@@ -48,9 +48,9 @@ def __init__(
     def run(self) -> str:
         self.validate()
         try:
-            DashboardDAO.get_by_id_or_slug(self.dashboard_id)
+            dashboard = DashboardDAO.get_by_id_or_slug(self.dashboard_id)
             value = {
-                "dashboardId": self.dashboard_id,
+                "dashboardId": str(dashboard.uuid),
                 "state": self.state,
             }
             user_id = get_user_id()

superset/models/dashboard.py

@@ -18,6 +18,7 @@
 
 import json
 import logging
+import uuid
 from collections import defaultdict
 from functools import partial
 from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Type, Union
@@ -450,11 +451,27 @@ def get(cls, id_or_slug: Union[str, int]) -> Dashboard:
         return qry.one_or_none()
 
 
+def is_uuid(value: Union[str, int]) -> bool:
+    try:
+        uuid.UUID(str(value))
+        return True
+    except ValueError:
+        return False
+
+
+def is_int(value: Union[str, int]) -> bool:
+    try:
+        int(value)
+        return True
+    except ValueError:
+        return False
+
+
 def id_or_slug_filter(id_or_slug: Union[int, str]) -> BinaryExpression:
-    if isinstance(id_or_slug, int):
-        return Dashboard.id == id_or_slug
-    if id_or_slug.isdigit():
+    if is_int(id_or_slug):
         return Dashboard.id == int(id_or_slug)
+    if is_uuid(id_or_slug):
+        return Dashboard.uuid == uuid.UUID(str(id_or_slug))
     return Dashboard.slug == id_or_slug
 
 

superset/reports/commands/execute.py

@@ -179,15 +179,19 @@ def _get_url(
         dashboard_state = self._report_schedule.extra.get("dashboard")
         if dashboard_state:
             permalink_key = CreateDashboardPermalinkCommand(
-                dashboard_id=str(self._report_schedule.dashboard_id),
+                dashboard_id=str(self._report_schedule.dashboard.uuid),
                 state=dashboard_state,
             ).run()
             return get_url_path("Superset.dashboard_permalink", key=permalink_key)
 
+        dashboard = self._report_schedule.dashboard
+        dashboard_id_or_slug = (
+            dashboard.uuid if dashboard and dashboard.uuid else dashboard.id
+        )
         return get_url_path(
             "Superset.dashboard",
             user_friendly=user_friendly,
-            dashboard_id_or_slug=self._report_schedule.dashboard_id,
+            dashboard_id_or_slug=dashboard_id_or_slug,
             force=force,
             **kwargs,
         )

tests/integration_tests/dashboards/api_tests.py

@@ -505,6 +505,43 @@ def test_get_dashboard_no_data_access(self):
         db.session.delete(dashboard)
         db.session.commit()
 
+    def test_get_draft_dashboard_without_roles_by_uuid(self):
+        """
+        Dashboard API: Test get draft dashboard without roles by uuid
+        """
+        admin = self.get_user("admin")
+        dashboard = self.insert_dashboard("title", "slug1", [admin.id])
+        assert not dashboard.published
+        assert dashboard.roles == []
+
+        self.login(username="gamma")
+        uri = f"api/v1/dashboard/{dashboard.uuid}"
+        rv = self.client.get(uri)
+        assert rv.status_code == 200
+        # rollback changes
+        db.session.delete(dashboard)
+        db.session.commit()
+
+    def test_cannot_get_draft_dashboard_with_roles_by_uuid(self):
+        """
+        Dashboard API: Test get dashboard by uuid
+        """
+        admin = self.get_user("admin")
+        admin_role = self.get_role("Admin")
+        dashboard = self.insert_dashboard(
+            "title", "slug1", [admin.id], roles=[admin_role.id]
+        )
+        assert not dashboard.published
+        assert dashboard.roles == [admin_role]
+
+        self.login(username="gamma")
+        uri = f"api/v1/dashboard/{dashboard.uuid}"
+        rv = self.client.get(uri)
+        assert rv.status_code == 403
+        # rollback changes
+        db.session.delete(dashboard)
+        db.session.commit()
+
     def test_get_dashboards_changed_on(self):
         """
         Dashboard API: Test get dashboards changed on

tests/integration_tests/dashboards/permalink/api_tests.py

@@ -107,7 +107,8 @@ def test_get(test_client, login_as_admin, dashboard_id: int, permalink_salt: str
     resp = test_client.get(f"api/v1/dashboard/permalink/{key}")
     assert resp.status_code == 200
     result = resp.json
-    assert result["dashboardId"] == str(dashboard_id)
+    dashboard_uuid = result["dashboardId"]
+    assert Dashboard.get(dashboard_uuid).id == dashboard_id
     assert result["state"] == STATE
     id_ = decode_permalink_id(key, permalink_salt)
     db.session.query(KeyValueEntry).filter_by(id=id_).delete()