feat(BA-6860): RBAC-gate app_config fragment writes at the processors

Attach the RBAC validators to the app_config_fragment write processors
(create -> scope, update/purge -> single-entity, bulk_* -> bulk) and pass
validators through the factory. Reads (get/admin_search/scoped_search) carry
no validator — they stay gated by the allow list and scope visibility. The
fragment write routes move from superadmin_required to auth_required and the
handler/adapter methods drop their admin_ prefix; RBAC resolves who may write
(a user their own user-scope fragment via their system role and the fragment's
scope association from BA-6859, a domain admin their domain's, a superadmin
any scope — public stays superadmin-only, having no RBAC scope).

Replaces the interim fixed-scope _may_write service check entirely.

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

Author: jopemachine

changes/12759.feature.md

@@ -0,0 +1 @@
+Gate AppConfig fragment writes with RBAC: create/update/purge are validated by the RBAC processors (a user writes their own user-scope fragment, a domain admin their domain's, a superadmin any), while reads stay gated by the allow list and scope visibility; fragment write endpoints move from superadmin-only to authenticated (BEP-1052).

src/ai/backend/manager/api/adapters/app_config/adapter.py

@@ -99,15 +99,14 @@ class AppConfigAdapter(BaseAdapter):
 
     # --- admin fragment CRUD ---
 
-    async def admin_create(
-        self, input: CreateAppConfigFragmentInput
-    ) -> CreateAppConfigFragmentPayload:
+    async def create(self, input: CreateAppConfigFragmentInput) -> CreateAppConfigFragmentPayload:
         spec = AppConfigFragmentCreatorSpec(
             config_name=input.config_name,
             scope_type=AppConfigScopeType(input.scope_type.value),
             scope_id=input.scope_id,
             config=input.config,
         )
+        # Write authorization is enforced by the RBAC validators on the processor.
         action_result = await self._processors.app_config_fragment.create.wait_for_complete(
             CreateAppConfigFragmentAction(creator_spec=spec)
         )
@@ -121,28 +120,26 @@ async def admin_get(self, fragment_id: AppConfigFragmentID) -> AppConfigFragment
         )
         return self._fragment_to_node(action_result.fragment)
 
-    async def admin_update(
+    async def update(
         self, fragment_id: AppConfigFragmentID, input: UpdateAppConfigFragmentInput
     ) -> UpdateAppConfigFragmentPayload:
         updater = Updater(
             spec=AppConfigFragmentUpdaterSpec(config=OptionalState.update(input.config)),
             pk_value=fragment_id,
         )
-        # No allow-list gate: a fragment row exists only while its entry does (FK with
-        # cascade), so an existing fragment is always writable at its own scope.
+        # Write authorization is enforced by the RBAC validators on the processor (the
+        # fragment's scope is resolved via its RBAC scope association).
         action_result = await self._processors.app_config_fragment.update.wait_for_complete(
             UpdateAppConfigFragmentAction(updater=updater)
         )
         return UpdateAppConfigFragmentPayload(
             app_config_fragment=self._fragment_to_node(action_result.fragment),
         )
 
-    async def admin_purge(
-        self, input: PurgeAppConfigFragmentInput
-    ) -> PurgeAppConfigFragmentPayload:
+    async def purge(self, input: PurgeAppConfigFragmentInput) -> PurgeAppConfigFragmentPayload:
         fragment_id = AppConfigFragmentID(input.id)
         purger = Purger(row_class=AppConfigFragmentRow, pk_value=fragment_id)
-        # No allow-list gate — see ``admin_update``.
+        # Write authorization is enforced by the RBAC validators — see ``update``.
         action_result = await self._processors.app_config_fragment.purge.wait_for_complete(
             PurgeAppConfigFragmentAction(purger=purger)
         )

src/ai/backend/manager/api/rest/v2/app_config_fragment/handler.py

@@ -30,12 +30,12 @@ class V2AppConfigFragmentHandler:
     def __init__(self, *, adapter: AppConfigAdapter) -> None:
         self._adapter = adapter
 
-    async def admin_create(
+    async def create(
         self,
         body: BodyParam[CreateAppConfigFragmentInput],
     ) -> APIResponse:
-        """Create a fragment at any scope (superadmin only)."""
-        result = await self._adapter.admin_create(body.parsed)
+        """Create a fragment; authorized by the layer's permission policy (auth required)."""
+        result = await self._adapter.create(body.parsed)
         return APIResponse.build(status_code=HTTPStatus.CREATED, response_model=result)
 
     async def admin_get(
@@ -46,25 +46,23 @@ async def admin_get(
         result = await self._adapter.admin_get(AppConfigFragmentID(path.parsed.fragment_id))
         return APIResponse.build(status_code=HTTPStatus.OK, response_model=result)
 
-    async def admin_update(
+    async def update(
         self,
         path: PathParam[AppConfigFragmentIdPathParam],
         body: BodyParam[UpdateAppConfigFragmentInput],
     ) -> APIResponse:
-        """Update a fragment's config document by id (superadmin only)."""
-        result = await self._adapter.admin_update(
+        """Update a fragment's config; authorized by the layer's permission policy (auth required)."""
+        result = await self._adapter.update(
             AppConfigFragmentID(path.parsed.fragment_id), body.parsed
         )
         return APIResponse.build(status_code=HTTPStatus.OK, response_model=result)
 
-    async def admin_purge(
+    async def purge(
         self,
         path: PathParam[AppConfigFragmentIdPathParam],
     ) -> APIResponse:
-        """Purge a fragment by id (superadmin only)."""
-        result = await self._adapter.admin_purge(
-            PurgeAppConfigFragmentInput(id=path.parsed.fragment_id)
-        )
+        """Purge a fragment by id; authorized by the layer's permission policy (auth required)."""
+        result = await self._adapter.purge(PurgeAppConfigFragmentInput(id=path.parsed.fragment_id))
         return APIResponse.build(status_code=HTTPStatus.OK, response_model=result)
 
     async def admin_search(

src/ai/backend/manager/api/rest/v2/app_config_fragment/registry.py

@@ -20,20 +20,23 @@ def register_v2_app_config_fragment_routes(
     """Register all REST v2 app config fragment routes.
 
     Layout:
-        POST   /                  create a fragment              (superadmin)
+        POST   /                  create a fragment              (auth; RBAC)
         POST   /search            admin paginated search         (superadmin)
         POST   /scoped-search     principal-visible search       (auth)
         GET    /{fragment_id}     get by id                      (superadmin)
-        PATCH  /{fragment_id}     update config by id            (superadmin)
-        DELETE /{fragment_id}     purge by id                    (superadmin)
+        PATCH  /{fragment_id}     update config by id            (auth; RBAC)
+        DELETE /{fragment_id}     purge by id                    (auth; RBAC)
     """
     registry = RouteRegistry.create("app-config-fragments", route_deps.cors_options)
 
-    registry.add("POST", "/", handler.admin_create, middlewares=[superadmin_required])
+    # Write routes are auth_required; write authorization is RBAC (validators on the
+    # processors): a user writes their own user-scope fragment, a domain admin their
+    # domain's, and only a superadmin the public scope. Reads stay allowlist/visibility.
+    registry.add("POST", "/", handler.create, middlewares=[auth_required])
     registry.add("POST", "/search", handler.admin_search, middlewares=[superadmin_required])
     registry.add("POST", "/scoped-search", handler.scoped_search, middlewares=[auth_required])
     registry.add("GET", "/{fragment_id}", handler.admin_get, middlewares=[superadmin_required])
-    registry.add("PATCH", "/{fragment_id}", handler.admin_update, middlewares=[superadmin_required])
-    registry.add("DELETE", "/{fragment_id}", handler.admin_purge, middlewares=[superadmin_required])
+    registry.add("PATCH", "/{fragment_id}", handler.update, middlewares=[auth_required])
+    registry.add("DELETE", "/{fragment_id}", handler.purge, middlewares=[auth_required])
 
     return registry

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

@@ -7,6 +7,7 @@
 from ai.backend.manager.actions.processor.scope import ScopeActionProcessor
 from ai.backend.manager.actions.processor.single_entity import SingleEntityActionProcessor
 from ai.backend.manager.actions.types import AbstractProcessorPackage, ActionSpec
+from ai.backend.manager.actions.validators import ActionValidators
 from ai.backend.manager.services.app_config_fragment.actions.admin_search import (
     AdminSearchAppConfigFragmentAction,
     AdminSearchAppConfigFragmentActionResult,
@@ -77,16 +78,35 @@ def __init__(
         self,
         service: AppConfigFragmentService,
         action_monitors: list[ActionMonitor],
+        validators: ActionValidators,
     ) -> None:
-        self.create = ScopeActionProcessor(service.create, action_monitors)
+        # Writes are RBAC-gated (a non-admin may write only their own user-scope fragment,
+        # resolved via the caller's roles + the scope→entity association). Reads are gated by
+        # the allow-list / visibility, not RBAC, so they carry no validator.
+        scope_v = validators.rbac.scope
+        single_v = validators.rbac.single_entity
+        bulk_v = validators.rbac.bulk
+        self.create = ScopeActionProcessor(
+            service.create, monitors=action_monitors, validators=[scope_v]
+        )
         self.get = SingleEntityActionProcessor(service.get, action_monitors)
         self.admin_search = ScopeActionProcessor(service.admin_search, action_monitors)
         self.scoped_search = BulkActionProcessor(service.scoped_search, monitors=action_monitors)
-        self.update = SingleEntityActionProcessor(service.update, action_monitors)
-        self.purge = SingleEntityActionProcessor(service.purge, action_monitors)
-        self.bulk_create = BulkActionProcessor(service.bulk_create, monitors=action_monitors)
-        self.bulk_update = BulkActionProcessor(service.bulk_update, monitors=action_monitors)
-        self.bulk_purge = BulkActionProcessor(service.bulk_purge, monitors=action_monitors)
+        self.update = SingleEntityActionProcessor(
+            service.update, monitors=action_monitors, validators=[single_v]
+        )
+        self.purge = SingleEntityActionProcessor(
+            service.purge, monitors=action_monitors, validators=[single_v]
+        )
+        self.bulk_create = BulkActionProcessor(
+            service.bulk_create, monitors=action_monitors, validators=[bulk_v]
+        )
+        self.bulk_update = BulkActionProcessor(
+            service.bulk_update, monitors=action_monitors, validators=[bulk_v]
+        )
+        self.bulk_purge = BulkActionProcessor(
+            service.bulk_purge, monitors=action_monitors, validators=[bulk_v]
+        )
 
     @override
     def supported_actions(self) -> list[ActionSpec]:

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

@@ -465,7 +465,7 @@ def create_processors(
             services.app_config_allow_list, action_monitors
         ),
         app_config_fragment=AppConfigFragmentProcessors(
-            services.app_config_fragment, action_monitors
+            services.app_config_fragment, action_monitors, validators
         ),
         domain=DomainProcessors(services.domain, action_monitors, validators),
         dotfile=DotfileProcessors(services.dotfile, action_monitors, validators),