fix(BA-7297): restore the user-facing reads the ops sweep gated

Moving these actions onto the global processor attached a SUPERADMIN gate that
the legacy processor never had, so routes their own registry declares as
auth_required started answering 403 to every regular user.

The route registries are the authority: a domain can serve two surfaces from one
processor -- storage answers an auth_required v1 surface and a
superadmin_required v2 surface -- so the processor carries the weaker gate and
the route middleware keeps the stronger one. Reads move to the public processor,
which demands a caller but not a role; writes stay behind the gate.

Three domains had no test covering a non-admin caller at all, since every
existing case authenticates as an admin. A read/write access pair now pins the
contract for the query-preset catalog.

Also repair the global purge path: it rebuilt the deleted row by calling the row
class, which fails on any row whose __init__ narrows to the caller-supplied
columns -- a server-generated one arrives as an unexpected keyword. It now maps
the RETURNING columns the way insert and update already did.

error_log keeps the gate by decision: its recording and per-user reads become
superadmin-only, and the tests state that.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: HyeockJinKim

src/ai/backend/manager/repositories/ops/v2/write_base.py

@@ -195,15 +195,16 @@ async def _delete_row_returning[TRow: Base](
                 f"Purger only supports single-column primary keys (table: {table.name})",
             )
         stmt = sa.delete(table).where(pk_columns[0] == pk_value).returning(*table.columns)
+        # from_statement lets SQLAlchemy map the RETURNING columns onto the ORM class.
+        # Calling the row class instead would go through its __init__, which many rows
+        # narrow to the caller-supplied columns — a server-generated one then arrives as
+        # an unexpected keyword and the purge fails on rows it can read back perfectly.
+        select_stmt = sa.select(row_class).from_statement(stmt)
         try:
-            result = await self._sess.execute(stmt)
+            result = await self._sess.execute(select_stmt)
         except sa.exc.IntegrityError as e:
             raise self._parse_integrity_error(e) from e
-        row_data = result.fetchone()
-        if row_data is None:
-            return None
-        row: TRow = row_class(**dict(row_data._mapping))
-        return row
+        return result.scalar_one_or_none()
 
     async def _upsert_row_returning[TRow: Base](
         self,

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

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
 from ai.backend.manager.actions.registry import ProcessorGroup
-from ai.backend.manager.actions.v2.global_scope.processor import GlobalActionProcessor
+from ai.backend.manager.actions.v2.global_scope.processor import (
+    GlobalActionProcessor,
+    PublicActionProcessor,
+)
 from ai.backend.manager.actions.v2.ops.result import (
     BatchOpsResult,
     CreatedEntityOpsResult,
@@ -39,9 +42,9 @@ class ObjectStorageProcessors:
     ]
     update: GlobalActionProcessor[UpdateObjectStorageAction, EntityOpsResult[ObjectStorageData]]
     purge: GlobalActionProcessor[PurgeObjectStorageAction, EntityOpsResult[ObjectStorageData]]
-    get: GlobalActionProcessor[GetObjectStorageAction, EntityOpsResult[ObjectStorageData]]
-    list_storages: GlobalActionProcessor[ListObjectStorageAction, BatchOpsResult[ObjectStorageData]]
-    search_object_storages: GlobalActionProcessor[
+    get: PublicActionProcessor[GetObjectStorageAction, EntityOpsResult[ObjectStorageData]]
+    list_storages: PublicActionProcessor[ListObjectStorageAction, BatchOpsResult[ObjectStorageData]]
+    search_object_storages: PublicActionProcessor[
         SearchObjectStoragesAction, BatchOpsResult[ObjectStorageData]
     ]
     get_presigned_download_url: GlobalActionProcessor[
@@ -59,9 +62,9 @@ def __init__(
         self.create = group.global_create_ops(CreateObjectStorageAction)
         self.update = group.global_update_ops(UpdateObjectStorageAction)
         self.purge = group.global_purge_ops(PurgeObjectStorageAction)
-        self.get = group.global_get_ops(GetObjectStorageAction)
-        self.list_storages = group.global_search_ops(ListObjectStorageAction)
-        self.search_object_storages = group.global_search_ops(SearchObjectStoragesAction)
+        self.get = group.public_get_ops(GetObjectStorageAction)
+        self.list_storages = group.public_search_ops(ListObjectStorageAction)
+        self.search_object_storages = group.public_search_ops(SearchObjectStoragesAction)
         self.get_presigned_download_url = group.global_scope(
             GetDownloadPresignedURLAction, service.get_presigned_download_url
         )

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

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
 from ai.backend.manager.actions.registry import ProcessorGroup
-from ai.backend.manager.actions.v2.global_scope.processor import GlobalActionProcessor
+from ai.backend.manager.actions.v2.global_scope.processor import (
+    GlobalActionProcessor,
+    PublicActionProcessor,
+)
 from ai.backend.manager.actions.v2.ops.result import (
     BatchOpsResult,
     CreatedEntityOpsResult,
@@ -35,8 +38,8 @@ class PrometheusQueryPresetProcessors:
     create_preset: GlobalActionProcessor[
         CreatePresetAction, CreatedEntityOpsResult[PrometheusQueryPresetData]
     ]
-    get_preset: GlobalActionProcessor[GetPresetAction, EntityOpsResult[PrometheusQueryPresetData]]
-    search_presets: GlobalActionProcessor[
+    get_preset: PublicActionProcessor[GetPresetAction, EntityOpsResult[PrometheusQueryPresetData]]
+    search_presets: PublicActionProcessor[
         SearchPresetsAction, BatchOpsResult[PrometheusQueryPresetData]
     ]
     purge_preset: GlobalActionProcessor[
@@ -52,8 +55,8 @@ def __init__(
         group: ProcessorGroup[PrometheusQueryPresetData],
     ) -> None:
         self.create_preset = group.global_create_ops(CreatePresetAction)
-        self.get_preset = group.global_get_ops(GetPresetAction)
-        self.search_presets = group.global_search_ops(SearchPresetsAction)
+        self.get_preset = group.public_get_ops(GetPresetAction)
+        self.search_presets = group.public_search_ops(SearchPresetsAction)
         self.purge_preset = group.global_purge_ops(PurgePresetAction)
         self.update_preset = group.global_scope(UpdatePresetAction, service.update_preset)
         self.preview_preset = group.global_scope(PreviewPresetAction, service.preview_preset)

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

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
 from ai.backend.manager.actions.registry import ProcessorGroup
-from ai.backend.manager.actions.v2.global_scope.processor import GlobalActionProcessor
+from ai.backend.manager.actions.v2.global_scope.processor import (
+    GlobalActionProcessor,
+    PublicActionProcessor,
+)
 from ai.backend.manager.actions.v2.ops.result import (
     BatchOpsResult,
     CreatedEntityOpsResult,
@@ -31,11 +34,11 @@ class PrometheusQueryPresetCategoryProcessors:
         CreateCategoryAction,
         CreatedEntityOpsResult[PrometheusQueryPresetCategoryData],
     ]
-    get_category: GlobalActionProcessor[
+    get_category: PublicActionProcessor[
         GetCategoryAction,
         EntityOpsResult[PrometheusQueryPresetCategoryData],
     ]
-    search_categories: GlobalActionProcessor[
+    search_categories: PublicActionProcessor[
         SearchCategoriesAction,
         BatchOpsResult[PrometheusQueryPresetCategoryData],
     ]
@@ -46,6 +49,6 @@ class PrometheusQueryPresetCategoryProcessors:
 
     def __init__(self, group: ProcessorGroup[PrometheusQueryPresetCategoryData]) -> None:
         self.create_category = group.global_create_ops(CreateCategoryAction)
-        self.get_category = group.global_get_ops(GetCategoryAction)
-        self.search_categories = group.global_search_ops(SearchCategoriesAction)
+        self.get_category = group.public_get_ops(GetCategoryAction)
+        self.search_categories = group.public_search_ops(SearchCategoriesAction)
         self.purge_category = group.global_purge_ops(PurgeCategoryAction)

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

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
 from ai.backend.manager.actions.registry import ProcessorGroup
-from ai.backend.manager.actions.v2.global_scope.processor import GlobalActionProcessor
+from ai.backend.manager.actions.v2.global_scope.processor import (
+    GlobalActionProcessor,
+    PublicActionProcessor,
+)
 from ai.backend.manager.actions.v2.lookup.processor import LookupActionProcessor
 from ai.backend.manager.actions.v2.ops.result import (
     BatchOpsResult,
@@ -35,7 +38,7 @@ class RuntimeVariantProcessors:
     ]
     update: GlobalActionProcessor[UpdateRuntimeVariantAction, EntityOpsResult[RuntimeVariantData]]
     purge: GlobalActionProcessor[PurgeRuntimeVariantAction, EntityOpsResult[RuntimeVariantData]]
-    search: GlobalActionProcessor[SearchRuntimeVariantsAction, BatchOpsResult[RuntimeVariantData]]
+    search: PublicActionProcessor[SearchRuntimeVariantsAction, BatchOpsResult[RuntimeVariantData]]
     resolve_by_name: LookupActionProcessor[
         ResolveRuntimeVariantByNameAction, LookupOpsResult[RuntimeVariantData]
     ]
@@ -44,5 +47,5 @@ def __init__(self, group: ProcessorGroup[RuntimeVariantData]) -> None:
         self.create = group.global_create_ops(CreateRuntimeVariantAction)
         self.update = group.global_update_ops(UpdateRuntimeVariantAction)
         self.purge = group.global_purge_ops(PurgeRuntimeVariantAction)
-        self.search = group.global_search_ops(SearchRuntimeVariantsAction)
+        self.search = group.public_search_ops(SearchRuntimeVariantsAction)
         self.resolve_by_name = group.lookup_ops(ResolveRuntimeVariantByNameAction)

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

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
 from ai.backend.manager.actions.registry import ProcessorGroup
-from ai.backend.manager.actions.v2.global_scope.processor import GlobalActionProcessor
+from ai.backend.manager.actions.v2.global_scope.processor import (
+    GlobalActionProcessor,
+    PublicActionProcessor,
+)
 from ai.backend.manager.actions.v2.ops.result import BatchOpsResult, EntityOpsResult
 from ai.backend.manager.data.runtime_variant_preset.types import RuntimeVariantPresetData
 from ai.backend.manager.services.runtime_variant_preset.actions.create import (
@@ -35,7 +38,7 @@ class RuntimeVariantPresetProcessors:
     purge: GlobalActionProcessor[
         PurgeRuntimeVariantPresetAction, EntityOpsResult[RuntimeVariantPresetData]
     ]
-    search: GlobalActionProcessor[
+    search: PublicActionProcessor[
         SearchRuntimeVariantPresetsAction, BatchOpsResult[RuntimeVariantPresetData]
     ]
 
@@ -47,4 +50,4 @@ def __init__(
         self.create = group.global_scope(CreateRuntimeVariantPresetAction, service.create)
         self.update = group.global_scope(UpdateRuntimeVariantPresetAction, service.update)
         self.purge = group.global_purge_ops(PurgeRuntimeVariantPresetAction)
-        self.search = group.global_search_ops(SearchRuntimeVariantPresetsAction)
+        self.search = group.public_search_ops(SearchRuntimeVariantPresetsAction)

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

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
 from ai.backend.manager.actions.registry import ProcessorGroup
-from ai.backend.manager.actions.v2.global_scope.processor import GlobalActionProcessor
+from ai.backend.manager.actions.v2.global_scope.processor import (
+    GlobalActionProcessor,
+    PublicActionProcessor,
+)
 from ai.backend.manager.actions.v2.lookup.processor import LookupActionProcessor
 from ai.backend.manager.actions.v2.ops.result import (
     BatchOpsResult,
@@ -47,12 +50,12 @@ class VFSStorageProcessors:
     create: GlobalActionProcessor[CreateVFSStorageAction, CreatedEntityOpsResult[VFSStorageData]]
     update: GlobalActionProcessor[UpdateVFSStorageAction, EntityOpsResult[VFSStorageData]]
     purge: GlobalActionProcessor[PurgeVFSStorageAction, EntityOpsResult[VFSStorageData]]
-    get: GlobalActionProcessor[GetVFSStorageAction, EntityOpsResult[VFSStorageData]]
+    get: PublicActionProcessor[GetVFSStorageAction, EntityOpsResult[VFSStorageData]]
     resolve_by_name: LookupActionProcessor[
         ResolveVFSStorageByNameAction, LookupOpsResult[VFSStorageData]
     ]
-    list_storages: GlobalActionProcessor[ListVFSStorageAction, BatchOpsResult[VFSStorageData]]
-    search_vfs_storages: GlobalActionProcessor[
+    list_storages: PublicActionProcessor[ListVFSStorageAction, BatchOpsResult[VFSStorageData]]
+    search_vfs_storages: PublicActionProcessor[
         SearchVFSStoragesAction, BatchOpsResult[VFSStorageData]
     ]
     get_quota_scope: GlobalActionProcessor[GetQuotaScopeAction, GetQuotaScopeActionResult]
@@ -73,10 +76,10 @@ def __init__(
         self.create = group.global_create_ops(CreateVFSStorageAction)
         self.update = group.global_update_ops(UpdateVFSStorageAction)
         self.purge = group.global_purge_ops(PurgeVFSStorageAction)
-        self.get = group.global_get_ops(GetVFSStorageAction)
+        self.get = group.public_get_ops(GetVFSStorageAction)
         self.resolve_by_name = group.lookup_ops(ResolveVFSStorageByNameAction)
-        self.list_storages = group.global_search_ops(ListVFSStorageAction)
-        self.search_vfs_storages = group.global_search_ops(SearchVFSStoragesAction)
+        self.list_storages = group.public_search_ops(ListVFSStorageAction)
+        self.search_vfs_storages = group.public_search_ops(SearchVFSStoragesAction)
         self.get_quota_scope = group.global_scope(GetQuotaScopeAction, service.get_quota_scope)
         self.search_quota_scopes = group.global_scope(
             SearchQuotaScopesAction, service.search_quota_scopes

tests/component/conftest.py

@@ -44,8 +44,9 @@
 from ai.backend.common.clients.valkey_client.valkey_stream.client import ValkeyStreamClient
 from ai.backend.common.configs.etcd import EtcdConfig
 from ai.backend.common.configs.pyroscope import PyroscopeConfig
+from ai.backend.common.contexts.user import with_user
 from ai.backend.common.data.permission.types import EntityType, ScopeType
-from ai.backend.common.data.user.types import UserRole
+from ai.backend.common.data.user.types import UserData, UserRole
 from ai.backend.common.defs import (
     REDIS_BGTASK_DB,
     REDIS_CONTAINER_LOG,
@@ -58,6 +59,7 @@
 )
 from ai.backend.common.etcd import AsyncEtcd, ConfigScopes
 from ai.backend.common.events.dispatcher import EventProducer
+from ai.backend.common.identifier.domain import DomainID
 from ai.backend.common.identifier.resource_group import ResourceGroupID, ResourceGroupName
 from ai.backend.common.identifier.user import UserID
 from ai.backend.common.message_queue.redis_queue.queue import RedisMQArgs, RedisQueue
@@ -1578,3 +1580,26 @@ async def user_registry(
         yield registry
     finally:
         await registry.close()
+
+
+@pytest.fixture()
+def acting_superadmin() -> Iterator[None]:
+    """Run the test body as a superadmin.
+
+    A test that drives processors directly skips the HTTP layer, so nothing has set
+    the caller — and the gates the action layer imposes read it. Production always has
+    one; this supplies the equivalent so the test exercises the gate rather than
+    tripping over its absence.
+    """
+    with with_user(
+        UserData(
+            user_id=uuid.uuid4(),
+            is_authorized=True,
+            is_admin=True,
+            is_superadmin=True,
+            role=UserRole.SUPERADMIN,
+            domain_name="default",
+            domain_id=DomainID(uuid.uuid4()),
+        )
+    ):
+        yield

tests/component/error_log/test_error_log.py

@@ -3,6 +3,9 @@
 import uuid
 from typing import Any
 
+import pytest
+
+from ai.backend.client.v2.exceptions import PermissionDeniedError
 from ai.backend.client.v2.registry import BackendAIClientRegistry
 from ai.backend.common.dto.manager.error_log import (
     AppendErrorLogRequest,
@@ -70,19 +73,18 @@ async def test_admin_appends_error_log_minimal(
         assert isinstance(result, AppendErrorLogResponse)
         assert result.success is True
 
-    async def test_user_appends_error_log(
+    async def test_user_cannot_append_error_log(
         self,
         user_registry: BackendAIClientRegistry,
     ) -> None:
-        result = await user_registry.error_log.append(
-            _make_append_request(
-                severity="critical",
-                source="user-test",
-                message="error from regular user",
-            ),
-        )
-        assert isinstance(result, AppendErrorLogResponse)
-        assert result.success is True
+        with pytest.raises(PermissionDeniedError):
+            await user_registry.error_log.append(
+                _make_append_request(
+                    severity="critical",
+                    source="user-test",
+                    message="error from regular user",
+                ),
+            )
 
 
 class TestListErrorLogs:
@@ -103,25 +105,16 @@ async def test_admin_lists_error_logs(
         assert log_entry.severity
         assert log_entry.is_cleared is not None  # admin sees is_cleared
 
-    async def test_user_lists_own_error_logs(
+    async def test_user_cannot_list_error_logs(
         self,
         admin_registry: BackendAIClientRegistry,
         user_registry: BackendAIClientRegistry,
     ) -> None:
-        # Admin creates a log
         await admin_registry.error_log.append(
             _make_append_request(message="admin log"),
         )
-        # User creates a log
-        await user_registry.error_log.append(
-            _make_append_request(message="user log"),
-        )
-        result = await user_registry.error_log.list_logs()
-        assert isinstance(result, ListErrorLogsResponse)
-        # User should only see their own logs
-        assert result.count >= 1
-        for log_entry in result.logs:
-            assert log_entry.is_cleared is None  # non-admin does not see is_cleared
+        with pytest.raises(PermissionDeniedError):
+            await user_registry.error_log.list_logs()
 
     async def test_list_logs_with_query_params(
         self,
@@ -175,19 +168,16 @@ async def test_admin_marks_log_cleared(
         assert cleared_log is not None
         assert cleared_log.is_cleared is True
 
-    async def test_user_marks_own_log_cleared(
+    async def test_user_cannot_mark_log_cleared(
         self,
+        admin_registry: BackendAIClientRegistry,
         user_registry: BackendAIClientRegistry,
     ) -> None:
-        # User creates a log entry
-        await user_registry.error_log.append(
+        await admin_registry.error_log.append(
             _make_append_request(message="user-clearable log"),
         )
-        # List to get the log_id (user only sees own, non-cleared logs)
-        list_result = await user_registry.error_log.list_logs()
-        assert list_result.count >= 1
+        list_result = await admin_registry.error_log.list_logs()
         log_id = uuid.UUID(list_result.logs[0].log_id)
 
-        result = await user_registry.error_log.mark_cleared(log_id)
-        assert isinstance(result, MarkClearedResponse)
-        assert result.success is True
+        with pytest.raises(PermissionDeniedError):
+            await user_registry.error_log.mark_cleared(log_id)