fix: chart import validation (#26993)

This commit is contained in:
Daniel Vaz Gaspar
2024-02-06 12:14:02 +00:00
committed by Michael S. Molina
parent e772915bb8
commit c029475f60
13 changed files with 404 additions and 146 deletions

View File

@@ -17,104 +17,130 @@
# pylint: disable=unused-argument, import-outside-toplevel, unused-import, invalid-name
import copy
from collections.abc import Generator
import pytest
from flask_appbuilder.security.sqla.models import Role, User
from pytest_mock import MockFixture
from sqlalchemy.orm.session import Session
from superset import security_manager
from superset.charts.commands.importers.v1.utils import import_chart
from superset.commands.exceptions import ImportFailedError
from superset.connectors.sqla.models import Database, SqlaTable
from superset.models.slice import Slice
from superset.utils.core import override_user
from tests.integration_tests.fixtures.importexport import chart_config
def test_import_chart(mocker: MockFixture, session: Session) -> None:
@pytest.fixture
def session_with_data(session: Session) -> Generator[Session, None, None]:
engine = session.get_bind()
SqlaTable.metadata.create_all(engine) # pylint: disable=no-member
dataset = SqlaTable(
table_name="test_table",
metrics=[],
main_dttm_col=None,
database=Database(database_name="my_database", sqlalchemy_uri="sqlite://"),
)
session.add(dataset)
session.flush()
slice = Slice(
id=1,
datasource_id=dataset.id,
datasource_type="table",
datasource_name="tmp_perm_table",
slice_name="slice_name",
uuid=chart_config["uuid"],
)
session.add(slice)
session.flush()
yield session
session.rollback()
@pytest.fixture
def session_with_schema(session: Session) -> Generator[Session, None, None]:
from superset.connectors.sqla.models import SqlaTable
engine = session.get_bind()
SqlaTable.metadata.create_all(engine) # pylint: disable=no-member
yield session
def test_import_chart(mocker: MockFixture, session_with_schema: Session) -> None:
"""
Test importing a chart.
"""
from superset import security_manager
from superset.charts.commands.importers.v1.utils import import_chart
from superset.connectors.sqla.models import SqlaTable
from superset.models.core import Database
from superset.models.slice import Slice
from tests.integration_tests.fixtures.importexport import chart_config
mocker.patch.object(security_manager, "can_access", return_value=True)
engine = session.get_bind()
Slice.metadata.create_all(engine) # pylint: disable=no-member
config = copy.deepcopy(chart_config)
config["datasource_id"] = 1
config["datasource_type"] = "table"
chart = import_chart(session, config)
chart = import_chart(session_with_schema, config)
assert chart.slice_name == "Deck Path"
assert chart.viz_type == "deck_path"
assert chart.is_managed_externally is False
assert chart.external_url is None
# Assert that the can write to chart was checked
security_manager.can_access.assert_called_once_with("can_write", "Chart")
def test_import_chart_managed_externally(mocker: MockFixture, session: Session) -> None:
def test_import_chart_managed_externally(
mocker: MockFixture, session_with_schema: Session
) -> None:
"""
Test importing a chart that is managed externally.
"""
from superset import security_manager
from superset.charts.commands.importers.v1.utils import import_chart
from superset.connectors.sqla.models import SqlaTable
from superset.models.core import Database
from superset.models.slice import Slice
from tests.integration_tests.fixtures.importexport import chart_config
mocker.patch.object(security_manager, "can_access", return_value=True)
engine = session.get_bind()
Slice.metadata.create_all(engine) # pylint: disable=no-member
config = copy.deepcopy(chart_config)
config["datasource_id"] = 1
config["datasource_type"] = "table"
config["is_managed_externally"] = True
config["external_url"] = "https://example.org/my_chart"
chart = import_chart(session, config)
chart = import_chart(session_with_schema, config)
assert chart.is_managed_externally is True
assert chart.external_url == "https://example.org/my_chart"
# Assert that the can write to chart was checked
security_manager.can_access.assert_called_once_with("can_write", "Chart")
def test_import_chart_without_permission(
mocker: MockFixture,
session: Session,
session_with_schema: Session,
) -> None:
"""
Test importing a chart when a user doesn't have permissions to create.
"""
from superset import security_manager
from superset.charts.commands.importers.v1.utils import import_chart
from superset.connectors.sqla.models import SqlaTable
from superset.models.core import Database
from superset.models.slice import Slice
from tests.integration_tests.fixtures.importexport import chart_config
mocker.patch.object(security_manager, "can_access", return_value=False)
engine = session.get_bind()
Slice.metadata.create_all(engine) # pylint: disable=no-member
config = copy.deepcopy(chart_config)
config["datasource_id"] = 1
config["datasource_type"] = "table"
with pytest.raises(ImportFailedError) as excinfo:
import_chart(session, config)
import_chart(session_with_schema, config)
assert (
str(excinfo.value)
== "Chart doesn't exist and user doesn't have permission to create charts"
)
# Assert that the can write to chart was checked
security_manager.can_access.assert_called_once_with("can_write", "Chart")
def test_filter_chart_annotations(mocker: MockFixture, session: Session) -> None:
def test_filter_chart_annotations(session: Session) -> None:
"""
Test importing a chart.
"""
from superset import security_manager
from superset.charts.commands.importers.v1.utils import filter_chart_annotations
from tests.integration_tests.fixtures.importexport import (
chart_config_with_mixed_annotations,
@@ -127,3 +153,67 @@ def test_filter_chart_annotations(mocker: MockFixture, session: Session) -> None
assert len(annotation_layers) == 1
assert all([al["annotationType"] == "FORMULA" for al in annotation_layers])
def test_import_existing_chart_without_permission(
mocker: MockFixture,
session_with_data: Session,
) -> None:
"""
Test importing a chart when a user doesn't have permissions to modify.
"""
mocker.patch.object(security_manager, "can_access", return_value=True)
mocker.patch.object(security_manager, "can_access_chart", return_value=False)
slice = (
session_with_data.query(Slice)
.filter(Slice.uuid == chart_config["uuid"])
.one_or_none()
)
with override_user("admin"):
with pytest.raises(ImportFailedError) as excinfo:
import_chart(session_with_data, chart_config, overwrite=True)
assert (
str(excinfo.value)
== "A chart already exists and user doesn't have permissions to overwrite it"
)
# Assert that the can write to chart was checked
security_manager.can_access.assert_called_once_with("can_write", "Chart")
security_manager.can_access_chart.assert_called_once_with(slice)
def test_import_existing_chart_with_permission(
mocker: MockFixture,
session_with_data: Session,
) -> None:
"""
Test importing a chart that exists when a user has access permission to that chart.
"""
mocker.patch.object(security_manager, "can_access", return_value=True)
mocker.patch.object(security_manager, "can_access_chart", return_value=True)
admin = User(
first_name="Alice",
last_name="Doe",
email="adoe@example.org",
username="admin",
roles=[Role(name="Admin")],
)
config = copy.deepcopy(chart_config)
config["datasource_id"] = 1
config["datasource_type"] = "table"
slice = (
session_with_data.query(Slice)
.filter(Slice.uuid == config["uuid"])
.one_or_none()
)
with override_user(admin):
import_chart(session_with_data, config, overwrite=True)
# Assert that the can write to chart was checked
security_manager.can_access.assert_called_once_with("can_write", "Chart")
security_manager.can_access_chart.assert_called_once_with(slice)

View File

@@ -17,32 +17,57 @@
# pylint: disable=unused-argument, import-outside-toplevel, unused-import, invalid-name
import copy
from collections.abc import Generator
import pytest
from flask_appbuilder.security.sqla.models import Role, User
from pytest_mock import MockFixture
from sqlalchemy.orm.session import Session
from superset import security_manager
from superset.commands.exceptions import ImportFailedError
from superset.dashboards.commands.importers.v1.utils import import_dashboard
from superset.models.dashboard import Dashboard
from superset.utils.core import override_user
from tests.integration_tests.fixtures.importexport import dashboard_config
def test_import_dashboard(mocker: MockFixture, session: Session) -> None:
@pytest.fixture
def session_with_data(session: Session) -> Generator[Session, None, None]:
engine = session.get_bind()
Dashboard.metadata.create_all(engine) # pylint: disable=no-member
dashboard = Dashboard(
id=100,
dashboard_title="Test dash",
slug=None,
slices=[],
published=True,
uuid=dashboard_config["uuid"],
)
session.add(dashboard)
session.flush()
yield session
session.rollback()
@pytest.fixture
def session_with_schema(session: Session) -> Generator[Session, None, None]:
engine = session.get_bind()
Dashboard.metadata.create_all(engine) # pylint: disable=no-member
yield session
session.rollback()
def test_import_dashboard(mocker: MockFixture, session_with_schema: Session) -> None:
"""
Test importing a dashboard.
"""
from superset import security_manager
from superset.dashboards.commands.importers.v1.utils import import_dashboard
from superset.models.slice import Slice
from tests.integration_tests.fixtures.importexport import dashboard_config
mocker.patch.object(security_manager, "can_access", return_value=True)
engine = session.get_bind()
Slice.metadata.create_all(engine) # pylint: disable=no-member
config = copy.deepcopy(dashboard_config)
dashboard = import_dashboard(session, config)
dashboard = import_dashboard(session_with_schema, dashboard_config)
assert dashboard.dashboard_title == "Test dash"
assert dashboard.description is None
assert dashboard.is_managed_externally is False
@@ -53,26 +78,18 @@ def test_import_dashboard(mocker: MockFixture, session: Session) -> None:
def test_import_dashboard_managed_externally(
mocker: MockFixture,
session: Session,
session_with_schema: Session,
) -> None:
"""
Test importing a dashboard that is managed externally.
"""
from superset import security_manager
from superset.dashboards.commands.importers.v1.utils import import_dashboard
from superset.models.slice import Slice
from tests.integration_tests.fixtures.importexport import dashboard_config
mocker.patch.object(security_manager, "can_access", return_value=True)
engine = session.get_bind()
Slice.metadata.create_all(engine) # pylint: disable=no-member
config = copy.deepcopy(dashboard_config)
config["is_managed_externally"] = True
config["external_url"] = "https://example.org/my_dashboard"
dashboard = import_dashboard(session, config)
dashboard = import_dashboard(session_with_schema, config)
assert dashboard.is_managed_externally is True
assert dashboard.external_url == "https://example.org/my_dashboard"
@@ -82,25 +99,15 @@ def test_import_dashboard_managed_externally(
def test_import_dashboard_without_permission(
mocker: MockFixture,
session: Session,
session_with_schema: Session,
) -> None:
"""
Test importing a dashboard when a user doesn't have permissions to create.
"""
from superset import security_manager
from superset.dashboards.commands.importers.v1.utils import import_dashboard
from superset.models.slice import Slice
from tests.integration_tests.fixtures.importexport import dashboard_config
mocker.patch.object(security_manager, "can_access", return_value=False)
engine = session.get_bind()
Slice.metadata.create_all(engine) # pylint: disable=no-member
config = copy.deepcopy(dashboard_config)
with pytest.raises(ImportFailedError) as excinfo:
import_dashboard(session, config)
import_dashboard(session_with_schema, dashboard_config)
assert (
str(excinfo.value)
== "Dashboard doesn't exist and user doesn't have permission to create dashboards"
@@ -112,72 +119,43 @@ def test_import_dashboard_without_permission(
def test_import_existing_dashboard_without_permission(
mocker: MockFixture,
session: Session,
session_with_data: Session,
) -> None:
"""
Test importing a dashboard when a user doesn't have permissions to create.
"""
from superset import security_manager
from superset.dashboards.commands.importers.v1.utils import g, import_dashboard
from superset.models.dashboard import Dashboard
from superset.models.slice import Slice
from tests.integration_tests.fixtures.importexport import dashboard_config
mocker.patch.object(security_manager, "can_access", return_value=True)
mocker.patch.object(security_manager, "can_access_dashboard", return_value=False)
mock_g = mocker.patch(
"superset.dashboards.commands.importers.v1.utils.g"
) # Replace with the actual path to g
mock_g.user = mocker.MagicMock(return_value=True)
engine = session.get_bind()
Slice.metadata.create_all(engine) # pylint: disable=no-member
Dashboard.metadata.create_all(engine) # pylint: disable=no-member
dashboard_obj = Dashboard(
id=100,
dashboard_title="Test dash",
slug=None,
slices=[],
published=True,
uuid="c4b28c4e-a1fe-4cf8-a5ac-d6f11d6fdd51",
dashboard = (
session_with_data.query(Dashboard)
.filter(Dashboard.uuid == dashboard_config["uuid"])
.one_or_none()
)
session.add(dashboard_obj)
session.flush()
config = copy.deepcopy(dashboard_config)
with pytest.raises(ImportFailedError) as excinfo:
import_dashboard(session, config, overwrite=True)
assert (
str(excinfo.value)
== "A dashboard already exists and user doesn't have permissions to overwrite it"
)
with override_user("admin"):
with pytest.raises(ImportFailedError) as excinfo:
import_dashboard(session_with_data, dashboard_config, overwrite=True)
assert (
str(excinfo.value)
== "A dashboard already exists and user doesn't have permissions to overwrite it"
)
# Assert that the can write to dashboard was checked
security_manager.can_access.assert_called_once_with("can_write", "Dashboard")
security_manager.can_access_dashboard.assert_called_once_with(dashboard_obj)
security_manager.can_access_dashboard.assert_called_once_with(dashboard)
def test_import_existing_dashboard_with_permission(
mocker: MockFixture,
session: Session,
session_with_data: Session,
) -> None:
"""
Test importing a dashboard when a user doesn't have permissions to create.
Test importing a dashboard that exists when a user has access permission to that dashboard.
"""
from flask_appbuilder.security.sqla.models import Role, User
from superset import security_manager
from superset.dashboards.commands.importers.v1.utils import import_dashboard
from superset.models.dashboard import Dashboard
from tests.integration_tests.fixtures.importexport import dashboard_config
mocker.patch.object(security_manager, "can_access", return_value=True)
mocker.patch.object(security_manager, "can_access_dashboard", return_value=True)
engine = session.get_bind()
Dashboard.metadata.create_all(engine) # pylint: disable=no-member
admin = User(
first_name="Alice",
last_name="Doe",
@@ -186,20 +164,14 @@ def test_import_existing_dashboard_with_permission(
roles=[Role(name="Admin")],
)
dashboard_obj = Dashboard(
id=100,
dashboard_title="Test dash",
slug=None,
slices=[],
published=True,
uuid="c4b28c4e-a1fe-4cf8-a5ac-d6f11d6fdd51",
dashboard = (
session_with_data.query(Dashboard)
.filter(Dashboard.uuid == dashboard_config["uuid"])
.one_or_none()
)
session.add(dashboard_obj)
session.flush()
config = copy.deepcopy(dashboard_config)
with override_user(admin):
import_dashboard(session, config, overwrite=True)
import_dashboard(session_with_data, dashboard_config, overwrite=True)
# Assert that the can write to dashboard was checked
security_manager.can_access.assert_called_once_with("can_write", "Dashboard")
security_manager.can_access_dashboard.assert_called_once_with(dashboard_obj)
security_manager.can_access_dashboard.assert_called_once_with(dashboard)

View File

@@ -16,12 +16,16 @@
# under the License.
import pytest
from flask_appbuilder.security.sqla.models import Role, User
from pytest_mock import MockFixture
from superset.common.query_object import QueryObject
from superset.connectors.sqla.models import Database, SqlaTable
from superset.exceptions import SupersetSecurityException
from superset.extensions import appbuilder
from superset.models.slice import Slice
from superset.security.manager import SupersetSecurityManager
from superset.utils.core import override_user
def test_security_manager(app_context: None) -> None:
@@ -164,3 +168,139 @@ def test_raise_for_access_query_default_schema(
== """You need access to the following tables: `public.ab_user`,
`all_database_access` or `all_datasource_access` permission"""
)
def test_raise_for_access_chart_for_datasource_permission(
mocker: MockFixture,
app_context: None,
) -> None:
"""
Test that the security manager can raise an exception for chart access,
when the user does not have access to the chart datasource
"""
sm = SupersetSecurityManager(appbuilder)
session = sm.get_session
engine = session.get_bind()
Slice.metadata.create_all(engine) # pylint: disable=no-member
alpha = User(
first_name="Alice",
last_name="Doe",
email="adoe@example.org",
username="admin",
roles=[Role(name="Alpha")],
)
dataset = SqlaTable(
table_name="test_table",
metrics=[],
main_dttm_col=None,
database=Database(database_name="my_database", sqlalchemy_uri="sqlite://"),
)
session.add(dataset)
session.flush()
slice = Slice(
id=1,
datasource_id=dataset.id,
datasource_type="table",
datasource_name="tmp_perm_table",
slice_name="slice_name",
)
session.add(slice)
session.flush()
mocker.patch.object(sm, "can_access_datasource", return_value=False)
with override_user(alpha):
with pytest.raises(SupersetSecurityException) as excinfo:
sm.raise_for_access(
chart=slice,
)
assert str(excinfo.value) == "You don't have access to this chart."
mocker.patch.object(sm, "can_access_datasource", return_value=True)
with override_user(alpha):
sm.raise_for_access(
chart=slice,
)
def test_raise_for_access_chart_on_admin(
app_context: None,
) -> None:
"""
Test that the security manager can raise an exception for chart access,
when the user does not have access to the chart datasource
"""
from flask_appbuilder.security.sqla.models import Role, User
from superset.models.slice import Slice
from superset.utils.core import override_user
sm = SupersetSecurityManager(appbuilder)
session = sm.get_session
engine = session.get_bind()
Slice.metadata.create_all(engine) # pylint: disable=no-member
admin = User(
first_name="Alice",
last_name="Doe",
email="adoe@example.org",
username="admin",
roles=[Role(name="Admin")],
)
slice = Slice(
id=1,
datasource_id=1,
datasource_type="table",
datasource_name="tmp_perm_table",
slice_name="slice_name",
)
session.add(slice)
session.flush()
with override_user(admin):
sm.raise_for_access(
chart=slice,
)
def test_raise_for_access_chart_owner(
app_context: None,
) -> None:
"""
Test that the security manager can raise an exception for chart access,
when the user does not have access to the chart datasource
"""
sm = SupersetSecurityManager(appbuilder)
session = sm.get_session
engine = session.get_bind()
Slice.metadata.create_all(engine) # pylint: disable=no-member
alpha = User(
first_name="Alice",
last_name="Doe",
email="adoe@example.org",
username="admin",
roles=[Role(name="Alpha")],
)
slice = Slice(
id=1,
datasource_id=1,
datasource_type="table",
datasource_name="tmp_perm_table",
slice_name="slice_name",
owners=[alpha],
)
session.add(slice)
session.flush()
with override_user(alpha):
sm.raise_for_access(
chart=slice,
)