test(themes): add the parallel is_system_dark admin-overwrite case

Mirrors the existing is_system_default test for full branch coverage
of the admin carve-out.

superset/commands/semantic_layer/create.py

@@ -29,7 +29,9 @@ from superset.commands.semantic_layer.exceptions import (
     SemanticLayerInvalidError,
     SemanticLayerNotFoundError,
     SemanticViewCreateFailedError,
+    SemanticViewForbiddenError,
 )
+from superset.commands.utils import current_user_can_modify_object
 from superset.daos.semantic_layer import SemanticLayerDAO, SemanticViewDAO
 from superset.semantic_layers.registry import registry
 from superset.utils import json
@@ -92,9 +94,13 @@ class CreateSemanticViewCommand(BaseCommand):
 
     def validate(self) -> None:
         layer_uuid: str = self._properties.get("semantic_layer_uuid", "")
-        if not SemanticLayerDAO.find_by_uuid(layer_uuid):
+        layer = SemanticLayerDAO.find_by_uuid(layer_uuid)
+        if not layer:
             raise SemanticLayerNotFoundError()
 
+        if not current_user_can_modify_object(layer):
+            raise SemanticViewForbiddenError()
+
         name: str = self._properties.get("name", "")
         configuration: dict[str, Any] = self._properties.get("configuration") or {}
         if not SemanticViewDAO.validate_uniqueness(name, layer_uuid, configuration):

superset/commands/semantic_layer/delete.py

@@ -21,17 +21,17 @@ from functools import partial
 
 from sqlalchemy.exc import SQLAlchemyError
 
-from superset import security_manager
 from superset.commands.base import BaseCommand
 from superset.commands.semantic_layer.exceptions import (
     SemanticLayerDeleteFailedError,
+    SemanticLayerForbiddenError,
     SemanticLayerNotFoundError,
     SemanticViewDeleteFailedError,
     SemanticViewForbiddenError,
     SemanticViewNotFoundError,
 )
+from superset.commands.utils import current_user_can_modify_object
 from superset.daos.semantic_layer import SemanticLayerDAO, SemanticViewDAO
-from superset.exceptions import SupersetSecurityException
 from superset.semantic_layers.models import SemanticLayer, SemanticView
 from superset.utils.decorators import on_error, transaction
 
@@ -60,6 +60,9 @@ class DeleteSemanticLayerCommand(BaseCommand):
         if not self._model:
             raise SemanticLayerNotFoundError()
 
+        if not current_user_can_modify_object(self._model):
+            raise SemanticLayerForbiddenError()
+
 
 class DeleteSemanticViewCommand(BaseCommand):
     def __init__(self, pk: int):
@@ -82,10 +85,8 @@ class DeleteSemanticViewCommand(BaseCommand):
         self._model = SemanticViewDAO.find_by_id(self._pk, id_column="id")
         if not self._model:
             raise SemanticViewNotFoundError()
-        try:
-            security_manager.raise_for_editorship(self._model)
-        except SupersetSecurityException as ex:
-            raise SemanticViewForbiddenError() from ex
+        if not current_user_can_modify_object(self._model):
+            raise SemanticViewForbiddenError()
 
 
 class BulkDeleteSemanticViewCommand(BaseCommand):
@@ -109,7 +110,5 @@ class BulkDeleteSemanticViewCommand(BaseCommand):
         if len(self._models) != len(self._model_ids):
             raise SemanticViewNotFoundError()
         for model in self._models:
-            try:
-                security_manager.raise_for_editorship(model)
-            except SupersetSecurityException as ex:
-                raise SemanticViewForbiddenError() from ex
+            if not current_user_can_modify_object(model):
+                raise SemanticViewForbiddenError()

superset/commands/semantic_layer/update.py

@@ -23,9 +23,9 @@ from typing import Any
 from flask_appbuilder.models.sqla import Model
 from sqlalchemy.exc import SQLAlchemyError
 
