fix(agents): restore token narrowing and scope the private-access suppressions

The managed-model check lost its valid_token narrowing when it moved to the
shared helper. Make the caller-access resolver public rather than reaching
into it from module scope, and give each remaining private access a reason.
This commit is contained in:
Joshua Valluru 2026-09-29 16:19:27 -07:00
parent 2d2782e4a6
commit d462634aea
3 changed files with 7 additions and 7 deletions

View file

@ -93,7 +93,7 @@ class AgentRequestHandler:
if managed_agent_policy(user_api_key_auth) is not None:
return await _managed_actor_agent_access(user_api_key_auth)
key_team_access: Final = await AgentRequestHandler.resolve_key_team_agent_access(user_api_key_auth)
caller_access: Final = await AgentRequestHandler._agent_caller_access(user_api_key_auth)
caller_access: Final = await AgentRequestHandler.agent_caller_access(user_api_key_auth)
own_access: Final = _intersect_agent_access(key_team_access, caller_access)
agent_ceiling: Final = await AgentRequestHandler._agent_access_group_ceiling(user_api_key_auth, resolve_ceiling)
if agent_ceiling is None:
@ -103,7 +103,7 @@ class AgentRequestHandler:
return RestrictedAgentAccess(own_access.agent_ids & agent_ceiling)
@staticmethod
async def _agent_caller_access(user_api_key_auth: UserAPIKeyAuth | None) -> AgentAccess:
async def agent_caller_access(user_api_key_auth: UserAPIKeyAuth | None) -> AgentAccess:
caller_auth: Final = agent_caller_auth(user_api_key_auth) if user_api_key_auth else None
if caller_auth is None:
return UnrestrictedAgentAccess()
@ -182,7 +182,7 @@ class AgentRequestHandler:
key_hash: Final = user_api_key_auth.api_key or user_api_key_auth.token
authority: Final = (
await MCPRequestHandler._reload_admitted_key(key_hash, check_db_only=True)
await MCPRequestHandler._reload_admitted_key(key_hash, check_db_only=True) # pyright: ignore[reportPrivateUsage] # the authoritative key reload has no public seam
if key_hash
and managed_agent_policy(user_api_key_auth) is None
and not user_api_key_auth.is_session_token
@ -651,7 +651,7 @@ async def _managed_actor_agent_access(auth: UserAPIKeyAuth) -> AgentAccess:
ceilings: Final = await resolve_managed_agent_ceilings(agent)
grouped: Final = frozenset(target for target in own if all(target in ceiling.agent_ids for ceiling in ceilings))
caller: Final = await AgentRequestHandler._agent_caller_access(auth)
caller: Final = await AgentRequestHandler.agent_caller_access(auth)
capped: Final = grouped if isinstance(caller, UnrestrictedAgentAccess) else grouped & caller.agent_ids
context: Final = auth.managed_agent_context
if context is None or context.mode == "autonomous":

View file

@ -18,8 +18,8 @@ def managed_agent_policy(auth: "UserAPIKeyAuth | None") -> AgentResponse | None:
async def admit_managed_actor(auth: UserAPIKeyAuth, store: AgentIdentityStore | None) -> None:
delegation_verified: Final = auth._managed_delegation_verified
auth._managed_delegation_verified = False
delegation_verified: Final = auth._managed_delegation_verified # pyright: ignore[reportPrivateUsage] # the one-shot delegation marker is a PrivateAttr by design
auth._managed_delegation_verified = False # pyright: ignore[reportPrivateUsage] # consumed here so a replayed token cannot reuse it
if auth.agent_id is None:
return
if store is None:

View file

@ -1060,7 +1060,7 @@ async def common_checks(
)
managed_policy: Final = managed_agent_policy(valid_token)
if _model and managed_policy is not None:
if _model and valid_token is not None and managed_policy is not None:
managed_models: Final = (managed_policy.object_permission or MappingProxyType({})).get("models", ())
if not isinstance(managed_models, (list, tuple)) or not managed_models:
raise HTTPException(403, "This agent has no model grants")