Skip to content

fix(authz): grant legacy agent register principal #325

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rpatel-scale merged 1 commit into from
Jun 18, 2026
+32 −2
Merged

fix(authz): grant legacy agent register principal #325

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions agentex/openapi.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5221,6 +5221,12 @@ components:
acp_type:
$ref: '#/components/schemas/ACPType'
description: The type of ACP to use for the agent.
principal_context:
anyOf:
- {}
- type: 'null'
title: Principal Context
description: Principal used for authorization
registration_metadata:
anyOf:
- additionalProperties: true
Expand Down
7 changes: 6 additions & 1 deletion agentex/src/api/routes/agents.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,11 +195,15 @@ 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.
"""
enforce_ownership = _has_resolvable_creator(authorization_service.principal_context)
principal_context = authorization_service.principal_context
if not _has_resolvable_creator(principal_context):
principal_context = request.principal_context
enforce_ownership = _has_resolvable_creator(principal_context)
Comment on lines +199 to +201

@greptile-apps greptile-apps Bot Jun 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 security Security / trust-boundary concern: unauthenticated body principal injection

/agents/register is in WHITELISTED_ROUTES, so the auth middleware sets request.state.principal_context = None and skips all credential validation for every request to this path — including requests from external, untrusted callers.

With this PR, any caller to the endpoint can supply:

{"principal_context": {"user_id": "any-user-id", "account_id": "any-account"}}

_has_resolvable_creator will return True for that dict, enforce_ownership becomes True, and the code proceeds to call authorization_service.grant(AgentexResource.agent(...), principal_context=<attacker_dict>). The AgentexAuthorizationProxy.grant then forwards this dict verbatim to agentex-auth /v1/authz/grant with no cryptographic proof that the caller is actually that user.

Whether this results in a successful ownership escalation depends on whether agentex-auth performs independent principal validation server-side. If it accepts the principal dict at face value (as a trusted internal call), any caller who can reach this endpoint can claim ownership of agents under any identity.

Recommendation: If the body-principal trust model depends on network isolation (only internal pods can reach this endpoint), document that assumption explicitly. Consider adding a comment on the principal_context field and here explaining the trust model, and/or adding a check that explicitly limits this fallback to only when AGENTEX_AUTH_URL is configured in a way that signals internal-only trust.

Prompt To Fix With AI 
This is a comment left during a code review.
Path: agentex/src/api/routes/agents.py
Line: 199-201

Comment:
**Security / trust-boundary concern: unauthenticated body principal injection**

`/agents/register` is in `WHITELISTED_ROUTES`, so the auth middleware sets `request.state.principal_context = None` and skips all credential validation for every request to this path — including requests from external, untrusted callers.

With this PR, any caller to the endpoint can supply:
```json
{"principal_context": {"user_id": "any-user-id", "account_id": "any-account"}}
```
`_has_resolvable_creator` will return `True` for that dict, `enforce_ownership` becomes `True`, and the code proceeds to call `authorization_service.grant(AgentexResource.agent(...), principal_context=<attacker_dict>)`. The `AgentexAuthorizationProxy.grant` then forwards this dict verbatim to `agentex-auth /v1/authz/grant` with no cryptographic proof that the caller is actually that user.

Whether this results in a successful ownership escalation depends on whether agentex-auth performs independent principal validation server-side. If it accepts the principal dict at face value (as a trusted internal call), any caller who can reach this endpoint can claim ownership of agents under any identity.

**Recommendation:** If the body-principal trust model depends on network isolation (only internal pods can reach this endpoint), document that assumption explicitly. Consider adding a comment on the `principal_context` field and here explaining the trust model, and/or adding a check that explicitly limits this fallback to only when `AGENTEX_AUTH_URL` is configured in a way that signals internal-only trust.

How can I resolve this? If you propose a fix, please make it concise.

Fix in Cursor Fix in Claude Code Fix in Codex

if enforce_ownership:
await authorization_service.check(
AgentexResource.agent("*"),
AuthorizedOperationType.create,
principal_context=principal_context,
)
logger.info(
"Registering agent name=%s agent_id=%s acp_type=%s",
Expand All @@ -220,6 +224,7 @@ async def register_agent(
if enforce_ownership:
await authorization_service.grant(
AgentexResource.agent(agent_entity.id),
principal_context=principal_context,
)
response_fields = agent_entity.model_dump()
existing_key = await api_keys_use_case.get_internal_api_key_by_agent_id(
Expand Down
3 changes: 3 additions & 0 deletions agentex/src/api/schemas/agents.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -88,6 +88,9 @@ class RegisterAgentRequest(BaseModel):
description="Optional agent ID if the agent already exists and needs to be updated.",
)
acp_type: ACPType = Field(..., description="The type of ACP to use for the agent.")
principal_context: Any | None = Field(
default=None, description="Principal used for authorization"
)
registration_metadata: dict[str, Any] | None = Field(
default=None,
description="The metadata for the agent's registration.",
Expand Down
18 changes: 17 additions & 1 deletion agentex/tests/unit/api/test_agents_authz.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -392,12 +392,13 @@ def _mocks(principal_context):
return authorization, agents_use_case, api_keys_use_case

@staticmethod
def _request():
def _request(principal_context=None):
return RegisterAgentRequest(
name="my-agent",
description="d",
acp_url="http://agent:5000",
acp_type=ACPType.ASYNC,
principal_context=principal_context,
)

@pytest.mark.parametrize("principal_context", [None, {}, {"account_id": "acct"}])
Expand All @@ -413,6 +414,21 @@ async def test_unresolvable_creator_skips_check_and_grant(self, principal_contex
use_case.register_agent.assert_awaited_once()
assert resp.agent_api_key == "internal-key"
Comment on lines 404 to 415

@greptile-apps greptile-apps Bot Jun 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Missing test: unresolvable non-None auth-service principal + resolvable body principal

The parametrize covers [None, {}, {"account_id":"acct"}] for the auth-service principal, but always calls self._request() with no body principal_context. This means the case:

  • auth service has {"account_id": "acct"} (unresolvable) AND body has {"user_id": "u"} (resolvable)

is never tested. With the new fallback logic, the body principal would be used here (enforce_ownership=True), which is the intended behavior — but there is zero coverage for it. A regression that skips enforcement in this case would pass all current tests.

Consider adding a parametrized variant like:

@pytest.mark.parametrize(
    "auth_principal,body_principal",
    [
        ({}, {"user_id": "u", "account_id": "acct"}),
        ({"account_id": "acct"}, {"user_id": "u", "account_id": "acct"}),
    ],
)
async def test_unresolvable_auth_service_with_resolvable_body_enforces(self, auth_principal, body_principal):
    authz, use_case, api_keys = self._mocks(principal_context=auth_principal)
    await register_agent(self._request(principal_context=body_principal), use_case, authz, api_keys)
    authz.check.assert_awaited_once()
    assert authz.check.await_args.kwargs["principal_context"] == body_principal
    authz.grant.assert_awaited_once()
    assert authz.grant.await_args.kwargs["principal_context"] == body_principal
Prompt To Fix With AI 
This is a comment left during a code review.
Path: agentex/tests/unit/api/test_agents_authz.py
Line: 404-415

Comment:
**Missing test: unresolvable non-None auth-service principal + resolvable body principal**

The parametrize covers `[None, {}, {"account_id":"acct"}]` for the auth-service principal, but always calls `self._request()` with no body `principal_context`. This means the case:

- auth service has `{"account_id": "acct"}` (unresolvable) **AND** body has `{"user_id": "u"}` (resolvable)

is never tested. With the new fallback logic, the body principal would be used here (`enforce_ownership=True`), which is the intended behavior — but there is zero coverage for it. A regression that skips enforcement in this case would pass all current tests.

Consider adding a parametrized variant like:
```python
@pytest.mark.parametrize(
    "auth_principal,body_principal",
    [
        ({}, {"user_id": "u", "account_id": "acct"}),
        ({"account_id": "acct"}, {"user_id": "u", "account_id": "acct"}),
    ],
)
async def test_unresolvable_auth_service_with_resolvable_body_enforces(self, auth_principal, body_principal):
    authz, use_case, api_keys = self._mocks(principal_context=auth_principal)
    await register_agent(self._request(principal_context=body_principal), use_case, authz, api_keys)
    authz.check.assert_awaited_once()
    assert authz.check.await_args.kwargs["principal_context"] == body_principal
    authz.grant.assert_awaited_once()
    assert authz.grant.await_args.kwargs["principal_context"] == body_principal
```

How can I resolve this? If you propose a fix, please make it concise.

Fix in Cursor Fix in Claude Code Fix in Codex


async def test_body_principal_enforces_check_and_grant(self):
# Legacy deployed pods send the deploy principal in the body because
# /agents/register is whitelisted and has no request-state principal.
body_principal = {"user_id": "u", "account_id": "acct"}
authz, use_case, api_keys = self._mocks(principal_context=None)

await register_agent(
self._request(principal_context=body_principal), use_case, authz, api_keys
)

authz.check.assert_awaited_once()
assert authz.check.await_args.kwargs["principal_context"] == body_principal
authz.grant.assert_awaited_once()
assert authz.grant.await_args.kwargs["principal_context"] == body_principal

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"}
Expand Down
Loading