feat(BA-6859): bind app_config fragments to their RBAC scope on write (repository layer)

Run the app_config_fragment repository on RBACOpsProvider and make every
fragment write flow through the RBAC scope machinery, branch-free:

- The shared RBAC creator/unbinder specs now model *global-scoped* entities:
  RBACEntityCreator.scope_ref and RBACScopeEntityUnbinder.scope_ref accept
  None, meaning the entity lives outside the RBAC scope hierarchy and carries
  no scope association.
- AppConfigScopeType.to_rbac_element_type() maps a fragment scope to its RBAC
  scope element (public -> None), next to its to_rbac_scope_type sibling.
- create binds a user/domain fragment to its scope via
  association_scopes_entities in the same tx; purge unbinds it through
  RBACWriteOps.unbind_scope_entities (implemented directly on the ops
  provider — the standalone execute_rbac_* helpers are transitional),
  deleting the row and its association atomically. A public fragment takes
  the same single code path with scope_ref=None.
- Registers APP_CONFIG_FRAGMENT as an owner-accessible RBAC resource type.
- Repository/db_source create now takes AppConfigFragmentCreatorSpec directly.

Data-layer foundation for RBAC-gating fragment writes; service-layer
validators + DI registration follow (BA-6860).

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

Author: jopemachine

changes/12801.feature.md

@@ -0,0 +1 @@
+Bind AppConfig fragments to their RBAC scope on write (repository layer): a `user` / `domain` fragment is associated to its RBAC scope on create and unbound on purge (`public` maps to GLOBAL and carries none), so fragment write authorization can be enforced by RBAC. The allow list becomes a read allowlist (registration + `rank`) while writes move to RBAC (BEP-1052).

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

@@ -4,7 +4,7 @@
 
 import enum
 
-from ai.backend.common.data.permission.types import ScopeType
+from ai.backend.common.data.permission.types import RBACElementType, ScopeType
 
 __all__ = ("AppConfigScopeType",)
 
@@ -34,6 +34,21 @@ def to_rbac_scope_type(self) -> ScopeType:
             case AppConfigScopeType.USER:
                 return ScopeType.USER
 
