Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions superset/commands/chart/update.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
from superset.daos.chart import ChartDAO
from superset.daos.dashboard import DashboardDAO
from superset.exceptions import SupersetSecurityException
from superset.models.dashboard import Dashboard
from superset.models.slice import Slice
from superset.tags.models import ObjectType
from superset.utils.decorators import on_error, transaction
Expand Down Expand Up @@ -71,6 +72,28 @@ def run(self) -> Model:

return ChartDAO.update(self._model, self._properties)

def _validate_new_dashboard_access(
self, requested_dashboards: list[Dashboard], exceptions: list[Exception]
) -> None:
"""
Validate user has access to any NEW dashboard relationships.
Existing relationships are preserved to maintain chart ownership rights.
"""
if not self._model:
return

existing_dashboard_ids = {d.id for d in self._model.dashboards}
requested_dashboard_ids = {d.id for d in requested_dashboards}

if new_dashboard_ids := requested_dashboard_ids - existing_dashboard_ids:
# For NEW dashboard relationships, verify user has access
accessible_dashboards = DashboardDAO.find_by_ids(list(new_dashboard_ids))
accessible_dashboard_ids = {d.id for d in accessible_dashboards}
unauthorized_dashboard_ids = new_dashboard_ids - accessible_dashboard_ids
Comment on lines +90 to +92

This comment was marked as resolved.


if unauthorized_dashboard_ids:
exceptions.append(DashboardsNotFoundValidationError())

def validate(self) -> None: # noqa: C901
exceptions: list[ValidationError] = []
dashboard_ids = self._properties.get("dashboards")
Expand Down Expand Up @@ -120,12 +143,16 @@ def validate(self) -> None: # noqa: C901

# Validate/Populate dashboards only if it's a list
if dashboard_ids is not None:
# First, verify all requested dashboards exist
dashboards = DashboardDAO.find_by_ids(
dashboard_ids,
skip_base_filter=True,
)
if len(dashboards) != len(dashboard_ids):
exceptions.append(DashboardsNotFoundValidationError())
else:
# Then, validate user has access to any NEW dashboard relationships
self._validate_new_dashboard_access(dashboards, exceptions)
self._properties["dashboards"] = dashboards

if exceptions:
Expand Down
151 changes: 151 additions & 0 deletions tests/integration_tests/charts/commands_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -417,6 +417,157 @@ def test_query_context_update_command(self, mock_sm_g, mock_g):
assert len(chart.owners) == 1
assert chart.owners[0] == admin

@patch("superset.commands.chart.update.g")
@patch("superset.utils.core.g")
@patch("superset.security.manager.g")
@pytest.mark.usefixtures("load_energy_table_with_slice")
def test_update_chart_dashboard_security_existing_relationship(
self, mock_sm_g, mock_u_g, mock_c_g
):
"""Test that chart owners can update charts linked to inaccessible
dashboards (existing relationships)"""
from superset.models.dashboard import Dashboard

# Create a chart owned by alpha
admin = security_manager.find_user(username="admin")
alpha = security_manager.find_user(username="alpha")

# Set user context for dashboard creation
mock_u_g.user = mock_c_g.user = mock_sm_g.user = admin

chart = db.session.query(Slice).first()
chart.owners = [alpha]

# Create a dashboard owned by admin (not accessible to alpha)
admin_dashboard = Dashboard(
dashboard_title="Admin Dashboard",
slug="admin-dashboard",
owners=[admin],
published=False,
)
db.session.add(admin_dashboard)

# Link chart to admin's dashboard (alpha owns chart, admin owns dashboard)
chart.dashboards.append(admin_dashboard)
db.session.commit()

# Alpha should still be able to update their chart
# even though it's linked to admin's dashboard
mock_u_g.user = mock_c_g.user = mock_sm_g.user = alpha

json_obj = {
"description": "Updated description",
"dashboards": [
d.id for d in chart.dashboards
], # Keep existing relationships
}
command = UpdateChartCommand(chart.id, json_obj)
command.run()

# Should succeed - alpha can update their chart
updated_chart = db.session.query(Slice).get(chart.id)
assert updated_chart.description == "Updated description"

# Clean up
db.session.delete(admin_dashboard)
db.session.commit()

@patch("superset.commands.chart.update.g")
@patch("superset.utils.core.g")
@patch("superset.security.manager.g")
@pytest.mark.usefixtures("load_energy_table_with_slice")
def test_update_chart_dashboard_security_new_unauthorized_relationship(
self, mock_sm_g, mock_u_g, mock_c_g
):
"""Test that users cannot add charts to dashboards they don't have access to"""
from superset.commands.chart.exceptions import ChartInvalidError
from superset.models.dashboard import Dashboard

admin = security_manager.find_user(username="admin")
alpha = security_manager.find_user(username="alpha")

# Set user context for dashboard creation
mock_u_g.user = mock_c_g.user = mock_sm_g.user = admin

# Create chart owned by alpha
chart = db.session.query(Slice).first()
chart.owners = [alpha]

# Create private dashboard owned by admin (not accessible to alpha)
admin_dashboard = Dashboard(
dashboard_title="Admin Private Dashboard",
slug="admin-private-dashboard",
owners=[admin],
published=False, # Private dashboard
)
db.session.add(admin_dashboard)
db.session.commit()

# Alpha tries to add their chart to admin's private dashboard
mock_u_g.user = mock_c_g.user = mock_sm_g.user = alpha

json_obj = {
"description": "Trying to add to unauthorized dashboard",
"dashboards": [admin_dashboard.id], # NEW unauthorized relationship
}
command = UpdateChartCommand(chart.id, json_obj)

# Should fail - alpha cannot access admin's private dashboard
with self.assertRaises(ChartInvalidError): # noqa: PT027
command.run()

# Clean up
db.session.delete(admin_dashboard)
db.session.commit()

@patch("superset.commands.chart.update.g")
@patch("superset.utils.core.g")
@patch("superset.security.manager.g")
@pytest.mark.usefixtures("load_energy_table_with_slice")
def test_update_chart_dashboard_security_admin_bypass(
self, mock_sm_g, mock_u_g, mock_c_g
):
"""Test that admins can add charts to any dashboard"""
from superset.models.dashboard import Dashboard

admin = security_manager.find_user(username="admin")
alpha = security_manager.find_user(username="alpha")

# Set user context for dashboard creation
mock_u_g.user = mock_c_g.user = mock_sm_g.user = alpha

# Create chart owned by admin
chart = db.session.query(Slice).first()
chart.owners = [admin]

# Create private dashboard owned by alpha
alpha_dashboard = Dashboard(
dashboard_title="Alpha Private Dashboard",
slug="alpha-private-dashboard",
owners=[alpha],
published=False,
)
db.session.add(alpha_dashboard)
db.session.commit()

# Admin should be able to add chart to any dashboard
mock_u_g.user = mock_c_g.user = mock_sm_g.user = admin

json_obj = {
"description": "Admin adding to any dashboard",
"dashboards": [alpha_dashboard.id],
}
command = UpdateChartCommand(chart.id, json_obj)
command.run()

# Should succeed - admin has access to all dashboards
updated_chart = db.session.query(Slice).get(chart.id)
assert alpha_dashboard in updated_chart.dashboards

# Clean up
db.session.delete(alpha_dashboard)
db.session.commit()


class TestChartWarmUpCacheCommand(SupersetTestCase):
def test_warm_up_cache_command_chart_not_found(self):
Expand Down
Loading