feat(BA-5765): add RBAC-enforced VFolder purge mutation

Add PurgeVFolderV2RBACAction (SingleEntityActionProcessor + RBAC).
Adapter purge() routes to RBAC path; bulk_purge() stays on legacy.

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

Author: fregataa

changes/11165.feature.md

@@ -0,0 +1 @@
+Add RBAC-enforced VFolder purge v2 mutation with SingleEntityActionProcessor.

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

@@ -129,6 +129,9 @@
     DeleteVFolderV2Action,
     PurgeVFolderV2Action,
 )
+from ai.backend.manager.services.vfolder.actions.vfolder_v2_rbac import (
+    PurgeVFolderV2RBACAction,
+)
 
 from .base import BaseAdapter
 
@@ -394,12 +397,9 @@ async def delete(self, vfolder_id: UUID) -> DeleteVFolderPayload:
         return DeleteVFolderPayload(id=vfolder_id)
 
     async def purge(self, vfolder_id: UUID) -> PurgeVFolderPayload:
-        """Permanently delete a vfolder."""
-        me = current_user()
-        if me is None:
-            raise UnreachableError("User context is not available")
-        action = PurgeVFolderV2Action(user_id=me.user_id, vfolder_id=vfolder_id)
-        await self._processors.vfolder.purge_v2.wait_for_complete(action)
+        """Permanently delete a vfolder. RBAC enforced."""
+        action = PurgeVFolderV2RBACAction(vfolder_id=vfolder_id)
+        await self._processors.vfolder.purge_v2_rbac.wait_for_complete(action)
         return PurgeVFolderPayload(id=vfolder_id)
 
     async def deploy(

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

@@ -0,0 +1,63 @@
+"""RBAC-enforced v2 VFolder actions (purge).
+
+Purge targets an existing vfolder by ID and uses
+``SingleEntityActionProcessor`` with ``single_entity_rbac_validators``.
+"""
+
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass
+from typing import override
+
+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.services.vfolder.actions.base import (
+    VFolderSingleEntityAction,
+    VFolderSingleEntityActionResult,
+)
+
+# ---------------------------------------------------------------------------
+# Purge v2 RBAC (single-entity -- scope chain resolves project)
+# ---------------------------------------------------------------------------
+
+
+@dataclass
+class PurgeVFolderV2RBACAction(VFolderSingleEntityAction):
+    """Permanently purge a vfolder by ID with RBAC enforcement."""
+
+    vfolder_id: uuid.UUID
+
+    @override
+    def entity_id(self) -> str | None:
+        return str(self.vfolder_id)
+
+    @override
+    @classmethod
+    def operation_type(cls) -> ActionOperationType:
+        return ActionOperationType.PURGE
+
+    @override
+    def target_entity_id(self) -> str:
+        return str(self.vfolder_id)
+
+    @override
+    def target_element(self) -> RBACElementRef:
+        return RBACElementRef(
+            element_type=RBACElementType.VFOLDER,
+            element_id=str(self.vfolder_id),
+        )
+
+
+@dataclass
+class PurgeVFolderV2RBACActionResult(VFolderSingleEntityActionResult):
+    vfolder_id: uuid.UUID
+
+    @override
+    def entity_id(self) -> str | None:
+        return str(self.vfolder_id)
+
+    @override
+    def target_entity_id(self) -> str:
+        return str(self.vfolder_id)

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

@@ -98,6 +98,10 @@
     PurgeVFolderV2Action,
     PurgeVFolderV2ActionResult,
 )
+from ai.backend.manager.services.vfolder.actions.vfolder_v2_rbac import (
+    PurgeVFolderV2RBACAction,
+    PurgeVFolderV2RBACActionResult,
+)
 from ai.backend.manager.services.vfolder.services.vfolder import VFolderService
 
 
@@ -163,6 +167,9 @@ class VFolderProcessors(AbstractProcessorPackage):
     delete_v2: ActionProcessor[DeleteVFolderV2Action, DeleteVFolderV2ActionResult]
     purge_v2: ActionProcessor[PurgeVFolderV2Action, PurgeVFolderV2ActionResult]
     clone_v2: ActionProcessor[CloneVFolderV2Action, CloneVFolderV2ActionResult]
+    purge_v2_rbac: SingleEntityActionProcessor[
+        PurgeVFolderV2RBACAction, PurgeVFolderV2RBACActionResult
+    ]
 
     def __init__(
         self,
@@ -257,6 +264,9 @@ def __init__(
         self.delete_v2 = ActionProcessor(service.delete_v2, action_monitors)
         self.purge_v2 = ActionProcessor(service.purge_v2, action_monitors)
         self.clone_v2 = ActionProcessor(service.clone_v2, action_monitors)
+        self.purge_v2_rbac = SingleEntityActionProcessor(
+            service.purge_v2_rbac, action_monitors, validators=single_entity_rbac_validators
+        )
 
     @override
     def supported_actions(self) -> list[ActionSpec]:
@@ -296,4 +306,5 @@ def supported_actions(self) -> list[ActionSpec]:
             DeleteVFolderV2Action.spec(),
             PurgeVFolderV2Action.spec(),
             CloneVFolderV2Action.spec(),
+            PurgeVFolderV2RBACAction.spec(),
         ]

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

@@ -162,6 +162,10 @@
     PurgeVFolderV2Action,
     PurgeVFolderV2ActionResult,
 )
