feat(BA-5797): add effective permissions resolver for entities

Implement repository, service, and action layers to resolve all
permitted operations a user can perform on given entities by
traversing the scope chain and evaluating role/permission assignments.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: fregataa

src/ai/backend/manager/data/permission/role.py

@@ -94,6 +94,27 @@ class ScopeChainPermissionCheckInput:
     permission_entity_type: EntityType | None
 
 
+@dataclass(frozen=True)
+class EffectivePermissionsInput:
+    """Input for resolving effective permissions per entity for a given user.
+
+    Given a user and a list of entity references (same RBACElementType),
+    returns the set of permitted operations per entity by traversing the
+    scope chain and evaluating all role/permission assignments.
+    """
+
+    user_id: uuid.UUID
+    target_element_refs: list[RBACElementRef]
+    permission_entity_type: EntityType | None = None
+
+
+@dataclass(frozen=True)
+class EffectivePermissionsResult:
+    """Mapping from entity ID to the set of operations the user is authorized to perform."""
+
+    permissions: dict[str, set[OperationType]]
+
+
 @dataclass(frozen=True)
 class UserRoleAssignmentInput:
     """

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

@@ -32,6 +32,8 @@
     BulkRoleRevocationFailure,
     BulkRoleRevocationResultData,
     BulkUserRoleRevocationInput,
+    EffectivePermissionsInput,
+    EffectivePermissionsResult,
     ProjectRoleCount,
     RoleListResult,
     RolePermissionsUpdateInput,
@@ -1121,6 +1123,121 @@ async def check_permission_with_scope_chain(
             result = await db_session.scalar(combined_query)
             return result or False
 
+    def _build_scope_chain_operations_query(
+        self,
+        user_id: uuid.UUID,
+        target_element_ref: RBACElementRef,
+        target_entity_type: EntityType,
+    ) -> sa.sql.Select[Any]:
+        """Build a query that returns all permitted operations via CTE scope chain traversal."""
+        permissions = PermissionRow.__table__
+        user_roles = UserRoleRow.__table__
+        roles = RoleRow.__table__
+
+        scope_chain_cte = self._build_scope_chain_cte(target_element_ref)
+        return (
+            sa.select(permissions.c.operation)
+            .select_from(
+                scope_chain_cte.join(
+                    permissions,
+                    sa.and_(
+                        permissions.c.scope_type == scope_chain_cte.c.scope_type,
+                        permissions.c.scope_id == scope_chain_cte.c.scope_id,
+                    ),
+                )
+                .join(
+                    roles,
+                    roles.c.id == permissions.c.role_id,
+                )
+                .join(
+                    user_roles,
+                    user_roles.c.role_id == roles.c.id,
+                )
+            )
+            .where(
+                sa.and_(
+                    user_roles.c.user_id == user_id,
+                    roles.c.status == RoleStatus.ACTIVE,
+                    permissions.c.entity_type == target_entity_type,
+                )
+            )
+            .distinct()
+        )
+
+    def _build_self_scope_operations_query(
+        self,
+        user_id: uuid.UUID,
+        target_element_ref: RBACElementRef,
+        target_entity_type: EntityType,
+        target_scope_type: ScopeType,
+    ) -> sa.sql.Select[Any]:
+        """Build a query that returns all permitted operations scoped to the target entity itself."""
+        permissions = PermissionRow.__table__
+        user_roles = UserRoleRow.__table__
+        roles = RoleRow.__table__
+
+        return (
+            sa.select(permissions.c.operation)
+            .select_from(
+                permissions.join(
+                    roles,
+                    roles.c.id == permissions.c.role_id,
+                ).join(
+                    user_roles,
+                    user_roles.c.role_id == roles.c.id,
+                )
+            )
+            .where(
+                sa.and_(
+                    user_roles.c.user_id == user_id,
+                    roles.c.status == RoleStatus.ACTIVE,
+                    permissions.c.scope_type == target_scope_type,
+                    permissions.c.scope_id == target_element_ref.element_id,
+                    permissions.c.entity_type == target_entity_type,
+                )
+            )
+            .distinct()
+        )
+
+    async def resolve_effective_permissions(
+        self,
+        data: EffectivePermissionsInput,
+    ) -> EffectivePermissionsResult:
+        """Resolve the effective permissions for a user across multiple entities.
+
+        For each target entity, traverses the scope chain (AUTO edges only) and
+        self-scope permissions to collect ALL operations the user can perform.
+
+        Returns a mapping from entity ID to the set of permitted operations.
+        """
+        permissions: dict[str, set[OperationType]] = {}
+
+        async with self._db.begin_readonly_session_read_committed() as db_session:
+            for ref in data.target_element_refs:
+                target_entity_type = (
+                    data.permission_entity_type or ref.element_type.to_entity_type()
+                )
+                target_scope_type = ref.element_type.to_scope_type()
+
+                scope_chain_query = self._build_scope_chain_operations_query(
+                    data.user_id,
+                    ref,
+                    target_entity_type,
+                )
+                self_scope_query = self._build_self_scope_operations_query(
+                    data.user_id,
+                    ref,
+                    target_entity_type,
+                    target_scope_type,
+                )
+                combined_query = sa.union(scope_chain_query, self_scope_query)
+
+                result = await db_session.execute(combined_query)
+                operations = {OperationType(row[0]) for row in result}
+                permissions[ref.element_id] = operations
+
+        return EffectivePermissionsResult(permissions=permissions)
+
     async def bulk_assign_role(
         self, bulk_creator: BulkCreator[UserRoleRow]
     ) -> BulkCreatorResultWithFailures[UserRoleRow]:

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

@@ -23,6 +23,8 @@
     BulkRoleAssignmentResultData,
     BulkRoleRevocationResultData,
     BulkUserRoleRevocationInput,
+    EffectivePermissionsInput,
+    EffectivePermissionsResult,
     RoleData,
     RoleDetailData,
     RoleListResult,
@@ -336,3 +338,15 @@ async def check_permission_with_scope_chain(
         scope. REF edges are not traversed.
         """
         return await self._db_source.check_permission_with_scope_chain(data)
+
+    @permission_controller_repository_resilience.apply()
+    async def resolve_effective_permissions(
+        self,
+        data: EffectivePermissionsInput,
+    ) -> EffectivePermissionsResult:
+        """Resolve the set of permitted operations per entity for a given user.
+
+        For each target entity, traverses the scope chain (AUTO edges) and
+        self-scope permissions to collect all operations the user can perform.
+        """
+        return await self._db_source.resolve_effective_permissions(data)

src/ai/backend/manager/services/permission_contoller/actions/__init__.py

@@ -7,6 +7,10 @@
 from .get_permission_matrix import GetPermissionMatrixAction, GetPermissionMatrixActionResult
 from .get_role_detail import GetRoleDetailAction, GetRoleDetailActionResult
 from .purge_role import PurgeRoleAction, PurgeRoleActionResult
+from .resolve_effective_permissions import (
+    ResolveEffectivePermissionsAction,
+    ResolveEffectivePermissionsActionResult,
+)
 from .revoke_role import RevokeRoleAction, RevokeRoleActionResult
 from .search_permissions import (
     SearchPermissionsAction,
@@ -47,6 +51,8 @@
     "GetRoleDetailActionResult",
     "PurgeRoleAction",
     "PurgeRoleActionResult",
+    "ResolveEffectivePermissionsAction",
+    "ResolveEffectivePermissionsActionResult",
     "RevokeRoleAction",
     "RevokeRoleActionResult",
     "SearchRolesAction",

src/ai/backend/manager/services/permission_contoller/actions/resolve_effective_permissions.py

@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import override
+from uuid import UUID
+
+from ai.backend.common.data.permission.types import EntityType, OperationType
+from ai.backend.manager.actions.action import BaseActionResult
+from ai.backend.manager.actions.types import ActionOperationType
+from ai.backend.manager.data.permission.types import RBACElementRef
+from ai.backend.manager.services.permission_contoller.actions.base import RoleAction
+
+
+@dataclass
+class ResolveEffectivePermissionsAction(RoleAction):
+    """Action to resolve effective permissions per entity for a given user.
+
+    Given a user ID and a list of entity references (same RBACElementType),
+    returns the set of permitted operations per entity by traversing the
+    scope chain and evaluating all role/permission assignments.
+    """
+
+    user_id: UUID
+    target_element_refs: list[RBACElementRef]
+    permission_entity_type: EntityType | None = None
+
+    @override
+    def entity_id(self) -> str | None:
+        return str(self.user_id)
+
+    @override
+    @classmethod
+    def operation_type(cls) -> ActionOperationType:
+        return ActionOperationType.GET
+
+
+@dataclass
+class ResolveEffectivePermissionsActionResult(BaseActionResult):
+    """Result containing the effective permissions per entity."""
+
+    permissions: dict[str, set[OperationType]] = field(default_factory=dict)
+
+    @override
+    def entity_id(self) -> str | None:
+        return None

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

@@ -9,7 +9,10 @@
     RBACActionName,
     RBACRequiredPermission,
 )
-from ai.backend.manager.data.permission.role import UserRoleRevocationData
+from ai.backend.manager.data.permission.role import (
+    EffectivePermissionsInput,
+    UserRoleRevocationData,
+)
 from ai.backend.manager.repositories.group.repository import GroupRepository
 from ai.backend.manager.repositories.permission_controller.creators import UserRoleCreatorSpec
 from ai.backend.manager.repositories.permission_controller.db_source.db_source import (
@@ -64,6 +67,10 @@
     PurgeRoleAction,
     PurgeRoleActionResult,
 )
+from ai.backend.manager.services.permission_contoller.actions.resolve_effective_permissions import (
+    ResolveEffectivePermissionsAction,
+    ResolveEffectivePermissionsActionResult,
+)
 from ai.backend.manager.services.permission_contoller.actions.revoke_role import (
     RevokeRoleAction,
     RevokeRoleActionResult,
@@ -349,3 +356,20 @@ async def get_permission_matrix(
             actions = entity_map.setdefault(perm.element_type, {})
             actions[action_cls.action_name()] = perm
         return GetPermissionMatrixActionResult(matrix=result)
+
+    async def resolve_effective_permissions(
+        self, action: ResolveEffectivePermissionsAction
+    ) -> ResolveEffectivePermissionsActionResult:
+        """Resolve the set of permitted operations per entity for a given user.
+
+        Traverses the scope chain and evaluates all role/permission assignments
+        to return all operations the user is authorized to perform on each entity.
+        """
+        result = await self._repository.resolve_effective_permissions(
+            EffectivePermissionsInput(
+                user_id=action.user_id,
+                target_element_refs=action.target_element_refs,
+                permission_entity_type=action.permission_entity_type,
+            )
+        )
+        return ResolveEffectivePermissionsActionResult(permissions=result.permissions)