feat(AGX1-244): enforce parent-agent authz on event routes

agentex/src/api/routes/events.py

@@ -3,11 +3,12 @@
 from src.api.schemas.authorization_types import (
     AgentexResourceType,
     AuthorizedOperationType,
-    TaskChildResourceType,
 )
 from src.api.schemas.events import Event
+from src.domain.services.authorization_service import DAuthorizationService
 from src.domain.use_cases.events_use_case import DEventUseCase
-from src.utils.authorization_shortcuts import DAuthorizedId, DAuthorizedQuery
+from src.utils.agent_authorization import check_agent_or_collapse_to_404
+from src.utils.authorization_shortcuts import DAuthorizedQuery
 from src.utils.logging import make_logger
 
 logger = make_logger(__name__)
@@ -20,10 +21,15 @@
     response_model=Event,
 )
 async def get_event(
-    event_id: DAuthorizedId(TaskChildResourceType.event, AuthorizedOperationType.read),
+    event_id: str,
     event_use_case: DEventUseCase,
+    authorization: DAuthorizationService,
 ) -> Event:
+    # Events delegate authz to the parent agent.
     event_entity = await event_use_case.get(event_id)
+    await check_agent_or_collapse_to_404(
+        authorization, event_entity.agent_id, AuthorizedOperationType.read
+    )
     return Event.model_validate(event_entity)
 
 

agentex/src/api/schemas/authorization_types.py

@@ -20,7 +20,6 @@ class AgentexResourceType(StrEnum):
 class TaskChildResourceType(StrEnum):
     """Resources that inherit permissions from their parent task."""
 
-    event = "event"
     state = "state"
 
 

agentex/src/utils/agent_authorization.py

@@ -0,0 +1,26 @@
+from src.adapters.authorization.exceptions import AuthorizationError
+from src.adapters.crud_store.exceptions import ItemDoesNotExist
+from src.api.schemas.authorization_types import (
+    AgentexResource,
+    AuthorizedOperationType,
+)
+
+
+async def check_agent_or_collapse_to_404(
+    authorization,
+    agent_id: str,
+    operation: AuthorizedOperationType,
+) -> None:
+    """Check an agent resource; collapse any denial to 404 to avoid leaking
+    cross-tenant existence. Mirrors ``check_task_or_collapse_to_404`` in
+    ``task_authorization.py`` — see that docstring for the full rationale.
+
+    TODO(AGX1-290): restore 403/404 split once agents carry tenant scope.
+    """
+    try:
+        await authorization.check(
+            resource=AgentexResource.agent(agent_id),
+            operation=operation,
+        )
+    except AuthorizationError:
+        raise ItemDoesNotExist(f"Item with id '{agent_id}' does not exist.") from None

agentex/src/utils/authorization_shortcuts.py

@@ -2,14 +2,15 @@
 
 from fastapi import Depends, Path, Query, Request
 
+from src.adapters.authorization.exceptions import AuthorizationError
+from src.adapters.crud_store.exceptions import ItemDoesNotExist
 from src.api.schemas.authorization_types import (
     AgentexResource,
     AgentexResourceType,
     AuthorizedOperationType,
     TaskChildResourceType,
 )
 from src.domain.repositories.agent_repository import DAgentRepository
-from src.domain.repositories.event_repository import DEventRepository
 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
@@ -18,13 +19,11 @@
 async def _get_parent_task_id(
     resource_type: TaskChildResourceType,
     resource_id: str,
-    event_repository: DEventRepository,
     state_repository: DTaskStateRepository,
 ) -> str:
-    """Get the parent task ID for a child resource."""
+    """Get the parent task ID for a task-child resource."""
     registry = {
         TaskChildResourceType.state: state_repository,
-        TaskChildResourceType.event: event_repository,
     }
 
     repository = registry[resource_type]
@@ -42,19 +41,25 @@ def DAuthorizedId(
 
     async def _ensure_authorized_id(
         authorization: DAuthorizationService,
-        event_repository: DEventRepository,
         state_repository: DTaskStateRepository,
         resource_id: str = Path(..., alias=param_name),
     ) -> str:
-        # For child resources, check the parent task
+        # For child resources, check the parent task. Collapse a denied check
+        # into 404 so callers cannot use 403 vs 404 to probe whether a resource
+        # exists in another tenant.
         if isinstance(resource_type, TaskChildResourceType):
             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,
+                resource_type, resource_id, state_repository
             )
+            try:
+                await authorization.check(
+                    resource=AgentexResource.task(task_id),
+                    operation=operation,
+                )
+            except AuthorizationError:
+                raise ItemDoesNotExist(
+                    f"Item with id '{resource_id}' does not exist."
+                ) from None
         else:
             # For direct resources, check directly
             await authorization.check(
@@ -79,19 +84,25 @@ def DAuthorizedQuery(
 
     async def _ensure_authorized_query(
         authorization: DAuthorizationService,
-        event_repository: DEventRepository,
         state_repository: DTaskStateRepository,
         resource_id: str = Query(..., alias=param_name, description=description),
     ) -> str:
-        # For child resources, check the parent task
+        # For child resources, check the parent task. Collapse a denied check
+        # into 404 so callers cannot use 403 vs 404 to probe whether a resource
+        # exists in another tenant.
         if isinstance(resource_type, TaskChildResourceType):
             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,
+                resource_type, resource_id, state_repository
             )
+            try:
+                await authorization.check(
+                    resource=AgentexResource.task(task_id),
+                    operation=operation,
+                )
+            except AuthorizationError:
+                raise ItemDoesNotExist(
+                    f"Item with id '{resource_id}' does not exist."
+                ) from None
         else:
             # For direct resources, check directly
             await authorization.check(