|
self.pending_actions = [] |
|
|
|
# Should we let infrastructure tools through this turn? |
|
approval_granted = is_confirmation(user_input) |
|
|
|
messages = self.prompt.format_messages( |
|
chat_history=chat_history, input=user_input, |
|
) |
|
|
|
if callbacks: |
|
callbacks.on_thinking() |
|
|
|
response = self.llm.invoke(messages) |
|
return self._tool_loop(response, messages, approval_granted, callbacks) |
|
|
|
# ------------------------------------------------------------------ |
|
# Tool-calling loop |
|
# ------------------------------------------------------------------ |
|
def _tool_loop(self, response, messages: list, |
|
approval_granted: bool, callbacks) -> str: |
|
""" |
|
Execute tool calls in a loop until the LLM returns a text-only |
|
response or we hit the iteration limit. |
|
""" |
|
last_text = "" |
|
|
|
for _ in range(Config.MAX_ITERATIONS): |
|
# -- No tool calls -> return the final answer -- |
|
if not getattr(response, "tool_calls", None): |
|
text = extract_response_text(response) |
|
return text or last_text or "No response generated." |
|
|
|
# Keep any intermediate text the LLM produced alongside tools |
|
intermediate = extract_response_text(response) |
|
if intermediate: |
|
last_text = intermediate |
|
|
|
# Show reasoning in the UI |
|
if callbacks and intermediate: |
|
callbacks.on_reasoning(intermediate) |
|
|
|
# -- Execute each tool call -- |
|
tool_results = [] |
|
for tc in response.tool_calls: |
|
result_msg = self._execute_tool_call( |
|
tc, approval_granted, callbacks, |
|
) |
|
tool_results.append(result_msg) |
|
|
|
# Feed results back to the LLM |
|
messages.append(AIMessage( |
|
content=response.content, |
|
tool_calls=response.tool_calls, |
|
)) |
|
messages.extend(tool_results) |
|
|
|
if callbacks: |
|
callbacks.on_thinking() |
|
|
|
response = self.llm.invoke(messages) |
|
|
|
# Exhausted iterations |
|
text = extract_response_text(response) |
|
return text or last_text or "Reached maximum analysis steps." |
|
|
|
# ------------------------------------------------------------------ |
|
# Single tool-call execution |
|
# ------------------------------------------------------------------ |
|
def _execute_tool_call(self, tc: dict, approval_granted: bool, |
|
callbacks) -> ToolMessage: |
|
""" |
|
Execute one tool call. If the tool requires approval and the user |
|
has not confirmed, block it and record it in pending_actions. |
|
""" |
|
name, args, call_id = tc["name"], tc["args"], tc["id"] |
|
|
|
# -- Block infrastructure tools until user confirms -- |
|
if requires_approval(name) and not approval_granted: |
|
self.pending_actions.append(tc) |
|
if callbacks: |
|
callbacks.on_approval_skipped(name, args) |
|
return ToolMessage( |
|
content=( |
|
f"Action '{name}' requires human approval and was not executed. " |
|
"Present your findings and ask the user to confirm. " |
|
"When the user confirms, call this tool again -- " |
|
"the system will allow it through." |
|
), |
|
tool_call_id=call_id, |
|
) |
|
|
|
# -- Execute the tool -- |
|
if callbacks: |
|
callbacks.on_tool_start(name, args) |
|
|
|
tool_func = self._find_tool(name) |
|
if not tool_func: |
|
return ToolMessage(content=f"Tool '{name}' not found", |
|
tool_call_id=call_id) |
|
|
|
try: |
|
result = str(tool_func.invoke(args)) |
|
if callbacks: |
|
callbacks.on_tool_end(name, result, success=True) |
|
return ToolMessage(content=result, tool_call_id=call_id) |
Summary
In the Chapter 10 DevOps agent, tool approval is turn-scoped rather than action-scoped. A single "yes" authorizes every approval-gated tool call the model emits for the rest of the turn, with no check that the executed call matches the action the user reviewed. The gated tools are destructive (
reboot_rds_instance,restart_kubernetes_pod).Where (HEAD
main)03-ai-agent-for-devops/code/10/src/agents/log_analyzer.pyL67-171.devops-ai-guidelines/03-ai-agent-for-devops/code/10/src/agents/log_analyzer.py
Lines 67 to 171 in 2d07199
Mechanism
process_querysets a single turn-wide booleanapproval_granted = is_confirmation(user_input)(L70). In_execute_tool_call, an approval-required tool is blocked only whenapproval_grantedis false; the blocked call is appended topending_actions(L145), butpending_actionsis never read anywhere. Whenapproval_grantedis true, the code executes the current provider-generated tool name and args (L168) with no comparison against the action the user approved (no check of tool name, arguments, target, or call identity).approval_grantedpersists for the entire_tool_loop(up toMAX_ITERATIONS), which re-invokes the model after each tool round.The class docstring says "When the user confirms, the same tool is allowed through" (singular). The implementation does not enforce "the same tool," so it contradicts its own documented intent.
Impact / reachability
Reachable with ordinary multi-step remediation: any second approval-gated call the model emits after the first approval runs unchecked (a different pod/instance target, or a second destructive op), causing an unintended pod restart or RDS reboot while the user believes only the approved action ran. Because the agent analyzes untrusted log content, crafted log text can also steer the model to emit a different gated call in the same turn (prompt injection). Not fault-injection-only.
Note
This is tutorial/guidelines code and the approval logic pre-dates PR #33 (surfaced while reviewing it), but readers copy these patterns, so it is worth fixing.
Suggested fix
Bind approval to a specific pending action (tool name + arguments + target) and require a fresh confirmation for any call that does not match the approved one; clear approval after a single gated call executes.
Found via Ito automated code review. Full analysis: https://app.ito.ai/share/5c9a9cbd-4892-4713-ac15-6edc5bd50f64