feat(AGX1-263): agent_api_keys route migration to FGAC (404 collapse + two-factor)

agentex/src/api/routes/agent_api_keys.py

@@ -2,14 +2,29 @@
 
 from fastapi import APIRouter, HTTPException, Query
 
+from src.adapters.crud_store.exceptions import ItemDoesNotExist
 from src.api.schemas.agent_api_keys import (
     AgentAPIKey,
     CreateAPIKeyRequest,
     CreateAPIKeyResponse,
 )
+from src.api.schemas.authorization_types import (
+    AgentexResource,
+    AgentexResourceType,
+    AuthorizedOperationType,
+)
 from src.domain.entities.agent_api_keys import AgentAPIKeyType
+from src.domain.services.authorization_service import DAuthorizationService
 from src.domain.use_cases.agent_api_keys_use_case import DAgentAPIKeysUseCase
 from src.domain.use_cases.agents_use_case import DAgentsUseCase
+from src.utils.agent_api_key_authorization import (
+    API_KEY_NOT_FOUND_MESSAGE,
+    _check_api_key_or_collapse_to_404,
+)
+from src.utils.authorization_shortcuts import (
+    DAuthorizedId,
+    DAuthorizedResourceIds,
+)
 from src.utils.logging import make_logger
 
 logger = make_logger(__name__)