-from superset import security_manager
 from superset.commands.base import BaseCommand
 from superset.commands.semantic_layer.exceptions import (
+    SemanticLayerForbiddenError,
     SemanticLayerInvalidError,
     SemanticLayerNotFoundError,
     SemanticLayerUpdateFailedError,
@@ -33,8 +33,8 @@ from superset.commands.semantic_layer.exceptions import (
     SemanticViewNotFoundError,
     SemanticViewUpdateFailedError,
 )
+from superset.commands.utils import current_user_can_modify_object
 from superset.daos.semantic_layer import SemanticLayerDAO, SemanticViewDAO
-from superset.exceptions import SupersetSecurityException
 from superset.semantic_layers.models import SemanticLayer, SemanticView
 from superset.semantic_layers.registry import registry
 from superset.utils import json
@@ -66,10 +66,8 @@ class UpdateSemanticViewCommand(BaseCommand):
         if not self._model:
             raise SemanticViewNotFoundError()
 
-        try:
-            security_manager.raise_for_editorship(self._model)
-        except SupersetSecurityException as ex:
-            raise SemanticViewForbiddenError() from ex
+        if not current_user_can_modify_object(self._model):
+            raise SemanticViewForbiddenError()
 
         name = self._properties.get("name", self._model.name)
         layer_uuid = str(self._model.semantic_layer_uuid)
@@ -116,6 +114,9 @@ class UpdateSemanticLayerCommand(BaseCommand):
         if not self._model:
             raise SemanticLayerNotFoundError()
 
+        if not current_user_can_modify_object(self._model):
+            raise SemanticLayerForbiddenError()
+
         name = self._properties.get("name")
         if name and not SemanticLayerDAO.validate_update_uniqueness(self._uuid, name):
             raise SemanticLayerInvalidError(f"Name already exists: {name}")

superset/commands/tag/create.py

@@ -21,11 +21,8 @@ from typing import Any
 from superset import security_manager
 from superset.commands.base import BaseCommand, CreateMixin
 from superset.commands.tag.exceptions import TagCreateFailedError, TagInvalidError
-from superset.commands.tag.utils import (
-    current_user_can_modify_object,
-    to_object_model,
-    to_object_type,
-)
+from superset.commands.tag.utils import to_object_model, to_object_type
+from superset.commands.utils import current_user_can_modify_object
 from superset.daos.tag import TagDAO
 from superset.exceptions import SupersetSecurityException
 from superset.tags.models import ObjectType, TagType

superset/commands/tag/update.py

@@ -22,11 +22,8 @@ from flask_appbuilder.models.sqla import Model
 from superset import db
 from superset.commands.base import BaseCommand, UpdateMixin
 from superset.commands.tag.exceptions import TagInvalidError, TagNotFoundError
-from superset.commands.tag.utils import (
-    current_user_can_modify_object,
-    to_object_model,
-    to_object_type,
-)
+from superset.commands.tag.utils import to_object_model, to_object_type
+from superset.commands.utils import current_user_can_modify_object
 from superset.daos.tag import TagDAO
 from superset.tags.models import Tag
 from superset.utils.decorators import transaction

superset/commands/tag/utils.py

@@ -17,11 +17,9 @@
 
 from typing import Any, Optional, Union
 
-from superset import security_manager
 from superset.daos.chart import ChartDAO
 from superset.daos.dashboard import DashboardDAO
 from superset.daos.query import SavedQueryDAO
-from superset.exceptions import SupersetSecurityException
 from superset.models.dashboard import Dashboard
 from superset.models.slice import Slice
 from superset.models.sql_lab import SavedQuery
@@ -52,22 +50,3 @@ def to_object_model(
 
         return DatasetDAO.find_by_id(object_id, skip_base_filter=skip_base_filter)
     return None
-
-
-def current_user_can_modify_object(model: Any) -> bool:
-    """Whether the current user may create/modify tag relationships on ``model``.
-
-    Mirrors the editorship check the bulk-create path already applies, or the
-    object's creator, so the tag-update path enforces the same boundary.
-    Look the model up with
-    ``skip_base_filter=True`` before calling this, so an object the user cannot
-    access reaches the check instead of resolving to ``None`` and being written
-    without any check.
-    """
-    try:
-        security_manager.raise_for_editorship(model)
-        return True
-    except SupersetSecurityException:
-        return bool(
-            model.created_by and model.created_by == security_manager.current_user
-        )

superset/commands/theme/import_themes.py

@@ -44,6 +44,18 @@ def import_theme(config: dict[str, Any], overwrite: bool = False) -> "Theme | No
     if existing:
         if not overwrite or not can_write:
             return existing
+        if existing.is_system:
+            raise ThemeImportError("Cannot overwrite a system theme via import")
+        # The active system-default/dark theme slot may be overwritten by
+        # admins only; a non-admin overwriting it would change the theme
+        # rendered for every user, including the login page and other
+        # admins.
+        if (
+            existing.is_system_default or existing.is_system_dark
+        ) and not security_manager.is_admin():
+            raise ThemeImportError(
+                "Cannot overwrite the active system-default/dark theme via import"
+            )
         config["id"] = existing.id
     elif not can_write:
         raise ThemeImportError(

superset/commands/theme/update.py

@@ -18,8 +18,10 @@ import logging
 from functools import partial
 from typing import Any, Optional
 
+from superset import security_manager
 from superset.commands.base import UpdateMixin
 from superset.commands.theme.exceptions import (
+    SystemThemeInUseError,
     SystemThemeProtectedError,
     ThemeNotFoundError,
 )
@@ -52,3 +54,11 @@ class UpdateThemeCommand(UpdateMixin):
         # Check if it's a system theme
         if self._model.is_system:
             raise SystemThemeProtectedError()
+
+        # The active system-default/dark theme slot may be edited by admins
+        # only; a non-admin editing it would change the theme rendered for
+        # every user, including the login page and other admins.
+        if (
+            self._model.is_system_default or self._model.is_system_dark
+        ) and not security_manager.is_admin():
+            raise SystemThemeInUseError()

superset/commands/utils.py

@@ -32,6 +32,7 @@ from superset.commands.exceptions import (
 from superset.daos.datasource import DatasourceDAO
 from superset.daos.exceptions import DatasourceNotFound
 from superset.daos.tag import TagDAO
+from superset.exceptions import SupersetSecurityException
 from superset.subjects.exceptions import SubjectsNotFoundValidationError
 from superset.subjects.models import Subject
 from superset.subjects.utils import (
@@ -52,6 +53,30 @@ def _has_extra_editors_resolver() -> bool:
     return bool(has_app_context() and current_app.config.get("EXTRA_EDITORS_RESOLVER"))
 
 
+def current_user_can_modify_object(model: Any) -> bool:
+    """Whether the current user is authorized to create/modify ``model``.
+
+    Delegates to ``security_manager.raise_for_editorship``, which grants
+    access to admins and any subject in ``model.editors`` (when that
+    relationship exists). For models that don't carry an ``editors``
+    relationship, or when the current subject isn't one of them, this falls
+    back to allowing the object's creator.
+
+    Callers that need to distinguish "not found" from "no access" should
+    look the model up bypassing DAO base filters (e.g.
+    ``skip_base_filter=True``) before calling this, so an object the user
+    cannot access reaches the check instead of resolving to ``None`` and
+    being written without any check.
+    """
+    try:
+        security_manager.raise_for_editorship(model)
+        return True
+    except SupersetSecurityException:
+        return bool(
+            model.created_by and model.created_by == security_manager.current_user
+        )
+
+
 def populate_subject_list(
     subject_ids: list[int] | None,
     default_to_user: bool,

superset/security/manager.py

@@ -1408,6 +1408,7 @@ class SupersetSecurityManager(  # pylint: disable=too-many-public-methods
         "CssTemplate",
         "Dataset",
         "Datasource",
+        "Theme",
     } | READ_ONLY_MODEL_VIEWS
 
     GAMMA_EXCLUDED_PVMS = {

superset/semantic_layers/api.py

@@ -41,6 +41,7 @@ from superset.commands.semantic_layer.delete import (
 from superset.commands.semantic_layer.exceptions import (
     SemanticLayerCreateFailedError,
     SemanticLayerDeleteFailedError,
+    SemanticLayerForbiddenError,
     SemanticLayerInvalidError,
     SemanticLayerNotFoundError,
     SemanticLayerUpdateFailedError,
@@ -210,6 +211,8 @@ class SemanticViewRestApi(BaseSupersetModelRestApi):
               description: Semantic view structure
             401:
               $ref: '#/components/responses/401'
+            403:
+              $ref: '#/components/responses/403'
             404:
               $ref: '#/components/responses/404'
             422:
@@ -338,6 +341,8 @@ class SemanticViewRestApi(BaseSupersetModelRestApi):
                 errors.append(
                     {"name": view_data.get("name"), "error": "Semantic layer not found"}
                 )
+            except SemanticViewForbiddenError as ex:
+                errors.append({"name": view_data.get("name"), "error": str(ex)})
             except SemanticViewCreateFailedError as ex:
                 logger.error(
                     "Error creating semantic view: %s",
@@ -447,6 +452,8 @@ class SemanticViewRestApi(BaseSupersetModelRestApi):
               description: Semantic view deleted
             401:
               $ref: '#/components/responses/401'
+            403:
+              $ref: '#/components/responses/403'
             404:
               $ref: '#/components/responses/404'
             422:
@@ -665,6 +672,8 @@ class SemanticLayerRestApi(BaseSupersetApi):
               description: Runtime JSON Schema
             401:
               $ref: '#/components/responses/401'
+            403:
+              $ref: '#/components/responses/403'
             404:
               $ref: '#/components/responses/404'
         """
@@ -672,6 +681,11 @@ class SemanticLayerRestApi(BaseSupersetApi):
         if not layer:
             return self.response_404()
 
+        try:
+            layer.raise_for_access()
+        except SupersetSecurityException as ex:
+            return self.response(403, message=ex.message)
+
         body = request.get_json(silent=True) or {}
         runtime_data = body.get("runtime_data")
 
@@ -716,6 +730,8 @@ class SemanticLayerRestApi(BaseSupersetApi):
               description: Available views
             401:
               $ref: '#/components/responses/401'
+            403:
+              $ref: '#/components/responses/403'
             404:
               $ref: '#/components/responses/404'
         """
@@ -726,6 +742,11 @@ class SemanticLayerRestApi(BaseSupersetApi):
         if not layer:
             return self.response_404()
 
+        try:
+            layer.raise_for_access()
+        except SupersetSecurityException as ex:
+            return self.response(403, message=ex.message)
+
         body = request.get_json(silent=True) or {}
         runtime_data = body.get("runtime_data", {})
 
@@ -855,6 +876,8 @@ class SemanticLayerRestApi(BaseSupersetApi):
               $ref: '#/components/responses/400'
             401:
               $ref: '#/components/responses/401'
+            403:
+              $ref: '#/components/responses/403'
             404:
               $ref: '#/components/responses/404'
             422:
@@ -870,6 +893,8 @@ class SemanticLayerRestApi(BaseSupersetApi):
             return self.response(200, result={"uuid": str(changed_model.uuid)})
         except SemanticLayerNotFoundError:
             return self.response_404()
+        except SemanticLayerForbiddenError as ex:
+            return self.response(403, message=str(ex))
         except SemanticLayerInvalidError as ex:
             return self.response_422(message=str(ex))
         except SemanticLayerUpdateFailedError as ex:
@@ -899,6 +924,8 @@ class SemanticLayerRestApi(BaseSupersetApi):
               description: Semantic layer deleted
             401:
               $ref: '#/components/responses/401'
+            403:
+              $ref: '#/components/responses/403'
             404:
               $ref: '#/components/responses/404'
             422:
@@ -909,6 +936,8 @@ class SemanticLayerRestApi(BaseSupersetApi):
             return self.response(200, message="OK")
         except SemanticLayerNotFoundError:
             return self.response_404()
+        except SemanticLayerForbiddenError as ex:
+            return self.response(403, message=str(ex))
         except SemanticLayerDeleteFailedError as ex:
             logger.error(
                 "Error deleting semantic layer: %s",
@@ -1152,10 +1181,18 @@ class SemanticLayerRestApi(BaseSupersetApi):
               description: A semantic layer
             401:
               $ref: '#/components/responses/401'
+            403:
+              $ref: '#/components/responses/403'
             404:
               $ref: '#/components/responses/404'
         """
         layer = SemanticLayerDAO.find_by_uuid(uuid)
         if not layer:
             return self.response_404()
+
+        try:
+            layer.raise_for_access()
+        except SupersetSecurityException as ex:
+            return self.response(403, message=ex.message)
+
         return self.response(200, result=_serialize_layer(layer))

superset/semantic_layers/models.py

@@ -148,6 +148,26 @@ class SemanticLayer(AuditMixinNullable, Model):
         """Compute the permission string for this semantic layer."""
         return f"[{self.name}](id:{self.uuid.hex})"
 
+    def raise_for_access(self) -> None:
+        """Check that the user has access to this semantic layer."""
+        from superset import security_manager
+        from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
+        from superset.exceptions import SupersetSecurityException
+
+        if security_manager.can_access_all_datasources():
+            return
+
+        if self.perm and security_manager.can_access("datasource_access", self.perm):
+            return
+
+        raise SupersetSecurityException(
+            SupersetError(
+                error_type=SupersetErrorType.DATASOURCE_SECURITY_ACCESS_ERROR,
+                message=str(_("You don't have access to this semantic layer.")),
+                level=ErrorLevel.ERROR,
+            )
+        )
+
     @staticmethod
     def after_insert(
         mapper: Mapper,

superset/themes/api.py

@@ -350,6 +350,8 @@ class ThemeRestApi(BaseSupersetModelRestApi):
             return self.response_404()
         except SystemThemeProtectedError:
             return self.response_403()
+        except SystemThemeInUseError:
+            return self.response_403()
         except Exception as ex:
             logger.exception("Unexpected error in PUT /theme/%s", pk)
             return self.response_422(message=str(ex))

superset/utils/core.py

@@ -615,9 +615,21 @@ def sanitize_svg_content(svg_content: str) -> str:
         return ""
 
     # Minimal protection: remove obvious malicious content, preserve all SVG features
+    # The closing tag pattern tolerates attributes/whitespace after "script"
+    # (e.g. "</script foo>"), which browsers still parse as a valid closer.
     content = re.sub(
-        r"<script[^>]*>.*?</script>", "", svg_content, flags=re.IGNORECASE | re.DOTALL
+        r"<script\b[^>]*>.*?</script\b[^>]*>",
+        "",
+        svg_content,
+        flags=re.IGNORECASE | re.DOTALL,
     )
+    # Second pass: an unterminated <script ...> opener has no matching
+    # closer, so browsers treat everything after it as script content
+    # through end-of-file. Drop the opener and the remainder of the
+    # content with it, rather than leaving the payload text behind.
+    content = re.sub(r"<script\b[^>]*>.*", "", content, flags=re.IGNORECASE | re.DOTALL)
+    # Drop any orphaned closing </script ...> fragment too.
+    content = re.sub(r"</script\b[^>]*>?", "", content, flags=re.IGNORECASE)
     content = re.sub(r"javascript:", "", content, flags=re.IGNORECASE)
     content = re.sub(r"data:[^;]*;[^,]*,.*javascript", "", content, flags=re.IGNORECASE)
 

tests/integration_tests/themes/test_theme_api_permissions.py

@@ -23,7 +23,11 @@ from superset.models.core import Theme
 from superset.utils import json
 from tests.conftest import with_config
 from tests.integration_tests.base_tests import SupersetTestCase
-from tests.integration_tests.constants import ADMIN_USERNAME, GAMMA_USERNAME
+from tests.integration_tests.constants import (
+    ADMIN_USERNAME,
+    ALPHA_USERNAME,
+    GAMMA_USERNAME,
+)
 
 
 class TestThemeAPIPermissions(SupersetTestCase):
@@ -92,9 +96,11 @@ class TestThemeAPIPermissions(SupersetTestCase):
 
     @with_config({"ENABLE_UI_THEME_ADMINISTRATION": True})
     def test_non_admin_cannot_set_system_default(self):
-        """Test that non-admin users cannot set system themes"""
-        # Login as gamma user
-        self.login(GAMMA_USERNAME)
+        """Test that a non-admin user with theme write access (Alpha) still
+        cannot set system themes, since that is an admin-only action."""
+        # Login as alpha user, who has generic write access to themes but
+        # is not an admin
+        self.login(ALPHA_USERNAME)
 
         # Try to set theme as system default
         response = self.client.put(
@@ -110,6 +116,26 @@ class TestThemeAPIPermissions(SupersetTestCase):
         theme = db.session.query(Theme).filter_by(id=self.regular_theme.id).first()
         assert theme.is_system_default is False
 
+    @with_config({"ENABLE_UI_THEME_ADMINISTRATION": True})
+    def test_gamma_cannot_write_themes(self):
+        """Test that gamma users, who only have read access to themes, are
+        rejected before reaching the admin-only check."""
+        # Login as gamma user
+        self.login(GAMMA_USERNAME)
+
+        # Try to set theme as system default
+        response = self.client.put(
+            f"/api/v1/theme/{self.regular_theme.id}/set_system_default"
+        )
+
+        # Should be forbidden at the permission layer, since gamma has no
+        # write access to themes at all
+        assert response.status_code == 403
+
+        # Verify theme is not system default
+        theme = db.session.query(Theme).filter_by(id=self.regular_theme.id).first()
+        assert theme.is_system_default is False
+
     @with_config({"ENABLE_UI_THEME_ADMINISTRATION": False})
     def test_system_theme_requires_config_enabled(self):
         """Test that system theme APIs require configuration to be enabled"""

tests/unit_tests/commands/semantic_layer/create_test.py

@@ -25,6 +25,7 @@ from superset.commands.semantic_layer.exceptions import (
     SemanticLayerCreateFailedError,
     SemanticLayerInvalidError,
 )
+from superset.exceptions import SupersetSecurityException
 
 
 def test_create_semantic_layer_success(mocker: MockerFixture) -> None:
@@ -56,6 +57,36 @@ def test_create_semantic_layer_success(mocker: MockerFixture) -> None:
     mock_cls.from_configuration.assert_called_once_with({"account": "test"})
 
 
+def test_create_semantic_layer_configuration_already_serialized(
+    mocker: MockerFixture,
+) -> None:
+    """When ``configuration`` is already a JSON string, it is passed through
+    to the DAO unchanged instead of being re-serialized."""
+    new_model = MagicMock()
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.create.SemanticLayerDAO",
+    )
+    dao.validate_uniqueness.return_value = True
+    dao.create.return_value = new_model
+
+    mock_cls = MagicMock()
+    mocker.patch.dict(
+        "superset.commands.semantic_layer.create.registry",
+        {"snowflake": mock_cls},
+    )
+
+    data = {
+        "name": "My Layer",
+        "type": "snowflake",
+        "configuration": '{"account": "test"}',
+    }
+    result = CreateSemanticLayerCommand(data).run()
+
+    assert result == new_model
+    dao.create.assert_called_once_with(attributes=data)
+
+
 def test_create_semantic_layer_unknown_type(mocker: MockerFixture) -> None:
     """Test that SemanticLayerInvalidError is raised for unknown type."""
     mocker.patch(
@@ -166,6 +197,11 @@ def test_create_semantic_view_success(mocker: MockerFixture) -> None:
     mock_model.name = "orders"
     dao_view.create.return_value = mock_model
 
+    mocker.patch(
+        "superset.commands.semantic_layer.create.current_user_can_modify_object",
+        return_value=True,
+    )
+
     from superset.commands.semantic_layer.create import CreateSemanticViewCommand
 
     result = CreateSemanticViewCommand(
@@ -182,6 +218,42 @@ def test_create_semantic_view_success(mocker: MockerFixture) -> None:
     )
 
 
+def test_create_semantic_view_configuration_already_serialized(
+    mocker: MockerFixture,
+) -> None:
+    """When ``configuration`` is already a JSON string, it is passed through
+    to the DAO unchanged instead of being re-serialized."""
+    mock_layer = MagicMock()
+    dao_layer = mocker.patch(
+        "superset.commands.semantic_layer.create.SemanticLayerDAO",
+    )
+    dao_layer.find_by_uuid.return_value = mock_layer
+
+    dao_view = mocker.patch(
+        "superset.commands.semantic_layer.create.SemanticViewDAO",
+    )
+    dao_view.validate_uniqueness.return_value = True
+    mock_model = MagicMock()
+    dao_view.create.return_value = mock_model
+
+    mocker.patch(
+        "superset.commands.semantic_layer.create.current_user_can_modify_object",
+        return_value=True,
+    )
+
+    from superset.commands.semantic_layer.create import CreateSemanticViewCommand
+
+    data = {
+        "name": "orders",
+        "semantic_layer_uuid": "layer-uuid",
+        "configuration": '{"db": "prod"}',
+    }
+    result = CreateSemanticViewCommand(data).run()
+
+    assert result == mock_model
+    dao_view.create.assert_called_once_with(attributes=data)
+
+
 def test_create_semantic_view_layer_not_found(mocker: MockerFixture) -> None:
     """Test CreateSemanticViewCommand raises when layer not found."""
     dao_layer = mocker.patch(
@@ -215,6 +287,11 @@ def test_create_semantic_view_duplicate(mocker: MockerFixture) -> None:
     )
     dao_view.validate_uniqueness.return_value = False
 
+    mocker.patch(
+        "superset.commands.semantic_layer.create.current_user_can_modify_object",
+        return_value=True,
+    )
+
     from superset.commands.semantic_layer.create import CreateSemanticViewCommand
     from superset.commands.semantic_layer.exceptions import (
         SemanticViewCreateFailedError,
@@ -228,3 +305,110 @@ def test_create_semantic_view_duplicate(mocker: MockerFixture) -> None:
                 "configuration": {"db": "prod"},
             }
         ).run()
+
+
+def test_create_semantic_view_forbidden(mocker: MockerFixture) -> None:
+    """Test CreateSemanticViewCommand raises when the caller may not modify
+    the parent layer."""
+    from superset.commands.semantic_layer.create import CreateSemanticViewCommand
+    from superset.commands.semantic_layer.exceptions import SemanticViewForbiddenError
+
+    mock_layer = MagicMock()
+    dao_layer = mocker.patch(
+        "superset.commands.semantic_layer.create.SemanticLayerDAO",
+    )
+    dao_layer.find_by_uuid.return_value = mock_layer
+
+    dao_view = mocker.patch(
+        "superset.commands.semantic_layer.create.SemanticViewDAO",
+    )
+
+    mocker.patch(
+        "superset.commands.semantic_layer.create.current_user_can_modify_object",
+        return_value=False,
+    )
+
+    with pytest.raises(SemanticViewForbiddenError):
+        CreateSemanticViewCommand(
+            {
+                "name": "orders",
+                "semantic_layer_uuid": "layer-uuid",
+                "configuration": {"db": "prod"},
+            }
+        ).run()
+
+    dao_view.create.assert_not_called()
+
+
+def test_create_semantic_view_creator_of_layer_allowed(mocker: MockerFixture) -> None:
+    """A non-admin who created the parent layer, but holds no explicit
+    editorship on it, can still add a semantic view to it."""
+    from superset.commands.semantic_layer.create import CreateSemanticViewCommand
+
+    mock_layer = MagicMock()
+    dao_layer = mocker.patch(
+        "superset.commands.semantic_layer.create.SemanticLayerDAO",
+    )
+    dao_layer.find_by_uuid.return_value = mock_layer
+
+    dao_view = mocker.patch(
+        "superset.commands.semantic_layer.create.SemanticViewDAO",
+    )
+    dao_view.validate_uniqueness.return_value = True
+    mock_model = MagicMock()
+    dao_view.create.return_value = mock_model
+
+    sm = mocker.patch("superset.commands.utils.security_manager")
+    sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock()),
+    )
+    mock_layer.created_by = sm.current_user
+
+    result = CreateSemanticViewCommand(
+        {
+            "name": "orders",
+            "semantic_layer_uuid": "layer-uuid",
+            "configuration": {"db": "prod"},
+        }
+    ).run()
+
+    assert result == mock_model
+    dao_view.create.assert_called_once()
+
+
+def test_create_semantic_view_non_creator_non_editor_forbidden(
+    mocker: MockerFixture,
+) -> None:
+    """A non-admin who neither created the parent layer nor is an editor of
+    it is rejected."""
+    from superset.commands.semantic_layer.create import CreateSemanticViewCommand
+    from superset.commands.semantic_layer.exceptions import SemanticViewForbiddenError
+
+    mock_layer = MagicMock()
+    dao_layer = mocker.patch(
+        "superset.commands.semantic_layer.create.SemanticLayerDAO",
+    )
+    dao_layer.find_by_uuid.return_value = mock_layer
+
+    dao_view = mocker.patch(
+        "superset.commands.semantic_layer.create.SemanticViewDAO",
+    )
+
+    sm = mocker.patch(
+        "superset.commands.utils.security_manager",
+    )
+    sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock()),
+    )
+    mock_layer.created_by = MagicMock(name="someone_else")
+
+    with pytest.raises(SemanticViewForbiddenError):
+        CreateSemanticViewCommand(
+            {
+                "name": "orders",
+                "semantic_layer_uuid": "layer-uuid",
+                "configuration": {"db": "prod"},
+            }
+        ).run()
+
+    dao_view.create.assert_not_called()

tests/unit_tests/commands/semantic_layer/delete_test.py

@@ -21,7 +21,11 @@ import pytest
 from pytest_mock import MockerFixture
 
 from superset.commands.semantic_layer.delete import DeleteSemanticLayerCommand
-from superset.commands.semantic_layer.exceptions import SemanticLayerNotFoundError
+from superset.commands.semantic_layer.exceptions import (
+    SemanticLayerForbiddenError,
+    SemanticLayerNotFoundError,
+)
+from superset.exceptions import SupersetSecurityException
 
 
 def test_delete_semantic_layer_success(mocker: MockerFixture) -> None:
@@ -33,6 +37,11 @@ def test_delete_semantic_layer_success(mocker: MockerFixture) -> None:
     )
     dao.find_by_uuid.return_value = mock_model
 
+    mocker.patch(
+        "superset.commands.semantic_layer.delete.current_user_can_modify_object",
+        return_value=True,
+    )
+
     DeleteSemanticLayerCommand("some-uuid").run()
 
     dao.find_by_uuid.assert_called_once_with("some-uuid")
@@ -50,6 +59,71 @@ def test_delete_semantic_layer_not_found(mocker: MockerFixture) -> None:
         DeleteSemanticLayerCommand("missing-uuid").run()
 
 
+def test_delete_semantic_layer_forbidden(mocker: MockerFixture) -> None:
+    """Test that SemanticLayerForbiddenError is raised for non-editors."""
+    mock_model = MagicMock()
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.delete.SemanticLayerDAO",
+    )
+    dao.find_by_uuid.return_value = mock_model
+
+    mocker.patch(
+        "superset.commands.semantic_layer.delete.current_user_can_modify_object",
+        return_value=False,
+    )
+
+    with pytest.raises(SemanticLayerForbiddenError):
+        DeleteSemanticLayerCommand("some-uuid").run()
+
+    dao.delete.assert_not_called()
+
+
+def test_delete_semantic_layer_creator_allowed(mocker: MockerFixture) -> None:
+    """A non-admin who created the layer, but holds no explicit editorship
+    on it, can still delete it."""
+    mock_model = MagicMock()
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.delete.SemanticLayerDAO",
+    )
+    dao.find_by_uuid.return_value = mock_model
+
+    sm = mocker.patch("superset.commands.utils.security_manager")
+    sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock()),
+    )
+    mock_model.created_by = sm.current_user
+
+    DeleteSemanticLayerCommand("some-uuid").run()
+
+    dao.delete.assert_called_once_with([mock_model])
+
+
+def test_delete_semantic_layer_non_creator_non_editor_forbidden(
+    mocker: MockerFixture,
+) -> None:
+    """A non-admin who neither created the layer nor is an editor of it is
+    rejected."""
+    mock_model = MagicMock()
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.delete.SemanticLayerDAO",
+    )
+    dao.find_by_uuid.return_value = mock_model
+
+    sm = mocker.patch("superset.commands.utils.security_manager")
+    sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock()),
+    )
+    mock_model.created_by = MagicMock(name="someone_else")
+
+    with pytest.raises(SemanticLayerForbiddenError):
+        DeleteSemanticLayerCommand("some-uuid").run()
+
+    dao.delete.assert_not_called()
+
+
 def test_delete_semantic_view_success(mocker: MockerFixture) -> None:
     """Test successful deletion of a semantic view."""
     mock_model = MagicMock()
@@ -59,10 +133,11 @@ def test_delete_semantic_view_success(mocker: MockerFixture) -> None:
     )
     dao.find_by_id.return_value = mock_model
 
-    # Admin is owner of everything — no exception raised
+    # Admin (or an editor) can modify anything — no exception raised.
     mocker.patch(
-        "superset.commands.semantic_layer.delete.security_manager"
-    ).raise_for_editorship.return_value = None
+        "superset.commands.semantic_layer.delete.current_user_can_modify_object",
+        return_value=True,
+    )
 
     from superset.commands.semantic_layer.delete import DeleteSemanticViewCommand
 
@@ -76,12 +151,13 @@ def test_delete_semantic_view_forbidden(mocker: MockerFixture) -> None:
     """Test that SemanticViewForbiddenError is raised for non-owners."""
     from superset.commands.semantic_layer.delete import DeleteSemanticViewCommand
     from superset.commands.semantic_layer.exceptions import SemanticViewForbiddenError
-    from superset.exceptions import SupersetSecurityException
 
     dao = mocker.patch(
         "superset.commands.semantic_layer.delete.SemanticViewDAO",
     )
-    dao.find_by_id.return_value = MagicMock()
+    model = MagicMock()
+    model.created_by = None
+    dao.find_by_id.return_value = model
 
     mocker.patch(
         "superset.security_manager.raise_for_editorship",
@@ -92,6 +168,56 @@ def test_delete_semantic_view_forbidden(mocker: MockerFixture) -> None:
         DeleteSemanticViewCommand(42).run()
 
 
+def test_delete_semantic_view_creator_allowed(mocker: MockerFixture) -> None:
+    """A non-admin who created the view, but holds no explicit editorship on
+    it, can still delete it."""
+    from superset.commands.semantic_layer.delete import DeleteSemanticViewCommand
+
+    mock_model = MagicMock()
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.delete.SemanticViewDAO",
+    )
+    dao.find_by_id.return_value = mock_model
+
+    sm = mocker.patch("superset.commands.utils.security_manager")
+    sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock()),
+    )
+    mock_model.created_by = sm.current_user
+
+    DeleteSemanticViewCommand(42).run()
+
+    dao.delete.assert_called_once_with([mock_model])
+
+
+def test_delete_semantic_view_non_creator_non_editor_forbidden(
+    mocker: MockerFixture,
+) -> None:
+    """A non-admin who neither created the view nor is an editor of it is
+    rejected."""
+    from superset.commands.semantic_layer.delete import DeleteSemanticViewCommand
+    from superset.commands.semantic_layer.exceptions import SemanticViewForbiddenError
+
+    mock_model = MagicMock()
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.delete.SemanticViewDAO",
+    )
+    dao.find_by_id.return_value = mock_model
+
+    sm = mocker.patch("superset.commands.utils.security_manager")
+    sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock()),
+    )
+    mock_model.created_by = MagicMock(name="someone_else")
+
+    with pytest.raises(SemanticViewForbiddenError):
+        DeleteSemanticViewCommand(42).run()
+
+    dao.delete.assert_not_called()
+
+
 def test_delete_semantic_view_not_found(mocker: MockerFixture) -> None:
     """Test that SemanticViewNotFoundError is raised when view is missing."""
     dao = mocker.patch(
@@ -118,8 +244,9 @@ def test_bulk_delete_semantic_view_success(mocker: MockerFixture) -> None:
     dao.find_by_ids.return_value = mock_models
 
     mocker.patch(
-        "superset.commands.semantic_layer.delete.security_manager"
-    ).raise_for_editorship.return_value = None
+        "superset.commands.semantic_layer.delete.current_user_can_modify_object",
+        return_value=True,
+    )
 
     from superset.commands.semantic_layer.delete import BulkDeleteSemanticViewCommand
 
@@ -133,7 +260,6 @@ def test_bulk_delete_semantic_view_forbidden(mocker: MockerFixture) -> None:
     """Test that SemanticViewForbiddenError is raised for non-owners."""
     from superset.commands.semantic_layer.delete import BulkDeleteSemanticViewCommand
     from superset.commands.semantic_layer.exceptions import SemanticViewForbiddenError
-    from superset.exceptions import SupersetSecurityException
 
     dao = mocker.patch(
         "superset.commands.semantic_layer.delete.SemanticViewDAO",
@@ -141,14 +267,67 @@ def test_bulk_delete_semantic_view_forbidden(mocker: MockerFixture) -> None:
     dao.find_by_ids.return_value = [MagicMock(), MagicMock()]
 
     mocker.patch(
-        "superset.security_manager.raise_for_editorship",
-        side_effect=SupersetSecurityException(MagicMock()),
+        "superset.commands.semantic_layer.delete.current_user_can_modify_object",
+        return_value=False,
     )
 
     with pytest.raises(SemanticViewForbiddenError):
         BulkDeleteSemanticViewCommand([1, 2]).run()
 
 
+def test_bulk_delete_semantic_view_creator_allowed(mocker: MockerFixture) -> None:
+    """A non-admin who created every view in the batch, but holds no
+    explicit editorship on them, can still bulk-delete them."""
+    from superset.commands.semantic_layer.delete import BulkDeleteSemanticViewCommand
+
+    mock_models = [MagicMock(), MagicMock()]
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.delete.SemanticViewDAO",
+    )
+    dao.find_by_ids.return_value = mock_models
+
+    sm = mocker.patch("superset.commands.utils.security_manager")
+    sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock()),
+    )
+    for model in mock_models:
+        model.created_by = sm.current_user
+
+    BulkDeleteSemanticViewCommand([1, 2]).run()
+
+    dao.delete.assert_called_once_with(mock_models)
+
+
+def test_bulk_delete_semantic_view_non_creator_non_editor_forbidden(
+    mocker: MockerFixture,
+) -> None:
+    """A non-admin who is neither the creator of, nor an editor for, one of
+    the views in the batch is rejected."""
+    from superset.commands.semantic_layer.delete import BulkDeleteSemanticViewCommand
+    from superset.commands.semantic_layer.exceptions import SemanticViewForbiddenError
+
+    mock_models = [MagicMock(), MagicMock()]
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.delete.SemanticViewDAO",
+    )
+    dao.find_by_ids.return_value = mock_models
+
+    sm = mocker.patch("superset.commands.utils.security_manager")
+    sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock()),
+    )
+    # The first view belongs to the current user, the second doesn't.
+    mock_models[0].created_by = sm.current_user
+    mock_models[1].created_by = MagicMock(name="someone_else")
+
+    with pytest.raises(SemanticViewForbiddenError):
+        BulkDeleteSemanticViewCommand([1, 2]).run()
+
+    dao.delete.assert_not_called()
+
+
 def test_bulk_delete_semantic_view_not_found(mocker: MockerFixture) -> None:
     """Test that SemanticViewNotFoundError is raised when any id is missing."""
     dao = mocker.patch(

tests/unit_tests/commands/semantic_layer/update_test.py

@@ -21,6 +21,7 @@ import pytest
 from pytest_mock import MockerFixture
 
 from superset.commands.semantic_layer.exceptions import (
+    SemanticLayerForbiddenError,
     SemanticLayerInvalidError,
     SemanticLayerNotFoundError,
     SemanticViewForbiddenError,
@@ -46,7 +47,7 @@ def test_update_semantic_view_success(mocker: MockerFixture) -> None:
     dao.update.return_value = mock_model
 
     mocker.patch(
-        "superset.commands.semantic_layer.update.security_manager",
+        "superset.commands.semantic_layer.update.current_user_can_modify_object",
     )
 
     data = {"description": "Updated", "cache_timeout": 300}
@@ -77,18 +78,65 @@ def test_update_semantic_view_forbidden(mocker: MockerFixture) -> None:
     )
     dao.find_by_id.return_value = mock_model
 
-    sm = mocker.patch(
-        "superset.commands.semantic_layer.update.security_manager",
-    )
-    # Use a regular MagicMock for raise_for_editorship to avoid AsyncMock issues
-    sm.raise_for_editorship = MagicMock(
-        side_effect=SupersetSecurityException(MagicMock()),
+    mocker.patch(
+        "superset.commands.semantic_layer.update.current_user_can_modify_object",
+        return_value=False,
     )
 
     with pytest.raises(SemanticViewForbiddenError):
         UpdateSemanticViewCommand(1, {"description": "test"}).run()
 
 
+def test_update_semantic_view_creator_allowed(mocker: MockerFixture) -> None:
+    """A non-admin who created the view, but holds no explicit editorship on
+    it, can still update it."""
+    mock_model = MagicMock()
+    mock_model.id = 1
+    mock_model.configuration = "{}"
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.update.SemanticViewDAO",
+    )
+    dao.find_by_id.return_value = mock_model
+    dao.update.return_value = mock_model
+
+    sm = mocker.patch("superset.commands.utils.security_manager")
+    sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock()),
+    )
+    mock_model.created_by = sm.current_user
+
+    data = {"description": "Updated"}
+    result = UpdateSemanticViewCommand(1, data).run()
+
+    assert result == mock_model
+    dao.update.assert_called_once_with(mock_model, attributes=data)
+
+
+def test_update_semantic_view_non_creator_non_editor_forbidden(
+    mocker: MockerFixture,
+) -> None:
+    """A non-admin who neither created the view nor is an editor of it is
+    rejected."""
+    mock_model = MagicMock()
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.update.SemanticViewDAO",
+    )
+    dao.find_by_id.return_value = mock_model
+
+    sm = mocker.patch("superset.commands.utils.security_manager")
+    sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock()),
+    )
+    mock_model.created_by = MagicMock(name="someone_else")
+
+    with pytest.raises(SemanticViewForbiddenError):
+        UpdateSemanticViewCommand(1, {"description": "test"}).run()
+
+    dao.update.assert_not_called()
+
+
 def test_update_semantic_view_copies_data(mocker: MockerFixture) -> None:
     """Test that the command copies input data and does not mutate it."""
     mock_model = MagicMock()
@@ -101,7 +149,7 @@ def test_update_semantic_view_copies_data(mocker: MockerFixture) -> None:
     dao.update.return_value = mock_model
 
     mocker.patch(
-        "superset.commands.semantic_layer.update.security_manager",
+        "superset.commands.semantic_layer.update.current_user_can_modify_object",
     )
 
     original_data = {"description": "Original"}
@@ -127,6 +175,10 @@ def test_update_semantic_layer_success(mocker: MockerFixture) -> None:
     dao.find_by_uuid.return_value = mock_model
     dao.update.return_value = mock_model
 
+    mocker.patch(
+        "superset.commands.semantic_layer.update.current_user_can_modify_object",
+    )
+
     data = {"name": "Updated", "description": "New desc"}
     result = UpdateSemanticLayerCommand("some-uuid", data).run()
 
@@ -146,6 +198,77 @@ def test_update_semantic_layer_not_found(mocker: MockerFixture) -> None:
         UpdateSemanticLayerCommand("missing-uuid", {"name": "test"}).run()
 
 
+def test_update_semantic_layer_forbidden(mocker: MockerFixture) -> None:
+    """Test that SemanticLayerForbiddenError is raised on ownership failure."""
+    mock_model = MagicMock()
+    mock_model.type = "snowflake"
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.update.SemanticLayerDAO",
+    )
+    dao.find_by_uuid.return_value = mock_model
+
+    mocker.patch(
+        "superset.commands.semantic_layer.update.current_user_can_modify_object",
+        return_value=False,
+    )
+
+    with pytest.raises(SemanticLayerForbiddenError):
+        UpdateSemanticLayerCommand("some-uuid", {"name": "test"}).run()
+
+    dao.update.assert_not_called()
+
+
+def test_update_semantic_layer_creator_allowed(mocker: MockerFixture) -> None:
+    """A non-admin who created the layer, but holds no explicit editorship
+    on it, can still update it."""
+    mock_model = MagicMock()
+    mock_model.type = "snowflake"
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.update.SemanticLayerDAO",
+    )
+    dao.find_by_uuid.return_value = mock_model
+    dao.update.return_value = mock_model
+
+    sm = mocker.patch("superset.commands.utils.security_manager")
+    sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock()),
+    )
+    mock_model.created_by = sm.current_user
+
+    data = {"description": "Updated"}
+    result = UpdateSemanticLayerCommand("some-uuid", data).run()
+
+    assert result == mock_model
+    dao.update.assert_called_once_with(mock_model, attributes=data)
+
+
+def test_update_semantic_layer_non_creator_non_editor_forbidden(
+    mocker: MockerFixture,
+) -> None:
+    """A non-admin who neither created the layer nor is an editor of it is
+    rejected."""
+    mock_model = MagicMock()
+    mock_model.type = "snowflake"
+
+    dao = mocker.patch(
+        "superset.commands.semantic_layer.update.SemanticLayerDAO",
+    )
+    dao.find_by_uuid.return_value = mock_model
+
+    sm = mocker.patch("superset.commands.utils.security_manager")
+    sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock()),
+    )
+    mock_model.created_by = MagicMock(name="someone_else")
+
+    with pytest.raises(SemanticLayerForbiddenError):
+        UpdateSemanticLayerCommand("some-uuid", {"name": "test"}).run()
+
+    dao.update.assert_not_called()
+
+
 def test_update_semantic_layer_duplicate_name(mocker: MockerFixture) -> None:
     """Test that SemanticLayerInvalidError is raised for duplicate names."""
     mock_model = MagicMock()
@@ -157,6 +280,10 @@ def test_update_semantic_layer_duplicate_name(mocker: MockerFixture) -> None:
     dao.find_by_uuid.return_value = mock_model
     dao.validate_update_uniqueness.return_value = False
 
+    mocker.patch(
+        "superset.commands.semantic_layer.update.current_user_can_modify_object",
+    )
+
     with pytest.raises(SemanticLayerInvalidError):
         UpdateSemanticLayerCommand("some-uuid", {"name": "Duplicate"}).run()
 
@@ -174,6 +301,10 @@ def test_update_semantic_layer_validates_configuration(
     dao.find_by_uuid.return_value = mock_model
     dao.update.return_value = mock_model
 
+    mocker.patch(
+        "superset.commands.semantic_layer.update.current_user_can_modify_object",
+    )
+
     mock_cls = MagicMock()
     mocker.patch.dict(
         "superset.commands.semantic_layer.update.registry",
@@ -199,6 +330,10 @@ def test_update_semantic_layer_skips_name_check_when_no_name(
     dao.find_by_uuid.return_value = mock_model
     dao.update.return_value = mock_model
 
+    mocker.patch(
+        "superset.commands.semantic_layer.update.current_user_can_modify_object",
+    )
+
     UpdateSemanticLayerCommand("some-uuid", {"description": "Updated"}).run()
 
     dao.validate_update_uniqueness.assert_not_called()
@@ -215,6 +350,10 @@ def test_update_semantic_layer_copies_data(mocker: MockerFixture) -> None:
     dao.find_by_uuid.return_value = mock_model
     dao.update.return_value = mock_model
 
+    mocker.patch(
+        "superset.commands.semantic_layer.update.current_user_can_modify_object",
+    )
+
     original_data = {"description": "Original"}
     UpdateSemanticLayerCommand("some-uuid", original_data).run()
 
@@ -249,7 +388,7 @@ def test_update_uniqueness_different_config_same_name(
     dao.validate_update_uniqueness.return_value = True
 
     mocker.patch(
-        "superset.commands.semantic_layer.update.security_manager",
+        "superset.commands.semantic_layer.update.current_user_can_modify_object",
     )
 
     # Update to a config that differs from an existing view
@@ -279,7 +418,7 @@ def test_update_uniqueness_same_config_different_name(
     dao.validate_update_uniqueness.return_value = True
 
     mocker.patch(
-        "superset.commands.semantic_layer.update.security_manager",
+        "superset.commands.semantic_layer.update.current_user_can_modify_object",
     )
 
     data = {"name": "renamed_view", "configuration": {"schema": "prod"}}
@@ -307,7 +446,7 @@ def test_update_uniqueness_same_config_same_name_fails(
     dao.validate_update_uniqueness.return_value = False
 
     mocker.patch(
-        "superset.commands.semantic_layer.update.security_manager",
+        "superset.commands.semantic_layer.update.current_user_can_modify_object",
     )
 
     from superset.commands.semantic_layer.exceptions import (

tests/unit_tests/commands/test_utils.py

@@ -22,11 +22,13 @@ import pytest
 
 from superset.commands.exceptions import TagForbiddenError, TagNotFoundValidationError
 from superset.commands.utils import (
+    current_user_can_modify_object,
     Tag,
     TagType,
     update_tags,
     validate_tags,
 )
+from superset.exceptions import SupersetSecurityException
 from superset.tags.models import ObjectType
 
 OBJECT_TYPES = {ObjectType.chart, ObjectType.chart}
@@ -343,3 +345,60 @@ def test_update_tags_no_tags(mock_tag_dao, object_type):
     mock_tag_dao.create_custom_tagged_objects.assert_called_once_with(
         object_type, 1, new_tag_names
     )
+
+
+@patch("superset.commands.utils.security_manager")
+def test_current_user_can_modify_object_editor(mock_sm):
+    """
+    An editor of the resource (or an admin, since ``raise_for_editorship``
+    treats admins as editors of everything) is allowed to modify it.
+    """
+    mock_sm.raise_for_editorship.return_value = None
+    model = MagicMock()
+
+    assert current_user_can_modify_object(model) is True
+
+
+@patch("superset.commands.utils.security_manager")
+def test_current_user_can_modify_object_creator_fallback(mock_sm):
+    """
+    A resource without an ``editors`` relationship (or a user who isn't in
+    it) still allows the object's creator through.
+    """
+    mock_sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock())
+    )
+    model = MagicMock()
+    model.created_by = mock_sm.current_user
+
+    assert current_user_can_modify_object(model) is True
+
+
+@patch("superset.commands.utils.security_manager")
+def test_current_user_can_modify_object_denies_non_creator(mock_sm):
+    """
+    A user who is neither an editor nor the creator is denied.
+    """
+    mock_sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock())
+    )
+    mock_sm.current_user = MagicMock(name="current_user")
+    model = MagicMock()
+    model.created_by = MagicMock(name="someone_else")
+
+    assert current_user_can_modify_object(model) is False
+
+
+@patch("superset.commands.utils.security_manager")
+def test_current_user_can_modify_object_no_creator(mock_sm):
+    """
+    A resource with no ``created_by`` set (e.g. created programmatically)
+    is denied to non-editors.
+    """
+    mock_sm.raise_for_editorship = MagicMock(
+        side_effect=SupersetSecurityException(MagicMock())
+    )
+    model = MagicMock()
+    model.created_by = None
+
+    assert current_user_can_modify_object(model) is False

tests/unit_tests/commands/theme/test_import_themes.py

@@ -0,0 +1,193 @@
+# 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 unittest.mock import MagicMock, Mock, patch
+
+import pytest
+
+from superset.commands.theme.exceptions import ThemeImportError
+from superset.commands.theme.import_themes import import_theme
+from superset.models.core import Theme
+
+
+def _mock_existing(
+    is_system: bool = False,
+    is_system_default: bool = False,
+    is_system_dark: bool = False,
+) -> MagicMock:
+    theme = MagicMock(spec=Theme)
+    theme.id = 1
+    theme.is_system = is_system
+    theme.is_system_default = is_system_default
+    theme.is_system_dark = is_system_dark
+    return theme
+
+
+@patch("superset.security_manager")
+@patch("superset.db")
+def test_import_theme_refuses_system_theme_overwrite(mock_db, mock_security_manager):
+    """overwrite=True must not be able to replace a seeded system theme."""
+    mock_security_manager.can_access.return_value = True
+    existing = _mock_existing(is_system=True)
+    mock_db.session.query.return_value.filter_by.return_value.first.return_value = (
+        existing
+    )
+
+    config = {"uuid": "some-uuid", "theme_name": "hostile", "json_data": "{}"}
+
+    with pytest.raises(ThemeImportError):
+        import_theme(config, overwrite=True)
+
+
+@patch("superset.security_manager")
+@patch("superset.db")
+def test_import_theme_refuses_system_default_overwrite(mock_db, mock_security_manager):
+    """A non-admin overwrite=True must not replace the active default theme."""
+    mock_security_manager.can_access.return_value = True
+    # Use a regular Mock for is_admin to avoid AsyncMock auto-detection
+    mock_security_manager.is_admin = Mock(return_value=False)
+    existing = _mock_existing(is_system_default=True)
+    mock_db.session.query.return_value.filter_by.return_value.first.return_value = (
+        existing
+    )
+
+    config = {"uuid": "some-uuid", "theme_name": "hostile", "json_data": "{}"}
+
+    with pytest.raises(ThemeImportError):
+        import_theme(config, overwrite=True)
+
+
+@patch("superset.security_manager")
+@patch("superset.db")
+def test_import_theme_refuses_system_dark_overwrite(mock_db, mock_security_manager):
+    """A non-admin overwrite=True must not replace the active dark theme."""
+    mock_security_manager.can_access.return_value = True
+    # Use a regular Mock for is_admin to avoid AsyncMock auto-detection
+    mock_security_manager.is_admin = Mock(return_value=False)
+    existing = _mock_existing(is_system_dark=True)
+    mock_db.session.query.return_value.filter_by.return_value.first.return_value = (
+        existing
+    )
+
+    config = {"uuid": "some-uuid", "theme_name": "hostile", "json_data": "{}"}
+
+    with pytest.raises(ThemeImportError):
+        import_theme(config, overwrite=True)
+
+
+@patch("superset.utils.core.get_user")
+@patch("superset.security_manager")
+@patch("superset.db")
+def test_import_theme_admin_allows_system_default_overwrite(
+    mock_db, mock_security_manager, mock_get_user
+):
+    """An admin overwrite=True may still replace the active default theme,
+    mirroring UpdateThemeCommand's admin carve-out."""
+    mock_security_manager.can_access.return_value = True
+    # Use a regular Mock for is_admin to avoid AsyncMock auto-detection
+    mock_security_manager.is_admin = Mock(return_value=True)
+    mock_get_user.return_value = None
+    existing = _mock_existing(is_system_default=True)
+    mock_db.session.query.return_value.filter_by.return_value.first.return_value = (
+        existing
+    )
+
+    config = {"uuid": "some-uuid", "theme_name": "updated", "json_data": "{}"}
+
+    with patch("superset.models.core.Theme.import_from_dict") as mock_import_from_dict:
+        mock_theme = MagicMock(spec=Theme)
+        mock_theme.id = 1
+        mock_import_from_dict.return_value = mock_theme
+
+        result = import_theme(config, overwrite=True)
+
+    assert result is mock_theme
+    assert config["id"] == existing.id
+
+
+@patch("superset.utils.core.get_user")
+@patch("superset.security_manager")
+@patch("superset.db")
+def test_import_theme_admin_allows_system_dark_overwrite(
+    mock_db, mock_security_manager, mock_get_user
+):
+    """An admin overwrite=True may still replace the active dark theme,
+    mirroring UpdateThemeCommand's admin carve-out."""
+    mock_security_manager.can_access.return_value = True
+    # Use a regular Mock for is_admin to avoid AsyncMock auto-detection
+    mock_security_manager.is_admin = Mock(return_value=True)
+    mock_get_user.return_value = None
+    existing = _mock_existing(is_system_dark=True)
+    mock_db.session.query.return_value.filter_by.return_value.first.return_value = (
+        existing
+    )
+
+    config = {"uuid": "some-uuid", "theme_name": "updated", "json_data": "{}"}
+
+    with patch("superset.models.core.Theme.import_from_dict") as mock_import_from_dict:
+        mock_theme = MagicMock(spec=Theme)
+        mock_theme.id = 1
+        mock_import_from_dict.return_value = mock_theme
+
+        result = import_theme(config, overwrite=True)
+
+    assert result is mock_theme
+    assert config["id"] == existing.id
+
+
+@patch("superset.utils.core.get_user")
+@patch("superset.security_manager")
+@patch("superset.db")
+def test_import_theme_allows_regular_theme_overwrite(
+    mock_db, mock_security_manager, mock_get_user
+):
+    """A regular (non-system) theme can still be overwritten as before."""
+    mock_security_manager.can_access.return_value = True
+    mock_get_user.return_value = None
+    existing = _mock_existing()
+    mock_db.session.query.return_value.filter_by.return_value.first.return_value = (
+        existing
+    )
+
+    config = {"uuid": "some-uuid", "theme_name": "updated", "json_data": "{}"}
+
+    with patch("superset.models.core.Theme.import_from_dict") as mock_import_from_dict:
+        mock_theme = MagicMock(spec=Theme)
+        mock_theme.id = 1
+        mock_import_from_dict.return_value = mock_theme
+
+        result = import_theme(config, overwrite=True)
+
+    assert result is mock_theme
+    assert config["id"] == existing.id
+
+
+@patch("superset.security_manager")
+@patch("superset.db")
+def test_import_theme_no_overwrite_returns_existing(mock_db, mock_security_manager):
+    """Without overwrite=True, the existing theme is returned untouched."""
+    mock_security_manager.can_access.return_value = True
+    existing = _mock_existing(is_system_default=True)
+    mock_db.session.query.return_value.filter_by.return_value.first.return_value = (
+        existing
+    )
+
+    config = {"uuid": "some-uuid", "theme_name": "hostile", "json_data": "{}"}
+
+    result = import_theme(config, overwrite=False)
+
+    assert result is existing

tests/unit_tests/semantic_layers/api_test.py

@@ -36,6 +36,7 @@ from superset.commands.semantic_layer.exceptions import (
     SemanticViewNotFoundError,
     SemanticViewUpdateFailedError,
 )
+from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.exceptions import SupersetSecurityException
 from superset.semantic_layers.api import SemanticLayerRestApi, SemanticViewRestApi
 
@@ -529,6 +530,34 @@ def test_runtime_schema_not_found(
     assert response.status_code == 404
 
 
+@SEMANTIC_LAYERS_APP
+def test_runtime_schema_forbidden(
+    client: Any,
+    full_api_access: None,
+    mocker: MockerFixture,
+) -> None:
+    """Test POST /<uuid>/schema/runtime returns 403 when access is denied."""
+    test_uuid = str(uuid_lib.uuid4())
+    mock_layer = MagicMock()
+    mock_layer.raise_for_access.side_effect = SupersetSecurityException(
+        SupersetError(
+            error_type=SupersetErrorType.DATASOURCE_SECURITY_ACCESS_ERROR,
+            message="You don't have access to this semantic layer.",
+            level=ErrorLevel.ERROR,
+        )
+    )
+
+    mock_dao = mocker.patch("superset.semantic_layers.api.SemanticLayerDAO")
+    mock_dao.find_by_uuid.return_value = mock_layer
+
+    response = client.post(
+        f"/api/v1/semantic_layer/{test_uuid}/schema/runtime",
+    )
+
+    assert response.status_code == 403
+    mock_layer.raise_for_access.assert_called_once()
+
+
 @SEMANTIC_LAYERS_APP
 def test_runtime_schema_unknown_type(
     client: Any,
@@ -729,6 +758,30 @@ def test_put_semantic_layer_not_found(
     assert response.status_code == 404
 
 
+@SEMANTIC_LAYERS_APP
+def test_put_semantic_layer_forbidden(
+    client: Any,
+    full_api_access: None,
+    mocker: MockerFixture,
+) -> None:
+    """Test PUT /<uuid> returns 403 when the caller is not an editor."""
+    from superset.commands.semantic_layer.exceptions import (
+        SemanticLayerForbiddenError,
+    )
+
+    mock_command = mocker.patch(
+        "superset.semantic_layers.api.UpdateSemanticLayerCommand",
+    )
+    mock_command.return_value.run.side_effect = SemanticLayerForbiddenError()
+
+    response = client.put(
+        f"/api/v1/semantic_layer/{uuid_lib.uuid4()}",
+        json={"name": "New"},
+    )
+
+    assert response.status_code == 403
+
+
 @SEMANTIC_LAYERS_APP
 def test_put_semantic_layer_invalid(
     client: Any,
@@ -821,6 +874,27 @@ def test_delete_semantic_layer_not_found(
     assert response.status_code == 404
 
 
+@SEMANTIC_LAYERS_APP
+def test_delete_semantic_layer_forbidden(
+    client: Any,
+    full_api_access: None,
+    mocker: MockerFixture,
+) -> None:
+    """Test DELETE /<uuid> returns 403 when the caller is not an editor."""
+    from superset.commands.semantic_layer.exceptions import (
+        SemanticLayerForbiddenError,
+    )
+
+    mock_command = mocker.patch(
+        "superset.semantic_layers.api.DeleteSemanticLayerCommand",
+    )
+    mock_command.return_value.run.side_effect = SemanticLayerForbiddenError()
+
+    response = client.delete(f"/api/v1/semantic_layer/{uuid_lib.uuid4()}")
+
+    assert response.status_code == 403
+
+
 @SEMANTIC_LAYERS_APP
 def test_delete_semantic_layer_failed(
     client: Any,
@@ -939,6 +1013,33 @@ def test_get_semantic_layer_not_found(
     assert response.status_code == 404
 
 
+@SEMANTIC_LAYERS_APP
+def test_get_semantic_layer_forbidden(
+    client: Any,
+    full_api_access: None,
+    mocker: MockerFixture,
+) -> None:
+    """Test GET /<uuid> returns 403 when user lacks access to the layer."""
+    test_uuid = uuid_lib.uuid4()
+    layer = MagicMock()
+    layer.uuid = test_uuid
+    layer.raise_for_access.side_effect = SupersetSecurityException(
+        SupersetError(
+            error_type=SupersetErrorType.DATASOURCE_SECURITY_ACCESS_ERROR,
+            message="You don't have access to this semantic layer.",
+            level=ErrorLevel.ERROR,
+        )
+    )
+
+    mock_dao = mocker.patch("superset.semantic_layers.api.SemanticLayerDAO")
+    mock_dao.find_by_uuid.return_value = layer
+
+    response = client.get(f"/api/v1/semantic_layer/{test_uuid}")
+
+    assert response.status_code == 403
+    layer.raise_for_access.assert_called_once()
+
+
 @SEMANTIC_LAYERS_APP
 def test_serialize_layer_string_config(
     client: Any,
@@ -1645,6 +1746,35 @@ def test_post_semantic_view_layer_not_found(
     assert result["errors"][0]["error"] == "Semantic layer not found"
 
 
+@SEMANTIC_LAYERS_APP
+def test_post_semantic_view_forbidden(
+    client: Any,
+    full_api_access: None,
+    mocker: MockerFixture,
+) -> None:
+    """Test POST / collects forbidden errors instead of aborting the batch."""
+    mock_command = mocker.patch(
+        "superset.semantic_layers.api.CreateSemanticViewCommand",
+    )
+    mock_command.return_value.run.side_effect = SemanticViewForbiddenError()
+
+    payload = {
+        "views": [
+            {
+                "name": "View 1",
+                "semantic_layer_uuid": str(uuid_lib.uuid4()),
+                "configuration": {},
+            },
+        ],
+    }
+    response = client.post("/api/v1/semantic_view/", json=payload)
+
+    assert response.status_code == 422
+    result = response.json["result"]
+    assert len(result["errors"]) == 1
+    assert not result["created"]
+
+
 @SEMANTIC_LAYERS_APP
 def test_post_semantic_view_create_failed(
     client: Any,
@@ -1956,6 +2086,37 @@ def test_get_views(
     assert result[1]["name"] == "View B"
 
 
+@SEMANTIC_LAYERS_APP
+def test_get_views_forbidden(
+    client: Any,
+    full_api_access: None,
+    mocker: MockerFixture,
+) -> None:
+    """Test POST /<uuid>/views returns 403 when access is denied."""
+    test_uuid = str(uuid_lib.uuid4())
+    mock_layer = MagicMock()
+    mock_layer.uuid = uuid_lib.uuid4()
+    mock_layer.raise_for_access.side_effect = SupersetSecurityException(
+        SupersetError(
+            error_type=SupersetErrorType.DATASOURCE_SECURITY_ACCESS_ERROR,
+            message="You don't have access to this semantic layer.",
+            level=ErrorLevel.ERROR,
+        )
+    )
+
+    mock_dao = mocker.patch("superset.semantic_layers.api.SemanticLayerDAO")
+    mock_dao.find_by_uuid.return_value = mock_layer
+
+    response = client.post(
+        f"/api/v1/semantic_layer/{test_uuid}/views",
+        json={"runtime_data": {"database": "mydb"}},
+    )
+
+    assert response.status_code == 403
+    mock_layer.raise_for_access.assert_called_once()
+    mock_layer.implementation.get_semantic_views.assert_not_called()
+
+
 @SEMANTIC_LAYERS_APP
 def test_get_views_with_existing(
     client: Any,

tests/unit_tests/semantic_layers/models_test.py

@@ -1100,6 +1100,90 @@ def test_semantic_layer_get_perm_special_characters() -> None:
     )
 
 
+# =============================================================================
+# SemanticLayer.raise_for_access tests
+# =============================================================================
+
+
+def test_semantic_layer_raise_for_access_all_datasources(app: Any) -> None:
+    """Test raise_for_access passes when user has all_datasource_access."""
+    from superset import security_manager
+
+    layer = SemanticLayer()
+    layer.name = "Layer"
+    layer.uuid = uuid.UUID("abcdef12-3456-7890-abcd-ef1234567890")
+    layer.perm = layer.get_perm()
+
+    with patch.object(
+        security_manager, "can_access_all_datasources", return_value=True
+    ):
+        layer.raise_for_access()
+
+
+def test_semantic_layer_raise_for_access_perm(app: Any) -> None:
+    """Test raise_for_access passes when user has datasource_access to the
+    layer's perm."""
+    from superset import security_manager
+
+    layer = SemanticLayer()
+    layer.name = "Layer"
+    layer.uuid = uuid.UUID("abcdef12-3456-7890-abcd-ef1234567890")
+    layer.perm = layer.get_perm()
+
+    with (
+        patch.object(
+            security_manager, "can_access_all_datasources", return_value=False
+        ),
+        patch.object(
+            security_manager, "can_access", return_value=True
+        ) as mock_can_access,
+    ):
+        layer.raise_for_access()
+        mock_can_access.assert_called_once_with("datasource_access", layer.perm)
+
+
+def test_semantic_layer_raise_for_access_denied(app: Any) -> None:
+    """Test raise_for_access raises SupersetSecurityException when denied."""
+    from superset import security_manager
+    from superset.exceptions import SupersetSecurityException
+
+    layer = SemanticLayer()
+    layer.name = "Layer"
+    layer.uuid = uuid.UUID("abcdef12-3456-7890-abcd-ef1234567890")
+    layer.perm = layer.get_perm()
+
+    with (
+        patch.object(
+            security_manager, "can_access_all_datasources", return_value=False
+        ),
+        patch.object(security_manager, "can_access", return_value=False),
+    ):
+        with pytest.raises(SupersetSecurityException):
+            layer.raise_for_access()
+
+
+def test_semantic_layer_raise_for_access_no_perm_denied(app: Any) -> None:
+    """Test raise_for_access raises SupersetSecurityException when the layer
+    has no perm set, without even attempting a datasource_access check."""
+    from superset import security_manager
+    from superset.exceptions import SupersetSecurityException
+
+    layer = SemanticLayer()
+    layer.name = "Layer"
+    layer.uuid = uuid.UUID("abcdef12-3456-7890-abcd-ef1234567890")
+    layer.perm = None
+
+    with (
+        patch.object(
+            security_manager, "can_access_all_datasources", return_value=False
+        ),
+        patch.object(security_manager, "can_access") as mock_can_access,
+    ):
+        with pytest.raises(SupersetSecurityException):
+            layer.raise_for_access()
+        mock_can_access.assert_not_called()
+
+
 # =============================================================================
 # SemanticView.raise_for_access tests
 # =============================================================================

tests/unit_tests/themes/commands_test.py

@@ -20,6 +20,7 @@ from unittest.mock import Mock, patch
 import pytest
 
 from superset.commands.theme.exceptions import (
+    SystemThemeInUseError,
     SystemThemeProtectedError,
     ThemeNotFoundError,
 )
@@ -62,6 +63,8 @@ class TestUpdateThemeCommand:
         # Arrange
         mock_theme = Mock(spec=Theme)
         mock_theme.is_system = False
+        mock_theme.is_system_default = False
+        mock_theme.is_system_dark = False
         mock_theme_dao.find_by_id.return_value = mock_theme
         command = UpdateThemeCommand(123, {"theme_name": "test"})
 
@@ -77,6 +80,8 @@ class TestUpdateThemeCommand:
         # Arrange
         mock_theme = Mock(spec=Theme)
         mock_theme.is_system = False
+        mock_theme.is_system_default = False
+        mock_theme.is_system_dark = False
         mock_updated_theme = Mock(spec=Theme)
         mock_theme_dao.find_by_id.return_value = mock_theme
         mock_theme_dao.update.return_value = mock_updated_theme
@@ -92,6 +97,52 @@ class TestUpdateThemeCommand:
             mock_theme, {"theme_name": "updated_name"}
         )
 
+    @patch("superset.commands.theme.update.security_manager")
+    @patch("superset.commands.theme.update.ThemeDAO")
+    def test_validate_system_default_theme_blocks_non_admin(
+        self, mock_theme_dao, mock_security_manager
+    ):
+        """Non-admins cannot update the active system-default/dark theme slot."""
+        mock_theme = Mock(spec=Theme)
+        mock_theme.is_system = False
+        mock_theme.is_system_default = True
+        mock_theme.is_system_dark = False
+        mock_theme_dao.find_by_id.return_value = mock_theme
+        # Use a regular Mock for is_admin to avoid AsyncMock auto-detection
+        mock_security_manager.is_admin = Mock(return_value=False)
+
+        command = UpdateThemeCommand(123, {"json_data": "{}"})
+
+        with pytest.raises(SystemThemeInUseError):
+            command.validate()
+
+    @patch("superset.commands.theme.update.security_manager")
+    @patch("superset.commands.theme.update.ThemeDAO")
+    def test_validate_system_default_theme_allows_admin(
+        self, mock_theme_dao, mock_security_manager
+    ):
+        """Admins can still update the active system-default/dark theme slot."""
+        mock_theme = Mock(spec=Theme)
+        mock_theme.is_system = False
+        mock_theme.is_system_default = True
+        mock_theme.is_system_dark = False
+        mock_theme_dao.find_by_id.return_value = mock_theme
+        # Use a regular Mock for is_admin to avoid AsyncMock auto-detection
+        mock_security_manager.is_admin = Mock(return_value=True)
+
+        command = UpdateThemeCommand(123, {"json_data": "{}"})
+
+        command.validate()  # Should not raise any exception
+
+        assert command._model == mock_theme
+
+
+def test_theme_is_gamma_read_only():
+    """Theme writes must require at least Alpha; Gamma only gets read access."""
+    from superset import security_manager
+
+    assert "Theme" in security_manager.GAMMA_READ_ONLY_MODEL_VIEWS
+
 
 class TestSeedSystemThemesCommand:
     """Unit tests for SeedSystemThemesCommand"""

tests/unit_tests/utils/test_core.py

@@ -1942,13 +1942,29 @@ def test_sanitize_svg_content_safe():
 
 
 def test_sanitize_svg_content_removes_scripts():
-    """Test that nh3 removes dangerous script content."""
+    """Test that dangerous script content is removed."""
     malicious_svg = '<svg><script>alert("xss")</script><rect/></svg>'
     result = sanitize_svg_content(malicious_svg)
     assert "script" not in result.lower()
     assert "alert" not in result
 
 
+def test_sanitize_svg_content_removes_script_with_attributes_on_closer():
+    """A closing </script foo> tag is still a valid closer to browsers."""
+    malicious_svg = "<svg><script>fetch('/api/v1/me/')</script foo></svg>"
+    result = sanitize_svg_content(malicious_svg)
+    assert "script" not in result.lower()
+    assert "fetch" not in result
+
+
+def test_sanitize_svg_content_removes_unterminated_script():
+    """An unterminated <script> opener with no closing tag is still stripped."""
+    malicious_svg = "<svg><script>alert('xss')"
+    result = sanitize_svg_content(malicious_svg)
+    assert "script" not in result.lower()
+    assert "alert" not in result
+
+
 def test_sanitize_url_relative():
     """Test that relative URLs are allowed."""
     assert sanitize_url("/static/spinner.gif") == "/static/spinner.gif"