fix: constrain the RPC targets of a network-fetched A2A agent card

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 956644793

Author: GWeale

src/google/adk/a2a/_compat.py

@@ -564,6 +564,33 @@ def agent_card_url(
     return getattr(card, "url", None)
 
 
+def agent_card_rpc_urls(card: AgentCard) -> list[str]:
+  """Returns every URL on a card that a client may send RPC traffic to.
+
+  ``agent_card_url`` reports the single endpoint a given protocol binding
+  resolves to, but the client factory negotiates the endpoint across the
+  card's whole interface list, so it can pick a URL that helper never returns.
+  Callers that need to constrain the destination must consider all of them.
+
+  1.x: every ``supported_interfaces[i].url``, in card order.
+  0.3.x: the top-level ``url`` followed by every
+  ``additional_interfaces[i].url``.
+  """
+  if IS_A2A_V1:
+    candidates = [iface.url for iface in card.supported_interfaces]
+  else:
+    candidates = [getattr(card, "url", None)]
+    candidates.extend(
+        iface.url
+        for iface in getattr(card, "additional_interfaces", None) or []
+    )
+  urls: list[str] = []
+  for url in candidates:
+    if url and url not in urls:
+      urls.append(url)
+  return urls
+
+
 # -----------------------------------------------------------------------------
 # Stream-item normalization
 # -----------------------------------------------------------------------------

src/google/adk/agents/remote_a2a_agent.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import ipaddress
 import json
 import logging
 from pathlib import Path
@@ -84,9 +85,44 @@
 A2A_METADATA_PREFIX = "a2a:"
 DEFAULT_TIMEOUT = 600.0
 
+_DEFAULT_PORTS = {"http": 80, "https": 443}
+
 logger = logging.getLogger("google_adk." + __name__)
 
 
+def _is_loopback_host(hostname: Optional[str]) -> bool:
+  """Returns whether a hostname names the local machine.
+
+  Covers ``localhost`` and the reserved ``*.localhost`` names as well as any
+  literal loopback address, so the local-development pattern the A2A helpers
+  emit -- a plain-http card served from ``localhost`` -- keeps working.
+  """
+  if not hostname:
+    return False
+  host = hostname.strip("[]").lower()
+  if host == "localhost" or host.endswith(".localhost"):
+    return True
+  try:
+    return ipaddress.ip_address(host).is_loopback
+  except ValueError:
+    return False
+
+
+def _url_origin(url: str) -> tuple[str, str, Optional[int]]:
+  """Returns the ``(scheme, host, port)`` origin triple for a URL.
+
+  Raises:
+    ValueError: If the URL carries a malformed port.
+  """
+  parsed = urlparse(url)
+  scheme = parsed.scheme.lower()
+  return (
+      scheme,
+      (parsed.hostname or "").lower(),
+      (parsed.port or _DEFAULT_PORTS.get(scheme)),
+  )
+
+
 @a2a_experimental
 class AgentCardResolutionError(Exception):
   """Raised when agent card resolution fails."""
@@ -320,6 +356,54 @@ async def _validate_agent_card(self, agent_card: AgentCard) -> None:
           f"Invalid RPC URL in agent card: {card_url}, error: {e}"
       ) from e
 
+    self._validate_card_rpc_targets(agent_card)
+
+  def _validate_card_rpc_targets(self, agent_card: AgentCard) -> None:
+    """Constrains where a card fetched over the network may aim RPC traffic.
+
+    Every URL the card offers is checked, not only the one this ADK version
+    would select, because the client factory negotiates the endpoint across
+    the card's whole interface list. Each must be https and share the origin
+    the card was fetched from; plain http stays allowed on a loopback host,
+    the local-development shape the A2A helpers emit.
+
+    A card passed in directly or read from a local file did not come off the
+    network here, so its target is left to the caller.
+    """
+    source = self._agent_card_source
+    if not source or not source.startswith(("http://", "https://")):
+      return
+
+    try:
+      source_origin = _url_origin(source)
+    except ValueError as e:
+      raise AgentCardResolutionError(
+          f"Invalid agent card source URL: {source}, error: {e}"
+      ) from e
+
+    for card_url in _compat.agent_card_rpc_urls(agent_card):
+      parsed_card = urlparse(card_url)
+      if parsed_card.scheme.lower() != "https" and not _is_loopback_host(
+          parsed_card.hostname
+      ):
+        raise AgentCardResolutionError(
+            "Agent card RPC URL must use https, or http on a loopback host:"
+            f" {card_url}"
+        )
+
+      try:
+        card_origin = _url_origin(card_url)
+      except ValueError as e:
+        raise AgentCardResolutionError(
+            f"Invalid RPC URL in agent card: {card_url}, error: {e}"
+        ) from e
+
+      if card_origin != source_origin:
+        raise AgentCardResolutionError(
+            "Agent card RPC URL must have the same origin as the location the"
+            f" card was fetched from ({source}): {card_url}"
+        )
+
   async def _ensure_resolved(
       self, ctx: Optional[InvocationContext] = None
   ) -> A2AClient:

tests/unittests/agents/test_remote_a2a_agent.py

@@ -26,6 +26,7 @@
 from a2a.client.client_factory import ClientFactory
 from a2a.types import AgentCapabilities
 from a2a.types import AgentCard
+from a2a.types import AgentInterface
 from a2a.types import AgentSkill
 from a2a.types import Artifact
 from a2a.types import Message as A2AMessage
@@ -165,6 +166,37 @@ def create_test_agent_card(
   )
 
 
+def _make_multi_interface_card(interfaces) -> AgentCard:
+  """Build a card offering several RPC endpoints, version-agnostically.
+
+  ``interfaces`` is a list of ``(url, transport)`` pairs; the first pair is the
+  card's primary endpoint. On 1.x every pair becomes a ``supported_interfaces``
+  entry; on 0.3.x the first pair is the top-level ``url``/``preferredTransport``
+  and the rest land in ``additional_interfaces``.
+  """
+  if _compat.IS_A2A_V1:
+    return _compat.parse_agent_card({
+        "name": "test-agent",
+        "description": "Test agent",
+        "version": "1.0",
+        "supported_interfaces": [
+            {"url": url, "protocol_binding": transport}
+            for url, transport in interfaces
+        ],
+        "default_input_modes": ["text/plain"],
+        "default_output_modes": ["text/plain"],
+    })
+  (primary_url, primary_transport), *extra = interfaces
+  return _make_agent_card(
+      url=primary_url,
+      preferred_transport=primary_transport,
+      additional_interfaces=[
+          AgentInterface(url=url, transport=transport)
+          for url, transport in extra
+      ],
+  )
+
+
 class TestRemoteA2aAgentInit:
   """Test RemoteA2aAgent initialization and validation."""
 
@@ -784,6 +816,141 @@ async def test_validate_agent_card_invalid_url(self):
     with pytest.raises(AgentCardResolutionError, match="Invalid RPC URL"):
       await agent._validate_agent_card(invalid_card)
 
+  @pytest.mark.asyncio
+  async def test_validate_agent_card_accepts_same_origin_https_rpc_url(self):
+    """A fetched card pointing back at its own origin is accepted."""
+    agent = RemoteA2aAgent(
+        name="test_agent", agent_card="https://example.com/agent.json"
+    )
+
+    # Should not raise any exception.
+    await agent._validate_agent_card(
+        create_test_agent_card(url="https://example.com/rpc")
+    )
+
+  @pytest.mark.asyncio
+  async def test_validate_agent_card_rejects_cross_origin_rpc_url(self):
+    """A fetched card cannot redirect RPC traffic to an unrelated host."""
+    agent = RemoteA2aAgent(
+        name="test_agent", agent_card="https://example.com/agent.json"
+    )
+
+    with pytest.raises(AgentCardResolutionError, match="same origin"):
+      await agent._validate_agent_card(
+          create_test_agent_card(url="https://attacker.example.net/rpc")
+      )
+
+  @pytest.mark.asyncio
+  async def test_validate_agent_card_rejects_plain_http_rpc_url(self):
+    """A fetched card cannot downgrade RPC traffic to cleartext."""
+    agent = RemoteA2aAgent(
+        name="test_agent", agent_card="https://example.com/agent.json"
+    )
+
+    with pytest.raises(AgentCardResolutionError, match="must use https"):
+      await agent._validate_agent_card(
+          create_test_agent_card(url="http://example.com/rpc")
+      )
+
+  @pytest.mark.asyncio
+  @pytest.mark.parametrize(
+      "rpc_url",
+      [
+          "http://127.0.0.1:8080/rpc",
+          "http://[::1]:8080/rpc",
+          "http://169.254.169.254/rpc",
+          "http://metadata.internal/rpc",
+      ],
+  )
+  async def test_validate_agent_card_rejects_internal_rpc_url(self, rpc_url):
+    """A fetched card cannot aim RPC traffic at host-local or internal hosts."""
+    agent = RemoteA2aAgent(
+        name="test_agent", agent_card="https://example.com/agent.json"
+    )
+
+    with pytest.raises(AgentCardResolutionError):
+      await agent._validate_agent_card(create_test_agent_card(url=rpc_url))
+
+  @pytest.mark.asyncio
+  async def test_validate_agent_card_allows_local_development_http(self):
+    """Plain http stays allowed for a same-origin loopback card."""
+    agent = RemoteA2aAgent(
+        name="test_agent",
+        agent_card="http://localhost:8000/.well-known/agent.json",
+    )
+
+    # Should not raise any exception.
+    await agent._validate_agent_card(
+        create_test_agent_card(url="http://localhost:8000/a2a")
+    )
+
+  @pytest.mark.asyncio
+  async def test_validate_agent_card_file_source_is_not_origin_checked(self):
+    """A card read from a local file is configuration, not remote data."""
+    agent = RemoteA2aAgent(name="test_agent", agent_card="/path/to/agent.json")
+
+    # Should not raise any exception.
+    await agent._validate_agent_card(
+        create_test_agent_card(url="http://internal-host:8080/rpc")
+    )
+
+  @pytest.mark.asyncio
+  @pytest.mark.parametrize(
+      "interfaces",
+      [
+          # A second interface on the transport the client already prefers
+          # displaces the benign endpoint during transport negotiation.
+          [
+              ("https://example.com/rpc", "JSONRPC"),
+              ("http://169.254.169.254/", "JSONRPC"),
+          ],
+          # The primary endpoint advertises a transport the client cannot
+          # speak, so negotiation falls through to the second interface.
+          [
+              ("https://example.com/rpc", "GRPC"),
+              ("http://127.0.0.1:9000/", "HTTP+JSON"),
+          ],
+      ],
+      ids=["displaces_primary", "primary_transport_unsupported"],
+  )
+  async def test_validate_agent_card_rejects_off_origin_extra_interface(
+      self, interfaces
+  ):
+    """Every endpoint the card offers is constrained, not just the first."""
+    agent = RemoteA2aAgent(
+        name="test_agent", agent_card="https://example.com/agent.json"
+    )
+
+    with pytest.raises(AgentCardResolutionError):
+      await agent._validate_agent_card(_make_multi_interface_card(interfaces))
+
+  @pytest.mark.asyncio
+  async def test_validate_agent_card_accepts_same_origin_extra_interface(self):
+    """A card may still offer several endpoints on its own origin."""
+    agent = RemoteA2aAgent(
+        name="test_agent", agent_card="https://example.com/agent.json"
+    )
+
+    # Should not raise any exception.
+    await agent._validate_agent_card(
+        _make_multi_interface_card([
+            ("https://example.com/rpc", "JSONRPC"),
+            ("https://example.com/rest", "HTTP+JSON"),
+        ])
+    )
+
+  def test_agent_card_rpc_urls_lists_every_endpoint(self):
+    """Validation enumerates every endpoint on the card, in card order."""
+    card = _make_multi_interface_card([
+        ("https://example.com/rpc", "JSONRPC"),
+        ("https://example.com/rest", "HTTP+JSON"),
+    ])
+
+    assert _compat.agent_card_rpc_urls(card) == [
+        "https://example.com/rpc",
+        "https://example.com/rest",
+    ]
+
   @pytest.mark.asyncio
   async def test_ensure_resolved_with_direct_agent_card(self):
     """Test _ensure_resolved with direct agent card."""