From 3596558b7668be694de9ad146e6a377d55eae0a4 Mon Sep 17 00:00:00 2001 From: Harvey Han Date: Tue, 9 Jun 2026 09:50:20 -0700 Subject: [PATCH 1/2] fix(authz): skip create-check/grant on unauthenticated agent self-register /agents/register is whitelisted, so deployed pods self-register on startup with no principal context. #270 added an enforced check(create)/grant on that path, which forwarded a None principal to agentex-auth -> _validate_principal raised 422 -> agent pods stuck in CrashLoopBackOff on startup. The agent already exists and is owned from build time (register-build runs before deploy-time register), so the create check and ownership grant don't apply on the self-registration path. Skip them when no creator is resolvable; an authenticated caller is still enforced. --- agentex/src/api/routes/agents.py | 39 +++++++++-- agentex/tests/unit/api/test_agents_authz.py | 71 +++++++++++++++++++++ 2 files changed, 104 insertions(+), 6 deletions(-) diff --git a/agentex/src/api/routes/agents.py b/agentex/src/api/routes/agents.py index 89578542..63f0989b 100644 --- a/agentex/src/api/routes/agents.py +++ b/agentex/src/api/routes/agents.py @@ -50,6 +50,28 @@ router = APIRouter(prefix="/agents", tags=["Agents"]) +def _has_resolvable_creator(principal_context) -> bool: + """Whether a creator identity (user or service account) is present. + + ``/agents/register`` is whitelisted, so deployed pods self-register on + startup with no principal context (``None``). The agent already exists and + is owned from build time (``register-build`` runs before deploy-time + register), so the create authorization check and ownership grant don't + apply on that path -- and forwarding a ``None`` principal to agentex-auth + raises 422, crash-looping the pod. Skip the check/grant when no creator is + resolvable; enforce normally for an authenticated caller. + """ + if isinstance(principal_context, dict): + return bool( + principal_context.get("user_id") + or principal_context.get("service_account_id") + ) + return bool( + getattr(principal_context, "user_id", None) + or getattr(principal_context, "service_account_id", None) + ) + + @router.get( "/{agent_id}", summary="Get Agent by ID", @@ -173,10 +195,14 @@ async def register_agent( If agent_id is not provided, the system will look for an existing agent by name and update it, or create a new one if it doesn't exist. """ - await authorization_service.check( - AgentexResource.agent("*"), - AuthorizedOperationType.create, + enforce_ownership = _has_resolvable_creator( + authorization_service.principal_context ) + if enforce_ownership: + await authorization_service.check( + AgentexResource.agent("*"), + AuthorizedOperationType.create, + ) logger.info(f"Registering agent: {request}") try: agent_entity = await agents_use_case.register_agent( @@ -188,9 +214,10 @@ async def register_agent( registration_metadata=request.registration_metadata, agent_input_type=request.agent_input_type, ) - await authorization_service.grant( - AgentexResource.agent(agent_entity.id), - ) + if enforce_ownership: + await authorization_service.grant( + AgentexResource.agent(agent_entity.id), + ) response_fields = agent_entity.model_dump() existing_key = await api_keys_use_case.get_internal_api_key_by_agent_id( agent_id=agent_entity.id diff --git a/agentex/tests/unit/api/test_agents_authz.py b/agentex/tests/unit/api/test_agents_authz.py index cd6cd4e0..ed5f2ee5 100644 --- a/agentex/tests/unit/api/test_agents_authz.py +++ b/agentex/tests/unit/api/test_agents_authz.py @@ -7,6 +7,7 @@ from __future__ import annotations +from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock import pytest @@ -17,6 +18,9 @@ AgentexResourceType, AuthorizedOperationType, ) +from src.api.routes.agents import register_agent +from src.api.schemas.agents import RegisterAgentRequest +from src.domain.entities.agents import ACPType, AgentEntity, AgentStatus from src.utils.authorization_shortcuts import ( DAuthorizedId, DAuthorizedName, @@ -261,3 +265,70 @@ async def test_none_authorized_ids_passes_through_unfiltered(self): order_by=None, order_direction="desc", ) + + +@pytest.mark.unit +@pytest.mark.asyncio +class TestRegisterAgentOwnershipEnforcement: + """``/agents/register`` is whitelisted, so deployed pods self-register with + no principal. The create check / ownership grant must be skipped on that + path (the agent is already owned from build time), so a ``None`` principal + does not 422 and crash-loop the pod. An authenticated caller is enforced.""" + + @staticmethod + def _mocks(principal_context): + authorization = MagicMock() + authorization.principal_context = principal_context + authorization.check = AsyncMock(return_value=True) + authorization.grant = AsyncMock() + + now = datetime.now(timezone.utc) + agent = AgentEntity( + id="agent-1", + name="my-agent", + description="d", + acp_type=ACPType.ASYNC, + status=AgentStatus.READY, + acp_url="http://agent:5000", + created_at=now, + updated_at=now, + ) + agents_use_case = MagicMock() + agents_use_case.register_agent = AsyncMock(return_value=agent) + + api_keys_use_case = MagicMock() + existing_key = MagicMock() + existing_key.api_key = "internal-key" + api_keys_use_case.get_internal_api_key_by_agent_id = AsyncMock( + return_value=existing_key + ) + return authorization, agents_use_case, api_keys_use_case + + @staticmethod + def _request(): + return RegisterAgentRequest( + name="my-agent", + description="d", + acp_url="http://agent:5000", + acp_type=ACPType.ASYNC, + ) + + async def test_no_principal_skips_check_and_grant(self): + authz, use_case, api_keys = self._mocks(principal_context=None) + + resp = await register_agent(self._request(), use_case, authz, api_keys) + + authz.check.assert_not_awaited() + authz.grant.assert_not_awaited() + use_case.register_agent.assert_awaited_once() + assert resp.agent_api_key == "internal-key" + + async def test_authenticated_caller_enforces_check_and_grant(self): + authz, use_case, api_keys = self._mocks( + principal_context={"user_id": "u", "account_id": "acct"} + ) + + await register_agent(self._request(), use_case, authz, api_keys) + + authz.check.assert_awaited_once() + authz.grant.assert_awaited_once() From 84e758e060c730a5ea4294cb590132f34bb1bf8d Mon Sep 17 00:00:00 2001 From: Harvey Han Date: Tue, 9 Jun 2026 10:00:41 -0700 Subject: [PATCH 2/2] test(authz): align register authz tests with self-reg skip (CI) The legacy-authz integration test mocks agentex-auth, so it asserted the None-principal check/grant the real service 422s on -- masking the deployed-pod crashloop. Align the register/create portion: the whitelisted self-registration path forwards no check/grant (the agent is already owned from build time); the delete portion (authenticated) is unchanged. Unit tests cover the route guard directly: no/empty/creatorless principal skips check+grant (still returns the agent, no 422), while a dict- or object-shaped principal with a user/service id enforces both. --- .../api/agents/test_agents_auth_api.py | 23 +++++--------- agentex/tests/unit/api/test_agents_authz.py | 31 ++++++++++++++----- 2 files changed, 32 insertions(+), 22 deletions(-) diff --git a/agentex/tests/integration/api/agents/test_agents_auth_api.py b/agentex/tests/integration/api/agents/test_agents_auth_api.py index fd8ac821..676ddd93 100644 --- a/agentex/tests/integration/api/agents/test_agents_auth_api.py +++ b/agentex/tests/integration/api/agents/test_agents_auth_api.py @@ -188,7 +188,7 @@ async def test_agent_check( "src.utils.http_request_handler.HttpRequestHandler.post_with_error_handling", side_effect=_mock_post_with_error_handling, ) - async def test_agent_create_and_delete_gate_and_write_legacy_authz( + async def test_agent_register_skips_authz_and_delete_gates_legacy_authz( self, post_with_error_handling_mock, is_enabled_mock, @@ -214,20 +214,13 @@ def _payloads(path: str): assert response.status_code == 200 agent = response.json() - create_checks = _payloads("/v1/authz/check") - assert len(create_checks) == 1 - assert create_checks[0]["resource"]["type"] == AgentexResourceType.agent.value - assert create_checks[0]["resource"]["selector"] == "*" - assert create_checks[0]["operation"] == "create" - - agent_grants = [ - payload - for payload in _payloads("/v1/authz/grant") - if payload["resource"]["type"] == AgentexResourceType.agent.value - ] - assert len(agent_grants) == 1 - assert agent_grants[0]["resource"]["selector"] == agent["id"] - assert agent_grants[0]["operation"] == "create" + # /agents/register is whitelisted: deployed pods self-register on + # startup without a principal, so register does not gate on `create` + # or write an ownership grant -- the agent is already owned from build + # time (register-build runs first). Forwarding a None principal here + # previously 422'd and crash-looped the pod. + assert _payloads("/v1/authz/check") == [] + assert _payloads("/v1/authz/grant") == [] post_with_error_handling_mock.reset_mock() response = await isolated_client.delete("/agents/name/test-agent") diff --git a/agentex/tests/unit/api/test_agents_authz.py b/agentex/tests/unit/api/test_agents_authz.py index ed5f2ee5..a73658fc 100644 --- a/agentex/tests/unit/api/test_agents_authz.py +++ b/agentex/tests/unit/api/test_agents_authz.py @@ -7,19 +7,20 @@ from __future__ import annotations -from datetime import datetime, timezone +from datetime import UTC, datetime +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest from src.adapters.authorization.exceptions import AuthorizationError from src.adapters.crud_store.exceptions import ItemDoesNotExist +from src.api.routes.agents import register_agent +from src.api.schemas.agents import RegisterAgentRequest from src.api.schemas.authorization_types import ( AgentexResource, AgentexResourceType, AuthorizedOperationType, ) -from src.api.routes.agents import register_agent -from src.api.schemas.agents import RegisterAgentRequest from src.domain.entities.agents import ACPType, AgentEntity, AgentStatus from src.utils.authorization_shortcuts import ( DAuthorizedId, @@ -282,7 +283,7 @@ def _mocks(principal_context): authorization.check = AsyncMock(return_value=True) authorization.grant = AsyncMock() - now = datetime.now(timezone.utc) + now = datetime.now(UTC) agent = AgentEntity( id="agent-1", name="my-agent", @@ -313,8 +314,11 @@ def _request(): acp_type=ACPType.ASYNC, ) - async def test_no_principal_skips_check_and_grant(self): - authz, use_case, api_keys = self._mocks(principal_context=None) + @pytest.mark.parametrize("principal_context", [None, {}, {"account_id": "acct"}]) + async def test_unresolvable_creator_skips_check_and_grant(self, principal_context): + # None (whitelisted self-reg), an empty dict, or a principal with no + # user_id/service_account_id all mean "no creator" -> skip, don't 422. + authz, use_case, api_keys = self._mocks(principal_context=principal_context) resp = await register_agent(self._request(), use_case, authz, api_keys) @@ -323,7 +327,7 @@ async def test_no_principal_skips_check_and_grant(self): use_case.register_agent.assert_awaited_once() assert resp.agent_api_key == "internal-key" - async def test_authenticated_caller_enforces_check_and_grant(self): + async def test_dict_principal_enforces_check_and_grant(self): authz, use_case, api_keys = self._mocks( principal_context={"user_id": "u", "account_id": "acct"} ) @@ -332,3 +336,16 @@ async def test_authenticated_caller_enforces_check_and_grant(self): authz.check.assert_awaited_once() authz.grant.assert_awaited_once() + + async def test_object_principal_enforces_check_and_grant(self): + # Object-shaped principal exercises the getattr branch of the helper. + authz, use_case, api_keys = self._mocks( + principal_context=SimpleNamespace( + user_id="u", service_account_id=None, account_id="acct" + ) + ) + + await register_agent(self._request(), use_case, authz, api_keys) + + authz.check.assert_awaited_once() + authz.grant.assert_awaited_once()