# Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. """Unit tests for Superset""" import json import prison import pytest from flask import escape from superset import app from superset.dashboards.dao import DashboardDAO from superset.models import core as models from tests.integration_tests.dashboards.base_case import DashboardTestCase from tests.integration_tests.dashboards.consts import * from tests.integration_tests.dashboards.dashboard_test_utils import * from tests.integration_tests.dashboards.superset_factory_util import * from tests.integration_tests.fixtures.energy_dashboard import ( load_energy_table_data, load_energy_table_with_slice, ) class TestDashboardDatasetSecurity(DashboardTestCase): @pytest.fixture def load_dashboard(self): with app.app_context(): table = ( db.session.query(SqlaTable).filter_by(table_name="energy_usage").one() ) # get a slice from the allowed table slice = db.session.query(Slice).filter_by(slice_name="Energy Sankey").one() self.grant_public_access_to_table(table) pytest.hidden_dash_slug = f"hidden_dash_{random_slug()}" pytest.published_dash_slug = f"published_dash_{random_slug()}" # Create a published and hidden dashboard and add them to the database published_dash = Dashboard() published_dash.dashboard_title = "Published Dashboard" published_dash.slug = pytest.published_dash_slug published_dash.slices = [slice] published_dash.published = True hidden_dash = Dashboard() hidden_dash.dashboard_title = "Hidden Dashboard" hidden_dash.slug = pytest.hidden_dash_slug hidden_dash.slices = [slice] hidden_dash.published = False db.session.merge(published_dash) db.session.merge(hidden_dash) yield db.session.commit() self.revoke_public_access_to_table(table) db.session.delete(published_dash) db.session.delete(hidden_dash) db.session.commit() def test_dashboard_access__admin_can_access_all(self): # arrange self.login(username=ADMIN_USERNAME) dashboard_title_by_url = { dash.url: dash.dashboard_title for dash in get_all_dashboards() } # act responses_by_url = { url: self.client.get(url) for url in dashboard_title_by_url.keys() } # assert for dashboard_url, get_dashboard_response in responses_by_url.items(): self.assert200(get_dashboard_response) def test_get_dashboards__users_are_dashboards_owners(self): # arrange username = "gamma" user = security_manager.find_user(username) my_owned_dashboard = create_dashboard_to_db( dashboard_title="My Dashboard", published=False, owners=[user], ) not_my_owned_dashboard = create_dashboard_to_db( dashboard_title="Not My Dashboard", published=False, ) self.login(user.username) # act get_dashboards_response = self.get_resp(DASHBOARDS_API_URL) # assert self.assertIn(my_owned_dashboard.url, get_dashboards_response) self.assertNotIn(not_my_owned_dashboard.url, get_dashboards_response) def test_get_dashboards__owners_can_view_empty_dashboard(self): # arrange dash = create_dashboard_to_db("Empty Dashboard", slug="empty_dashboard") dashboard_url = dash.url gamma_user = security_manager.find_user("gamma") self.login(gamma_user.username) # act get_dashboards_response = self.get_resp(DASHBOARDS_API_URL) # assert self.assertNotIn(dashboard_url, get_dashboards_response) def test_get_dashboards__users_can_view_favorites_dashboards(self): # arrange user = security_manager.find_user("gamma") fav_dash_slug = f"my_favorite_dash_{random_slug()}" regular_dash_slug = f"regular_dash_{random_slug()}" favorite_dash = Dashboard() favorite_dash.dashboard_title = "My Favorite Dashboard" favorite_dash.slug = fav_dash_slug regular_dash = Dashboard() regular_dash.dashboard_title = "A Plain Ol Dashboard" regular_dash.slug = regular_dash_slug db.session.add(favorite_dash) db.session.add(regular_dash) db.session.commit() dash = db.session.query(Dashboard).filter_by(slug=fav_dash_slug).first() favorites = models.FavStar() favorites.obj_id = dash.id favorites.class_name = "Dashboard" favorites.user_id = user.id db.session.add(favorites) db.session.commit() self.login(user.username) # act get_dashboards_response = self.get_resp(DASHBOARDS_API_URL) # cleanup db.session.delete(favorites) db.session.delete(favorite_dash) db.session.delete(regular_dash) db.session.commit() # assert self.assertIn(f"/superset/dashboard/{fav_dash_slug}/", get_dashboards_response) def test_get_dashboards__user_can_not_view_unpublished_dash(self): # arrange admin_user = security_manager.find_user(ADMIN_USERNAME) gamma_user = security_manager.find_user(GAMMA_USERNAME) admin_and_draft_dashboard = create_dashboard_to_db( dashboard_title="admin_owned_unpublished_dash", owners=[admin_user] ) self.login(gamma_user.username) # act - list dashboards as a gamma user get_dashboards_response_as_gamma = self.get_resp(DASHBOARDS_API_URL) # assert self.assertNotIn( admin_and_draft_dashboard.url, get_dashboards_response_as_gamma ) @pytest.mark.usefixtures("load_energy_table_with_slice", "load_dashboard") def test_get_dashboards__users_can_view_permitted_dashboard(self): # arrange username = random_str() new_role = f"role_{random_str()}" self.create_user_with_roles(username, [new_role], should_create_roles=True) accessed_table = get_sql_table_by_name("energy_usage") self.grant_role_access_to_table(accessed_table, new_role) # get a slice from the allowed table slice_to_add_to_dashboards = get_slice_by_name("Energy Sankey") # Create a published and hidden dashboard and add them to the database first_dash = create_dashboard_to_db( dashboard_title="Published Dashboard", published=True, slices=[slice_to_add_to_dashboards], ) second_dash = create_dashboard_to_db( dashboard_title="Hidden Dashboard", published=True, slices=[slice_to_add_to_dashboards], ) try: self.login(username) # act get_dashboards_response = self.get_resp(DASHBOARDS_API_URL) # assert self.assertIn(second_dash.url, get_dashboards_response) self.assertIn(first_dash.url, get_dashboards_response) finally: self.revoke_public_access_to_table(accessed_table) def test_get_dashboards_api_no_data_access(self): """ Dashboard API: Test get dashboards no data access """ admin = self.get_user("admin") title = f"title{random_str()}" dashboard = create_dashboard_to_db(title, "slug1", owners=[admin]) self.login(username="gamma") arguments = { "filters": [{"col": "dashboard_title", "opr": "sw", "value": title[0:8]}] } uri = DASHBOARDS_API_URL_WITH_QUERY_FORMAT.format(prison.dumps(arguments)) rv = self.client.get(uri) self.assert200(rv) data = json.loads(rv.data.decode("utf-8")) self.assertEqual(0, data["count"]) DashboardDAO.delete(dashboard)