feat(BA-6836): authorize fragment writes via allow_list write_access tier

Enforce AppConfig's standalone write authorization: each fragment write is
checked against the write_access tier of the target (config_name, scope_type)
allow_list entry (AppConfigAccessLevel.is_satisfied_by) — no RBAC validator.
A user may manage their own user-scope fragment; public/domain default to
admin-only; superadmin passes any tier. create authorizes from the creator
spec; update/purge load the fragment first; bulk writes authorize per item
(unauthorized items become partial failures). Fragment write routes switch
from superadmin_required to auth_required; the service gains the allow_list
repository (by_config_and_scope) for the tier lookup.

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

Author: jopemachine

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

@@ -5,6 +5,7 @@
 import enum
 
 from ai.backend.common.data.permission.types import ScopeType
+from ai.backend.common.data.user.types import UserData
 
 __all__ = ("AppConfigScopeType", "AppConfigAccessLevel")
 
@@ -27,6 +28,39 @@ class AppConfigAccessLevel(enum.StrEnum):
     OWNER = "owner"
     ADMIN = "admin"
 
+    def is_satisfied_by(
+        self, user: UserData | None, scope_type: AppConfigScopeType, scope_id: str
+    ) -> bool:
+        """Whether ``user`` meets this access tier for a fragment at ``(scope_type, scope_id)``.
+
+        Anonymous (``user is None``) satisfies only ``public``. A superadmin satisfies every
+        tier. ``owner`` is instance-relative: the user itself for ``user`` scope, a domain
+        admin for ``domain`` scope; ``public`` scope has no owner.
+        """
+        if user is None:
+            return self is AppConfigAccessLevel.PUBLIC
+        if user.is_superadmin:
+            return True
+        match self:
+            case AppConfigAccessLevel.PUBLIC:
+                return True
+            case AppConfigAccessLevel.AUTHENTICATED:
+                return user.is_authorized
+            case AppConfigAccessLevel.OWNER:
+                match scope_type:
+                    case AppConfigScopeType.USER:
+                        return str(user.user_id) == scope_id
+                    case AppConfigScopeType.DOMAIN:
+                        # NOTE: assumes a domain fragment's scope_id holds the domain *name*
+                        # (matching UserData.domain_name). Not exercised by the default policy
+                        # (domain write defaults to ADMIN); verify when domain-owner writes land.
+                        return user.is_admin and user.domain_name == scope_id
+                    case AppConfigScopeType.PUBLIC:
+                        return False
+            case AppConfigAccessLevel.ADMIN:
+                # superadmin already returned True above; no one else satisfies ADMIN
+                return False
+
 
 class AppConfigScopeType(enum.StrEnum):
     """Scope at which an app config fragment is written (BEP-1052).

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

@@ -99,17 +99,16 @@ 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,
         )
+        # Authorization is enforced in the service against the layer's write_access tier.
         action_result = await self._processors.app_config_fragment.create.wait_for_complete(
-            CreateAppConfigFragmentAction(creator_spec=spec)
+            CreateAppConfigFragmentAction(creator_spec=spec, requester=current_user())
         )
         return CreateAppConfigFragmentPayload(
             app_config_fragment=self._fragment_to_node(action_result.fragment),
@@ -121,30 +120,28 @@ 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.
+        # Authorization is enforced in the service: it loads the fragment and checks the
+        # caller against the layer's write_access tier before updating.
         action_result = await self._processors.app_config_fragment.update.wait_for_complete(
-            UpdateAppConfigFragmentAction(updater=updater)
+            UpdateAppConfigFragmentAction(updater=updater, requester=current_user())
         )
         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``.
+        # Authorization is enforced in the service — see ``update``.
         action_result = await self._processors.app_config_fragment.purge.wait_for_complete(
-            PurgeAppConfigFragmentAction(purger=purger)
+            PurgeAppConfigFragmentAction(purger=purger, requester=current_user())
         )
         return PurgeAppConfigFragmentPayload(id=action_result.fragment.id)
 

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 write_access tier (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 write_access tier (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 write_access tier (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,22 @@ 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; write_access tier)
         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; write_access tier)
+        DELETE /{fragment_id}     purge by id                    (auth; write_access tier)
     """
     registry = RouteRegistry.create("app-config-fragments", route_deps.cors_options)
 
-    registry.add("POST", "/", handler.admin_create, middlewares=[superadmin_required])
+    # Write routes are auth_required; the service authorizes each write against the target
+    # layer's allow_list write_access tier (a superadmin passes any tier).
+    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/repositories/app_config_allow_list/db_source/db_source.py

@@ -9,6 +9,8 @@
 
 import sqlalchemy as sa
 
+from ai.backend.common.data.app_config.types import AppConfigScopeType
+from ai.backend.common.data.filter_specs import StringMatchSpec
 from ai.backend.common.exception import BackendAIError
 from ai.backend.common.identifier.app_config_allow_list import AppConfigAllowListID
 from ai.backend.common.metrics.metric import DomainType, LayerType
@@ -20,10 +22,14 @@
     AppConfigAllowListSearchResult,
 )
 from ai.backend.manager.errors.app_config import AppConfigAllowListNotFound
+from ai.backend.manager.models.app_config_allow_list.conditions import (
+    AppConfigAllowListConditions,
+)
 from ai.backend.manager.models.app_config_allow_list.row import AppConfigAllowListRow
 from ai.backend.manager.repositories.base import (
     BatchQuerier,
     Creator,
+    OffsetPagination,
     Purger,
     Querier,
     Updater,
@@ -94,6 +100,30 @@ async def purge(self, purger: Purger[AppConfigAllowListRow]) -> AppConfigAllowLi
                 raise AppConfigAllowListNotFound("App config allow-list entry not found")
             return result.row.to_data()
 
+    @app_config_allow_list_db_source_resilience.apply()
+    async def by_config_and_scope(
+        self, config_name: str, scope_type: AppConfigScopeType
+    ) -> AppConfigAllowListData | None:
+        """The allow-list entry for the unique ``(config_name, scope_type)`` pair, or ``None``.
+
+        Used by write/read authorization to read the layer's access tiers.
+        """
+        querier = BatchQuerier(
+            pagination=OffsetPagination(limit=1, offset=0),
+            conditions=[
+                AppConfigAllowListConditions.by_config_name_equals(
+                    StringMatchSpec(config_name, case_insensitive=False, negated=False)
+                ),
+                AppConfigAllowListConditions.by_scope_type_equals(scope_type),
+            ],
+        )
+        async with self._ops.read_ops() as r:
+            result = await r.batch_query_in_global(sa.select(AppConfigAllowListRow), querier)
+            if not result.rows:
+                return None
+            data: AppConfigAllowListData = result.rows[0].AppConfigAllowListRow.to_data()
+            return data
+
     @app_config_allow_list_db_source_resilience.apply()
     async def search(self, querier: BatchQuerier) -> AppConfigAllowListSearchResult:
         async with self._ops.read_ops() as r:

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

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from ai.backend.common.data.app_config.types import AppConfigScopeType
 from ai.backend.common.exception import BackendAIError
 from ai.backend.common.identifier.app_config_allow_list import AppConfigAllowListID
 from ai.backend.common.metrics.metric import DomainType, LayerType
@@ -63,6 +64,12 @@ async def create(
     async def get_by_id(self, allow_list_id: AppConfigAllowListID) -> AppConfigAllowListData:
         return await self._db_source.get_by_id(allow_list_id)
 
+    @app_config_allow_list_repository_resilience.apply()
+    async def by_config_and_scope(
+        self, config_name: str, scope_type: AppConfigScopeType
+    ) -> AppConfigAllowListData | None:
+        return await self._db_source.by_config_and_scope(config_name, scope_type)
+
     @app_config_allow_list_repository_resilience.apply()
     async def search(self, querier: BatchQuerier) -> AppConfigAllowListSearchResult:
         return await self._db_source.search(querier)

src/ai/backend/manager/services/app_config_fragment/actions/bulk_create.py

@@ -4,6 +4,7 @@
 from dataclasses import dataclass
 from typing import override
 
+from ai.backend.common.data.user.types import UserData
 from ai.backend.manager.actions.types import ActionOperationType
 from ai.backend.manager.repositories.app_config_fragment.creators import (
     AppConfigFragmentCreatorSpec,
@@ -20,6 +21,7 @@ class BulkCreateAppConfigFragmentAction(AppConfigFragmentBulkAction):
     """Create many fragments with per-item partial success; the FK to the allow-list gates each write."""
 
     creator_specs: Sequence[AppConfigFragmentCreatorSpec]
+    requester: UserData | None = None
 
     @override
     @classmethod

src/ai/backend/manager/services/app_config_fragment/actions/bulk_purge.py

@@ -4,6 +4,7 @@
 from dataclasses import dataclass
 from typing import cast, override
 
+from ai.backend.common.data.user.types import UserData
 from ai.backend.common.identifier.app_config_fragment import AppConfigFragmentID
 from ai.backend.manager.actions.types import ActionOperationType
 from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
@@ -20,6 +21,7 @@ class BulkPurgeAppConfigFragmentAction(AppConfigFragmentBulkAction):
     """Purge many fragments with per-item partial success (no gate — purging the allow-list entry itself cascades to its fragments separately)."""
 
     purgers: Sequence[Purger[AppConfigFragmentRow]]
+    requester: UserData | None = None
 
     @override
     @classmethod

src/ai/backend/manager/services/app_config_fragment/actions/bulk_update.py

@@ -4,6 +4,7 @@
 from dataclasses import dataclass
 from typing import cast, override
 
+from ai.backend.common.data.user.types import UserData
 from ai.backend.common.identifier.app_config_fragment import AppConfigFragmentID
 from ai.backend.manager.actions.types import ActionOperationType
 from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
@@ -20,6 +21,7 @@ class BulkUpdateAppConfigFragmentAction(AppConfigFragmentBulkAction):
     """Update many fragments' ``config`` with per-item partial success (no gate — an existing fragment is always writable at its own scope)."""
 
     updaters: Sequence[Updater[AppConfigFragmentRow]]
+    requester: UserData | None = None
 
     @override
     @classmethod

src/ai/backend/manager/services/app_config_fragment/actions/create.py

@@ -4,6 +4,7 @@
 from typing import override
 
 from ai.backend.common.data.permission.types import RBACElementType, ScopeType
+from ai.backend.common.data.user.types import UserData
 from ai.backend.manager.actions.types import ActionOperationType
 from ai.backend.manager.data.app_config_fragment.types import (
     AppConfigFragmentData,
@@ -28,6 +29,7 @@ class CreateAppConfigFragmentAction(AppConfigFragmentScopeAction):
     """
 
     creator_spec: AppConfigFragmentCreatorSpec
+    requester: UserData | None = None
 
     @override
     @classmethod