@@ -28,6 +43,7 @@ async def create_api_key(
     request: CreateAPIKeyRequest,
     agent_api_key_use_case: DAgentAPIKeysUseCase,
     agent_use_case: DAgentsUseCase,
+    authorization_service: DAuthorizationService,
 ) -> CreateAPIKeyResponse:
     if not request.agent_id and not request.agent_name:
         raise HTTPException(
@@ -40,6 +56,14 @@ async def create_api_key(
             detail="Only one of 'agent_id' or 'agent_name' should be provided to create an agent api_key.",
         )
     agent = await agent_use_case.get(id=request.agent_id, name=request.agent_name)
+
+    # No api_key resource exists yet, so SpiceDB can't gate transitively —
+    # gate on parent ``agent.update`` directly.
+    await authorization_service.check(
+        resource=AgentexResource.agent(agent.id),
+        operation=AuthorizedOperationType.update,
+    )
+
     # Check if external agent API key already exists for this name and agent ID
     existing_api_key = await agent_api_key_use_case.get_by_agent_id_and_name(
         agent_id=agent.id,
@@ -77,6 +101,9 @@ async def create_api_key(
 async def list_agent_api_keys(
     agent_api_key_use_case: DAgentAPIKeysUseCase,
     agent_use_case: DAgentsUseCase,
+    authorized_api_key_ids: DAuthorizedResourceIds(
+        AgentexResourceType.api_key, AuthorizedOperationType.read
+    ),
     agent_id: str | None = None,
     agent_name: str | None = None,
     limit: int = Query(default=50, ge=1, le=1000),
@@ -93,8 +120,13 @@ async def list_agent_api_keys(
             detail="Only one of 'agent_id' or 'agent_name' should be provided to list agent api_keys.",
         )
     agent = await agent_use_case.get(id=agent_id, name=agent_name)
+    # ``id`` filter runs at the SQL layer so limit/offset apply post-filter.
+    # ``None`` = authz declined to enumerate (e.g. bypass); pass through.
     agent_api_key_entities = await agent_api_key_use_case.list(
-        agent_id=agent.id, limit=limit, page_number=page_number
+        agent_id=agent.id,
+        limit=limit,
+        page_number=page_number,
+        id=authorized_api_key_ids,
     )
     return [
         AgentAPIKey.model_validate(agent_api_key_entity)
@@ -112,6 +144,7 @@ async def get_agent_api_key_by_name(
     name: str,
     agent_api_key_use_case: DAgentAPIKeysUseCase,
     agent_use_case: DAgentsUseCase,
+    authorization_service: DAuthorizationService,
     agent_id: str | None = None,
     agent_name: str | None = None,
     api_key_type: AgentAPIKeyType = AgentAPIKeyType.EXTERNAL,
@@ -131,10 +164,16 @@ async def get_agent_api_key_by_name(
         agent_id=agent.id, name=name, api_key_type=api_key_type
     )
     if not agent_api_key_entity:
-        raise HTTPException(
-            status_code=404,
-            detail=f"Agent api_key '{name}' not found for agent ID {agent.id}",
-        )
+        # Absent and denied 404s must be byte-for-byte identical — see
+        # ``API_KEY_NOT_FOUND_MESSAGE``.
+        raise ItemDoesNotExist(API_KEY_NOT_FOUND_MESSAGE)
+    # Composite lookup key ``(agent_id, name, api_key_type)`` doesn't fit
+    # ``DAuthorizedName`` — apply the collapse inline.
+    await _check_api_key_or_collapse_to_404(
+        authorization_service,
+        agent_api_key_entity.id,
+        AuthorizedOperationType.read,
+    )
     return AgentAPIKey.model_validate(agent_api_key_entity)
 
 
@@ -147,6 +186,11 @@ async def get_agent_api_key_by_name(
 async def get_agent_api_key(
     id: str,
     agent_api_key_use_case: DAgentAPIKeysUseCase,
+    _authorized_id: DAuthorizedId(
+        AgentexResourceType.api_key,
+        AuthorizedOperationType.read,
+        param_name="id",
+    ),
 ) -> AgentAPIKey:
     agent_api_key_entity = await agent_api_key_use_case.get(id=id)
     return AgentAPIKey.model_validate(agent_api_key_entity)
@@ -161,7 +205,14 @@ async def get_agent_api_key(
 async def delete_agent_api_key(
     id: str,
     agent_api_key_use_case: DAgentAPIKeysUseCase,
+    _authorized_id: DAuthorizedId(
+        AgentexResourceType.api_key,
+        AuthorizedOperationType.delete,
+        param_name="id",
+    ),
 ) -> str:
+    # SpiceDB's ``api_key.delete`` expands transitively to
+    # ``parent_agent->update``, so the dep above enforces both factors.
     await agent_api_key_use_case.delete(id=id)
     return f"Agent API key with ID {id} deleted"
 
@@ -176,6 +227,7 @@ async def delete_agent_api_key_by_name(
     api_key_name: str,
     agent_api_key_use_case: DAgentAPIKeysUseCase,
     agent_use_case: DAgentsUseCase,
+    authorization_service: DAuthorizationService,
     agent_id: str | None = None,
     agent_name: str | None = None,
     api_key_type: AgentAPIKeyType = AgentAPIKeyType.EXTERNAL,
@@ -191,10 +243,21 @@ async def delete_agent_api_key_by_name(
             detail="Only one of 'agent_id' or 'agent_name' should be provided to delete an agent api_key.",
         )
     agent = await agent_use_case.get(id=agent_id, name=agent_name)
-    await agent_api_key_use_case.delete_by_agent_id_and_key_name(
-        agent_id=agent.id,
-        key_name=api_key_name,
-        api_key_type=api_key_type,
+
+    # Resolve name -> id, check, then delete by the resolved id. Deleting by
+    # name would race: if the row were replaced between check and delete, the
+    # check would evaluate the old id but the mutation would land on the new
+    # one.
+    existing = await agent_api_key_use_case.get_by_agent_id_and_name(
+        agent_id=agent.id, name=api_key_name, api_key_type=api_key_type
+    )
+    if not existing:
+        raise ItemDoesNotExist(API_KEY_NOT_FOUND_MESSAGE)
+    await _check_api_key_or_collapse_to_404(
+        authorization_service,
+        existing.id,
+        AuthorizedOperationType.delete,
     )
+    await agent_api_key_use_case.delete(id=existing.id)
 
     return f"Agent api_key '{api_key_name}' deleted"

agentex/src/api/schemas/authorization_types.py

@@ -66,7 +66,5 @@ def api_key(cls, selector: str | None = None) -> "AgentexResourceOptionalSelecto
         return cls(type=AgentexResourceType.api_key, selector=selector)
 
     @classmethod
-    def schedule(
-        cls, selector: str | None = None
-    ) -> "AgentexResourceOptionalSelector":
+    def schedule(cls, selector: str | None = None) -> "AgentexResourceOptionalSelector":
         return cls(type=AgentexResourceType.schedule, selector=selector)

agentex/src/domain/use_cases/agent_api_keys_use_case.py

@@ -246,10 +246,23 @@ async def delete_by_agent_name_and_key_name(
             await self._deregister_api_key_from_auth(api_key_id=existing.id)
 
     async def list(
-        self, agent_id: str, limit: int, page_number: int
+        self,
+        agent_id: str,
+        limit: int,
+        page_number: int,
+        id: list[str] | None = None,
     ) -> list[AgentAPIKeyEntity]:
+        # ``id`` is the FGAC-filter shape used by route deps (DAuthorizedResourceIds).
+        # ``None`` means "no filter" (e.g. authz bypass); an empty list means
+        # "caller can see no api_keys" and must short-circuit to avoid the
+        # base repo translating ``id=[]`` into an unfiltered query.
+        if id is not None and not id:
+            return []
+        filters: dict = {"agent_id": agent_id}
+        if id is not None:
+            filters["id"] = id
         return await self.agent_api_key_repo.list(
-            limit=limit, page_number=page_number, filters={"agent_id": agent_id}
+            limit=limit, page_number=page_number, filters=filters
         )
 
     async def forward_agent_request(

agentex/src/utils/agent_api_key_authorization.py

@@ -0,0 +1,29 @@
+from src.adapters.authorization.exceptions import AuthorizationError
+from src.adapters.crud_store.exceptions import ItemDoesNotExist
+from src.api.schemas.authorization_types import (
+    AgentexResource,
+    AuthorizedOperationType,
+)
+
+# Identifier-free 404 detail, reused by the denied-resource branch below and
+# by the name routes when the row is absent — keeps both 404s indistinguishable.
+API_KEY_NOT_FOUND_MESSAGE = "Agent api_key not found."
+
+
+async def _check_api_key_or_collapse_to_404(
+    authorization,
+    api_key_id: str,
+    operation: AuthorizedOperationType,
+) -> None:
+    """Check an api_key resource; collapse any denial to 404 to avoid leaking
+    cross-tenant existence. Mirrors ``_check_task_or_collapse_to_404``.
+
+    TODO(AGX1-290): restore the 403/404 split once api_keys carry tenant scope.
+    """
+    try:
+        await authorization.check(
+            resource=AgentexResource.api_key(api_key_id),
+            operation=operation,
+        )
+    except AuthorizationError:
+        raise ItemDoesNotExist(API_KEY_NOT_FOUND_MESSAGE) from None

agentex/src/utils/authorization_shortcuts.py

@@ -16,6 +16,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.agent_api_key_authorization import _check_api_key_or_collapse_to_404
 
 
 async def _get_parent_task_id(
@@ -72,6 +73,12 @@ async def _ensure_authorized_id(
                 raise ItemDoesNotExist(
                     f"Item with id '{resource_id}' does not exist."
                 ) from None
+        elif resource_type == AgentexResourceType.api_key:
+            # Collapse api_key denials to 404 so name/id probes can't
+            # distinguish "present in another tenant" from "absent".
+            await _check_api_key_or_collapse_to_404(
+                authorization, resource_id, operation
+            )
         else:
             # For direct resources, check directly
             await authorization.check(
@@ -157,7 +164,9 @@ async def _ensure_authorized_body_field(
                     operation=operation,
                 )
             except AuthorizationError:
-                raise ItemDoesNotExist(f"Item with id '{field_value}' does not exist.") from None
+                raise ItemDoesNotExist(
+                    f"Item with id '{field_value}' does not exist."
+                ) from None
         else:
             await authorization.check(
                 resource=AgentexResource(type=resource_type, selector=field_value),