fix: reject tool confirmations arriving over A2A

Merge https://github.com/google/adk-python/pull/6462

A remote peer could self-approve a pending dangerous tool call via A2A because
inbound messages are treated as user role. This change ignores confirmations
arriving over A2A channels.

Closes #6461

PiperOrigin-RevId: 966208121

src/google/adk/a2a/converters/request_converter.py

@@ -98,10 +98,12 @@ def convert_a2a_request_to_agent_run_request(
   if not request.message:
     raise ValueError('Request message cannot be None')
 
-  custom_metadata = {}
-  request_metadata = _compat.meta_to_dict(request.metadata)
-  if request_metadata:
-    custom_metadata[A2A_METADATA_KEY] = request_metadata
+  # Always mark the invocation as A2A-originated, even when the peer sent no
+  # protocol metadata. Trust decisions downstream (e.g. refusing a
+  # human-in-the-loop tool confirmation that arrived from a remote peer) key
+  # off the PRESENCE of this marker, so it must not depend on the peer
+  # choosing to send metadata.
+  custom_metadata = {A2A_METADATA_KEY: _compat.meta_to_dict(request.metadata)}
 
   output_parts = []
   for a2a_part in request.message.parts:

src/google/adk/flows/llm_flows/request_confirmation.py

@@ -41,6 +41,8 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger("google_adk." + __name__)
 
+_A2A_METADATA_KEY = "a2a_metadata"
+
 
 def _parse_tool_confirmation(response: dict[str, Any]) -> ToolConfirmation:
   """Parses ToolConfirmation from a function response dict."""
@@ -262,6 +264,18 @@ class _RequestConfirmationLlmRequestProcessor(BaseLlmRequestProcessor):
 
     agent = invocation_context.agent
 
+    # A human-in-the-loop confirmation must not be satisfiable by a
+    # function_response that arrived over A2A: a remote peer is not the human
+    # operator and could self-approve a pending dangerous tool call.
+    run_config = invocation_context.run_config
+    custom_metadata = run_config.custom_metadata if run_config else None
+    if custom_metadata is not None and _A2A_METADATA_KEY in custom_metadata:
+      logger.warning(
+          "Ignoring tool confirmation(s) that arrived over A2A: a remote peer"
+          " cannot satisfy a human-in-the-loop confirmation."
+      )
+      return
+
     # Only look at events in the current branch.
     events = invocation_context._get_events(current_branch=True)
     if not events:

tests/unittests/a2a/converters/test_request_converter.py

@@ -259,8 +259,13 @@ class TestConvertA2aRequestToAgentRunRequest:
     # Numbers may come back as float on 1.x (proto Struct) -> tolerate both.
     assert float(stored["n"]) == 1.0
 
-  def test_convert_a2a_request_empty_metadata_omitted(self):
-    """Empty request metadata must not add an ``a2a_metadata`` entry."""
+  def test_convert_a2a_request_empty_metadata_still_marks_a2a(self):
+    """Empty request metadata must STILL mark the invocation as A2A-originated.
+
+    The marker is a trust signal, not a data carrier: downstream checks key off
+    its presence, so a peer must not be able to suppress it by omitting
+    metadata.
+    """
     empty_meta_msg = _compat.make_message(
         message_id="m1", role=_compat.ROLE_USER, parts=[]
     )
@@ -276,7 +281,7 @@ class TestConvertA2aRequestToAgentRunRequest:
 
     result = convert_a2a_request_to_agent_run_request(request, Mock())
 
-    assert "a2a_metadata" not in result.run_config.custom_metadata
+    assert result.run_config.custom_metadata == {"a2a_metadata": {}}
 
   def test_convert_a2a_request_no_message_raises_error(self):
     """Test that conversion raises ValueError when message is None."""

tests/unittests/flows/llm_flows/test_request_confirmation.py

@@ -13,8 +13,11 @@
 # limitations under the License.
 
 import json
+from unittest import mock
 from unittest.mock import patch
 
+from a2a.server.agent_execution import RequestContext
+from google.adk.a2a.converters.request_converter import convert_a2a_request_to_agent_run_request
 from google.adk.agents.llm_agent import LlmAgent
 from google.adk.events.event import Event
 from google.adk.events.event_actions import EventActions
@@ -1320,3 +1323,95 @@ async def test_resolve_confirmation_targets_requires_adk_name():
 
   assert set(tool_confirmation_dict) == {"requested_fc_id"}
   assert set(original_fcs_dict) == {"requested_fc_id"}
+
+
+@pytest.mark.asyncio
+async def test_request_confirmation_processor_ignores_a2a_confirmation():
+  """A confirmation arriving over A2A must not satisfy the HITL gate."""
+  await _assert_a2a_confirmation_ignored({"a2a:task_id": "t1"})
+
+
+@pytest.mark.asyncio
+async def test_request_confirmation_processor_ignores_a2a_without_metadata():
+  """The guard must hold when the peer sends no protocol metadata.
+
+  Regression test for #6461: the A2A marker used to be set only when
+  request.metadata was non-empty, so a peer that sent none produced an empty
+  custom_metadata and slipped past the guard.
+  """
+  await _assert_a2a_confirmation_ignored(None)
+
+
+async def _assert_a2a_confirmation_ignored(request_metadata):
+  """Asserts that a confirmation arriving over A2A is ignored."""
+  request = mock.Mock(spec=RequestContext)
+  request.message = mock.Mock()
+  request.message.parts = []
+  request.metadata = request_metadata
+  request.context_id = "ctx"
+  request.call_context = None
+  run_config = convert_a2a_request_to_agent_run_request(
+      request, mock.Mock()
+  ).run_config
+  assert "a2a_metadata" in run_config.custom_metadata
+
+  agent = LlmAgent(name="test_agent", tools=[mock_tool])
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent, run_config=run_config
+  )
+
+  original_function_call = types.FunctionCall(
+      name=MOCK_TOOL_NAME, args={"param1": "test"}, id=MOCK_FUNCTION_CALL_ID
+  )
+  tool_confirmation = ToolConfirmation(confirmed=False, hint="test hint")
+  tool_confirmation_args = {
+      "originalFunctionCall": original_function_call.model_dump(
+          exclude_none=True, by_alias=True
+      ),
+      "toolConfirmation": tool_confirmation.model_dump(
+          by_alias=True, exclude_none=True
+      ),
+  }
+  invocation_context.session.events.append(
+      Event(
+          author="agent",
+          content=types.Content(
+              parts=[
+                  types.Part(
+                      function_call=types.FunctionCall(
+                          name=functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME,
+                          args=tool_confirmation_args,
+                          id=MOCK_CONFIRMATION_FUNCTION_CALL_ID,
+                      )
+                  )
+              ]
+          ),
+      )
+  )
+  user_confirmation = ToolConfirmation(confirmed=True)
+  invocation_context.session.events.append(
+      Event(
+          author="user",
+          content=types.Content(
+              parts=[
+                  types.Part(
+                      function_response=types.FunctionResponse(
+                          name=functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME,
+                          id=MOCK_CONFIRMATION_FUNCTION_CALL_ID,
+                          response={
+                              "response": user_confirmation.model_dump_json()
+                          },
+                      )
+                  )
+              ]
+          ),
+      )
+  )
+
+  events = []
+  async for event in request_processor.run_async(
+      invocation_context, LlmRequest()
+  ):
+    events.append(event)
+
+  assert not events