refactor(BA-5071): Apply RBAC Creator pattern to ImageAlias

- Add IMAGE_ALIAS to RBACElementType enum
- Replace Creator/execute_creator with RBACEntityCreator/execute_rbac_entity_creator
  for ImageAlias creation in image domain
- Scope association: ImageAlias → IMAGE (parent scope)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: fregataa

src/ai/backend/common/data/permission/types.py

@@ -385,6 +385,9 @@ class RBACElementType(enum.StrEnum):
     # === Auto-only entities used in permissions ===
     NOTIFICATION_RULE = "notification_rule"
 
+    # === Auto sub-entities with direct GET APIs ===
+    IMAGE_ALIAS = "image:alias"
+
     # === Entity-level scopes (for entity-scope permissions) ===
     ARTIFACT_REVISION = "artifact_revision"
 

src/ai/backend/manager/repositories/image/db_source/db_source.py

@@ -12,6 +12,7 @@
 from sqlalchemy.orm import selectinload
 
 from ai.backend.common.bgtask.reporter import ProgressReporter
+from ai.backend.common.data.permission.types import RBACElementType
 from ai.backend.common.docker import ImageRef
 from ai.backend.common.exception import UnknownImageReference
 from ai.backend.common.types import ImageAlias, ImageID
@@ -26,6 +27,7 @@
     RescanImagesResult,
     ResourceLimitInput,
 )
+from ai.backend.manager.data.permission.types import RBACElementRef
 from ai.backend.manager.errors.image import (
     AliasImageActionDBError,
     AliasImageActionValueError,
@@ -45,8 +47,11 @@
 )
 from ai.backend.manager.models.kernel.row import KernelRow
 from ai.backend.manager.models.utils import ExtendedAsyncSAEngine
-from ai.backend.manager.repositories.base import BatchQuerier, Creator, execute_batch_querier
-from ai.backend.manager.repositories.base.creator import execute_creator
+from ai.backend.manager.repositories.base import BatchQuerier, execute_batch_querier
+from ai.backend.manager.repositories.base.rbac.entity_creator import (
+    RBACEntityCreator,
+    execute_rbac_entity_creator,
+)
 from ai.backend.manager.repositories.base.updater import Updater, execute_updater
 from ai.backend.manager.repositories.image.creators import ImageAliasCreatorSpec
 
@@ -244,10 +249,20 @@ async def insert_image_alias(
                 image_row = await ImageRow.resolve(
                     session, [ImageIdentifier(image_canonical, architecture)]
                 )
-                image_alias = ImageAliasRow(alias=alias, image_id=image_row.id)
-                image_row.aliases.append(image_alias)
+                rbac_creator = RBACEntityCreator(
+                    spec=ImageAliasCreatorSpec(
+                        alias=alias,
+                        image_id=image_row.id,
+                    ),
+                    element_type=RBACElementType.IMAGE_ALIAS,
+                    scope_ref=RBACElementRef(
+                        element_type=RBACElementType.IMAGE,
+                        element_id=str(image_row.id),
+                    ),
+                )
+                result = await execute_rbac_entity_creator(session, rbac_creator)
                 row_id = image_row.id
-                alias_data = ImageAliasData(id=image_alias.id, alias=image_alias.alias or "")
+                alias_data = ImageAliasData(id=result.row.id, alias=result.row.alias or "")
             return row_id, alias_data
         except ValueError as e:
             raise AliasImageActionValueError from e
@@ -327,16 +342,18 @@ async def clear_image_resource_limits(
             image_row._resources = {}
             return image_row.to_dataclass()
 
-    async def insert_image_alias_by_id(self, creator: Creator[ImageAliasRow]) -> ImageAliasData:
+    async def insert_image_alias_by_id(
+        self, creator: RBACEntityCreator[ImageAliasRow]
+    ) -> ImageAliasData:
         """
-        Creates an image alias using the Creator pattern.
+        Creates an image alias using the RBACEntityCreator pattern.
         """
-        spec = cast(ImageAliasCreatorSpec, creator.spec)
         try:
             async with self._db.begin_session() as session:
+                spec = cast(ImageAliasCreatorSpec, creator.spec)
                 # Validate that the image exists
                 await self._get_image_by_id(session, spec.image_id)
-                result = await execute_creator(session, creator)
+                result = await execute_rbac_entity_creator(session, creator)
                 return ImageAliasData(id=result.row.id, alias=result.row.alias or "")
         except ValueError as e:
             raise AliasImageActionValueError from e

src/ai/backend/manager/repositories/image/repository.py

@@ -32,7 +32,8 @@
 )
 from ai.backend.manager.models.image.row import ImageAliasRow
 from ai.backend.manager.models.utils import ExtendedAsyncSAEngine
-from ai.backend.manager.repositories.base import BatchQuerier, Creator
+from ai.backend.manager.repositories.base import BatchQuerier
+from ai.backend.manager.repositories.base.rbac.entity_creator import RBACEntityCreator
 from ai.backend.manager.repositories.base.updater import Updater
 from ai.backend.manager.repositories.image.db_source.db_source import ImageDBSource
 from ai.backend.manager.repositories.image.stateful_source.stateful_source import (
@@ -281,9 +282,11 @@ async def clear_image_custom_resource_limit(
         return await self._db_source.clear_image_resource_limits(image_canonical, architecture)
 
     @image_repository_resilience.apply()
-    async def add_image_alias_by_id(self, creator: Creator[ImageAliasRow]) -> ImageAliasData:
+    async def add_image_alias_by_id(
+        self, creator: RBACEntityCreator[ImageAliasRow]
+    ) -> ImageAliasData:
         """
-        Creates an image alias using the Creator pattern.
+        Creates an image alias using the RBACEntityCreator pattern.
         """
         return await self._db_source.insert_image_alias_by_id(creator)
 

src/ai/backend/manager/services/image/service.py

@@ -2,21 +2,23 @@
 from uuid import UUID
 
 from ai.backend.common.contexts.user import current_user
+from ai.backend.common.data.permission.types import RBACElementType
 from ai.backend.common.docker import ImageRef
 from ai.backend.common.dto.manager.rpc_request import PurgeImagesReq
 from ai.backend.common.exception import UnknownImageReference
 from ai.backend.common.types import AgentId, ImageAlias, ImageID
 from ai.backend.logging.utils import BraceStyleAdapter
 from ai.backend.manager.config.provider import ManagerConfigProvider
 from ai.backend.manager.data.image.types import ImageWithAgentInstallStatus
+from ai.backend.manager.data.permission.types import RBACElementRef
 from ai.backend.manager.errors.image import ImageAccessForbiddenError, ImageNotFound
 from ai.backend.manager.models.image import (
     ImageIdentifier,
     ImageRow,
 )
 from ai.backend.manager.models.user import UserRole
 from ai.backend.manager.registry import AgentRegistry
-from ai.backend.manager.repositories.base import Creator
+from ai.backend.manager.repositories.base.rbac.entity_creator import RBACEntityCreator
 from ai.backend.manager.repositories.base.updater import Updater
 from ai.backend.manager.repositories.image.creators import ImageAliasCreatorSpec
 from ai.backend.manager.repositories.image.repository import ImageRepository
@@ -405,13 +407,18 @@ async def alias_image_by_id(self, action: AliasImageByIdAction) -> AliasImageByI
         """
         Creates an alias for an image by its ID.
         """
-        creator = Creator(
+        rbac_creator = RBACEntityCreator(
             spec=ImageAliasCreatorSpec(
                 alias=action.alias,
                 image_id=action.image_id,
-            )
+            ),
+            element_type=RBACElementType.IMAGE_ALIAS,
+            scope_ref=RBACElementRef(
+                element_type=RBACElementType.IMAGE,
+                element_id=str(action.image_id),
+            ),
         )
-        image_alias = await self._image_repository.add_image_alias_by_id(creator)
+        image_alias = await self._image_repository.add_image_alias_by_id(rbac_creator)
         return AliasImageByIdActionResult(
             image_id=action.image_id,
             image_alias=image_alias,

tests/unit/manager/services/image/test_image_service.py

@@ -16,6 +16,7 @@
 
 from ai.backend.common.container_registry import ContainerRegistryType
 from ai.backend.common.contexts.user import with_user
+from ai.backend.common.data.permission.types import RBACElementType
 from ai.backend.common.data.user.types import UserData
 from ai.backend.common.dto.agent.response import PurgeImageResp, PurgeImagesResp
 from ai.backend.common.exception import UnknownImageReference
@@ -31,14 +32,16 @@
     RescanImagesResult,
     ResourceLimitInput,
 )
+from ai.backend.manager.data.permission.types import RBACElementRef
 from ai.backend.manager.errors.image import (
     ImageAccessForbiddenError,
     ImageAliasNotFound,
     ImageNotFound,
 )
 from ai.backend.manager.models.image import ImageStatus, ImageType
 from ai.backend.manager.models.user import UserRole
-from ai.backend.manager.repositories.base import BatchQuerier, Creator, OffsetPagination
+from ai.backend.manager.repositories.base import BatchQuerier, OffsetPagination
+from ai.backend.manager.repositories.base.rbac.entity_creator import RBACEntityCreator
 from ai.backend.manager.repositories.image.creators import ImageAliasCreatorSpec
 from ai.backend.manager.repositories.image.repository import ImageRepository
 from ai.backend.manager.repositories.image.updaters import ImageUpdaterSpec
@@ -992,10 +995,15 @@ async def test_alias_image_by_id_success(
         assert result.image_alias == image_alias_data
         mock_image_repository.add_image_alias_by_id.assert_called_once()
         creator_arg = mock_image_repository.add_image_alias_by_id.call_args[0][0]
-        assert isinstance(creator_arg, Creator)
+        assert isinstance(creator_arg, RBACEntityCreator)
         assert isinstance(creator_arg.spec, ImageAliasCreatorSpec)
         assert creator_arg.spec.alias == "python"
         assert creator_arg.spec.image_id == image_id
+        assert creator_arg.element_type == RBACElementType.IMAGE_ALIAS
+        assert creator_arg.scope_ref == RBACElementRef(
+            element_type=RBACElementType.IMAGE,
+            element_id=str(image_id),
+        )
 
 
 class TestClearImageCustomResourceLimitById(ImageServiceBaseFixtures):