+from ai.backend.manager.services.vfolder.actions.vfolder_v2_rbac import (
+    PurgeVFolderV2RBACAction,
+    PurgeVFolderV2RBACActionResult,
+)
 from ai.backend.manager.services.vfolder.types import (
     VFolderBaseInfo,
     VFolderOwnershipInfo,
@@ -1664,6 +1668,15 @@ async def purge_v2(self, action: PurgeVFolderV2Action) -> PurgeVFolderV2ActionRe
         await self._remove_vfolder_from_storage(vfolder_data)
         return PurgeVFolderV2ActionResult(vfolder_id=action.vfolder_id)
 
+    async def purge_v2_rbac(
+        self, action: PurgeVFolderV2RBACAction
+    ) -> PurgeVFolderV2RBACActionResult:
+        """Permanently purge a vfolder by ID. RBAC enforced at processor level."""
+        vfolder_data = await self._vfolder_repository.get_by_id(action.vfolder_id)
+        await self._vfolder_repository.delete_vfolders_forever([action.vfolder_id])
+        await self._remove_vfolder_from_storage(vfolder_data)
+        return PurgeVFolderV2RBACActionResult(vfolder_id=action.vfolder_id)
+
     async def clone_v2(self, action: CloneVFolderV2Action) -> CloneVFolderV2ActionResult:
         """Clone a vfolder (v2). Resolves policy internally from user_id."""
         allowed_vfolder_types = (

tests/component/vfolder_v2/test_vfolder_mutation.py

@@ -0,0 +1,197 @@
+"""Component tests for v2 VFolder RBAC-enforced purge mutation via SDK.
+
+Exercises purge mutation through the real HTTP server + V2ClientRegistry SDK.
+"""
+
+from __future__ import annotations
+
+import secrets
+import uuid
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+from unittest.mock import MagicMock
+
+import pytest
+import yarl
+
+from ai.backend.client.v2.auth import HMACAuth
+from ai.backend.client.v2.config import ClientConfig
+from ai.backend.client.v2.exceptions import PermissionDeniedError
+from ai.backend.client.v2.v2_registry import V2ClientRegistry
+from ai.backend.common.types import QuotaScopeID, QuotaScopeType, VFolderUsageMode
+from ai.backend.manager.actions.validators import ActionValidators
+from ai.backend.manager.actions.validators.rbac import RBACValidators
+from ai.backend.manager.actions.validators.rbac.scope import ScopeActionRBACValidator
+from ai.backend.manager.actions.validators.rbac.single_entity import (
+    SingleEntityActionRBACValidator,
+)
+from ai.backend.manager.api.adapters.vfolder import VFolderAdapter
+from ai.backend.manager.api.rest.routing import RouteRegistry
+from ai.backend.manager.api.rest.types import RouteDeps
+from ai.backend.manager.api.rest.v2.vfolder.handler import V2VFolderHandler
+from ai.backend.manager.api.rest.v2.vfolder.registry import register_v2_vfolder_routes
+from ai.backend.manager.data.vfolder.types import (
+    VFolderMountPermission,
+    VFolderOperationStatus,
+    VFolderOwnershipType,
+)
+from ai.backend.manager.models.utils import ExtendedAsyncSAEngine
+from ai.backend.manager.models.vfolder import vfolders
+from ai.backend.manager.repositories.permission_controller.repository import (
+    PermissionControllerRepository,
+)
+from ai.backend.manager.repositories.user.repository import UserRepository
+from ai.backend.manager.repositories.vfolder.repository import VfolderRepository
+from ai.backend.manager.services.processors import Processors
+from ai.backend.manager.services.vfolder.processors.vfolder import VFolderProcessors
+from ai.backend.manager.services.vfolder.services.vfolder import VFolderService
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio.engine import AsyncEngine as SAEngine
+    from tests.component.conftest import ServerInfo, UserFixtureData
+
+
+@dataclass(frozen=True)
+class ProjectVFolderFixtureData:
+    id: uuid.UUID
+    project_id: uuid.UUID
+
+
+# -- Fixtures ------------------------------------------------------------------
+
+
+@pytest.fixture()
+def rbac_permission_repo(
+    database_engine: ExtendedAsyncSAEngine,
+) -> PermissionControllerRepository:
+    return PermissionControllerRepository(database_engine)
+
+
+@pytest.fixture()
+def vfolder_processors(
+    database_engine: ExtendedAsyncSAEngine,
+    rbac_permission_repo: PermissionControllerRepository,
+) -> VFolderProcessors:
+    """Override: real scope + single-entity RBAC validators."""
+    vfolder_repository = VfolderRepository(database_engine)
+    user_repository = UserRepository(database_engine)
+    service = VFolderService(
+        config_provider=MagicMock(),
+        etcd=MagicMock(),
+        storage_manager=MagicMock(),
+        background_task_manager=MagicMock(),
+        vfolder_repository=vfolder_repository,
+        user_repository=user_repository,
+        valkey_stat_client=MagicMock(),
+    )
+    return VFolderProcessors(
+        service=service,
+        action_monitors=[],
+        validators=ActionValidators(
+            rbac=RBACValidators(
+                scope=ScopeActionRBACValidator(rbac_permission_repo),
+                single_entity=SingleEntityActionRBACValidator(rbac_permission_repo),
+            )
+        ),
+    )
+
+
+@pytest.fixture()
+def server_module_registries(
+    route_deps: RouteDeps,
+    vfolder_processors: VFolderProcessors,
+) -> list[RouteRegistry]:
+    """Register v2 vfolder REST routes with real RBAC."""
+    processors = MagicMock(spec=Processors)
+    processors.vfolder = vfolder_processors
+    adapter = VFolderAdapter(processors)
+    handler = V2VFolderHandler(adapter=adapter)
+    v2_reg = RouteRegistry.create("v2", route_deps.cors_options)
+    v2_reg.add_subregistry(register_v2_vfolder_routes(handler, route_deps))
+    return [v2_reg]
+
+
+@pytest.fixture()
+async def admin_v2_registry(
+    server: ServerInfo,
+    admin_user_fixture: UserFixtureData,
+) -> AsyncIterator[V2ClientRegistry]:
+    registry = await V2ClientRegistry.create(
+        ClientConfig(endpoint=yarl.URL(server.url)),
+        HMACAuth(
+            access_key=admin_user_fixture.keypair.access_key,
+            secret_key=admin_user_fixture.keypair.secret_key,
+        ),
+    )
+    try:
+        yield registry
+    finally:
+        await registry.close()
+
+
+@pytest.fixture()
+async def user_v2_registry(
+    server: ServerInfo,
+    regular_user_fixture: UserFixtureData,
+) -> AsyncIterator[V2ClientRegistry]:
+    registry = await V2ClientRegistry.create(
+        ClientConfig(endpoint=yarl.URL(server.url)),
+        HMACAuth(
+            access_key=regular_user_fixture.keypair.access_key,
+            secret_key=regular_user_fixture.keypair.secret_key,
+        ),
+    )
+    try:
+        yield registry
+    finally:
+        await registry.close()
+
+
+@pytest.fixture()
+async def project_vfolder(
+    db_engine: SAEngine,
+    domain_fixture: str,
+    group_fixture: uuid.UUID,
+    vfolder_host_permission_fixture: None,
+) -> AsyncIterator[ProjectVFolderFixtureData]:
+    """Seed a project-owned vfolder row directly in the DB."""
+    unique = secrets.token_hex(4)
+    vfolder_id = uuid.uuid4()
+    quota_scope_id = QuotaScopeID(scope_type=QuotaScopeType.PROJECT, scope_id=group_fixture)
+    async with db_engine.begin() as conn:
+        await conn.execute(
+            vfolders.insert().values(
+                id=vfolder_id,
+                name=f"test-proj-vfolder-{unique}",
+                host="local",
+                domain_name=domain_fixture,
+                quota_scope_id=str(quota_scope_id),
+                usage_mode=VFolderUsageMode.GENERAL,
+                permission=VFolderMountPermission.READ_WRITE,
+                ownership_type=VFolderOwnershipType.GROUP,
+                user=None,
+                group=group_fixture,
+                creator="admin-test@test.local",
+                status=VFolderOperationStatus.READY,
+                cloneable=False,
+            )
+        )
+    yield ProjectVFolderFixtureData(id=vfolder_id, project_id=group_fixture)
+    async with db_engine.begin() as conn:
+        await conn.execute(vfolders.delete().where(vfolders.c.id == vfolder_id))
+
+
+# -- Tests ---------------------------------------------------------------------
+
+
+class TestPurgeVFolderRBAC:
+    """POST /v2/vfolders/{id}/purge -- SingleEntityActionProcessor RBAC."""
+
+    async def test_regular_user_denied(
+        self,
+        user_v2_registry: V2ClientRegistry,
+        project_vfolder: ProjectVFolderFixtureData,
+    ) -> None:
+        with pytest.raises(PermissionDeniedError):
+            await user_v2_registry.vfolder.purge(project_vfolder.id)