feat(AGX1-275): per-RPC task permission rewire and 404/403 wrap

asherfink · asherfink · commit 3009275ca47c · 2026-06-01T14:42:56.000-04:00
Rewires the operation literal sent to agentex-auth on task RPC routes so each RPC checks the permission that actually matches its side effect, instead of using `execute` everywhere: - `MESSAGE_SEND` / `EVENT_SEND` -> `update` - `TASK_CANCEL` -> `cancel` - `TASK_CREATE` stays `create` - Unknown `AgentRPCMethod` values now raise `NotImplementedError` rather than falling through authz-free (defense-in-depth: a new RPC must be explicitly wired before it can dispatch). The same `execute -> update` swap is applied across `messages.py`, `checkpoints.py`, and `states.py` so the editor role can perform routine mutations without needing owner. The task SpiceDB schema defines `permission update = (editor + owner) & internal_tenant_gate`, so leaving these on `execute` (owner-only) would lock editors out of normal flows. Adds `check_task_or_collapse_to_404` in `src/utils/task_authorization.py` and routes every task-resource denial path through it: path id, query id, body id, and the name surface in `authorization_shortcuts.py`. `tasks.name` is globally unique, so a 403/404 split on the name route would let any authenticated caller probe the whole system for task existence — collapsing both denial cases into 404 closes that leak at the cost of an in-tenant UX regression on permission-gap updates (tracked under AGX1-290). The `MESSAGE_SEND` task-name branch is restructured to `try/else`: a denied update on an existing task must surface as 404 and NOT fall through to the create check, which would promote "denied update" into create access. Cross-repo wire dependency: the `update` and `cancel` literals must resolve against the existing OWNER grant in SGP's task permission schema before this deploys. `update` is already in SGP's `AgentexOperation` enum; `cancel` is added by scaleapi/scaleapi sgp-agentex-cancel-enum. Held behind that PR shipping everywhere. Part of the AGX1-264 stack: scaleapi/scaleapi#145000 (FGAC_AGENTEX_AUTH_SPARK flag) -> scaleapi/scaleapi sgp-agentex-cancel-enum (cancel enum on SGP backend) -> scaleapi/agentex#353 (agentex-auth per-account routing + cancel) -> #246 (task FGAC dual-write + audit columns) -> this PR.
diff --git a/agentex/src/api/routes/agents.py b/agentex/src/api/routes/agents.py
@@ -38,6 +38,7 @@
     DAuthorizedResourceIds,
 )
 from src.utils.logging import make_logger
+from src.utils.task_authorization import check_task_or_collapse_to_404
 
 logger = make_logger(__name__)
 
