feat(dashboard-rbac): dashboard lists (#12680)

This commit is contained in:
Amit Miran
2021-01-31 09:17:46 +02:00
committed by GitHub
parent 2b9d5795ad
commit 9a7fba810e
15 changed files with 1345 additions and 30 deletions

View File

@@ -0,0 +1,16 @@
# 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.

View File

@@ -0,0 +1,86 @@
# 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.
from typing import List, Optional
from flask import escape, Response
from superset.models.dashboard import Dashboard
from tests.dashboards.base_case import DashboardTestCase
class BaseTestDashboardSecurity(DashboardTestCase):
def tearDown(self) -> None:
self.clean_created_objects()
def assert_dashboard_view_response(
self, response: Response, dashboard_to_access: Dashboard
) -> None:
self.assert200(response)
assert escape(dashboard_to_access.dashboard_title) in response.data.decode(
"utf-8"
)
def assert_dashboard_api_response(
self, response: Response, dashboard_to_access: Dashboard
) -> None:
self.assert200(response)
assert response.json["id"] == dashboard_to_access.id
def assert_dashboards_list_view_response(
self,
response: Response,
expected_counts: int,
expected_dashboards: Optional[List[Dashboard]] = None,
not_expected_dashboards: Optional[List[Dashboard]] = None,
) -> None:
self.assert200(response)
response_html = response.data.decode("utf-8")
if expected_counts == 0:
assert "No records found" in response_html
else:
# # a way to parse number of dashboards returns
# in the list view as an html response
assert (
"Record Count:</strong> {count}".format(count=str(expected_counts))
in response_html
)
expected_dashboards = expected_dashboards or []
for dashboard in expected_dashboards:
assert dashboard.url in response_html
not_expected_dashboards = not_expected_dashboards or []
for dashboard in not_expected_dashboards:
assert dashboard.url not in response_html
def assert_dashboards_api_response(
self,
response: Response,
expected_counts: int,
expected_dashboards: Optional[List[Dashboard]] = None,
not_expected_dashboards: Optional[List[Dashboard]] = None,
) -> None:
self.assert200(response)
response_data = response.json
assert response_data["count"] == expected_counts
response_dashboards_url = set(
map(lambda dash: dash["url"], response_data["result"])
)
expected_dashboards = expected_dashboards or []
for dashboard in expected_dashboards:
assert dashboard.url in response_dashboards_url
not_expected_dashboards = not_expected_dashboards or []
for dashboard in not_expected_dashboards:
assert dashboard.url not in response_dashboards_url

View File

@@ -0,0 +1,241 @@
# 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.models import core as models
from tests.dashboards.base_case import DashboardTestCase
from tests.dashboards.consts import *
from tests.dashboards.dashboard_test_utils import *
from tests.dashboards.superset_factory_util import *
from tests.fixtures.energy_dashboard import 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).data.decode("utf-8")
for url in dashboard_title_by_url.keys()
}
# assert
for dashboard_url, get_dashboard_response in responses_by_url.items():
assert (
escape(dashboard_title_by_url[dashboard_url]) in 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.merge(favorite_dash)
db.session.merge(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.merge(favorites)
db.session.commit()
self.login(user.username)
# act
get_dashboards_response = self.get_resp(DASHBOARDS_API_URL)
# 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_not_published_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_not_published_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_dashboard_api_no_data_access(self):
"""
Dashboard API: Test get dashboard without data access
"""
admin = self.get_user("admin")
dashboard = create_dashboard_to_db(
random_title(), random_slug(), owners=[admin]
)
self.login(username="gamma")
uri = DASHBOARD_API_URL_FORMAT.format(dashboard.id)
rv = self.client.get(uri)
self.assert404(rv)
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()}"
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"])

View File

@@ -0,0 +1,308 @@
# 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"""
from unittest import mock
from tests.dashboards.dashboard_test_utils import *
from tests.dashboards.security.base_case import BaseTestDashboardSecurity
from tests.dashboards.superset_factory_util import (
create_dashboard_to_db,
create_database_to_db,
create_datasource_table_to_db,
create_slice_to_db,
)
@mock.patch.dict(
"superset.extensions.feature_flag_manager._feature_flags", DASHBOARD_RBAC=True,
)
class TestDashboardRoleBasedSecurity(BaseTestDashboardSecurity):
def test_get_dashboards_list__admin_get_all_dashboards(self):
# arrange
create_dashboard_to_db(
owners=[], slices=[create_slice_to_db()], published=False
)
dashboard_counts = count_dashboards()
self.login("admin")
# act
response = self.get_dashboards_list_response()
# assert
self.assert_dashboards_list_view_response(response, dashboard_counts)
def test_get_dashboards_list__owner_get_all_owned_dashboards(self):
# arrange
username = random_str()
new_role = f"role_{random_str()}"
owner = self.create_user_with_roles(
username, [new_role], should_create_roles=True
)
database = create_database_to_db()
table = create_datasource_table_to_db(db_id=database.id, owners=[owner])
first_dash = create_dashboard_to_db(
owners=[owner], slices=[create_slice_to_db(datasource_id=table.id)]
)
second_dash = create_dashboard_to_db(
owners=[owner], slices=[create_slice_to_db(datasource_id=table.id)]
)
owned_dashboards = [first_dash, second_dash]
not_owned_dashboards = [
create_dashboard_to_db(
slices=[create_slice_to_db(datasource_id=table.id)], published=True
)
]
self.login(username)
# act
response = self.get_dashboards_list_response()
# assert
self.assert_dashboards_list_view_response(
response, 2, owned_dashboards, not_owned_dashboards
)
def test_get_dashboards_list__user_without_any_permissions_get_empty_list(self):
# arrange
username = random_str()
new_role = f"role_{random_str()}"
self.create_user_with_roles(username, [new_role], should_create_roles=True)
create_dashboard_to_db(published=True)
self.login(username)
# act
response = self.get_dashboards_list_response()
# assert
self.assert_dashboards_list_view_response(response, 0)
def test_get_dashboards_list__user_get_only_published_permitted_dashboards(self):
# arrange
username = random_str()
new_role = f"role_{random_str()}"
self.create_user_with_roles(username, [new_role], should_create_roles=True)
published_dashboards = [
create_dashboard_to_db(published=True),
create_dashboard_to_db(published=True),
]
not_published_dashboards = [
create_dashboard_to_db(published=False),
create_dashboard_to_db(published=False),
]
for dash in published_dashboards + not_published_dashboards:
grant_access_to_dashboard(dash, new_role)
self.login(username)
# act
response = self.get_dashboards_list_response()
# assert
self.assert_dashboards_list_view_response(
response,
len(published_dashboards),
published_dashboards,
not_published_dashboards,
)
# post
for dash in published_dashboards + not_published_dashboards:
revoke_access_to_dashboard(dash, new_role)
def test_get_dashboards_list__public_user_without_any_permissions_get_empty_list(
self,
):
create_dashboard_to_db(published=True)
# act
response = self.get_dashboards_list_response()
# assert
self.assert_dashboards_list_view_response(response, 0)
def test_get_dashboards_list__public_user_get_only_published_permitted_dashboards(
self,
):
# arrange
published_dashboards = [
create_dashboard_to_db(published=True),
create_dashboard_to_db(published=True),
]
not_published_dashboards = [
create_dashboard_to_db(published=False),
create_dashboard_to_db(published=False),
]
for dash in published_dashboards + not_published_dashboards:
grant_access_to_dashboard(dash, "Public")
# act
response = self.get_dashboards_list_response()
# assert
self.assert_dashboards_list_view_response(
response,
len(published_dashboards),
published_dashboards,
not_published_dashboards,
)
# post
for dash in published_dashboards + not_published_dashboards:
revoke_access_to_dashboard(dash, "Public")
def test_get_dashboards_api__admin_get_all_dashboards(self):
# arrange
create_dashboard_to_db(
owners=[], slices=[create_slice_to_db()], published=False
)
dashboard_counts = count_dashboards()
self.login("admin")
# act
response = self.get_dashboards_api_response()
# assert
self.assert_dashboards_api_response(response, dashboard_counts)
def test_get_dashboards_api__owner_get_all_owned_dashboards(self):
# arrange
username = random_str()
new_role = f"role_{random_str()}"
owner = self.create_user_with_roles(
username, [new_role], should_create_roles=True
)
database = create_database_to_db()
table = create_datasource_table_to_db(db_id=database.id, owners=[owner])
first_dash = create_dashboard_to_db(
owners=[owner], slices=[create_slice_to_db(datasource_id=table.id)]
)
second_dash = create_dashboard_to_db(
owners=[owner], slices=[create_slice_to_db(datasource_id=table.id)]
)
owned_dashboards = [first_dash, second_dash]
not_owned_dashboards = [
create_dashboard_to_db(
slices=[create_slice_to_db(datasource_id=table.id)], published=True
)
]
self.login(username)
# act
response = self.get_dashboards_api_response()
# assert
self.assert_dashboards_api_response(
response, 2, owned_dashboards, not_owned_dashboards
)
def test_get_dashboards_api__user_without_any_permissions_get_empty_list(self):
username = random_str()
new_role = f"role_{random_str()}"
self.create_user_with_roles(username, [new_role], should_create_roles=True)
create_dashboard_to_db(published=True)
self.login(username)
# act
response = self.get_dashboards_api_response()
# assert
self.assert_dashboards_api_response(response, 0)
def test_get_dashboards_api__user_get_only_published_permitted_dashboards(self):
username = random_str()
new_role = f"role_{random_str()}"
self.create_user_with_roles(username, [new_role], should_create_roles=True)
# arrange
published_dashboards = [
create_dashboard_to_db(published=True),
create_dashboard_to_db(published=True),
]
not_published_dashboards = [
create_dashboard_to_db(published=False),
create_dashboard_to_db(published=False),
]
for dash in published_dashboards + not_published_dashboards:
grant_access_to_dashboard(dash, new_role)
self.login(username)
# act
response = self.get_dashboards_api_response()
# assert
self.assert_dashboards_api_response(
response,
len(published_dashboards),
published_dashboards,
not_published_dashboards,
)
# post
for dash in published_dashboards + not_published_dashboards:
revoke_access_to_dashboard(dash, new_role)
def test_get_dashboards_api__public_user_without_any_permissions_get_empty_list(
self,
):
create_dashboard_to_db(published=True)
# act
response = self.get_dashboards_api_response()
# assert
self.assert_dashboards_api_response(response, 0)
def test_get_dashboards_api__public_user_get_only_published_permitted_dashboards(
self,
):
# arrange
published_dashboards = [
create_dashboard_to_db(published=True),
create_dashboard_to_db(published=True),
]
not_published_dashboards = [
create_dashboard_to_db(published=False),
create_dashboard_to_db(published=False),
]
for dash in published_dashboards + not_published_dashboards:
grant_access_to_dashboard(dash, "Public")
# act
response = self.get_dashboards_api_response()
# assert
self.assert_dashboards_api_response(
response,
len(published_dashboards),
published_dashboards,
not_published_dashboards,
)
# post
for dash in published_dashboards + not_published_dashboards:
revoke_access_to_dashboard(dash, "Public")