fix(BA-5737): implement create_in_project without legacy role check

Replace the create_v2() delegation with an independent implementation
that skips the UserRole.ADMIN/SUPERADMIN gate. Project CREATE permission
is validated by the ScopeActionProcessor RBAC — the service no longer
blocks regular users who hold project-level grants.

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

Author: fregataa

src/ai/backend/manager/api/adapters/vfolder.py

@@ -102,7 +102,10 @@
     ProjectVFolderSearchScope,
     UserVFolderSearchScope,
 )
-from ai.backend.manager.repositories.vfolder.updaters import VFolderAttributeUpdaterSpec
+from ai.backend.manager.repositories.vfolder.updaters import (
+    VFolderAttributeUpdaterSpec,
+    VFolderTrashUpdaterSpec,
+)
 from ai.backend.manager.services.deployment.actions.create_deployment import CreateDeploymentAction
 from ai.backend.manager.services.vfolder.actions.admin_search_vfolders import (
     AdminSearchVFoldersAction,
@@ -445,7 +448,8 @@ async def update(self, vfolder_id: UUID, input: UpdateVFolderInput) -> UpdateVFo
 
     async def delete(self, vfolder_id: UUID) -> DeleteVFolderPayload:
         """Soft-delete a vfolder (move to trash). RBAC enforced."""
-        action = DeleteVFolderV2Action(vfolder_id=vfolder_id)
+        updater = Updater(spec=VFolderTrashUpdaterSpec(), pk_value=vfolder_id)
+        action = DeleteVFolderV2Action(vfolder_id=vfolder_id, updater=updater)
         await self._processors.vfolder.delete_v2.wait_for_complete(action)
         return DeleteVFolderPayload(id=vfolder_id)
 
@@ -537,7 +541,8 @@ async def deploy(
     async def bulk_delete(self, input: BulkDeleteVFoldersInput) -> BulkDeleteVFoldersPayload:
         """Soft-delete multiple vfolders. RBAC enforced per item."""
         for vfolder_id in input.ids:
-            action = DeleteVFolderV2Action(vfolder_id=vfolder_id)
+            updater = Updater(spec=VFolderTrashUpdaterSpec(), pk_value=vfolder_id)
+            action = DeleteVFolderV2Action(vfolder_id=vfolder_id, updater=updater)
             await self._processors.vfolder.delete_v2.wait_for_complete(action)
         return BulkDeleteVFoldersPayload(deleted_count=len(input.ids))
 

src/ai/backend/manager/services/vfolder/actions/vfolder_v2.py

@@ -12,6 +12,8 @@
 from ai.backend.common.data.permission.types import RBACElementType
 from ai.backend.manager.actions.types import ActionOperationType
 from ai.backend.manager.data.permission.types import RBACElementRef
+from ai.backend.manager.models.vfolder import VFolderRow
+from ai.backend.manager.repositories.base.updater import Updater
 from ai.backend.manager.services.vfolder.actions.base import (
     VFolderSingleEntityAction,
     VFolderSingleEntityActionResult,
@@ -23,6 +25,7 @@ class DeleteVFolderV2Action(VFolderSingleEntityAction):
     """Soft-delete a vfolder by ID with RBAC enforcement."""
 
     vfolder_id: uuid.UUID
+    updater: Updater[VFolderRow]
 
     @override
     def entity_id(self) -> str | None:

src/ai/backend/manager/services/vfolder/services/vfolder.py

@@ -65,13 +65,9 @@
     verify_vfolder_name,
     vfolder_status_map,
 )
-from ai.backend.manager.repositories.base.updater import Updater
 from ai.backend.manager.repositories.user.repository import UserRepository
 from ai.backend.manager.repositories.vfolder.repository import VfolderRepository
-from ai.backend.manager.repositories.vfolder.updaters import (
-    VFolderAttributeUpdaterSpec,
-    VFolderTrashUpdaterSpec,
-)
+from ai.backend.manager.repositories.vfolder.updaters import VFolderAttributeUpdaterSpec
 from ai.backend.manager.services.vfolder.actions.base import (
     CloneVFolderAction,
     CloneVFolderActionResult,
@@ -1636,25 +1632,102 @@ async def get_v2(self, action: GetVFolderV2Action) -> GetVFolderV2ActionResult:
     async def create_in_project(
         self, action: CreateVFolderInProjectAction
     ) -> CreateVFolderInProjectActionResult:
-        """Create a vfolder scoped to a project.
+        """Create a vfolder owned by a project.
 
-        RBAC is enforced by the ScopeActionProcessor wiring; this method
-        delegates the heavy lifting to ``create_v2`` with ``project_id`` set.
+        RBAC is enforced by the ScopeActionProcessor at the processor level.
+        Unlike ``create_v2`` this method does NOT check ``UserRole`` — project
+        CREATE permission is validated by the scope RBAC validator.
         """
-        inner_action = CreateVFolderV2Action(
+        user_uuid = action.user_id
+        domain_name = action.domain_name
+        project_id = action.project_id
+
+        # Resolve host
+        folder_host = action.host
+        if not folder_host:
+            folder_host = self._config_provider.config.volumes.default_host
+            if not folder_host:
+                raise VFolderInvalidParameter(
+                    "You must specify the vfolder host because the default host is not configured."
+                )
+
+        if action.name.startswith(".") and action.name != ".local":
+            raise VFolderInvalidParameter("dot-prefixed vfolders cannot be a group folder.")
+
+        # Resolve project info
+        group_info = await self._vfolder_repository.get_group_resource_info(project_id, domain_name)
+        if not group_info:
+            raise ProjectNotFound(f"Project with {project_id} not found.")
+        group_uuid, max_vfolder_count, max_quota_scope_size, group_type = group_info
+
+        quota_scope_id = QuotaScopeID(QuotaScopeType.PROJECT, group_uuid)
+
+        if group_type == ProjectType.MODEL_STORE:
+            if action.usage_mode != VFolderUsageMode.MODEL:
+                raise VFolderInvalidParameter(
+                    "Only Model VFolder can be created under the model store project"
+                )
+
+        # Host permission check
+        await self._vfolder_repository.ensure_host_permission_allowed_by_user(
+            folder_host,
+            permission=VFolderHostPermission.CREATE,
+            user_uuid=user_uuid,
+            group_id=group_uuid,
+        )
+
+        # Quota check
+        if max_vfolder_count > 0:
+            current_count = await self._vfolder_repository.count_vfolders_by_group(group_uuid)
+            if current_count >= max_vfolder_count:
+                raise VFolderInvalidParameter("You cannot create more vfolders.")
+
+        # Create in storage
+        folder_id = uuid.uuid4()
+        mount_permission = action.permission
+        if group_type == ProjectType.MODEL_STORE:
+            mount_permission = VFolderPermission.READ_ONLY
+
+        try:
+            vfid = VFolderID(quota_scope_id, folder_id)
+            proxy_name, volume_name = self._storage_manager.get_proxy_and_volume(folder_host, False)
+            manager_client = self._storage_manager.get_manager_facing_client(proxy_name)
+            await manager_client.create_folder(volume_name, str(vfid), max_quota_scope_size, None)
+        except aiohttp.ClientResponseError as e:
+            raise VFolderCreationFailure from e
+
+        # Create in DB
+        params = VFolderCreateParams(
+            id=folder_id,
             name=action.name,
-            user_id=action.user_id,
-            domain_name=action.domain_name,
-            project_id=action.project_id,
-            host=action.host,
+            domain_name=domain_name,
+            quota_scope_id=str(quota_scope_id),
             usage_mode=action.usage_mode,
-            permission=action.permission,
+            permission=mount_permission,
+            host=folder_host,
+            creator=str(user_uuid),
+            creator_id=user_uuid,
+            ownership_type=VFolderOwnershipType.GROUP,
+            user=None,
+            group=group_uuid,
+            unmanaged_path=None,
             cloneable=action.cloneable,
+            status=VFolderOperationStatus.READY,
         )
-        inner_result = await self.create_v2(inner_action)
+
+        try:
+            create_owner_permission = group_type == ProjectType.MODEL_STORE
+            await self._vfolder_repository.create_vfolder_with_permission(
+                params, create_owner_permission=create_owner_permission
+            )
+        except sa_exc.DataError as e:
+            raise VFolderInvalidParameter from e
+
+        # Fetch created vfolder data for response
+        vfolder_data = await self._vfolder_repository.get_by_id(folder_id)
         return CreateVFolderInProjectActionResult(
             project_id=action.project_id,
-            vfolder=inner_result.vfolder,
+            vfolder=vfolder_data,
         )
 
     async def delete_v2(self, action: DeleteVFolderV2Action) -> DeleteVFolderV2ActionResult:
@@ -1666,8 +1739,7 @@ async def delete_v2(self, action: DeleteVFolderV2Action) -> DeleteVFolderV2Actio
         matching row exists, which the repository translates to
         ``VFolderNotFound``.
         """
-        updater = Updater(spec=VFolderTrashUpdaterSpec(), pk_value=action.vfolder_id)
-        await self._vfolder_repository.trash_vfolder(updater)
+        await self._vfolder_repository.trash_vfolder(action.updater)
         return DeleteVFolderV2ActionResult(vfolder_id=action.vfolder_id)
 
     async def purge_v2(self, action: PurgeVFolderV2Action) -> PurgeVFolderV2ActionResult: