From 4fb768fcdae5e31ad8fcbf77e62c804fab9c3556 Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Thu, 1 Oct 2026 18:28:10 +0530 Subject: [PATCH] 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. --- plugins/platforms/a2a/adapter.py | 2 +- plugins/platforms/a2a/security.py | 15 ++++++++------- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/plugins/platforms/a2a/adapter.py b/plugins/platforms/a2a/adapter.py index bd3dbda516..a69b68b3c6 100644 --- a/plugins/platforms/a2a/adapter.py +++ b/plugins/platforms/a2a/adapter.py @@ -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 diff --git a/plugins/platforms/a2a/security.py b/plugins/platforms/a2a/security.py index b6f0d9d0a2..e6215021d4 100644 --- a/plugins/platforms/a2a/security.py +++ b/plugins/platforms/a2a/security.py @@ -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."""