diff --git a/superset/models/dashboard.py b/superset/models/dashboard.py index a63594371e910..f827cd42c8e33 100644 --- a/superset/models/dashboard.py +++ b/superset/models/dashboard.py @@ -22,7 +22,6 @@ from typing import Any, Callable, Dict, List, Set, Union import sqlalchemy as sqla -from flask import g from flask_appbuilder import Model from flask_appbuilder.models.decorators import renders from flask_appbuilder.security.sqla.models import User @@ -48,7 +47,6 @@ from superset.connectors.base.models import BaseDatasource from superset.connectors.druid.models import DruidColumn, DruidMetric from superset.connectors.sqla.models import SqlMetric, TableColumn -from superset.dashboards.commands.exceptions import DashboardAccessDeniedError from superset.extensions import cache_manager from superset.models.helpers import AuditMixinNullable, ImportExportMixin from superset.models.slice import Slice @@ -422,22 +420,3 @@ def clear_dashboard_cache( sqla.event.listen(TableColumn, "after_update", clear_dashboard_cache) sqla.event.listen(DruidMetric, "after_update", clear_dashboard_cache) sqla.event.listen(DruidColumn, "after_update", clear_dashboard_cache) - - -def raise_for_dashboard_access(dashboard: Dashboard) -> None: - from superset.views.base import get_user_roles, is_user_admin - from superset.views.utils import is_owner - - if is_feature_enabled("DASHBOARD_RBAC"): - has_rbac_access = any( - dashboard_role.id in [user_role.id for user_role in get_user_roles()] - for dashboard_role in dashboard.roles - ) - can_access = ( - is_user_admin() - or is_owner(dashboard, g.user) - or (dashboard.published and has_rbac_access) - ) - - if not can_access: - raise DashboardAccessDeniedError() diff --git a/superset/security/manager.py b/superset/security/manager.py index 2bf8cd01d12a1..1c4419cfc5241 100644 --- a/superset/security/manager.py +++ b/superset/security/manager.py @@ -55,6 +55,7 @@ from superset.common.query_context import QueryContext from superset.connectors.base.models import BaseDatasource from superset.connectors.druid.models import DruidCluster + from superset.models.dashboard import Dashboard from superset.models.core import Database from superset.models.sql_lab import Query from superset.sql_parse import Table @@ -1097,3 +1098,30 @@ def get_rls_ids(self, table: "BaseDatasource") -> List[int]: ids = [f.id for f in self.get_rls_filters(table)] ids.sort() # Combinations rather than permutations return ids + + # pylint: disable=no-self-use + def raise_for_dashboard_access(self, dashboard: "Dashboard") -> None: + """ + Raise an exception if the user cannot access the dashboard. + + :param dashboard: Dashboard the user wants access to + :raises DashboardAccessDeniedError: If the user cannot access the resource + """ + from superset.dashboards.commands.exceptions import DashboardAccessDeniedError + from superset.views.base import get_user_roles, is_user_admin + from superset.views.utils import is_owner + from superset import is_feature_enabled + + if is_feature_enabled("DASHBOARD_RBAC"): + has_rbac_access = any( + dashboard_role.id in [user_role.id for user_role in get_user_roles()] + for dashboard_role in dashboard.roles + ) + can_access = ( + is_user_admin() + or is_owner(dashboard, g.user) + or (dashboard.published and has_rbac_access) + ) + + if not can_access: + raise DashboardAccessDeniedError() diff --git a/superset/utils/decorators.py b/superset/utils/decorators.py index e5cf713696fdd..27abab30e9a74 100644 --- a/superset/utils/decorators.py +++ b/superset/utils/decorators.py @@ -19,7 +19,7 @@ from typing import Any, Callable, Dict, Iterator, Union from contextlib2 import contextmanager -from flask import Response +from flask import current_app, Response from superset import is_feature_enabled from superset.dashboards.commands.exceptions import DashboardAccessDeniedError @@ -87,15 +87,12 @@ def check_dashboard_access( def decorator(f: Callable[..., Any]) -> Callable[..., Any]: @wraps(f) def wrapper(self: Any, *args: Any, **kwargs: Any) -> Any: - from superset.models.dashboard import ( - Dashboard, - raise_for_dashboard_access, - ) + from superset.models.dashboard import Dashboard dashboard = Dashboard.get(str(kwargs["dashboard_id_or_slug"])) if is_feature_enabled("DASHBOARD_RBAC"): try: - raise_for_dashboard_access(dashboard) + current_app.appbuilder.sm.raise_for_dashboard_access(dashboard) except DashboardAccessDeniedError as ex: return on_error(self, ex) except Exception as exception: