fix: Prevent continuation forgery in tool confirmation

An attacker who could manipulate or inject events into the session history could execute unauthorized tools by forging a tool confirmation response. This fixes the vulnerability by:
- When resolving confirmation targets, the processor verifies if the tool is registered in the executing agent's tools_dict
- Validate that the tool actually requires confirmation, supporting both static definitions and dynamic confirmation requests
- Verify that the original tool call event exists in the session history with the matching ID, and that its name and arguments match the confirmation request's originalFunctionCall exactly to prevent argument tampering.

Co-authored-by: Xuan Yang <xygoogle@google.com>
PiperOrigin-RevId: 953540969

Author: xuanyang15

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

@@ -13,7 +13,6 @@
 # limitations under the License.
 from __future__ import annotations
 
-import json
 import logging
 from typing import Any
 from typing import AsyncGenerator
@@ -27,68 +26,157 @@
 from ...agents.readonly_context import ReadonlyContext
 from ...events.event import Event
 from ...models.llm_request import LlmRequest
+from ...tools.base_tool import BaseTool
 from ...tools.tool_confirmation import ToolConfirmation
+from ...tools.tool_context import ToolContext
 from ._base_llm_processor import BaseLlmRequestProcessor
 from .functions import REQUEST_CONFIRMATION_FUNCTION_CALL_NAME
 
 if TYPE_CHECKING:
   pass
 
 
-logger = logging.getLogger('google_adk.' + __name__)
+logger = logging.getLogger("google_adk." + __name__)
 
 
 def _parse_tool_confirmation(response: dict[str, Any]) -> ToolConfirmation:
-  """Parse ToolConfirmation from a function response dict.
+  """Parses ToolConfirmation from a function response dict."""
+  return ToolConfirmation.from_response_dict(response)
 
-  Handles both the direct dict format and the ADK client's
-  ``{'response': json_string}`` wrapper format.
 
-  """
-  if response and len(response.values()) == 1 and 'response' in response.keys():
-    return ToolConfirmation.model_validate(json.loads(response['response']))
-  return ToolConfirmation.model_validate(response)
-
-
-def _resolve_confirmation_targets(
+async def _resolve_confirmation_targets(
+    invocation_context: InvocationContext,
     events: list[Event],
     confirmation_fc_ids: set[str],
     confirmations_by_fc_id: dict[str, ToolConfirmation],
+    tools_dict: dict[str, BaseTool],
 ) -> tuple[dict[str, ToolConfirmation], dict[str, types.FunctionCall]]:
-  """Find original function calls for confirmed tools.
+  """Find original function calls for confirmed tools and validate them.
 
   Scans events for ``adk_request_confirmation`` function calls whose IDs
   are in *confirmation_fc_ids*, extracts the ``originalFunctionCall`` from
-  their args, and maps each confirmation to the original FC ID.
+  their args, validates that they are registered, actually require confirmation,
+  and match the original function calls in history, and maps each confirmation
+  to the original FC ID.
 
   Args:
+    invocation_context: Current invocation context.
     events: Session events to scan.
     confirmation_fc_ids: IDs of ``adk_request_confirmation`` function calls.
     confirmations_by_fc_id: Mapping of confirmation FC ID ->
       ``ToolConfirmation``.
+    tools_dict: Dictionary of registered tools.
 
   Returns:
     Tuple of ``(tool_confirmation_dict, original_fcs_dict)`` where both
     are keyed by the ORIGINAL function call IDs.
+
+  Raises:
+    ValueError: If validation of any confirmation target fails.
   """
   tool_confirmation_dict: dict[str, ToolConfirmation] = {}
   original_fcs_dict: dict[str, types.FunctionCall] = {}
 
+  history_fcs = {
+      fc.id: (fc, ev)
+      for ev in events
+      for fc in ev.get_function_calls()
+      if fc.id and fc.name != REQUEST_CONFIRMATION_FUNCTION_CALL_NAME
+  }
+  history_fr_events = {
+      fr.id: ev for ev in events for fr in ev.get_function_responses() if fr.id
+  }
+
   for event in events:
     event_function_calls = event.get_function_calls()
     if not event_function_calls:
       continue
 
     for function_call in event_function_calls:
-      if function_call.id not in confirmation_fc_ids:
+      if not function_call.id or function_call.id not in confirmation_fc_ids:
         continue
 
       args = function_call.args
-      if 'originalFunctionCall' not in args:
+      if not args or "originalFunctionCall" not in args:
         continue
       original_function_call = types.FunctionCall(
-          **args['originalFunctionCall']
+          **args["originalFunctionCall"]
+      )
+      if not original_function_call.id:
+        raise ValueError("Original function call ID is missing.")
+      tool_name = original_function_call.name
+      if not tool_name:
+        raise ValueError("Original function call name is missing.")
+
+      # Check 1: Is the tool registered?
+      original_fc_info = history_fcs.get(original_function_call.id)
+      if not original_fc_info:
+        raise ValueError(
+            f"Original function call for ID '{original_function_call.id}' not"
+            " found in session history."
+        )
+      original_fc_in_history, original_fc_event = original_fc_info
+
+      # If this tool call was authored by another agent, skip it to let that
+      # agent's processor handle it.
+      agent = invocation_context.agent
+      if agent and original_fc_event.author != agent.name:
+        continue
+
+      tool = tools_dict.get(tool_name)
+      if not tool:
+        raise ValueError(
+            f"Tool '{original_function_call.name}' is not registered."
+        )
+
+      # Check 2: Does the tool require confirmation for these arguments?
+      # We check if it is either statically required, or if it was dynamically
+      # requested in the session history.
+      temp_tool_context = ToolContext(
+          invocation_context=invocation_context,
+          function_call_id=original_function_call.id,
       )
+      requires_confirmation = await tool.check_require_confirmation(
+          original_function_call.args or {}, temp_tool_context
+      )
+
+      requested_in_history = False
+      if not requires_confirmation:
+        # Search the history for the response event of the original tool call
+        original_response_event = history_fr_events.get(
+            original_function_call.id
+        )
+        if (
+            original_response_event
+            and original_response_event.actions.requested_tool_confirmations
+        ):
+          requested_in_history = (
+              original_function_call.id
+              in original_response_event.actions.requested_tool_confirmations
+          )
+
+      if not requires_confirmation and not requested_in_history:
+        raise ValueError(
+            f"Tool '{original_function_call.name}' does not require"
+            " confirmation."
+        )
+
+      # Check 3: Does the original function call match name and arguments?
+      if original_fc_in_history.name != original_function_call.name:
+        raise ValueError(
+            f"Function call name mismatch for ID '{original_function_call.id}':"
+            f" history has '{original_fc_in_history.name}', confirmation has"
+            f" '{original_function_call.name}'."
+        )
+
+      hist_args = original_fc_in_history.args or {}
+      conf_args = original_function_call.args or {}
+      if hist_args != conf_args:
+        raise ValueError(
+            "Function call arguments mismatch for ID"
+            f" '{original_function_call.id}'."
+        )
+
       tool_confirmation_dict[original_function_call.id] = (
           confirmations_by_fc_id[function_call.id]
       )
@@ -115,10 +203,9 @@ async def run_async(
     # Step 1: Find the last user-authored event and parse confirmation
     # responses from it.
     confirmations_by_fc_id: dict[str, ToolConfirmation] = {}
-    confirmation_event_index = -1
     for k in range(len(events) - 1, -1, -1):
       event = events[k]
-      if not event.author or event.author != 'user':
+      if not event.author or event.author != "user":
         continue
       responses = event.get_function_responses()
       if not responses:
@@ -127,29 +214,45 @@ async def run_async(
       for function_response in responses:
         if function_response.name != REQUEST_CONFIRMATION_FUNCTION_CALL_NAME:
           continue
+        if not function_response.id or function_response.response is None:
+          continue
         confirmations_by_fc_id[function_response.id] = _parse_tool_confirmation(
             function_response.response
         )
-      confirmation_event_index = k
       break
 
     if not confirmations_by_fc_id:
       return
 
+    # Resolve all canonical tools and build tools_dict
+    tools_dict = {}
+    if agent is not None and hasattr(agent, "canonical_tools"):
+      tools_dict = {
+          tool.name: tool
+          for tool in await agent.canonical_tools(
+              ReadonlyContext(invocation_context)
+          )
+      }
+
     # Step 2: Resolve confirmation targets using extracted helper.
     confirmation_fc_ids = set(confirmations_by_fc_id.keys())
     tools_to_resume_with_confirmation, tools_to_resume_with_args = (
-        _resolve_confirmation_targets(
-            events, confirmation_fc_ids, confirmations_by_fc_id
+        await _resolve_confirmation_targets(
+            invocation_context,
+            events,
+            confirmation_fc_ids,
+            confirmations_by_fc_id,
+            tools_dict,
         )
     )
 
     if not tools_to_resume_with_confirmation:
       return
 
     # Step 3: Remove tools that have already been confirmed (dedup).
-    for i in range(len(events) - 1, confirmation_event_index, -1):
-      event = events[i]
+    for event in reversed(events):
+      if event.author == "user":
+        break
       fr_list = event.get_function_responses()
       if not fr_list:
         continue
@@ -167,14 +270,9 @@ async def run_async(
     # Step 4: Re-execute the confirmed tools.
     if function_response_event := await functions.handle_function_call_list_async(
         invocation_context,
-        tools_to_resume_with_args.values(),
-        {
-            tool.name: tool
-            for tool in await agent.canonical_tools(
-                ReadonlyContext(invocation_context)
-            )
-        },
-        tools_to_resume_with_confirmation.keys(),
+        list(tools_to_resume_with_args.values()),
+        tools_dict,
+        set(tools_to_resume_with_confirmation.keys()),
         tools_to_resume_with_confirmation,
     ):
       yield function_response_event

src/google/adk/tools/base_tool.py

@@ -168,6 +168,12 @@ async def process_llm_request(
     # Use the consolidated logic in LlmRequest.append_tools
     llm_request.append_tools([self])
 
+  async def check_require_confirmation(
+      self, args: dict[str, Any], tool_context: ToolContext
+  ) -> bool:
+    """Returns whether the tool requires confirmation for the given args."""
+    return False
+
   @property
   def _api_variant(self) -> GoogleLLMVariant:
     return get_google_llm_variant()

src/google/adk/tools/function_tool.py

@@ -18,6 +18,7 @@
 import logging
 from typing import Any
 from typing import Callable
+from typing import cast
 from typing import get_args
 from typing import get_origin
 from typing import get_type_hints
@@ -194,36 +195,52 @@ def _preprocess_args(self, args: dict[str, Any]) -> dict[str, Any]:
               args[param_name], list
           ):
             item_type = get_list_inner_type(target_type)
-            try:
-              converted_args[param_name] = [
-                  item_type.model_validate(item)
-                  if isinstance(item, dict)
-                  else item
-                  for item in args[param_name]
-              ]
-            except Exception as e:
-              logger.warning(
-                  f"Failed to convert argument '{param_name}' to"
-                  f' list[{item_type.__name__}]: {e}'
-              )
-              pass
+            if item_type is not None:
+              try:
+                converted_args[param_name] = [
+                    item_type.model_validate(item)
+                    if isinstance(item, dict)
+                    else item
+                    for item in args[param_name]
+                ]
+              except Exception as e:
+                logger.warning(
+                    f"Failed to convert argument '{param_name}' to"
+                    f' list[{item_type.__name__}]: {e}'
+                )
+                pass
 
     return converted_args
 
-  @override
-  async def run_async(
-      self, *, args: dict[str, Any], tool_context: ToolContext
-  ) -> Any:
-    # Preprocess arguments (includes Pydantic model conversion)
+  def _prepare_invocation_args(
+      self, args: dict[str, Any], tool_context: ToolContext
+  ) -> dict[str, Any]:
+    """Prepare args for function invocation (preprocesses, injects context and filters)."""
     args_to_call = self._preprocess_args(args)
-
     signature = inspect.signature(self.func)
-    valid_params = {param for param in signature.parameters}
+    valid_params = set(signature.parameters.keys())
     if self._context_param_name in valid_params:
       args_to_call[self._context_param_name] = tool_context
+    return {k: v for k, v in args_to_call.items() if k in valid_params}
 
-    # Filter args_to_call to only include valid parameters for the function
-    args_to_call = {k: v for k, v in args_to_call.items() if k in valid_params}
+  @override
+  async def check_require_confirmation(
+      self, args: dict[str, Any], tool_context: ToolContext
+  ) -> bool:
+    if callable(self._require_confirmation):
+      args_to_call = self._prepare_invocation_args(args, tool_context)
+      return cast(
+          bool,
+          await self._invoke_callable(self._require_confirmation, args_to_call),
+      )
+    return bool(self._require_confirmation)
+
+  @override
+  async def run_async(
+      self, *, args: dict[str, Any], tool_context: ToolContext
+  ) -> Any:
+    # Preprocess arguments (includes Pydantic model conversion)
+    args_to_call = self._prepare_invocation_args(args, tool_context)
 
     # Before invoking the function, we check for if the list of args passed in
     # has all the mandatory arguments or not.
@@ -242,12 +259,9 @@ async def run_async(
 You could retry calling this tool, but it is IMPORTANT for you to provide all the mandatory parameters."""
       return {'error': error_str}
 
-    if isinstance(self._require_confirmation, Callable):
-      require_confirmation = await self._invoke_callable(
-          self._require_confirmation, args_to_call
-      )
-    else:
-      require_confirmation = bool(self._require_confirmation)
+    require_confirmation = await self.check_require_confirmation(
+        args, tool_context
+    )
 
     if require_confirmation:
       if not tool_context.tool_confirmation: