diff --git a/docs/guides/README.md b/docs/guides/README.md index 58bc331f003..cf550ba4dcb 100644 --- a/docs/guides/README.md +++ b/docs/guides/README.md @@ -12,6 +12,7 @@ This directory contains specific developer guides for the ADK Python implementat ### Agents * [BaseAgent](agents/base_agent/index.md) - The foundational base class for custom agents, container orchestrators, and lifecycle callbacks. +* [CallerPrincipal](agents/caller_principal/index.md) - The caller identity a serving layer establishes for an invocation, and how tool confirmation uses it. * [Context](agents/context/index.md) - The runtime interface for state, artifacts, memory, credentials, and dynamic execution. * [Creating Agents with Configurations](agents/config/index.md) - Building and wiring multi-agent graphs from external YAML configuration files. * [InvocationContext](agents/invocation_context/index.md) - The runtime dependency container and execution state for a single invocation turn. diff --git a/docs/guides/agents/caller_principal/index.md b/docs/guides/agents/caller_principal/index.md new file mode 100644 index 00000000000..bc90776e999 --- /dev/null +++ b/docs/guides/agents/caller_principal/index.md @@ -0,0 +1,237 @@ +# CallerPrincipal + +`CallerPrincipal` records whether a serving layer authenticated the caller of +an invocation, and as whom. The human-in-the-loop tool confirmation flow reads +it to decide whether an approval that arrived over the wire is one the agent +may act on. + +## Introduction + +A tool that requires confirmation pauses the agent until a user approves the +call. When the agent is served remotely, the approval arrives as a message on +the same channel as everything else the remote caller sends, so without a +principal the framework cannot tell an operator's approval from a remote peer +approving the dangerous tool call it just caused. `CallerPrincipal` carries the +one fact that decides this: whether the layer that received the request +verified who sent it. + +The principal lives on `InvocationContext.caller_principal`. `Runner.run_async` +accepts it as a keyword argument, and the A2A executor sets it for every +request it serves, so an application that serves agents over A2A gets it +without writing any code. An application with its own serving layer sets it +from its own authenticator. + +The design keeps the principal separate from the transport on purpose. A +transport is a proxy for identity, and a proxy for identity fails in both +directions: it refuses authenticated callers because of how they connected, +and it admits anyone who reaches a path the proxy does not cover. Asking the +authentication question directly avoids both failures. + +## Get started + +This example serves an agent behind an application's own request handler. The +handler runs its authenticator first, then tells the runner what it found. A +confirmation-gated tool is included so that the principal has something to +decide. + +```python +from google.adk.agents import CallerPrincipal +from google.adk.agents import LlmAgent +from google.adk.runners import InMemoryRunner +from google.adk.tools.bash_tool import ExecuteBashTool + +agent = LlmAgent(name="ops_agent", tools=[ExecuteBashTool()]) +runner = InMemoryRunner(agent=agent, app_name="ops") + + +async def handle_request(request, new_message): + # Your authenticator runs before this point. Report what it established; + # do not infer anything from the message the caller sent. + user = request.authenticated_user + principal = CallerPrincipal( + authenticated=user is not None, + user_name=user.name if user else None, + source="gateway", + ) + async for event in runner.run_async( + user_id=request.user_id, + session_id=request.session_id, + new_message=new_message, + caller_principal=principal, + ): + yield event +``` + +An in-process caller, such as a test or a command-line tool that calls +`runner.run_async` directly, does not pass a principal. Leaving it unset says +that no remote trust boundary was crossed, which is true for that caller. + +## How it works + +`InvocationContext.caller_principal` has three meaningful states, and the tool +confirmation flow treats each differently. + +| State | Meaning | Tool confirmation | +| :--- | :--- | :--- | +| `None` | The invocation started in process. Nothing had to vouch for the caller, because the caller is the operator. | Honored. | +| `authenticated=False` | A serving layer handled the request and could not say who sent it. | Warned about by default; refused when strict mode is on. | +| `authenticated=True` | The serving layer verified the caller. | Honored. | + +A refusal is explicit rather than silent. The confirmation is delivered to the +tool with `confirmed=False`, which is the same state a user's decline produces, +so a tool that follows the confirmation contract returns its rejection response +and the turn ends with a reason the caller can see. The pending call never +hangs waiting for an approval that will not come. + +The same decision applies wherever a confirmation answer enters the framework: +a confirmation carried in a user event to an `LlmAgent`, and an answer that +resumes a paused workflow node (`RequestInput` with interrupt id +`wf_tool_confirmation:`), whether the node wraps a tool or a +function declared with `@node(require_confirmation=True)`. In a workflow node +the answer is rewritten before it is stored on the context, so tools that read +`tool_confirmation` themselves, such as `ExecuteBashTool`, see the refusal too. + +The principal is derived only from what the serving layer established. It is +never read from message content, from event authorship, or from transport +metadata, because a remote caller controls all three and could set any of them +to whatever the framework wanted to see. + +### A2A + +`A2aAgentExecutor` builds the principal from the A2A server's call context on +every request. The caller is authenticated exactly when the A2A server +authenticated it, which is also the case in which the invocation's user id is +the authenticated user name rather than a generated `A2A_USER_` value. The two +cannot disagree about the same request. + +An A2A server that runs without an authenticator produces an unauthenticated +principal for every request. With the default settings that yields a warning +per approval and the approval is honored, which is the behavior those +deployments had before the principal existed. + +### ADK API server + +The `/run` and `/run_sse` routes of the ADK API server (`adk api_server`, +`adk web`, `get_fast_api_app`) record a principal on every invocation as +well. The server does not authenticate callers itself, and the `user_id` in +the request body is a claim the caller made rather than one the server +checked, so without anything in front of these routes every request runs as +`CallerPrincipal(authenticated=False, source="api_server")`. With the default +settings that is a warning per approval, as for an unauthenticated A2A caller. + +A deployment that authenticates callers in front of these routes, with an +identity-aware proxy, a gateway or its own middleware, states the result on +`request.state.caller_principal`, and the server uses that value as is: + +```python +from google.adk.agents import CallerPrincipal +from google.adk.cli.fast_api import get_fast_api_app + +app = get_fast_api_app(agents_dir="./agents", web=False) + + +@app.middleware("http") +async def record_caller(request, call_next): + # Your authenticator decides who sent this request; report that outcome. + user = verify_request(request) # None when the request is not authenticated. + if user is not None: + request.state.caller_principal = CallerPrincipal( + authenticated=True, user_name=user.name, source="gateway" + ) + return await call_next(request) +``` + +Under strict mode, approvals arriving on `/run` or `/run_sse` are refused +unless the deployment sets the principal this way. The dev UI served by `adk +web` has no authentication of its own, so run it without strict mode, or add +middleware that recognizes its operator. + +### Strict mode + +Strict mode refuses confirmations from unauthenticated callers instead of +warning about them. It is the `STRICT_CALLER_PRINCIPAL` feature and is off by +default, because turning it on changes what an existing deployment does with an +approval, and an upgrade should not do that on its own. The intent is to make +strict mode the default at the next major version. + +Turn it on with the environment variable ADK uses for every feature: + +```bash +export ADK_ENABLE_STRICT_CALLER_PRINCIPAL=1 +``` + +Or programmatically, when environment variables are not practical in your +deployment: + +```python +from google.adk.features import FeatureName +from google.adk.features import override_feature_enabled + +override_feature_enabled(FeatureName.STRICT_CALLER_PRINCIPAL, True) +``` + +Strict mode never affects an invocation with no principal. An in-process +caller crossed no boundary, so there is nothing to refuse. + +## Configuration options + +`CallerPrincipal` is a frozen model with three fields. It rejects unknown +fields, so a typo cannot silently produce a principal that says nothing. + +| Option | Type | Default | Description | +| :--- | :--- | :--- | :--- | +| `authenticated` | `bool` | required | Whether a serving-layer authenticator verified the caller. | +| `user_name` | `str \| None` | `None` | The verified identity when `authenticated` is `True`. | +| `source` | `str \| None` | `None` | A short label for the serving edge that set the principal. | + +`authenticated` is the only field a trust decision reads. Set it to `True` only +when your authenticator actually verified the caller. A request that carries a +name it asserted about itself, without verification, is not authenticated, and +saying otherwise reopens the problem the principal exists to close. + +`user_name` is informational. It is the identity the authenticator established, +kept alongside the decision so that a log line or an audit record can say who +approved a tool call. Leave it `None` when `authenticated` is `False`; a name +without verification behind it is a claim, not an identity. + +`source` is a label such as `"a2a"` or `"gateway"` that names the edge that +set the principal. It appears in the confirmation flow's log messages so that +an operator reading a warning can tell which entry point produced it. Nothing +makes a trust decision on it, so a transport cannot become a proxy for identity +through this field either. + +## Advanced applications + +### Custom A2A request converters + +`A2aAgentExecutorConfig.request_converter` lets an application replace the +function that turns an A2A request into a run request. The executor sets the +principal from the server's call context after the converter returns, +regardless of what the converter produced. A converter can therefore neither +drop the principal nor assert one the server never established, which keeps +the trust decision structural rather than dependent on every converter +remembering to make it. + +### Serving layers with more than one entry point + +An application that exposes an agent through several edges can give each a +distinct `source` and let them share the same rule for `authenticated`. The +confirmation flow's log messages then identify the edge, and the trust +decision stays uniform across all of them. + +## Limitations + +- The principal describes the caller of the invocation as a whole. ADK does + not verify the identity behind each individual event within a session. +- Only tool confirmation answers are gated on the principal today, on every + path that accepts one: the `LlmAgent` confirmation flow and the resume of a + workflow tool node or `require_confirmation` function node. Other consumers + can read `InvocationContext.caller_principal`, but the framework does not yet + gate anything else on it. +- The synchronous `Runner.run` wrapper and live sessions do not accept a + principal. Both behave as in-process callers. +- Setting a principal does not authenticate anything by itself. The ADK API + server's `/run` and `/run_sse` routes record an unauthenticated principal + unless the deployment's own authentication sets one; the principal records + the outcome of authentication, it does not perform it. `/run_live` sessions + do not carry a principal. diff --git a/src/google/adk/a2a/converters/request_converter.py b/src/google/adk/a2a/converters/request_converter.py index ee0628540b2..7d92496b3eb 100644 --- a/src/google/adk/a2a/converters/request_converter.py +++ b/src/google/adk/a2a/converters/request_converter.py @@ -25,6 +25,7 @@ from pydantic import BaseModel from .. import _compat +from ...agents._caller_principal import CallerPrincipal from ...agents.run_config import RunConfig from ..experimental import a2a_experimental from .part_converter import A2APartToGenAIPartConverter @@ -45,6 +46,7 @@ class AgentRunRequest(BaseModel): new_message: Optional[genai_types.Content] = None state_delta: Optional[dict[str, Any]] = None run_config: Optional[RunConfig] = None + caller_principal: Optional[CallerPrincipal] = None A2ARequestToAgentRunRequestConverter = Callable[ @@ -86,6 +88,35 @@ def _get_user_id(request: RequestContext) -> str: return f'A2A_USER_{request.context_id}' +def build_caller_principal(request: RequestContext) -> CallerPrincipal: + """Records what the A2A serving layer established about the caller. + + ``_get_user_id`` above already reads ``call_context.user``, but it collapses + the result to a bare string and drops the one bit that matters downstream: + whether an authenticator produced that name or whether it was synthesized + from the caller-supplied context id. This keeps that bit. + + Args: + request: The incoming request context from the A2A server. + + Returns: + An authenticated principal when the A2A server authenticated the caller, + and an unauthenticated one otherwise. Never ``None``: an A2A request has + crossed a remote trust boundary either way, and ``None`` has to keep + meaning that no boundary was crossed at all. + """ + user = request.call_context.user if request.call_context else None + user_name = getattr(user, 'user_name', None) if user is not None else None + # Authenticated exactly when _get_user_id above uses the name: a non-empty + # user_name on a principal that reports is_authenticated. The principal and + # the user id can therefore never disagree about the same request. + if not isinstance(user_name, str) or not user_name: + return CallerPrincipal(authenticated=False, source='a2a') + if not getattr(user, 'is_authenticated', False): + return CallerPrincipal(authenticated=False, source='a2a') + return CallerPrincipal(authenticated=True, user_name=user_name, source='a2a') + + @functools.lru_cache(maxsize=1) def _warn_unauthenticated_user_name_once() -> None: logger.warning( @@ -137,4 +168,5 @@ def convert_a2a_request_to_agent_run_request( parts=output_parts, ), run_config=RunConfig(custom_metadata=custom_metadata), + caller_principal=build_caller_principal(request), ) diff --git a/src/google/adk/a2a/executor/a2a_agent_executor.py b/src/google/adk/a2a/executor/a2a_agent_executor.py index 9b655c39412..db4e9594acd 100644 --- a/src/google/adk/a2a/executor/a2a_agent_executor.py +++ b/src/google/adk/a2a/executor/a2a_agent_executor.py @@ -34,6 +34,7 @@ from ...utils.context_utils import Aclosing from ..agent.interceptors.new_integration_extension import _NEW_A2A_ADK_INTEGRATION_EXTENSION from ..converters.request_converter import AgentRunRequest +from ..converters.request_converter import build_caller_principal from ..converters.utils import _get_adk_metadata_key from ..experimental import a2a_experimental from .a2a_agent_executor_impl import _A2aAgentExecutor as ExecutorImpl @@ -189,6 +190,11 @@ async def _handle_request( context, self._config.a2a_part_converter, ) + # The request converter is pluggable, so the principal is re-derived here + # from the RequestContext the A2A server handed us. A custom converter must + # not be able to drop the principal, and must not be able to assert one + # the server never established. + run_request.caller_principal = build_caller_principal(context) # ensure the session exists session = await self._prepare_session(context, run_request, runner) @@ -198,6 +204,7 @@ async def _handle_request( session=session, new_message=run_request.new_message, run_config=run_request.run_config, + caller_principal=run_request.caller_principal, ) executor_context = ExecutorContext( diff --git a/src/google/adk/a2a/executor/a2a_agent_executor_impl.py b/src/google/adk/a2a/executor/a2a_agent_executor_impl.py index c34ecf8d03d..fcd5d05c4ce 100644 --- a/src/google/adk/a2a/executor/a2a_agent_executor_impl.py +++ b/src/google/adk/a2a/executor/a2a_agent_executor_impl.py @@ -37,6 +37,7 @@ from ..converters.long_running_functions import handle_user_input from ..converters.long_running_functions import LongRunningFunctions from ..converters.request_converter import AgentRunRequest +from ..converters.request_converter import build_caller_principal from ..converters.utils import _get_adk_metadata_key from ..experimental import a2a_experimental from .config import A2aAgentExecutorConfig @@ -102,6 +103,11 @@ async def execute( context, self._config.a2a_part_converter, ) + # The request converter is pluggable, so the principal is re-derived + # here from the RequestContext the A2A server handed us. A custom + # converter must not be able to drop the principal, and must not be + # able to assert one the server never established. + run_request.caller_principal = build_caller_principal(context) session_id = await self._resolve_session(run_request, runner) executor_context = ExecutorContext( diff --git a/src/google/adk/agents/__init__.py b/src/google/adk/agents/__init__.py index 6a2f464913e..ecbc54abf6c 100644 --- a/src/google/adk/agents/__init__.py +++ b/src/google/adk/agents/__init__.py @@ -42,6 +42,7 @@ 'Agent': '.llm_agent', 'BaseAgent': '.base_agent', 'BaseAgentConfig': '.base_agent_config', + 'CallerPrincipal': '._caller_principal', 'Context': '.context', 'InvocationContext': '.invocation_context', 'LiveRequest': '.live_request_queue', @@ -68,6 +69,7 @@ 'McpInstructionProvider', 'ParallelAgent', 'SequentialAgent', + 'CallerPrincipal', 'InvocationContext', 'LiveRequest', 'LiveRequestQueue', diff --git a/src/google/adk/agents/_caller_principal.py b/src/google/adk/agents/_caller_principal.py new file mode 100644 index 00000000000..119f0cbf761 --- /dev/null +++ b/src/google/adk/agents/_caller_principal.py @@ -0,0 +1,126 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""The caller identity that the serving layer established for an invocation.""" + +from __future__ import annotations + +import logging +from typing import Optional + +from pydantic import BaseModel +from pydantic import ConfigDict + +from ..features import FeatureName +from ..features import is_feature_enabled + +logger = logging.getLogger("google_adk." + __name__) + + +class CallerPrincipal(BaseModel): + """Who sent this invocation, and whether a serving layer vouched for them. + + A serving edge (for example the A2A executor) sets this from the + authentication it actually performed on the inbound request. It is never + derived from message content, event authorship, or transport metadata, + because the caller controls all of those. + + The confirmation flow uses it to decide whether a human-in-the-loop approval + can be honored. Three states are meaningful: + + - ``None`` on the invocation context: no remote trust boundary was crossed + (for example an in-process ``Runner.run_async`` call). The caller is the + operator by construction. + - ``authenticated=True``: the edge verified the caller's identity. + - ``authenticated=False``: the edge saw the request but could not vouch for + who sent it. + """ + + model_config = ConfigDict(extra="forbid", frozen=True) + """The pydantic model config.""" + + authenticated: bool + """True only if a serving-layer authenticator verified the caller.""" + + user_name: Optional[str] = None + """The verified identity when ``authenticated`` is True, else None.""" + + source: Optional[str] = None + """Short label for the edge that set this, for example ``"a2a"``. + + Informational only. Trust decisions must key off ``authenticated``, never + off the source, so a transport can not be used as a proxy for identity. + """ + + +def caller_may_confirm(principal: Optional[CallerPrincipal]) -> bool: + """Decides whether a tool confirmation from this caller may be honored. + + Every consumer of a human-in-the-loop answer asks this one question, so the + agent pipeline and the workflow tool node refuse the same callers. Three + states, and only the middle one is a refusal: + + - No principal: nothing vouched for the caller because nothing had to. The + invocation was started in process, so the caller is the operator. + - Principal present, not authenticated: a serving layer handled this request + and could not say who sent it, so the approval is not known to be the + operator's. + - Principal present and authenticated: the serving layer verified the caller. + + The question asked here is deliberately about authentication and not about + transport. A transport is a proxy for identity, and a proxy for identity + fails in both directions: it refuses authenticated peers that happen to + arrive over the wire, and it admits anyone who reaches an ungated path. + + Strict mode is the STRICT_CALLER_PRINCIPAL feature, off by default for now + so that upgrading cannot break a working deployment that runs its A2A server + without an authenticator; those get a warning instead. The intent is to flip + the registry default at the next major version. + + Callers that get ``False`` should store the answer as ``confirmed=False`` + rather than drop it. Dropping it leaves the confirmation request pending with + nothing left to resolve it, which is what made an earlier attempt at this + guard stall every human-in-the-loop tool. ``confirmed=False`` is a state the + framework already has a contract for -- it is what a human decline produces + -- so the tool returns its rejection response and the turn ends with a + reason the caller can see. + + Args: + principal: The ``caller_principal`` of the current invocation context. + + Returns: + True if the answer may be honored as given, False if it must be treated as + a rejection. + """ + if principal is None or principal.authenticated: + return True + + if not is_feature_enabled(FeatureName.STRICT_CALLER_PRINCIPAL): + logger.warning( + "Honoring a tool confirmation from an unauthenticated caller" + " (principal source %r). The serving layer could not say who sent this" + " approval, so it is not known to be the operator's. Enable the" + " STRICT_CALLER_PRINCIPAL feature to refuse these instead; that is" + " intended to become the default in a future major version.", + principal.source, + ) + return True + + logger.error( + "Refusing a tool confirmation from an unauthenticated caller (principal" + " source %r). Enable authentication on the serving layer, or disable the" + " STRICT_CALLER_PRINCIPAL feature to downgrade this to a warning.", + principal.source, + ) + return False diff --git a/src/google/adk/agents/invocation_context.py b/src/google/adk/agents/invocation_context.py index 338f1e23874..1c5354fe27a 100644 --- a/src/google/adk/agents/invocation_context.py +++ b/src/google/adk/agents/invocation_context.py @@ -51,6 +51,7 @@ from ..tools.base_tool import BaseTool from ..utils._agent_mode import AgentMode from ..workflow._base_node import BaseNode +from ._caller_principal import CallerPrincipal from .base_agent import _agent_state_key from .base_agent import BaseAgent from .base_agent import BaseAgentState @@ -246,6 +247,18 @@ class InvocationContext(BaseModel): run_config: RunConfig | None = None """Configurations for live agents under this invocation.""" + caller_principal: CallerPrincipal | None = None + """Who the serving layer says submitted this invocation. + + ``None`` means no remote trust boundary was crossed for this invocation: it + was started in process (for example ``Runner.run_async`` called directly), so + the caller is the operator by construction. + + A serving edge sets this from the authentication it actually performed on the + inbound request. It is never derived from message content, event authorship, + or transport metadata, because a remote caller controls all of those. + """ + resumability_config: ResumabilityConfig | None = None """The resumability config that applies to all agents under this invocation.""" diff --git a/src/google/adk/cli/api_server.py b/src/google/adk/cli/api_server.py index 407b9892c56..d05bb6cfcea 100644 --- a/src/google/adk/cli/api_server.py +++ b/src/google/adk/cli/api_server.py @@ -70,6 +70,7 @@ from typing_extensions import override import yaml +from ..agents._caller_principal import CallerPrincipal from ..agents._streaming_mode import StreamingMode from ..agents.base_agent import BaseAgent from ..agents.llm_agent import LlmAgent @@ -450,23 +451,68 @@ def _accepts_kwargs(func: Callable[..., Any], kwargs: dict[str, Any]) -> bool: return True +def _run_async_accepts(runner: Runner, name: str) -> bool: + """Whether runner.run_async takes the keyword ``name`` (or ``**kwargs``).""" + try: + params = inspect.signature(runner.run_async).parameters + except (ValueError, TypeError): + return True + return name in params or any( + p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values() + ) + + def _with_abort_signal_kwarg( runner: Runner, kwargs: dict[str, Any], abort_signal: asyncio.Event, ) -> dict[str, Any]: """Adds abort_signal to kwargs when runner.run_async accepts it.""" - try: - params = inspect.signature(runner.run_async).parameters - if "abort_signal" in params or any( - p.kind == inspect.Parameter.VAR_KEYWORD for p in params.values() - ): - kwargs["abort_signal"] = abort_signal - except (ValueError, TypeError): + if _run_async_accepts(runner, "abort_signal"): kwargs["abort_signal"] = abort_signal return kwargs +def _with_caller_principal_kwarg( + runner: Runner, + kwargs: dict[str, Any], + caller_principal: CallerPrincipal, +) -> dict[str, Any]: + """Adds caller_principal to kwargs when runner.run_async accepts it.""" + if _run_async_accepts(runner, "caller_principal"): + kwargs["caller_principal"] = caller_principal + return kwargs + + +_API_SERVER_PRINCIPAL_SOURCE = "api_server" + + +def _caller_principal_for_request( + request: Optional[Request], +) -> CallerPrincipal: + """Returns the caller identity to run a /run or /run_sse request as. + + The API server does not authenticate callers itself. A deployment that does, + in middleware or a dependency in front of these routes, states the result on + ``request.state.caller_principal`` and that value is used as is. Without + one, the request is recorded as unauthenticated from the ``api_server`` + edge, so a tool confirmation it carries is warned about (default) or refused + (strict mode), the same decision made for an unauthenticated A2A caller. + The ``user_id`` in the request body is a claim the caller made rather than + one the server checked, so it is not consulted here. + """ + principal = ( + getattr(request.state, "caller_principal", None) + if request is not None + else None + ) + if isinstance(principal, CallerPrincipal): + return principal + return CallerPrincipal( + authenticated=False, source=_API_SERVER_PRINCIPAL_SOURCE + ) + + _current_session_options: contextvars.ContextVar[Optional[dict[str, Any]]] = ( contextvars.ContextVar("current_session_options", default=None) ) @@ -2112,6 +2158,7 @@ async def run_agent(req: RunAgentRequest, request: Request) -> Response: ) abort_signal = asyncio.Event() + caller_principal = _caller_principal_for_request(request) async def worker() -> list[Event]: run_async_kwargs = _with_abort_signal_kwarg( @@ -2126,6 +2173,9 @@ async def worker() -> list[Event]: }, abort_signal, ) + run_async_kwargs = _with_caller_principal_kwarg( + runner, run_async_kwargs, caller_principal + ) try: async with Aclosing(runner.run_async(**run_async_kwargs)) as agen: return [public_event(event) async for event in agen] @@ -2175,7 +2225,9 @@ async def monitor() -> None: monitor_task.cancel() @app.post("/run_sse") - async def run_agent_sse(req: RunAgentRequest) -> StreamingResponse: + async def run_agent_sse( + req: RunAgentRequest, request: Request = None + ) -> StreamingResponse: app_name = req.app_name or self.default_app_name if not app_name: raise HTTPException( @@ -2224,6 +2276,8 @@ async def run_agent_sse(req: RunAgentRequest) -> StreamingResponse: detail=f"Session not found: {req.session_id}", ) + caller_principal = _caller_principal_for_request(request) + # Convert the events to properly formatted SSE async def event_generator() -> AsyncIterator[str]: is_closing = False @@ -2246,6 +2300,9 @@ async def _produce_events() -> None: }, abort_signal, ) + run_async_kwargs = _with_caller_principal_kwarg( + runner, run_async_kwargs, caller_principal + ) try: async with Aclosing(runner.run_async(**run_async_kwargs)) as agen: try: diff --git a/src/google/adk/features/_feature_registry.py b/src/google/adk/features/_feature_registry.py index 7ab9dec9b62..3b792eac1cd 100644 --- a/src/google/adk/features/_feature_registry.py +++ b/src/google/adk/features/_feature_registry.py @@ -79,6 +79,7 @@ class FeatureName(str, Enum): SPANNER_ADMIN_TOOLSET = "SPANNER_ADMIN_TOOLSET" SPANNER_TOOL_SETTINGS = "SPANNER_TOOL_SETTINGS" SPANNER_VECTOR_STORE = "SPANNER_VECTOR_STORE" + STRICT_CALLER_PRINCIPAL = "STRICT_CALLER_PRINCIPAL" TOOL_CONFIG = "TOOL_CONFIG" TOOL_CONFIRMATION = "TOOL_CONFIRMATION" PLUGGABLE_AUTH = "PLUGGABLE_AUTH" @@ -251,6 +252,14 @@ class FeatureConfig: FeatureName.SPANNER_VECTOR_STORE: FeatureConfig( FeatureStage.EXPERIMENTAL, default_on=True ), + # Refuse, rather than warn about, a human-in-the-loop tool confirmation + # from a caller the serving layer did not authenticate. Off by default so + # that upgrading cannot break a deployment that runs without an + # authenticator; flipping default_on is the whole of the planned change + # at the next major version. + FeatureName.STRICT_CALLER_PRINCIPAL: FeatureConfig( + FeatureStage.EXPERIMENTAL, default_on=False + ), FeatureName.TOOL_CONFIG: FeatureConfig( FeatureStage.EXPERIMENTAL, default_on=True ), diff --git a/src/google/adk/flows/llm_flows/tools/_confirmation.py b/src/google/adk/flows/llm_flows/tools/_confirmation.py index afb1cdab085..8256ca0021f 100644 --- a/src/google/adk/flows/llm_flows/tools/_confirmation.py +++ b/src/google/adk/flows/llm_flows/tools/_confirmation.py @@ -22,6 +22,7 @@ from google.genai import types from typing_extensions import override +from ....agents._caller_principal import caller_may_confirm from ....agents.invocation_context import InvocationContext from ....agents.readonly_context import ReadonlyContext from ....events.event import Event @@ -256,6 +257,36 @@ def _map_confirmation_to_original_fc_ids( return mapping +def _apply_caller_principal_gate( + invocation_context: InvocationContext, + confirmations_by_fc_id: dict[str, ToolConfirmation], +) -> dict[str, ToolConfirmation]: + """Decides whether the caller of this invocation may approve a tool call. + + The decision itself is ``caller_may_confirm``, shared with the workflow tool + and function nodes so that every consumer of a human-in-the-loop answer + refuses the same callers. A refusal rewrites each confirmation to + ``confirmed=False`` instead of dropping it: dropping it leaves the + ``adk_request_confirmation`` call pending with nothing left to resolve it, + which is what made an earlier attempt at this guard stall every + human-in-the-loop tool. + + Args: + invocation_context: Current invocation context. + confirmations_by_fc_id: Confirmations parsed from the last user event. + + Returns: + The confirmations to act on, with refused ones forced to + ``confirmed=False``. + """ + if caller_may_confirm(invocation_context.caller_principal): + return confirmations_by_fc_id + return { + confirmation_fc_id: confirmation.model_copy(update={"confirmed": False}) + for confirmation_fc_id, confirmation in confirmations_by_fc_id.items() + } + + class _RequestConfirmationLlmRequestProcessor(BaseLlmRequestProcessor): """Handles tool confirmation information to build the LLM request.""" @@ -354,6 +385,13 @@ async def run_async( invocation_context.end_invocation = True return + # An approval is only worth acting on if it came from the operator this + # agent answers to. Everything above this point establishes that a + # confirmation was sent; this establishes who sent it. + confirmations_by_fc_id = _apply_caller_principal_gate( + invocation_context, confirmations_by_fc_id + ) + # Resolve all canonical tools and build tools_dict. Deliberately after the # dedup above so a consumed confirmation does not force a toolset # resolution, which can be a remote call for e.g. MCP toolsets. diff --git a/src/google/adk/runners.py b/src/google/adk/runners.py index b4e24bd63ef..b4edb107ef9 100644 --- a/src/google/adk/runners.py +++ b/src/google/adk/runners.py @@ -36,6 +36,7 @@ from google.genai import types from typing_extensions import Self +from .agents._caller_principal import CallerPrincipal from .agents.base_agent import BaseAgent from .agents.context_cache_config import ContextCacheConfig from .agents.invocation_context import InvocationContext @@ -583,6 +584,7 @@ async def _run_node_async( node: BaseNode | None = None, session: Optional[Session] = None, abort_signal: Optional[asyncio.Event] = None, + caller_principal: Optional[CallerPrincipal] = None, ) -> AsyncGenerator[Event, None]: """Run a BaseNode through NodeRunner. @@ -603,6 +605,7 @@ async def _run_node_async( node=node, session=session, abort_signal=abort_signal, + caller_principal=caller_principal, ) ) as agen: async for event in agen: @@ -1133,6 +1136,7 @@ async def run_async( new_message: Optional[types.Content] = None, state_delta: Optional[dict[str, Any]] = None, run_config: Optional[RunConfig] = None, + caller_principal: Optional[CallerPrincipal] = None, yield_user_message: bool = False, abort_signal: Optional[asyncio.Event] = None, ) -> AsyncGenerator[Event, None]: @@ -1152,6 +1156,10 @@ async def run_async( new_message: A new message to append to the session. state_delta: Optional state changes to apply to the session. run_config: The run config for the agent. + caller_principal: Set by a serving layer to record whether it + authenticated the caller of this invocation, and as whom. Leave it + unset for in-process callers: no remote trust boundary is crossed, so + there is nothing to vouch for. yield_user_message: If True, yield the user message event before agent/node events. abort_signal: Optional asyncio.Event to cancel the invocation. @@ -1225,6 +1233,7 @@ async def run_async( node=agent_to_run, session=session, abort_signal=abort_signal, + caller_principal=caller_principal, ) ) as agen: async for event in agen: @@ -1246,6 +1255,7 @@ async def run_async( run_config=run_config, yield_user_message=yield_user_message, abort_signal=abort_signal, + caller_principal=caller_principal, ) ) as agen: async for event in agen: @@ -1299,6 +1309,7 @@ async def _run_with_trace( state_delta=state_delta, invocation_id=invocation_id, abort_signal=abort_signal, + caller_principal=caller_principal, ) else: invocation_id = self._resolve_invocation_id( @@ -1315,6 +1326,7 @@ async def _run_with_trace( run_config=run_config, state_delta=state_delta, abort_signal=abort_signal, + caller_principal=caller_principal, ) else: invocation_context = await self._setup_context_for_resumed_invocation( @@ -1324,6 +1336,7 @@ async def _run_with_trace( run_config=run_config, state_delta=state_delta, abort_signal=abort_signal, + caller_principal=caller_principal, ) active_agent = invocation_context.agent if not isinstance(active_agent, BaseAgent): @@ -1983,6 +1996,7 @@ async def _setup_context_for_new_invocation( state_delta: Optional[dict[str, Any]], invocation_id: Optional[str] = None, abort_signal: Optional[asyncio.Event] = None, + caller_principal: Optional[CallerPrincipal] = None, ) -> InvocationContext: """Sets up the context for a new invocation. @@ -1993,6 +2007,8 @@ async def _setup_context_for_new_invocation( state_delta: Optional state changes to apply to the session. invocation_id: Optional invocation identifier. abort_signal: Optional abort signal to cancel this invocation. + caller_principal: Optional caller identity established by a serving + layer. Returns: The invocation context for the new invocation. @@ -2003,6 +2019,7 @@ async def _setup_context_for_new_invocation( new_message=new_message, run_config=run_config, invocation_id=invocation_id, + caller_principal=caller_principal, ) if abort_signal is not None: # Attached here rather than passed to `_new_invocation_context`, whose @@ -2045,6 +2062,7 @@ async def _setup_context_for_resumed_invocation( run_config: RunConfig, state_delta: Optional[dict[str, Any]], abort_signal: Optional[asyncio.Event] = None, + caller_principal: Optional[CallerPrincipal] = None, ) -> InvocationContext: """Sets up the context for a resumed invocation. @@ -2055,6 +2073,8 @@ async def _setup_context_for_resumed_invocation( run_config: The run config of the agent. state_delta: Optional state changes to apply to the session. abort_signal: Optional abort signal to cancel this invocation. + caller_principal: Optional caller identity established by a serving + layer. Returns: The invocation context for the resumed invocation. @@ -2080,6 +2100,7 @@ async def _setup_context_for_resumed_invocation( new_message=user_message, run_config=run_config, invocation_id=invocation_id, + caller_principal=caller_principal, ) if abort_signal is not None: # See `_setup_context_for_new_invocation` for why this is attached @@ -2134,6 +2155,7 @@ def _new_invocation_context( new_message: Optional[types.Content] = None, live_request_queue: Optional[LiveRequestQueue] = None, run_config: Optional[RunConfig] = None, + caller_principal: Optional[CallerPrincipal] = None, ) -> InvocationContext: """Creates a new invocation context. @@ -2149,6 +2171,8 @@ def _new_invocation_context( new_message: The new message for the context. live_request_queue: The live request queue for the context. run_config: The run config for the context. + caller_principal: The caller identity a serving layer established for + this invocation, or None when no serving layer was involved. Returns: The new invocation context. @@ -2183,6 +2207,7 @@ def _new_invocation_context( user_content=new_message, live_request_queue=live_request_queue, run_config=run_config, + caller_principal=caller_principal, resumability_config=self.resumability_config, ) diff --git a/src/google/adk/workflow/_node_runner_utils.py b/src/google/adk/workflow/_node_runner_utils.py index 97138aec682..3077ffeec76 100644 --- a/src/google/adk/workflow/_node_runner_utils.py +++ b/src/google/adk/workflow/_node_runner_utils.py @@ -41,6 +41,7 @@ from ._workflow import _LoopState if TYPE_CHECKING: + from ..agents._caller_principal import CallerPrincipal from ..runners import Runner from ._base_node import BaseNode @@ -60,6 +61,7 @@ async def run_node_async( node: BaseNode | None = None, session: Optional[Session] = None, abort_signal: Optional[asyncio.Event] = None, + caller_principal: Optional[CallerPrincipal] = None, ) -> AsyncGenerator[Event, None]: """Runs a BaseNode or Workflow in async mode.""" from ..runners import _CALLER_CLOSED_EARLY_MSG @@ -115,6 +117,12 @@ async def _run() -> AsyncGenerator[Event, None]: # signature subclasses override. Safe at this point because nothing has # derived a sub-context from this one yet. ic._attach_abort_signal(abort_signal) # pylint: disable=protected-access + if caller_principal is not None: + # Set here rather than passed to `_new_invocation_context`, for the + # same reason as the abort signal. The serving layer established this + # principal from the request it authenticated; nothing on this path + # derives one from message content or event authorship. + ic.caller_principal = caller_principal if node and node is not runner.agent: ic.agent = node runner._restore_branch_from_history( # pylint: disable=protected-access diff --git a/src/google/adk/workflow/utils/_workflow_hitl_utils.py b/src/google/adk/workflow/utils/_workflow_hitl_utils.py index 2dc7f8b4059..9256a8f7a23 100644 --- a/src/google/adk/workflow/utils/_workflow_hitl_utils.py +++ b/src/google/adk/workflow/utils/_workflow_hitl_utils.py @@ -22,6 +22,7 @@ from google.genai import types +from ...agents._caller_principal import caller_may_confirm from ...auth.auth_credential import AuthCredential from ...auth.auth_credential import AuthCredentialTypes as _AuthCredentialTypes from ...auth.auth_credential import OAuth2Auth @@ -74,11 +75,22 @@ def parse_tool_confirmation(response: Any) -> ToolConfirmation: def apply_confirmation_resume(ctx: Context) -> bool: - """Applies resumed tool confirmation input onto ctx if present.""" + """Applies resumed tool confirmation input onto ctx if present. + + An answer from a caller the serving layer could not authenticate is stored + as a rejection, the same decision the agent pipeline makes for a + confirmation carried in a user event. It is made here, before the answer is + stored on the context, because tools such as `ExecuteBashTool` read + `tool_confirmation` themselves rather than through + `apply_confirmation_gate`. + """ response = ctx.resume_inputs.get(confirmation_interrupt_id(ctx.node_path)) if response is None: return False - ctx.tool_confirmation = parse_tool_confirmation(response) + confirmation = parse_tool_confirmation(response) + if not caller_may_confirm(ctx.get_invocation_context().caller_principal): + confirmation = confirmation.model_copy(update={'confirmed': False}) + ctx.tool_confirmation = confirmation return True diff --git a/tests/unittests/a2a/converters/test_request_converter.py b/tests/unittests/a2a/converters/test_request_converter.py index 7e8dccf1dfe..5cdcff5f3a8 100644 --- a/tests/unittests/a2a/converters/test_request_converter.py +++ b/tests/unittests/a2a/converters/test_request_converter.py @@ -20,6 +20,7 @@ from google.adk.a2a import _compat from google.adk.a2a.converters.request_converter import _get_user_id from google.adk.a2a.converters.request_converter import _warn_unauthenticated_user_name_once +from google.adk.a2a.converters.request_converter import build_caller_principal from google.adk.a2a.converters.request_converter import convert_a2a_request_to_agent_run_request from google.adk.runners import RunConfig from google.genai import types as genai_types @@ -537,3 +538,112 @@ def test_end_to_end_conversion_with_fallback_user(self): assert result.new_message.role == "user" assert result.new_message.parts == [mock_genai_part] assert isinstance(result.run_config, RunConfig) + + +class TestBuildCallerPrincipal: + """Test cases for build_caller_principal.""" + + def _request_with_user(self, user): + """Builds a RequestContext whose call context carries the given user.""" + mock_call_context = Mock() + mock_call_context.user = user + request = Mock(spec=RequestContext) + request.call_context = mock_call_context + request.context_id = "test_context" + return request + + def test_authenticated_when_call_context_names_the_user(self): + """The serving layer verified the caller, so the principal says so.""" + mock_user = Mock() + mock_user.user_name = "authenticated_user" + + principal = build_caller_principal(self._request_with_user(mock_user)) + + assert principal.authenticated + assert principal.user_name == "authenticated_user" + assert principal.source == "a2a" + + def test_unauthenticated_without_call_context(self): + """No call context means no authenticator ran.""" + request = Mock(spec=RequestContext) + request.call_context = None + request.context_id = "test_context" + + principal = build_caller_principal(request) + + assert not principal.authenticated + assert principal.user_name is None + assert principal.source == "a2a" + + def test_unauthenticated_when_call_context_has_no_user(self): + """A call context without a user vouches for nobody.""" + principal = build_caller_principal(self._request_with_user(None)) + + assert not principal.authenticated + assert principal.user_name is None + + def test_unauthenticated_with_empty_user_name(self): + """An a2a UnauthenticatedUser reports an empty name.""" + mock_user = Mock() + mock_user.user_name = "" + + principal = build_caller_principal(self._request_with_user(mock_user)) + + assert not principal.authenticated + assert principal.user_name is None + + def test_unauthenticated_when_the_user_says_so_despite_a_name(self): + """is_authenticated False wins over a populated name.""" + mock_user = Mock() + mock_user.user_name = "looks_real" + mock_user.is_authenticated = False + + principal = build_caller_principal(self._request_with_user(mock_user)) + + assert not principal.authenticated + assert principal.user_name is None + + @pytest.mark.parametrize( + "user_name, expect_authenticated", + [("real_user", True), ("", False), (None, False)], + ) + def test_principal_never_disagrees_with_get_user_id( + self, user_name, expect_authenticated + ): + """The principal is authenticated exactly when _get_user_id trusts the name. + + Both functions read the same call context. Deriving them from the same + condition is what keeps them from drifting apart, which is the failure + mode behind the original bug: an authentication fact computed and then + thrown away. + """ + mock_user = Mock() + mock_user.user_name = user_name + request = self._request_with_user(mock_user) + + principal = build_caller_principal(request) + used_call_context_name = _get_user_id(request) == user_name + + assert principal.authenticated is expect_authenticated + assert principal.authenticated is used_call_context_name + + def test_conversion_attaches_the_principal(self): + """The converter must put the principal on the AgentRunRequest.""" + mock_message = Mock() + mock_message.parts = [] + mock_user = Mock() + mock_user.user_name = "authenticated_user" + mock_call_context = Mock() + mock_call_context.user = mock_user + + request = Mock(spec=RequestContext) + request.message = mock_message + request.context_id = "test_context" + request.call_context = mock_call_context + request.metadata = None + + result = convert_a2a_request_to_agent_run_request(request, Mock()) + + assert result.caller_principal is not None + assert result.caller_principal.authenticated + assert result.caller_principal.user_name == "authenticated_user" diff --git a/tests/unittests/cli/test_fast_api.py b/tests/unittests/cli/test_fast_api.py index d754a35fa8c..299bfb834bd 100755 --- a/tests/unittests/cli/test_fast_api.py +++ b/tests/unittests/cli/test_fast_api.py @@ -30,6 +30,7 @@ from fastapi import HTTPException from fastapi.testclient import TestClient from google.adk.a2a import _compat +from google.adk.agents._caller_principal import CallerPrincipal from google.adk.agents.base_agent import BaseAgent from google.adk.agents.llm_agent import LlmAgent from google.adk.agents.run_config import RunConfig @@ -64,7 +65,9 @@ from google.adk.sessions.base_session_service import ListSessionsResponse from google.adk.sessions.in_memory_session_service import InMemorySessionService from google.adk.sessions.session import Session +from google.adk.tools.function_tool import FunctionTool from google.adk.tools.tool_confirmation import ToolConfirmation +from google.adk.utils._function_call_names import REQUEST_CONFIRMATION_FUNCTION_CALL_NAME from google.api_core.exceptions import GoogleAPICallError from google.api_core.exceptions import InvalidArgument from google.genai import types @@ -74,6 +77,8 @@ import starlette.requests from starlette.routing import Mount +from tests.unittests import testing_utils + # Configure logging to help diagnose server startup issues logging.basicConfig( level=logging.INFO, @@ -2643,6 +2648,319 @@ async def run_async_capture( assert captured["run_config"].max_llm_calls == 37 +class _CallerPrincipalInjector: + """ASGI wrapper standing in for a deployment's own authentication. + + Sets ``request.state.caller_principal`` the way an auth middleware would, + before the request reaches the ADK routes. + """ + + def __init__(self, app, principal: CallerPrincipal): + self._app = app + self._principal = principal + + async def __call__(self, scope, receive, send): + if scope["type"] == "http": + scope.setdefault("state", {})["caller_principal"] = self._principal + await self._app(scope, receive, send) + + +def _capture_caller_principal(monkeypatch) -> dict[str, Any]: + """Patches Runner.run_async to record the caller_principal it is given.""" + captured: dict[str, Any] = {"calls": 0, "caller_principal": "unset"} + + async def run_async_capture( + self, + *, + user_id: str, + session_id: str, + invocation_id: Optional[str] = None, + new_message: Optional[types.Content] = None, + state_delta: Optional[dict[str, Any]] = None, + run_config: Optional[RunConfig] = None, + caller_principal: Optional[CallerPrincipal] = None, + **kwargs, + ): + del self, user_id, session_id, invocation_id, new_message, state_delta + del run_config, kwargs + captured["calls"] += 1 + captured["caller_principal"] = caller_principal + yield _event_1() + + monkeypatch.setattr(Runner, "run_async", run_async_capture) + return captured + + +@pytest.mark.parametrize("route", ["/run", "/run_sse"]) +def test_agent_run_records_unauthenticated_caller_by_default( + test_app, create_test_session, monkeypatch, route +): + """Without authentication in front of the server, a run is unauthenticated. + + The user_id in the body is a claim the caller made, so it must not turn + into an authenticated principal. + """ + info = create_test_session + captured = _capture_caller_principal(monkeypatch) + + payload = { + "app_name": info["app_name"], + "user_id": info["user_id"], + "session_id": info["session_id"], + "new_message": {"role": "user", "parts": [{"text": "Hello"}]}, + "streaming": False, + } + + response = test_app.post(route, json=payload) + + assert response.status_code == 200 + assert captured["calls"] == 1 + assert captured["caller_principal"] == CallerPrincipal( + authenticated=False, source="api_server" + ) + + +@pytest.mark.parametrize("route", ["/run", "/run_sse"]) +def test_agent_run_uses_caller_principal_set_by_middleware( + test_app, create_test_session, monkeypatch, route +): + """A principal the deployment's own authentication set is used as is.""" + info = create_test_session + captured = _capture_caller_principal(monkeypatch) + principal = CallerPrincipal( + authenticated=True, user_name="operator", source="gateway" + ) + client = TestClient(_CallerPrincipalInjector(test_app.app, principal)) + + payload = { + "app_name": info["app_name"], + "user_id": info["user_id"], + "session_id": info["session_id"], + "new_message": {"role": "user", "parts": [{"text": "Hello"}]}, + "streaming": False, + } + + response = client.post(route, json=payload) + + assert response.status_code == 200 + assert captured["caller_principal"] == principal + + +def test_agent_run_omits_caller_principal_for_runner_without_the_parameter( + test_app, create_test_session, monkeypatch +): + """A Runner subclass whose run_async predates the parameter still works.""" + info = create_test_session + captured: dict[str, Any] = {"kwargs": None} + + async def run_async_capture( + self, + *, + user_id: str, + session_id: str, + invocation_id: Optional[str] = None, + new_message: Optional[types.Content] = None, + state_delta: Optional[dict[str, Any]] = None, + run_config: Optional[RunConfig] = None, + ): + del self, user_id, session_id, new_message, state_delta, run_config + captured["kwargs"] = {"invocation_id": invocation_id} + yield _event_1() + + monkeypatch.setattr(Runner, "run_async", run_async_capture) + + payload = { + "app_name": info["app_name"], + "user_id": info["user_id"], + "session_id": info["session_id"], + "new_message": {"role": "user", "parts": [{"text": "Hello"}]}, + "streaming": False, + } + + response = test_app.post("/run", json=payload) + + assert response.status_code == 200 + assert captured["kwargs"] == {"invocation_id": None} + + +def _confirmation_gated_app(info, calls: list[str]) -> App: + """An App whose agent calls a confirmation-gated tool on the first turn.""" + + def rm_all() -> dict: + calls.append("rm_all") + return {"status": "deleted"} + + model = testing_utils.MockModel.create( + responses=[ + types.Part.from_function_call(name="rm_all", args={}), + "after tool", + "after tool 2", + ] + ) + return App( + name=info["app_name"], + root_agent=LlmAgent( + name="ops_agent", + model=model, + tools=[FunctionTool(func=rm_all, require_confirmation=True)], + ), + ) + + +def _run_payload(info, parts): + return { + "app_name": info["app_name"], + "user_id": info["user_id"], + "session_id": info["session_id"], + "new_message": {"role": "user", "parts": parts}, + "streaming": False, + } + + +def _approve_through_run(client, info) -> list[dict]: + """Triggers the gated tool over /run, then answers its confirmation.""" + first = client.post("/run", json=_run_payload(info, [{"text": "delete"}])) + assert first.status_code == 200 + confirmation_id = next( + part["functionCall"]["id"] + for event in first.json() + for part in (event.get("content") or {}).get("parts", []) + if part.get("functionCall", {}).get("name") + == REQUEST_CONFIRMATION_FUNCTION_CALL_NAME + ) + second = client.post( + "/run", + json=_run_payload( + info, + [{ + "functionResponse": { + "id": confirmation_id, + "name": REQUEST_CONFIRMATION_FUNCTION_CALL_NAME, + "response": {"confirmed": True}, + } + }], + ), + ) + assert second.status_code == 200 + return [ + part["functionResponse"]["response"] + for event in second.json() + for part in (event.get("content") or {}).get("parts", []) + if part.get("functionResponse", {}).get("name") == "rm_all" + ] + + +def test_agent_run_strict_mode_refuses_confirmation_without_authentication( + create_test_session, + mock_session_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, + monkeypatch, +): + """A /run caller cannot approve its own gated tool call in strict mode.""" + info = create_test_session + monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1") + monkeypatch.setattr(Runner, "run_async", _ORIGINAL_RUNNER_RUN_ASYNC) + calls: list[str] = [] + loaded_app = _confirmation_gated_app(info, calls) + monkeypatch.setattr( + mock_agent_loader, "load_agent", lambda app_name: loaded_app + ) + client = _create_test_client( + mock_session_service, + InMemoryArtifactService(), + InMemoryMemoryService(), + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, + ) + + results = _approve_through_run(client, info) + + assert calls == [] + assert results == [{"error": "This tool call is rejected."}] + + +def test_agent_run_strict_mode_honors_confirmation_from_authenticated_operator( + create_test_session, + mock_session_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, + monkeypatch, +): + """A principal set by the deployment's authentication is honored.""" + info = create_test_session + monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1") + monkeypatch.setattr(Runner, "run_async", _ORIGINAL_RUNNER_RUN_ASYNC) + calls: list[str] = [] + loaded_app = _confirmation_gated_app(info, calls) + monkeypatch.setattr( + mock_agent_loader, "load_agent", lambda app_name: loaded_app + ) + base = _create_test_client( + mock_session_service, + InMemoryArtifactService(), + InMemoryMemoryService(), + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, + ) + client = TestClient( + _CallerPrincipalInjector( + base.app, + CallerPrincipal( + authenticated=True, user_name="operator", source="gateway" + ), + ) + ) + + results = _approve_through_run(client, info) + + assert calls == ["rm_all"] + assert results == [{"status": "deleted"}] + + +def test_agent_run_default_mode_honors_confirmation_with_a_warning( + create_test_session, + mock_session_service, + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, + monkeypatch, + caplog, +): + """Without strict mode the behavior is unchanged apart from the warning.""" + info = create_test_session + monkeypatch.delenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", raising=False) + monkeypatch.setattr(Runner, "run_async", _ORIGINAL_RUNNER_RUN_ASYNC) + calls: list[str] = [] + loaded_app = _confirmation_gated_app(info, calls) + monkeypatch.setattr( + mock_agent_loader, "load_agent", lambda app_name: loaded_app + ) + client = _create_test_client( + mock_session_service, + InMemoryArtifactService(), + InMemoryMemoryService(), + mock_agent_loader, + mock_eval_sets_manager, + mock_eval_set_results_manager, + ) + + with caplog.at_level("WARNING", logger="google_adk"): + results = _approve_through_run(client, info) + + assert calls == ["rm_all"] + assert results == [{"status": "deleted"}] + assert any( + "unauthenticated caller" in record.getMessage() + and "api_server" in record.getMessage() + for record in caplog.records + ) + + def test_agent_run_sse_splits_artifact_delta( test_app, create_test_session, monkeypatch ): diff --git a/tests/unittests/flows/llm_flows/tools/test_confirmation.py b/tests/unittests/flows/llm_flows/tools/test_confirmation.py index 16dd3f714a4..fad3e132022 100644 --- a/tests/unittests/flows/llm_flows/tools/test_confirmation.py +++ b/tests/unittests/flows/llm_flows/tools/test_confirmation.py @@ -15,6 +15,7 @@ from unittest.mock import create_autospec from unittest.mock import patch +from google.adk.agents._caller_principal import CallerPrincipal from google.adk.agents.llm_agent import LlmAgent from google.adk.events.event import Event from google.adk.events.event_actions import EventActions @@ -1522,3 +1523,221 @@ async def test_request_confirmation_processor_skips_events_authored_by_other_age # Current agent should skip the other agent's confirmation request without error. assert not events + + +def _build_pending_confirmation_events(agent_name: str) -> list[Event]: + """Builds a gated tool call followed by an approval on the user turn. + + Args: + agent_name: Author to use for the agent-authored events. + + Returns: + The session events, in order. + """ + 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") + confirmation_request = types.FunctionCall( + name=functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME, + args={ + "originalFunctionCall": original_function_call.model_dump( + exclude_none=True, by_alias=True + ), + "toolConfirmation": tool_confirmation.model_dump( + by_alias=True, exclude_none=True + ), + }, + id=MOCK_CONFIRMATION_FUNCTION_CALL_ID, + ) + approval = types.FunctionResponse( + name=functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME, + id=MOCK_CONFIRMATION_FUNCTION_CALL_ID, + response={"response": ToolConfirmation(confirmed=True).model_dump_json()}, + ) + return [ + Event( + author=agent_name, + content=types.Content( + parts=[types.Part(function_call=original_function_call)] + ), + ), + Event( + author=agent_name, + content=types.Content( + parts=[types.Part(function_call=confirmation_request)] + ), + ), + Event( + author="user", + content=types.Content(parts=[types.Part(function_response=approval)]), + ), + ] + + +async def _run_with_caller_principal(caller_principal): + """Runs the processor over a pending approval under one caller principal. + + Args: + caller_principal: The principal to put on the invocation context. + + Returns: + A tuple of the yielded events and the ToolConfirmation the processor + handed to tool execution, or None for the confirmation if execution was + never reached. + """ + agent = LlmAgent( + name="test_agent", + tools=[FunctionTool(mock_tool, require_confirmation=True)], + ) + invocation_context = await testing_utils.create_invocation_context( + agent=agent + ) + invocation_context.caller_principal = caller_principal + invocation_context.session.events.extend( + _build_pending_confirmation_events(agent.name) + ) + + resolved_event = Event( + author="agent", + content=types.Content( + parts=[ + types.Part( + function_response=types.FunctionResponse( + name=MOCK_TOOL_NAME, + id=MOCK_FUNCTION_CALL_ID, + response={"result": "Mock tool result with test"}, + ) + ) + ] + ), + ) + + with patch( + "google.adk.flows.llm_flows.functions.handle_function_call_list_async" + ) as mock_handle_function_call_list_async: + mock_handle_function_call_list_async.return_value = resolved_event + + events = [] + async for event in request_processor.run_async( + invocation_context, LlmRequest() + ): + events.append(event) + + if not mock_handle_function_call_list_async.call_args: + return events, None + args, _ = mock_handle_function_call_list_async.call_args + return events, args[4][MOCK_FUNCTION_CALL_ID] + + +@pytest.mark.asyncio +async def test_confirmation_honored_without_caller_principal(): + """No principal means no remote boundary was crossed, so nothing changes.""" + events, confirmation = await _run_with_caller_principal(None) + + assert len(events) == 1 + assert confirmation is not None + assert confirmation.confirmed + + +@pytest.mark.asyncio +async def test_confirmation_honored_for_authenticated_caller(): + """An authenticated remote caller keeps its approval. + + This is the half a transport-keyed guard gets wrong in the permissive + direction being wrong the other way: arriving over A2A is not by itself a + reason to refuse. + """ + events, confirmation = await _run_with_caller_principal( + CallerPrincipal(authenticated=True, user_name="alice", source="a2a") + ) + + assert len(events) == 1 + assert confirmation is not None + assert confirmation.confirmed + + +@pytest.mark.asyncio +async def test_confirmation_honored_for_unauthenticated_caller_by_default(): + """Default is warn, not refuse. + + Upgrading must not break a deployment that runs its A2A server without an + authenticator. Those operators get a log line; the behavior is unchanged + until they opt in. + """ + events, confirmation = await _run_with_caller_principal( + CallerPrincipal(authenticated=False, source="a2a") + ) + + assert len(events) == 1 + assert confirmation is not None + assert confirmation.confirmed + + +@pytest.mark.asyncio +async def test_confirmation_refused_for_unauthenticated_caller_when_strict( + monkeypatch, +): + """Strict mode refuses, and the refusal is explicit rather than a stall. + + The call still reaches tool execution, carrying confirmed=False. That is the + state a human decline produces, so the pending adk_request_confirmation call + resolves with a rejection the caller can see. Dropping the confirmation + instead is what made the earlier transport-keyed guard hang every + human-in-the-loop tool. + """ + monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1") + + events, confirmation = await _run_with_caller_principal( + CallerPrincipal(authenticated=False, source="a2a") + ) + + assert len(events) == 1 + assert confirmation is not None + assert not confirmation.confirmed + + +@pytest.mark.asyncio +async def test_strict_mode_does_not_touch_in_process_callers(monkeypatch): + """Strict mode must not refuse an invocation that crossed no boundary.""" + monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1") + + events, confirmation = await _run_with_caller_principal(None) + + assert len(events) == 1 + assert confirmation is not None + assert confirmation.confirmed + + +@pytest.mark.asyncio +async def test_strict_mode_keeps_multi_turn_reentry_a_noop(monkeypatch): + """Re-entry within a turn stays a no-op, principal or not. + + The processor re-runs on every LLM step and the approval stays the last user + event for the rest of the turn. The gate deliberately runs after the + consumed-confirmation dedup, so a second pass over an approval that was + already acted on yields nothing rather than refusing or re-executing it. + This is the regression that got the previous guard reverted. + """ + monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1") + agent = LlmAgent( + name="test_agent", + tools=[FunctionTool(mock_tool, require_confirmation=False)], + ) + invocation_context = await testing_utils.create_invocation_context( + agent=agent + ) + invocation_context.caller_principal = CallerPrincipal( + authenticated=False, source="a2a" + ) + invocation_context.session.events.extend( + _build_consumed_dynamic_confirmation_events(agent.name) + ) + + events = [] + async for event in request_processor.run_async( + invocation_context, LlmRequest() + ): + events.append(event) + + assert not events diff --git a/tests/unittests/test_runners.py b/tests/unittests/test_runners.py index 7d6427b2490..3c8b7413512 100644 --- a/tests/unittests/test_runners.py +++ b/tests/unittests/test_runners.py @@ -29,6 +29,7 @@ from unittest.mock import patch from google.adk import runners +from google.adk.agents._caller_principal import CallerPrincipal from google.adk.agents.base_agent import BaseAgent from google.adk.agents.context import Context from google.adk.agents.context_cache_config import ContextCacheConfig @@ -6747,3 +6748,42 @@ async def test_run_async_tool_error_in_root_parallel_agent_keeps_each_author(): if __name__ == "__main__": pytest.main([__file__]) + + +@pytest.mark.asyncio +async def test_runner_root_llm_agent_carries_caller_principal_on_node_path(): + """run_async(caller_principal=...) reaches a root LlmAgent's invocation context. + + A root LlmAgent runs through the node runtime, which builds its own + invocation context. The principal a serving layer set has to survive that + hop, or the confirmation gate downstream never sees it. + """ + seen = [] + + def capture(callback_context): + seen.append(callback_context._invocation_context.caller_principal) + return None + + session_service = InMemorySessionService() + agent = LlmAgent( + name="chat_agent", + model=testing_utils.MockModel.create(responses=["hi"]), + before_agent_callback=capture, + ) + runner = Runner( + app_name=TEST_APP_ID, agent=agent, session_service=session_service + ) + await session_service.create_session( + app_name=TEST_APP_ID, user_id=TEST_USER_ID, session_id=TEST_SESSION_ID + ) + principal = CallerPrincipal(authenticated=False, source="test") + + async for _ in runner.run_async( + user_id=TEST_USER_ID, + session_id=TEST_SESSION_ID, + new_message=types.Content(role="user", parts=[types.Part(text="hello")]), + caller_principal=principal, + ): + pass + + assert seen == [principal] diff --git a/tests/unittests/workflow/test_tool_node.py b/tests/unittests/workflow/test_tool_node.py index bef64d730df..f7d4595d348 100644 --- a/tests/unittests/workflow/test_tool_node.py +++ b/tests/unittests/workflow/test_tool_node.py @@ -19,6 +19,7 @@ import re from typing import Any +from google.adk.agents._caller_principal import CallerPrincipal from google.adk.agents.context import Context from google.adk.apps.app import ResumabilityConfig from google.adk.events._interrupts import extract_event_interrupt_ids @@ -693,15 +694,36 @@ async def start(self) -> Event | None: events, REQUEST_INPUT_FUNCTION_CALL_NAME ) - async def answer(self, request: Event, response: dict[str, Any]) -> None: - """Resumes the workflow with the user's answer to the request.""" + async def answer( + self, + request: Event, + response: dict[str, Any], + *, + caller_principal: CallerPrincipal | None = None, + ) -> None: + """Resumes the workflow with the user's answer to the request. + + `caller_principal` stands in for what a serving layer would set on the + invocation that carries the answer; None is an in-process caller. + """ [interrupt_id] = extract_event_interrupt_ids(request) - await self.runner.run_async( - new_message=testing_utils.UserContent( - create_request_input_response(interrupt_id, response) - ), - invocation_id=request.invocation_id, + new_message = testing_utils.UserContent( + create_request_input_response(interrupt_id, response) ) + if caller_principal is None: + await self.runner.run_async( + new_message=new_message, invocation_id=request.invocation_id + ) + return + session = self.runner.session + async for _ in self.runner.runner.run_async( + user_id=session.user_id, + session_id=session.id, + invocation_id=request.invocation_id, + new_message=new_message, + caller_principal=caller_principal, + ): + pass @pytest.mark.parametrize("resumable", [False, True]) @@ -776,6 +798,74 @@ async def before_tool_callback(self, *, tool, tool_args, tool_context): assert wf.tool_calls == ["prod"] +@pytest.mark.parametrize("resumable", [False, True]) +@pytest.mark.asyncio +async def test_tool_node_confirmation_from_unauthenticated_caller_refused_in_strict_mode( + resumable: bool, monkeypatch +): + """An approval the serving layer could not attribute is stored as a rejection. + + Same decision as the agent confirmation flow: with strict mode on, a resume + that arrives with an unauthenticated principal must not run the tool, and + the node answers with the rejection response rather than hanging. + """ + monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1") + wf = _ConfirmationWorkflow(resumable=resumable) + request = await wf.start() + + await wf.answer( + request, + {"confirmed": True}, + caller_principal=CallerPrincipal(authenticated=False, source="a2a"), + ) + + assert not wf.tool_calls + assert wf.seen_downstream == [{"error": "This tool call is rejected."}] + + +@pytest.mark.asyncio +async def test_tool_node_confirmation_from_authenticated_caller_runs_tool( + monkeypatch, +): + """An approval from a caller the serving layer verified is honored.""" + monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1") + wf = _ConfirmationWorkflow(resumable=False) + request = await wf.start() + + await wf.answer( + request, + {"confirmed": True}, + caller_principal=CallerPrincipal( + authenticated=True, user_name="operator", source="a2a" + ), + ) + + assert wf.tool_calls == ["prod"] + assert wf.seen_downstream == [{"deleted": "prod"}] + + +@pytest.mark.asyncio +async def test_tool_node_confirmation_from_unauthenticated_caller_warns_by_default( + caplog, +): + """Without strict mode the approval is honored and logged, as in an agent.""" + wf = _ConfirmationWorkflow(resumable=False) + request = await wf.start() + + with caplog.at_level("WARNING", logger="google_adk"): + await wf.answer( + request, + {"confirmed": True}, + caller_principal=CallerPrincipal(authenticated=False, source="a2a"), + ) + + assert wf.tool_calls == ["prod"] + assert any( + "unauthenticated caller" in record.getMessage() + for record in caplog.records + ) + + @pytest.mark.asyncio async def test_tool_node_confirmation_not_required_for_args_runs_directly(): """Tests that a confirmation predicate returning False does not pause.""" @@ -885,6 +975,31 @@ async def test_tool_node_requested_confirmation_rejected_lets_tool_decide( assert wf.seen_downstream == [{"status": "declined"}] +@pytest.mark.parametrize("resumable", [False, True]) +@pytest.mark.asyncio +async def test_tool_node_requested_confirmation_from_unauthenticated_caller_refused_in_strict_mode( + resumable: bool, monkeypatch +): + """A tool that reads `tool_confirmation` itself sees the refusal too. + + `apply_confirmation_gate` never runs for a tool whose + `check_require_confirmation` is False, so the principal decision has to be + applied to the stored answer, not only to the gate. + """ + monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1") + wf = _RequestedConfirmationWorkflow(resumable=resumable) + request = await wf.start() + + await wf.answer( + request, + {"confirmed": True, "payload": {"limit": 10}}, + caller_principal=CallerPrincipal(authenticated=False, source="a2a"), + ) + + assert not wf.tool_calls + assert wf.seen_downstream == [{"status": "declined"}] + + def _make_runner( wf: Workflow, *, resumable: bool ) -> testing_utils.InMemoryRunner: diff --git a/tests/unittests/workflow/test_workflow_node.py b/tests/unittests/workflow/test_workflow_node.py index b59709c02c4..a6d28b9436a 100644 --- a/tests/unittests/workflow/test_workflow_node.py +++ b/tests/unittests/workflow/test_workflow_node.py @@ -18,10 +18,12 @@ from unittest import mock +from google.adk.agents._caller_principal import CallerPrincipal from google.adk.agents.base_agent import BaseAgent from google.adk.agents.llm_agent import LlmAgent from google.adk.apps import App from google.adk.events._interrupts import extract_event_interrupt_ids +from google.adk.events.event import Event from google.adk.runners import Runner from google.adk.sessions.in_memory_session_service import InMemorySessionService from google.adk.tools.base_tool import BaseTool @@ -576,6 +578,124 @@ def after_node(node_input): ) +class _ConfirmedFunctionNodeWorkflow: + """A Workflow with one `@node(require_confirmation=True)` node. + + Mirrors `_ConfirmationWorkflow` in test_tool_node.py for the FunctionNode + confirmation path, which stores the resumed answer the same way. + """ + + def __init__(self) -> None: + from google.adk.apps.app import ResumabilityConfig + + self.calls: list[str] = [] + self.downstream_seen: list[object] = [] + + class RemoveResult(BaseModel): + removed: bool + + @node(require_confirmation=True) + def remove_file() -> RemoveResult: + self.calls.append("removed") + return RemoveResult(removed=True) + + def after_node(node_input): + self.downstream_seen.append(node_input) + return node_input + + wf = Workflow( + name="confirm_wf", + edges=[(START, remove_file), (remove_file, after_node)], + ) + self.runner = testing_utils.InMemoryRunner( + app=App( + name="test_confirm_wf", + root_agent=wf, + resumability_config=ResumabilityConfig(is_resumable=True), + ) + ) + + async def start(self) -> Event: + """Runs the workflow and returns its confirmation request.""" + from google.adk.utils._function_call_names import REQUEST_INPUT_FUNCTION_CALL_NAME + + from . import workflow_testing_utils + + events = await self.runner.run_async("start") + request = workflow_testing_utils.find_function_call_event( + events, REQUEST_INPUT_FUNCTION_CALL_NAME + ) + assert request is not None + assert not self.calls + return request + + async def answer( + self, request: Event, *, caller_principal: CallerPrincipal + ) -> None: + """Approves the request on an invocation a serving layer attributed. + + `caller_principal` stands in for what the serving layer would set on the + invocation that carries the answer. + """ + from google.adk.workflow.utils._workflow_hitl_utils import create_request_input_response + + [interrupt_id] = extract_event_interrupt_ids(request) + session = self.runner.session + async for _ in self.runner.runner.run_async( + user_id=session.user_id, + session_id=session.id, + invocation_id=request.invocation_id, + new_message=testing_utils.UserContent( + create_request_input_response(interrupt_id, {"confirmed": True}) + ), + caller_principal=caller_principal, + ): + pass + + +@pytest.mark.asyncio +async def test_node_require_confirmation_from_unauthenticated_caller_refused_in_strict_mode( + monkeypatch, +): + """An approval the serving layer could not attribute is stored as a rejection. + + Same decision as the agent confirmation flow and the tool node: with strict + mode on, a resume that arrives with an unauthenticated principal must not + run the function, and the node answers with the rejection output. + """ + monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1") + wf = _ConfirmedFunctionNodeWorkflow() + request = await wf.start() + + await wf.answer( + request, + caller_principal=CallerPrincipal(authenticated=False, source="a2a"), + ) + + assert not wf.calls + assert wf.downstream_seen == [{"error": "This tool call is rejected."}] + + +@pytest.mark.asyncio +async def test_node_require_confirmation_from_authenticated_caller_runs_function( + monkeypatch, +): + """An approval from a caller the serving layer verified is honored.""" + monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1") + wf = _ConfirmedFunctionNodeWorkflow() + request = await wf.start() + + await wf.answer( + request, + caller_principal=CallerPrincipal( + authenticated=True, user_name="operator", source="a2a" + ), + ) + + assert wf.calls == ["removed"] + assert wf.downstream_seen == [{"removed": True}] + + @pytest.mark.parametrize( "target_path,should_pause", [("/dev/tmp", False), ("/prod/db", True)] ) diff --git a/tests/unittests/workflow/utils/test_workflow_hitl_utils.py b/tests/unittests/workflow/utils/test_workflow_hitl_utils.py index 01f64ab2344..f63b6bec8ee 100644 --- a/tests/unittests/workflow/utils/test_workflow_hitl_utils.py +++ b/tests/unittests/workflow/utils/test_workflow_hitl_utils.py @@ -21,6 +21,7 @@ from authlib.oauth2.rfc6749 import OAuth2Token from authlib.oauth2.rfc7636 import create_s256_code_challenge +from google.adk.agents._caller_principal import CallerPrincipal from google.adk.auth.auth_handler import AuthHandler from google.adk.events.event import Event from google.adk.events.event import NodeInfo @@ -56,6 +57,8 @@ def test_apply_confirmation_resume(): ctx.node_path = "wf@1/node@1" ctx.tool_confirmation = None ctx.resume_inputs = {} + # An in-process caller: no serving layer set a principal. + ctx.get_invocation_context.return_value.caller_principal = None assert apply_confirmation_resume(ctx) is False assert ctx.tool_confirmation is None @@ -72,6 +75,45 @@ def test_apply_confirmation_resume(): ) +def test_apply_confirmation_resume_refuses_unauthenticated_caller_in_strict_mode( + monkeypatch, +): + """An approval the serving layer could not attribute is stored as a rejection. + + Same decision as the agent confirmation flow; made here because tools that + read `tool_confirmation` themselves never pass through the confirmation + gate. + """ + monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1") + ctx = MagicMock() + ctx.node_path = "wf@1/node@1" + ctx.tool_confirmation = None + ctx.resume_inputs = {"wf_tool_confirmation:wf@1/node@1": {"confirmed": True}} + ctx.get_invocation_context.return_value.caller_principal = CallerPrincipal( + authenticated=False, source="a2a" + ) + + assert apply_confirmation_resume(ctx) is True + assert ctx.tool_confirmation == ToolConfirmation(confirmed=False) + + +def test_apply_confirmation_resume_honors_authenticated_caller_in_strict_mode( + monkeypatch, +): + """An approval from a caller the serving layer verified is kept as sent.""" + monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1") + ctx = MagicMock() + ctx.node_path = "wf@1/node@1" + ctx.tool_confirmation = None + ctx.resume_inputs = {"wf_tool_confirmation:wf@1/node@1": {"confirmed": True}} + ctx.get_invocation_context.return_value.caller_principal = CallerPrincipal( + authenticated=True, user_name="operator", source="a2a" + ) + + assert apply_confirmation_resume(ctx) is True + assert ctx.tool_confirmation == ToolConfirmation(confirmed=True) + + def test_create_confirmation_request_input(): """create_confirmation_request_input builds a ToolConfirmation RequestInput.""" default_req = create_confirmation_request_input(