refactor(a2a): name the fail-closed dispatch predicate once

connect() and is_trusted_peer() each spelled out the same
"network-exposed token bind with no allow-list" condition; drift between
the two would let the startup warning and the actual refusal disagree.
Extract A2ASecurityContext.dispatch_fails_closed() and use it in both
places (truth table unchanged). Also hoist the is_loopback_host import to
module top (no cycle) and drop the restating docstring.
This commit is contained in:
kshitijk4poor
2026-10-01 18:28:10 +05:30
committed by kshitij
parent 66ed823b5c
commit 4fb768fcda
2 changed files with 9 additions and 8 deletions

View File

@@ -331,7 +331,7 @@ class A2AAdapter(BasePlatformAdapter):
logger.info("A2A: serving Agent Card + JSON-RPC on http://%s:%s (%s) as %r; %d routed agent(s)", self.host, self.port,
"localhost-only" if self._security_context.localhost_only() else "REMOTE (bearer auth)", self.agent_name, len(self._agents))
sec = self._security_context
if not (sec.allow_all_users or sec.localhost_only() or sec.trusted_peers or sec.is_loopback_bind()):
if sec.dispatch_fails_closed():
logger.error("A2A: exposed on non-loopback bind %s with no A2A_TRUSTED_PEERS; every dispatch will be refused (403). "
"Set A2A_TRUSTED_PEERS, or A2A_ALLOW_ALL_USERS=true for a trusted network.", self.host)
self._wire_plugin_handlers(None) # plugin-registered native handlers

View File

@@ -16,6 +16,7 @@ import time
import urllib.parse
from dataclasses import dataclass
from typing import Optional
from agent.proxy_bypass import is_loopback_host
from gateway.platforms._shared import profile_scoped as _profile_scoped
logger = logging.getLogger(__name__)
@@ -100,18 +101,18 @@ class A2ASecurityContext:
return None
def is_loopback_bind(self) -> bool:
"""True when the resolved bind host is loopback."""
from agent.proxy_bypass import is_loopback_host
return is_loopback_host(self.resolve_bind_host())
def dispatch_fails_closed(self) -> bool:
"""Network-exposed token bind with no allow-list and no allow-all: refuse every peer."""
return not (self.allow_all_users or self.localhost_only() or self.trusted_peers or self.is_loopback_bind())
def is_trusted_peer(self, identity: str) -> bool:
"""Fail closed on network-exposed binds with no allow-list; loopback
binds without an allow-list stay open for backward compatibility."""
if self.allow_all_users or self.localhost_only():
return True
if not self.trusted_peers:
return self.is_loopback_bind() # the misconfiguration is logged once in A2AAdapter.connect()
return identity in self.trusted_peers
if self.dispatch_fails_closed():
return False # the misconfiguration is logged once in A2AAdapter.connect()
return self.allow_all_users or self.localhost_only() or not self.trusted_peers or identity in self.trusted_peers
def sign_push_payload(self, payload: dict) -> str:
"""HMAC-SHA256 hex over the sorted-key JSON body; "" when no secret."""