`require_confirmation` callable returning a non-bool skips the confirmation gate: `cast(bool, ...)` is not a runtime check
- Ngôn ngữ chính
- Python
- Star
- 21.5k
- Fork
- 4k
- Merge trung bình
- 1 ngày 14 giờ
- Pull request đã merge (30 ngày)
- 37
Mô tả
## Required Information
**Describe the Bug:**
`check_require_confirmation` returns the confirmation predicate's value through `cast(bool, ...)`. `typing.cast` is an assertion to the type checker and does nothing at runtime, so whatever the predicate returns arrives at the gate unchanged. A predicate that falls off the end of a branch returns `None`, `if require_confirmation:` is false, and the tool runs with no confirmation requested.
Two sites, both on `main` at `c7ffcfa8`:
```python
# src/google/adk/tools/function_tool.py:202-205
return cast(
bool,
await self._invoke_callable(self._require_confirmation, args_to_call),
)
```
```python
# src/google/adk/tools/mcp_tool/mcp_tool.py:470-473
return cast(
bool,
await self._invoke_callable(self._require_confirmation, args_to_call),
)
```
The line just below each of those, `return bool(self._require_confirmation)`, is the static branch and is fine. It is the callable branch that passes the value through.
The declared contract is a bool on both paths: `require_confirmation: Union[bool, Callable[..., bool]] = False` (`function_tool.py:104`), `require_confirmation: bool | Callable[..., bool] = False` (`mcp_tool.py:285`), and `BaseTool.check_require_confirmation` is annotated `-> bool` (`base_tool.py:182-184`). `_invoke_callable` is annotated `-> Any` and returns the callable's value as it is, so nothing between the predicate and `if require_confirmation:` narrows it (`function_tool.py:267-271`, `mcp_tool.py:487-491`).
**Steps to Reproduce:**
1. `pip install "google-adk @ git+https://github.com/google/adk-python@c7ffcfa85a8e8970f6318306479d9c4c110583b2"`
2. Save the script under Minimal Reproduction Code below as `repro.py`.
3. `python repro.py`
**Expected Behavior:**
A predicate that does not return a bool has not answered the question, so the gate should treat it as unanswered rather than as "no confirmation needed".
**Observed Behavior:**
Seven rows, six of them controls. The `answered` column is the value `check_require_confirmation` handed back, which is the part that shows the `cast` did nothing: the annotation says `bool`, the value is `None`.
```
require_confirmation -> did the guarded tool run unconfirmed?
ran=no True (control: must ask) -> asked for confirmation answered True (bool=True)
ran=YES False (control: must run) -> no confirmation asked answered False (bool=True)
ran=no callable -> True (control: must ask) -> asked for confirmation answered True (bool=True)
ran=YES callable -> False (control: must run) -> no confirmation asked answered False (bool=True)
ran=no callable raises (unanswerable) -> RuntimeError: policy backend unreachable answered RuntimeError
ran=no callable -> 'reason' (truthy non-bool) -> asked for confirmation answered 'amount over limit' (bool=False)
ran=YES callable -> None (unhandled branch) -> no confirmation asked answered None (bool=False)
```
Rows one to four are the ordinary paths and show the harness can tell a confirmation request from an execution. Row five is the other kind of non-answer: a predicate that raises propagates out of `run_async`, so the tool does not run. Row six matters for the fix rather than for the bug: a truthy non-bool asks for confirmation today. Row seven is the finding.
The same question put to `McpTool`, which needs the `mcp` extra installed and so is kept as a separate snippet:
```
callable -> True -> answered True (bool=True)
callable -> False -> answered False (bool=True)
callable -> None -> answered None (bool=False)
```
**Environment Details:**
- ADK Library Version (pip show google-adk): 2.8.0, installed from `main` at `c7ffcfa85a8e8970f6318306479d9c4c110583b2`. I checked that `function_tool.py` and `mcp_tool.py` in site-packages are byte identical (sha256) to the same paths at that commit.
- Desktop OS: Windows 11, 10.0.26200
- Python Version (python -V): Python 3.12.10
**Model Information:**
- Are you using LiteLLM: No
- Which model is being used: none. The repro calls the tool directly and needs no model.
---
## Optional Information
**Regression:**
I did not bisect. The same expression is in the 2.8.0 wheel on PyPI, so this is not new on `main`.
**Logs:**
The run prints `UserWarning: [EXPERIMENTAL] feature FeatureName.TOOL_CONFIRMATION is enabled`. I understand this surface is experimental. That is context for how you want to weigh the report, not an argument that it does not matter.
**Additional Context:**
What I read in this tracker before filing:
* #4327 (closed) and #6977 (closed) are both fixes on this same predicate path, so the callable form is live and maintained.
* #4625 asked for fail-closed behaviour when no confirmation policy is configured. It was closed, and the answer on the thread was that the current behaviour is by design, since `require_confirmation` defaults to `False` and is opt-in. This report is not that request. Here the caller has opted in with a callable, and the gate they configured still does not hold.
* #6461 (open) is a different bypass of the same gate, through A2A.
* Search check: `require_confirmation` returns 15 issues in this tracker and a nonsense token returns 0, so the search was working. I read the 15 and found no report of this case.
I am not proposing a specific fix, because each of the obvious ones is a behaviour change and the choice is yours. Replacing `cast(bool, ...)` with `bool(...)` would give row seven the same wrong answer with a different provenance. A strict `isinstance` check would start raising for the row six caller, who returns a reason string to mean yes and works today. Treating a falsy non-bool as unanswered changes only the dangerous half, but it also flips `require_confirmation=lambda amount: FLAGS.get("confirm")` from run to ask for anyone relying on that. Which non-bools stay legal is a design decision. I am happy to send the PR for whichever you choose, with tests.
**Minimal Reproduction Code:**
```python
"""Does a require_confirmation callable that returns a non-bool skip the gate?"""
import asyncio
from typing import Any
from unittest.mock import MagicMock
from google.adk.agents.invocation_context import InvocationContext
from google.adk.sessions.session import Session
from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.tool_context import ToolContext
EXECUTED: list[str] = []
def wire_money(amount: int) -> str:
"""Sends money."""
EXECUTED.append(f"wired {amount}")
return "wired"
def make_tool_context() -> ToolContext:
invocation_context = MagicMock(spec=InvocationContext)
invocation_context._state_schema = None
invocation_context.session = MagicMock(spec=Session)
invocation_context.session.state = MagicMock()
invocation_context.agent = MagicMock()
invocation_context.agent.name = "test_agent"
tool_context = ToolContext(invocation_context=invocation_context)
tool_context.function_call_id = "call_1"
return tool_context
async def probe(label: str, require_confirmation: Any) -> None:
EXECUTED.clear()
tool = FunctionTool(wire_money, require_confirmation=require_confirmation)
args = {"amount": 100}
try:
answer = await tool.check_require_confirmation(args, make_tool_context())
answered = f"{answer!r} (bool={isinstance(answer, bool)})"
except Exception as exc:
answered = type(exc).__name__
try:
result = await tool.run_async(args=args, tool_context=make_tool_context())
error = result.get("error", "") if isinstance(result, dict) else ""
asked = "requires confirmation" in str(error)
outcome = "asked for confirmation" if asked else "no confirmation asked"
except Exception as exc:
outcome = f"{type(exc).__name__}: {exc}"[:44]
ran = "YES" if EXECUTED else "no "
print(f" ran={ran} {label:38} -> {outcome:22} answered {answered}")
def returns_true(amount: int) -> bool:
return True
def returns_false(amount: int) -> bool:
return False
def raises(amount: int) -> bool:
raise RuntimeError("policy backend unreachable")
def returns_reason(amount: int) -> bool:
return "amount over limit"
def forgot_a_branch(amount: int) -> bool:
if amount > 1000:
return True
# every other amount falls through and returns None
async def main() -> None:
print("require_confirmation -> did the guarded tool run unconfirmed?")
await probe("True (control: must ask)", True)
await probe("False (control: must run)", False)
await probe("callable -> True (control: must ask)", returns_true)
await probe("callable -> False (control: must run)", returns_false)
await probe("callable raises (unanswerable)", raises)
await probe("callable -> 'reason' (truthy non-bool)", returns_reason)
await probe("callable -> None (unhandled branch)", forgot_a_branch)
asyncio.run(main())
```
The `McpTool` snippet, which additionally needs `pip install mcp`:
```python
"""Same question for McpTool. Needs the mcp extra."""
import asyncio
from unittest.mock import MagicMock
from google.adk.agents.invocation_context import InvocationContext
from google.adk.sessions.session import Session
from google.adk.tools.mcp_tool.mcp_tool import McpTool
from google.adk.tools.tool_context import ToolContext
from mcp.types import Tool as McpBaseTool
def make_tool_context() -> ToolContext:
invocation_context = MagicMock(spec=InvocationContext)
invocation_context._state_schema = None
invocation_context.session = MagicMock(spec=Session)
invocation_context.session.state = MagicMock()
invocation_context.agent = MagicMock()
invocation_context.agent.name = "test_agent"
tool_context = ToolContext(invocation_context=invocation_context)
tool_context.function_call_id = "call_1"
return tool_context
def forgot_a_branch(amount: int) -> bool:
if amount > 1000:
return True
async def main() -> None:
spec = McpBaseTool(
name="wire_money",
description="Sends money.",
inputSchema={"type": "object", "properties": {"amount": {"type": "integer"}}},
)
cases = [
("callable -> True ", lambda amount: True),
("callable -> False", lambda amount: False),
("callable -> None ", forgot_a_branch),
]
for label, predicate in cases:
tool = McpTool(
mcp_tool=spec,
mcp_session_manager=MagicMock(),
require_confirmation=predicate,
)
answer = await tool.check_require_confirmation({"amount": 100}, make_tool_context())
print(f" {label} -> answered {answer!r} (bool={isinstance(answer, bool)})")
asyncio.run(main())
```
**How often has this issue occurred?:**
- Always (100%)
Hướng dẫn đóng góp
Đánh giá
Issue này chưa được đánh giá.