+    def to_rbac_element_type(self) -> RBACElementType | None:
+        """The RBAC scope element a fragment at this scope belongs to.
+
+        ``public`` maps to the global scope, which has no RBAC scope element — a public
+        fragment is *global-scoped* (no scope association; superadmin-only writes), so it
+        returns ``None``.
+        """
+        match self:
+            case AppConfigScopeType.PUBLIC:
+                return None
+            case AppConfigScopeType.DOMAIN:
+                return RBACElementType.DOMAIN
+            case AppConfigScopeType.USER:
+                return RBACElementType.USER
+
     def to_rbac_scope_id(self, scope_id: str) -> str:
         """The RBAC scope id for a write at this fragment scope.
 

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

@@ -270,6 +270,7 @@ def _resource_types(cls) -> set[EntityType]:
             cls.ARTIFACT,
             cls.ARTIFACT_REGISTRY,
             cls.APP_CONFIG,
+            cls.APP_CONFIG_FRAGMENT,
             cls.NOTIFICATION_CHANNEL,
             cls.NOTIFICATION_RULE,
             cls.MODEL_DEPLOYMENT,

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

@@ -6,6 +6,7 @@
 
 import sqlalchemy as sa
 
+from ai.backend.common.data.permission.types import RBACElementType
 from ai.backend.common.exception import BackendAIError
 from ai.backend.common.identifier.app_config_fragment import AppConfigFragmentID
 from ai.backend.common.metrics.metric import DomainType, LayerType
@@ -25,19 +26,26 @@
 from ai.backend.manager.models.app_config_fragment.conditions import AppConfigFragmentConditions
 from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
 from ai.backend.manager.models.scopes import SearchScope
+from ai.backend.manager.repositories.app_config_fragment.creators import (
+    AppConfigFragmentCreatorSpec,
+)
+from ai.backend.manager.repositories.app_config_fragment.scope_binders import (
+    AppConfigFragmentScopeUnbinder,
+    fragment_rbac_scope_ref,
+)
 from ai.backend.manager.repositories.app_config_fragment.types import (
     AppConfigScopeArguments,
 )
 from ai.backend.manager.repositories.base import (
     BatchQuerier,
     BulkCreator,
-    Creator,
     NoPagination,
     Purger,
     Querier,
     Updater,
 )
-from ai.backend.manager.repositories.ops import DBOpsProvider
+from ai.backend.manager.repositories.base.rbac.entity_creator import RBACEntityCreator
+from ai.backend.manager.repositories.ops.rbac.provider import RBACOpsProvider
 
 __all__ = ("AppConfigFragmentDBSource",)
 
@@ -64,19 +72,27 @@
 class AppConfigFragmentDBSource:
     """Database source for app config fragment operations."""
 
-    _ops: DBOpsProvider
+    _ops: RBACOpsProvider
 
-    def __init__(self, ops_provider: DBOpsProvider) -> None:
+    def __init__(self, ops_provider: RBACOpsProvider) -> None:
         self._ops = ops_provider
 
     @app_config_fragment_db_source_resilience.apply()
-    async def create(self, creator: Creator[AppConfigFragmentRow]) -> AppConfigFragmentData:
-        # The FK to the allow-list is the gate: inserting a fragment with no
-        # allow-list row for its ``(config_name, scope_type)`` raises
-        # ``AppConfigFragmentWriteNotAllowed`` (see the spec's integrity checks).
+    async def create(self, spec: AppConfigFragmentCreatorSpec) -> AppConfigFragmentData:
+        # The FK to the allow-list gates *existence* (a missing allow-list row for the
+        # ``(config_name, scope_type)`` raises ``AppConfigFragmentWriteNotAllowed``). Create
+        # also binds the fragment to its owning RBAC scope via
+        # ``association_scopes_entities`` (same tx) so a later update/purge can resolve its
+        # scope for the RBAC write check; a ``public`` fragment is global-scoped
+        # (``scope_ref`` is ``None``) and carries no association.
+        rbac_creator = RBACEntityCreator(
+            spec=spec,
+            element_type=RBACElementType.APP_CONFIG_FRAGMENT,
+            scope_ref=fragment_rbac_scope_ref(spec.scope_type, spec.scope_id),
+        )
         async with self._ops.write_ops() as w:
-            created = await w.create(creator)
-            return created.row.to_data()
+            created = await w.bulk_create_scoped([rbac_creator])
+            return created.rows[0].to_data()
 
     @app_config_fragment_db_source_resilience.apply()
     async def get_by_id(self, fragment_id: AppConfigFragmentID) -> AppConfigFragmentData:
@@ -99,12 +115,23 @@ async def update(self, updater: Updater[AppConfigFragmentRow]) -> AppConfigFragm
 
     @app_config_fragment_db_source_resilience.apply()
     async def purge(self, purger: Purger[AppConfigFragmentRow]) -> AppConfigFragmentData:
-        # No write-gate here — see ``update``.
+        # A fragment is a config bound at a scope, so purging is an RBAC *unbind*: the row
+        # and its scope association are deleted atomically (the association table has no FK
+        # cascade). A public fragment is global-scoped and simply has no association. The
+        # row is fetched first to resolve its scope and to return its data.
         async with self._ops.write_ops() as w:
-            result = await w.purge(purger)
-            if result is None:
+            found = await w.query(Querier(row_class=AppConfigFragmentRow, pk_value=purger.pk_value))
+            if found is None:
                 raise AppConfigFragmentNotFound(f"App config fragment {purger.pk_value} not found")
-            return result.row.to_data()
+            data = found.row.to_data()
+            await w.unbind_scope_entities(
+                AppConfigFragmentScopeUnbinder(
+                    fragment_id=data.id,
+                    fragment_scope_type=data.scope_type,
+                    fragment_scope_id=data.scope_id,
+                )
+            )
+            return data
 
     @app_config_fragment_db_source_resilience.apply()
     async def bulk_create(

src/ai/backend/manager/repositories/app_config_fragment/repositories.py

@@ -4,6 +4,7 @@
 from ai.backend.manager.repositories.app_config_fragment.repository import (
     AppConfigFragmentRepository,
 )
+from ai.backend.manager.repositories.ops.rbac.provider import RBACOpsProvider
 from ai.backend.manager.repositories.types import RepositoryArgs
 
 
@@ -13,6 +14,8 @@ class AppConfigFragmentRepositories:
 
     @classmethod
     def create(cls, args: RepositoryArgs) -> Self:
+        # Fragment writes bind the fragment to its RBAC scope (see the db_source), so the
+        # repository runs on the RBAC-scoped ops provider rather than the plain one.
         return cls(
-            repository=AppConfigFragmentRepository(args.ops_provider),
+            repository=AppConfigFragmentRepository(RBACOpsProvider(args.db)),
         )

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

@@ -15,6 +15,9 @@
 )
 from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
 from ai.backend.manager.models.scopes import SearchScope
+from ai.backend.manager.repositories.app_config_fragment.creators import (
+    AppConfigFragmentCreatorSpec,
+)
 from ai.backend.manager.repositories.app_config_fragment.db_source import (
     AppConfigFragmentDBSource,
 )
@@ -24,11 +27,10 @@
 from ai.backend.manager.repositories.base import (
     BatchQuerier,
     BulkCreator,
-    Creator,
     Purger,
     Updater,
 )
-from ai.backend.manager.repositories.ops import DBOpsProvider
+from ai.backend.manager.repositories.ops.rbac.provider import RBACOpsProvider
 
 __all__ = ("AppConfigFragmentRepository",)
 
@@ -57,12 +59,12 @@ class AppConfigFragmentRepository:
 
     _db_source: AppConfigFragmentDBSource
 
-    def __init__(self, ops_provider: DBOpsProvider) -> None:
+    def __init__(self, ops_provider: RBACOpsProvider) -> None:
         self._db_source = AppConfigFragmentDBSource(ops_provider)
 
     @app_config_fragment_repository_resilience.apply()
-    async def create(self, creator: Creator[AppConfigFragmentRow]) -> AppConfigFragmentData:
-        return await self._db_source.create(creator)
+    async def create(self, spec: AppConfigFragmentCreatorSpec) -> AppConfigFragmentData:
+        return await self._db_source.create(spec)
 
     @app_config_fragment_repository_resilience.apply()
     async def get_by_id(self, fragment_id: AppConfigFragmentID) -> AppConfigFragmentData:

src/ai/backend/manager/repositories/app_config_fragment/scope_binders.py

@@ -0,0 +1,70 @@
+"""RBAC scope binding helpers for app config fragments."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import override
+
+import sqlalchemy as sa
+
+from ai.backend.common.data.app_config.types import AppConfigScopeType
+from ai.backend.common.data.permission.types import RBACElementType
+from ai.backend.common.identifier.app_config_fragment import AppConfigFragmentID
+from ai.backend.manager.data.permission.types import RBACElementRef
+from ai.backend.manager.models.app_config_fragment.row import AppConfigFragmentRow
+from ai.backend.manager.repositories.base.purger import BatchPurgerSpec
+from ai.backend.manager.repositories.base.rbac.scope_unbinder import RBACScopeEntityUnbinder
+
+
+def fragment_rbac_scope_ref(scope_type: AppConfigScopeType, scope_id: str) -> RBACElementRef | None:
+    """The RBAC scope a fragment belongs to; ``None`` for global-scoped (public) fragments."""
+    element = scope_type.to_rbac_element_type()
+    if element is None:
+        return None
+    return RBACElementRef(element, scope_id)
+
+
+@dataclass
+class AppConfigFragmentByIdPurgerSpec(BatchPurgerSpec[AppConfigFragmentRow]):
+    """Select a single fragment row (by id) for deletion."""
+
+    fragment_id: AppConfigFragmentID
+
+    @override
+    def build_subquery(self) -> sa.sql.Select[tuple[AppConfigFragmentRow]]:
+        return sa.select(AppConfigFragmentRow).where(AppConfigFragmentRow.id == self.fragment_id)
+
+
+@dataclass
+class AppConfigFragmentScopeUnbinder(RBACScopeEntityUnbinder[AppConfigFragmentRow]):
+    """Unbind (purge) one fragment from its owning scope.
+
+    A fragment is a config bound at a scope, so purging it is an unbind: the fragment row
+    and its RBAC scope association are deleted atomically. A ``public`` fragment is
+    global-scoped (``scope_ref`` is ``None``), so only its row is deleted — it never had
+    an association.
+    """
+
+    fragment_id: AppConfigFragmentID
+    fragment_scope_type: AppConfigScopeType
+    fragment_scope_id: str
+
+    @override
+    def build_purger_spec(self) -> BatchPurgerSpec[AppConfigFragmentRow]:
+        return AppConfigFragmentByIdPurgerSpec(fragment_id=self.fragment_id)
+
+    @property
+    @override
+    def entity_type(self) -> RBACElementType:
+        return RBACElementType.APP_CONFIG_FRAGMENT
+
+    @property
+    @override
+    def scope_ref(self) -> RBACElementRef | None:
+        return fragment_rbac_scope_ref(self.fragment_scope_type, self.fragment_scope_id)
+
+    @property
+    @override
+    def entity_ids(self) -> Sequence[str] | None:
+        return [str(self.fragment_id)]

src/ai/backend/manager/repositories/base/rbac/entity_creator.py

@@ -37,19 +37,21 @@ class RBACEntityCreator[TRow: Base]:
     """Creator for a single entity with scope associations for RBAC.
 
     Creates an entity row and associates it with one or more permission scopes.
-    The primary scope is required; additional scopes are optional.
 
     Attributes:
         spec: CreatorSpec implementation defining the row to create.
         element_type: The RBAC element type for this entity.
         scope_ref: Primary scope reference (scope_type + scope_id) for this entity.
+            ``None`` marks a *global-scoped* entity — one living outside the RBAC scope
+            hierarchy (e.g. a public app-config fragment) — which carries no scope
+            association (only a superadmin reaches it, bypassing scope resolution).
         additional_scope_refs: Additional scope references for multi-scope entities.
         relation_type: The relation type for the scope-entity association. Defaults to AUTO.
     """
 
     spec: CreatorSpec[TRow]
     element_type: RBACElementType
-    scope_ref: RBACElementRef
+    scope_ref: RBACElementRef | None
     additional_scope_refs: Sequence[RBACElementRef] = field(default_factory=list)
     relation_type: RelationType = RelationType.AUTO
 
@@ -106,7 +108,10 @@ async def execute_rbac_entity_creator[TRow: Base](
     instance_state = inspect(row)
     pk_value = instance_state.identity[0]
     entity_type = creator.element_type.to_entity_type()
-    all_scope_refs = [creator.scope_ref, *creator.additional_scope_refs]
+    # A None primary scope marks a global-scoped entity: no association row.
+    all_scope_refs = [
+        ref for ref in (creator.scope_ref, *creator.additional_scope_refs) if ref is not None
+    ]
     associations = [
         AssociationScopesEntitiesRow(
             scope_type=scope_ref.element_type.to_scope_type(),
@@ -117,7 +122,8 @@ async def execute_rbac_entity_creator[TRow: Base](
         )
         for scope_ref in all_scope_refs
     ]
-    await bulk_insert_on_conflict_do_nothing(db_sess, associations)
+    if associations:
+        await bulk_insert_on_conflict_do_nothing(db_sess, associations)
 
     return RBACEntityCreatorResult(row=row)
 
@@ -255,11 +261,14 @@ async def execute_rbac_entity_creators[TRow: Base](
         match_integrity_error(parsed, checks)
 
     # 3. Collect all associations from each creator's scope refs
+    # (a None primary scope marks a global-scoped entity: no association row)
     associations: list[AssociationScopesEntitiesRow] = []
     for creator, row in zip(creators, rows, strict=True):
         pk_value = inspect(row).identity[0]
         entity_type = creator.element_type.to_entity_type()
-        all_scope_refs = [creator.scope_ref, *creator.additional_scope_refs]
+        all_scope_refs = [
+            ref for ref in (creator.scope_ref, *creator.additional_scope_refs) if ref is not None
+        ]
         for scope_ref in all_scope_refs:
             associations.append(
                 AssociationScopesEntitiesRow(
@@ -270,6 +279,7 @@ async def execute_rbac_entity_creators[TRow: Base](
                     relation_type=creator.relation_type,
                 ),
             )
-    await bulk_insert_on_conflict_do_nothing(db_sess, associations)
+    if associations:
+        await bulk_insert_on_conflict_do_nothing(db_sess, associations)
 
     return RBACBulkEntityCreatorResult(rows=rows)

src/ai/backend/manager/repositories/base/rbac/scope_unbinder.py

@@ -53,8 +53,12 @@ def entity_type(self) -> RBACElementType:
 
     @property
     @abstractmethod
-    def scope_ref(self) -> RBACElementRef:
-        """RBAC element ref for the scope to unbind entities from."""
+    def scope_ref(self) -> RBACElementRef | None:
+        """RBAC element ref for the scope to unbind entities from.
+
+        ``None`` marks global-scoped entities (outside the RBAC scope hierarchy):
+        only the business rows are deleted — there is no scope association to remove.
+        """
         raise NotImplementedError
 
     @property
@@ -110,6 +114,12 @@ async def execute_rbac_scope_entity_unbinder[TRow: Base](
         db_sess, BatchPurger(spec=unbinder.build_purger_spec())
     )
     scope_ref = unbinder.scope_ref
+    if scope_ref is None:
+        # Global-scoped entities carry no scope association — nothing more to delete.
+        return RBACUnbinderResult(
+            deleted_count=purge_result.deleted_count,
+            association_rows=[],
+        )
     where_clauses = [
         AssociationScopesEntitiesRow.entity_type == unbinder.entity_type.to_entity_type(),
         AssociationScopesEntitiesRow.scope_type == scope_ref.element_type.to_scope_type(),