@@ -333,26 +334,31 @@ async def _authorize_rpc_request(
             task_name = request.params.task_name
 
             if task_id is not None:
-                # Direct task ID provided - check execute permission on that specific task
-                await authorization_service.check(
-                    resource=AgentexResource.task(task_id),
-                    operation=AuthorizedOperationType.execute,
+                await check_task_or_collapse_to_404(
+                    authorization_service,
+                    task_id,
+                    AuthorizedOperationType.update,
                 )
             elif task_name is not None:
-                # Task name provided - check if task exists
+                # try/else (not try/except wrapping the whole block): a denied
+                # update on an existing task must surface as 404 to the caller,
+                # NOT silently fall through to the create check below — that
+                # would let "I'm denied update" masquerade as "task is absent"
+                # and grant create access.
                 try:
                     existing_task = await task_service.get_task(name=task_name)
-                    # Task exists - require execute permission on the specific task
-                    await authorization_service.check(
-                        resource=AgentexResource.task(existing_task.id),
-                        operation=AuthorizedOperationType.execute,
-                    )
                 except ItemDoesNotExist:
                     # Task doesn't exist - will be created, require create permission
                     await authorization_service.check(
                         resource=AgentexResource.task("*"),
                         operation=AuthorizedOperationType.create,
                     )
+                else:
+                    await check_task_or_collapse_to_404(
+                        authorization_service,
+                        existing_task.id,
+                        AuthorizedOperationType.update,
+                    )
             else:
                 # No identifier provided - creating new task, require create permission
                 await authorization_service.check(
@@ -365,20 +371,22 @@ async def _authorize_rpc_request(
             task_name = request.params.task_name
 
             if task_id is not None:
-                # Direct task ID provided - check execute permission on that specific task
-                await authorization_service.check(
-                    resource=AgentexResource.task(task_id),
-                    operation=AuthorizedOperationType.execute,
+                await check_task_or_collapse_to_404(
+                    authorization_service,
+                    task_id,
+                    AuthorizedOperationType.update,
                 )
             elif task_name is not None:
-                # Task name provided - look up task and check execute permission
+                # `get_task` raises ItemDoesNotExist for absent rows (= 404
+                # naturally); the wrap collapses present-but-denied to the same
+                # shape so callers can't distinguish.
                 existing_task = await task_service.get_task(name=task_name)
-                await authorization_service.check(
-                    resource=AgentexResource.task(existing_task.id),
-                    operation=AuthorizedOperationType.execute,
+                await check_task_or_collapse_to_404(
+                    authorization_service,
+                    existing_task.id,
+                    AuthorizedOperationType.update,
                 )
             else:
-                # No identifier provided - this shouldn't happen but handle gracefully
                 raise ValueError(
                     "Either task_id or task_name must be provided for event/send"
                 )
@@ -388,25 +396,27 @@ async def _authorize_rpc_request(
             task_name = request.params.task_name
 
             if task_id is not None:
-                # Direct task ID provided - check execute permission on that specific task
-                await authorization_service.check(
-                    resource=AgentexResource.task(task_id),
-                    operation=AuthorizedOperationType.execute,
+                await check_task_or_collapse_to_404(
+                    authorization_service,
+                    task_id,
+                    AuthorizedOperationType.cancel,
                 )
             elif task_name is not None:
-                # Task name provided - look up task and check execute permission
                 existing_task = await task_service.get_task(name=task_name)
-                await authorization_service.check(
-                    resource=AgentexResource.task(existing_task.id),
-                    operation=AuthorizedOperationType.execute,
+                await check_task_or_collapse_to_404(
+                    authorization_service,
+                    existing_task.id,
+                    AuthorizedOperationType.cancel,
                 )
             else:
-                # No identifier provided - this shouldn't happen but handle gracefully
                 raise ValueError(
                     "Either task_id or task_name must be provided for task/cancel"
                 )
         case _:
-            pass
+            raise NotImplementedError(
+                f"_authorize_rpc_request has no case for {request.method}; "
+                "RPC methods must be wired explicitly before they can dispatch."
+            )
 
 
 async def _handle_agent_rpc(
diff --git a/agentex/src/api/routes/checkpoints.py b/agentex/src/api/routes/checkpoints.py
@@ -2,6 +2,10 @@
 
 from fastapi import APIRouter, Response
 
+from src.api.schemas.authorization_types import (
+    AgentexResourceType,
+    AuthorizedOperationType,
+)
 from src.api.schemas.checkpoints import (
     BlobResponse,
     CheckpointListItem,
@@ -14,10 +18,6 @@
     PutWritesRequest,
     WriteResponse,
 )
-from src.api.schemas.authorization_types import (
-    AgentexResourceType,
-    AuthorizedOperationType,
-)
 from src.domain.use_cases.checkpoints_use_case import DCheckpointsUseCase
 from src.utils.authorization_shortcuts import DAuthorizedBodyId
 from src.utils.logging import make_logger
@@ -95,7 +95,7 @@ async def put_checkpoint(
     request: PutCheckpointRequest,
     checkpoints_use_case: DCheckpointsUseCase,
     _authorized_task_id: DAuthorizedBodyId(
-        AgentexResourceType.task, AuthorizedOperationType.execute, field_name="thread_id"
+        AgentexResourceType.task, AuthorizedOperationType.update, field_name="thread_id"
     ),
 ) -> PutCheckpointResponse:
     blobs = [
@@ -133,7 +133,7 @@ async def put_writes(
     request: PutWritesRequest,
     checkpoints_use_case: DCheckpointsUseCase,
     _authorized_task_id: DAuthorizedBodyId(
-        AgentexResourceType.task, AuthorizedOperationType.execute, field_name="thread_id"
+        AgentexResourceType.task, AuthorizedOperationType.update, field_name="thread_id"
     ),
 ) -> Response:
     writes = [
diff --git a/agentex/src/api/routes/messages.py b/agentex/src/api/routes/messages.py
@@ -82,7 +82,7 @@ async def batch_create_messages(
     request: BatchCreateTaskMessagesRequest,
     message_use_case: DMessageUseCase,
     _authorized_task_id: DAuthorizedBodyId(
-        AgentexResourceType.task, AuthorizedOperationType.execute
+        AgentexResourceType.task, AuthorizedOperationType.update
     ),
 ) -> list[TaskMessage]:
     # Convert each content from API schema to entity schema
@@ -110,7 +110,7 @@ async def batch_update_messages(
     request: BatchUpdateTaskMessagesRequest,
     message_use_case: DMessageUseCase,
     _authorized_task_id: DAuthorizedBodyId(
-        AgentexResourceType.task, AuthorizedOperationType.execute
+        AgentexResourceType.task, AuthorizedOperationType.update
     ),
 ) -> list[TaskMessage]:
     task_message_entities = await message_use_case.update_batch(
@@ -131,7 +131,7 @@ async def create_message(
     request: CreateTaskMessageRequest,
     message_use_case: DMessageUseCase,
     _authorized_task_id: DAuthorizedBodyId(
-        AgentexResourceType.task, AuthorizedOperationType.execute
+        AgentexResourceType.task, AuthorizedOperationType.update
     ),
 ) -> TaskMessage:
     task_message_entity = await message_use_case.create(
@@ -152,7 +152,7 @@ async def update_message(
     message_id: str,
     message_use_case: DMessageUseCase,
     _authorized_task_id: DAuthorizedBodyId(
-        AgentexResourceType.task, AuthorizedOperationType.execute
+        AgentexResourceType.task, AuthorizedOperationType.update
     ),
 ) -> TaskMessage:
     task_message_entity = await message_use_case.update(
diff --git a/agentex/src/api/schemas/authorization_types.py b/agentex/src/api/schemas/authorization_types.py
@@ -9,6 +9,7 @@ class AuthorizedOperationType(StrEnum):
     update = "update"
     delete = "delete"
     execute = "execute"
+    cancel = "cancel"
 
 
 class AgentexResourceType(StrEnum):
diff --git a/agentex/src/utils/authorization_shortcuts.py b/agentex/src/utils/authorization_shortcuts.py
@@ -13,6 +13,7 @@
 from src.domain.repositories.task_repository import DTaskRepository
 from src.domain.repositories.task_state_repository import DTaskStateRepository
 from src.domain.services.authorization_service import DAuthorizationService
+from src.utils.task_authorization import check_task_or_collapse_to_404
 
 
 async def _get_parent_task_id(
@@ -51,12 +52,10 @@ async def _ensure_authorized_id(
             task_id = await _get_parent_task_id(
                 resource_type, resource_id, event_repository, state_repository
             )
-            await authorization.check(
-                resource=AgentexResource.task(task_id),
-                operation=operation,
-            )
+            await check_task_or_collapse_to_404(authorization, task_id, operation)
+        elif resource_type == AgentexResourceType.task:
+            await check_task_or_collapse_to_404(authorization, resource_id, operation)
         else:
-            # For direct resources, check directly
             await authorization.check(
                 resource=AgentexResource(type=resource_type, selector=resource_id),
                 operation=operation,
@@ -88,12 +87,10 @@ async def _ensure_authorized_query(
             task_id = await _get_parent_task_id(
                 resource_type, resource_id, event_repository, state_repository
             )
-            await authorization.check(
-                resource=AgentexResource.task(task_id),
-                operation=operation,
-            )
+            await check_task_or_collapse_to_404(authorization, task_id, operation)
+        elif resource_type == AgentexResourceType.task:
+            await check_task_or_collapse_to_404(authorization, resource_id, operation)
         else:
-            # For direct resources, check directly
             await authorization.check(
                 resource=AgentexResource(type=resource_type, selector=resource_id),
                 operation=operation,
@@ -118,10 +115,13 @@ async def _ensure_authorized_body_field(
         body = await request.json()
         field_value = body[field_name]
 
-        await authorization.check(
-            resource=AgentexResource(type=resource_type, selector=field_value),
-            operation=operation,
-        )
+        if resource_type == AgentexResourceType.task:
+            await check_task_or_collapse_to_404(authorization, field_value, operation)
+        else:
+            await authorization.check(
+                resource=AgentexResource(type=resource_type, selector=field_value),
+                operation=operation,
+            )
         return field_value
 
     return Annotated[str, Depends(_ensure_authorized_body_field)]
@@ -164,12 +164,21 @@ async def _ensure_authorized_name(
         resource_id = resource_name
         repository = registry[resource_type]
 
+        # Lookup-before-authz: if the name isn't present, `repository.get` raises
+        # ItemDoesNotExist (→ 404), which is what we want for absent resources.
+        # The present-but-denied case is handled per-resource below.
         resource = await repository.get(name=resource_id)
 
-        await authorization.check(
-            resource=AgentexResource(type=resource_type, selector=resource.id),
-            operation=operation,
-        )
+        if resource_type == AgentexResourceType.task:
+            # Tasks: collapse denial to 404 so name probes can't distinguish
+            # "present in another tenant" from "absent" (tasks.name is globally
+            # unique — any 403 leak here probes the whole system, not a tenant).
+            await check_task_or_collapse_to_404(authorization, resource.id, operation)
+        else:
+            await authorization.check(
+                resource=AgentexResource(type=resource_type, selector=resource.id),
+                operation=operation,
+            )
         return resource_id
 
     return Annotated[str, Depends(_ensure_authorized_name)]
diff --git a/agentex/src/utils/task_authorization.py b/agentex/src/utils/task_authorization.py
@@ -0,0 +1,40 @@
+from src.adapters.authorization.exceptions import AuthorizationError
+from src.adapters.crud_store.exceptions import ItemDoesNotExist
+from src.api.schemas.authorization_types import (
+    AgentexResource,
+    AuthorizedOperationType,
+)
+from src.domain.services.authorization_service import AuthorizationService
+
+
+async def check_task_or_collapse_to_404(
+    authorization: AuthorizationService,
+    task_id: str,
+    operation: AuthorizedOperationType,
+) -> None:
+    """Issue a check on a task resource. On any denial, surface 404 — even when
+    the task exists.
+
+    The 403/404 split cannot be done safely here: ``TaskORM`` has no tenant
+    column, ``task_repository.get`` does an unfiltered primary-key lookup, and
+    ``AuthorizationError`` carries no deny-reason discriminator. Returning 403
+    when the row exists and 404 when it doesn't leaks cross-tenant existence
+    (caller probes "does task X exist?" and gets distinguishable responses).
+
+    Until tasks carry tenant scope (or agentex-auth's deny distinguishes
+    "cross-tenant" from "in-tenant lacking permission"), the safer default is
+    to collapse both into 404. Trade-off: a user with ``read`` but not
+    ``update`` permission on an in-tenant task sees 404 on update attempts
+    instead of 403. UX regression for in-tenant permission granularity, but
+    eliminates the cross-tenant existence leak.
+
+    TODO(AGX1-290): Restore the 403/404 split for same-tenant calls once
+    tasks carry tenant/account_id scope at the data layer.
+    """
+    try:
+        await authorization.check(
+            resource=AgentexResource.task(task_id),
+            operation=operation,
+        )
+    except AuthorizationError:
+        raise ItemDoesNotExist(f"Item with id '{task_id}' does not exist.") from None
diff --git a/agentex/tests/unit/api/test_tasks_authz.py b/agentex/tests/unit/api/test_tasks_authz.py
