fix(dataset): validate datasource access during import (#39998)

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Shaitan
2026-05-15 12:06:47 +01:00
committed by GitHub
co-authored by Claude Sonnet 4.6
parent ffa32414ef
commit ee9eec25f9
2 changed files with 50 additions and 1 deletions
@@ -31,6 +31,7 @@ from sqlalchemy.orm.session import Session
from superset import db, security_manager
from superset.commands.dataset.exceptions import (
DatasetAccessDeniedError,
DatasetForbiddenDataURI,
)
from superset.commands.dataset.importers.v1.utils import (
@@ -744,6 +745,44 @@ def test_import_dataset_without_owner_permission(
mock_can_access.assert_called_with("can_write", "Dataset")
def test_import_dataset_access_check(
mocker: MockerFixture,
session: Session,
) -> None:
"""
Test that import_dataset raises DatasetAccessDeniedError when the user does not
have datasource-level access to the target dataset.
"""
from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
from superset.exceptions import SupersetSecurityException
mocker.patch.object(security_manager, "can_access", return_value=True)
mocker.patch.object(
security_manager,
"raise_for_access",
side_effect=SupersetSecurityException(
SupersetError(
error_type=SupersetErrorType.DATASOURCE_SECURITY_ACCESS_ERROR,
message="User does not have access to this datasource",
level=ErrorLevel.ERROR,
)
),
)
engine = db.session.get_bind()
SqlaTable.metadata.create_all(engine) # pylint: disable=no-member
database = Database(database_name="my_database", sqlalchemy_uri="sqlite://")
db.session.add(database)
db.session.flush()
config = copy.deepcopy(dataset_fixture)
config["database_id"] = database.id
with pytest.raises(DatasetAccessDeniedError):
import_dataset(config)
@pytest.mark.parametrize(
"allowed_urls, data_uri, expected, exception_class",
[