fix(tui): enforce shell safety gating

- honour ShellSafetyService verdicts before executing shell commands
- tighten TUI confirmation defaults and align warning messaging with spec
- split shell-safety Behave steps and add Robot coverage

ISSUES CLOSED: #6361
This commit is contained in:
2026-04-12 19:16:40 +00:00
committed by Forgejo
parent 6fe9b86b60
commit 8e8f2d5f89
7 changed files with 207 additions and 144 deletions
+2 -121
View File
@@ -1,26 +1,13 @@
"""Step definitions for tui_app_coverage.feature.
These steps target uncovered lines in cleveragents/tui/app.py:
- Lines 31-38: Textual import success path (mocked)
- Lines 81-100: _TextualCleverAgentsTuiApp class definition + __init__
- Lines 102-112: compose method
- Lines 114-121: on_mount method
- Lines 123-125: action_help method
- Lines 127-129: action_cycle_preset method
- Lines 131-142: _refresh_persona_bar method
- Lines 144-185: on_input_submitted (all branches)
- Line 189: CleverAgentsTuiApp alias
"""
"""Step definitions for tui_app_coverage.feature targeting cleveragents.tui.app."""
import importlib
import os
import shutil
import sys
import tempfile
from pathlib import Path
from types import ModuleType, SimpleNamespace
from typing import Any, cast
from unittest.mock import MagicMock, patch
from unittest.mock import MagicMock
from behave import given, then, when
@@ -211,20 +198,6 @@ def step_import_with_mock_textual(context):
context.add_cleanup(lambda: _cleanup_tmpdir(context))
@given("shell danger warnings are disabled in settings")
def step_disable_shell_warnings(context):
stub = SimpleNamespace(shell_warn_dangerous=False)
patcher = patch(
"cleveragents.config.settings.get_settings",
return_value=stub,
)
patcher.start()
context.add_cleanup(patcher.stop)
app_patcher = patch("cleveragents.tui.app.get_settings", return_value=stub)
app_patcher.start()
context.add_cleanup(app_patcher.stop)
# ---------------------------------------------------------------------------
# Module-level import gate (lines 31-38)
# ---------------------------------------------------------------------------
@@ -470,19 +443,6 @@ def _submit_text(context, text):
context._tui_app.on_input_submitted(event)
def _submit_text_with_mocked_shell(context, text, stdout: str = "mocked") -> None:
from cleveragents.tui.input.shell_exec import ShellResult
def fake_run(command: str, **_: Any) -> ShellResult:
return ShellResult(command=command, exit_code=0, stdout=stdout, stderr="")
with patch(
"cleveragents.tui.input.shell_exec.run_shell_command",
side_effect=fake_run,
):
_submit_text(context, text)
# ---------------------------------------------------------------------------
# on_input_submitted: empty text (lines 144-150)
# ---------------------------------------------------------------------------
@@ -504,85 +464,6 @@ def step_conv_not_updated(context):
# ---------------------------------------------------------------------------
# on_input_submitted: command mode (lines 152-167)
# ---------------------------------------------------------------------------
@when('I submit "{text}" to the app')
def step_submit_text(context, text):
os.environ["CLEVERAGENTS_ALLOW_DANGEROUS_SHELL"] = "1"
def restore_env():
os.environ.pop("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", None)
context.add_cleanup(restore_env)
_submit_text(context, text)
@when('I submit "{text}" to the app with shell execution mocked')
def step_submit_text_mocked_shell(context, text):
os.environ["CLEVERAGENTS_ALLOW_DANGEROUS_SHELL"] = "1"
def restore_env():
os.environ.pop("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", None)
context.add_cleanup(restore_env)
_submit_text_with_mocked_shell(context, text)
# ---------------------------------------------------------------------------
# on_input_submitted: shell returning None (lines 170-171)
# ---------------------------------------------------------------------------
@when("I submit shell text that produces a None shell result")
def step_submit_shell_none(context):
from cleveragents.tui.input.modes import InputMode, ModeResult
# We patch the InputModeRouter.process to return a shell result with None
with patch(
"cleveragents.tui.app.InputModeRouter.process",
return_value=ModeResult(
mode=InputMode.SHELL,
expanded_text="!nothing",
references=[],
shell_result=None,
command_result=None,
shell_warning=None,
),
):
_submit_text(context, "!nothing")
# ---------------------------------------------------------------------------
# Shell warning indicator assertions
# ---------------------------------------------------------------------------
@then("the shell warning indicator should be visible")
def step_shell_warning_visible(context):
MockStatic = context._tui_mock_static
banner = context._tui_app.query_one("#shell-warning", MockStatic)
assert banner.display is True, "Expected shell warning indicator to be visible"
assert banner._text, "Expected warning indicator text to be populated"
@then("the shell warning indicator should not be visible")
def step_shell_warning_hidden(context):
MockStatic = context._tui_mock_static
banner = context._tui_app.query_one("#shell-warning", MockStatic)
assert banner.display is False, "Expected shell warning indicator to be hidden"
assert banner._text == "", "Expected warning indicator text to be cleared"
@then("the prompt should be marked as dangerous")
def step_prompt_marked_dangerous(context):
from cleveragents.tui.widgets.prompt import PromptInput
prompt = context._tui_app.query_one("#prompt", PromptInput)
assert prompt.has_class("dangerous"), "Expected prompt to have dangerous class"
@then("the prompt should not be marked as dangerous")
def step_prompt_not_dangerous(context):
from cleveragents.tui.widgets.prompt import PromptInput
prompt = context._tui_app.query_one("#prompt", PromptInput)
assert not prompt.has_class("dangerous"), "Expected prompt to be safe"
# ---------------------------------------------------------------------------
# on_input_submitted: normal text with @ (lines 179-185)
# ---------------------------------------------------------------------------
+115
View File
@@ -0,0 +1,115 @@
"""Shell safety-related step definitions for the TUI app coverage suite."""
from __future__ import annotations
import os
from types import SimpleNamespace
from typing import Any
from unittest.mock import patch
from behave import given, then, when
from cleveragents.tui.input.modes import InputMode, ModeResult
from cleveragents.tui.input.shell_exec import ShellResult
def _submit_text(context, text: str) -> None:
from cleveragents.tui.widgets.prompt import PromptInput
prompt = context._tui_app.query_one("#prompt", PromptInput)
prompt.value = text
event = SimpleNamespace()
context._tui_app.on_input_submitted(event)
def _submit_text_with_mocked_shell(context, text: str, stdout: str = "mocked") -> None:
"""Submit *text* while faking shell execution."""
def fake_run(command: str, **_: Any) -> ShellResult:
return ShellResult(command=command, exit_code=0, stdout=stdout, stderr="")
with patch(
"cleveragents.tui.input.shell_exec.run_shell_command",
side_effect=fake_run,
):
_submit_text(context, text)
@when('I submit "{text}" to the app')
def step_submit_text(context, text):
os.environ["CLEVERAGENTS_ALLOW_DANGEROUS_SHELL"] = "1"
def restore_env() -> None:
os.environ.pop("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", None)
context.add_cleanup(restore_env)
_submit_text(context, text)
@when('I submit "{text}" to the app with shell execution mocked')
def step_submit_text_mocked_shell(context, text):
os.environ["CLEVERAGENTS_ALLOW_DANGEROUS_SHELL"] = "1"
def restore_env() -> None:
os.environ.pop("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", None)
context.add_cleanup(restore_env)
_submit_text_with_mocked_shell(context, text)
@when("I submit shell text that produces a None shell result")
def step_submit_shell_none(context):
# Patch the mode router to emulate a shell submission returning None.
with patch(
"cleveragents.tui.app.InputModeRouter.process",
return_value=ModeResult(
mode=InputMode.SHELL,
expanded_text="!nothing",
references=[],
shell_result=None,
command_result=None,
shell_warning=None,
),
):
_submit_text(context, "!nothing")
@then("the shell warning indicator should be visible")
def step_shell_warning_visible(context):
MockStatic = context._tui_mock_static
banner = context._tui_app.query_one("#shell-warning", MockStatic)
assert banner.display is True, "Expected shell warning indicator to be visible"
assert banner._text, "Expected warning indicator text to be populated"
@then("the shell warning indicator should not be visible")
def step_shell_warning_hidden(context):
MockStatic = context._tui_mock_static
banner = context._tui_app.query_one("#shell-warning", MockStatic)
assert banner.display is False, "Expected shell warning indicator to be hidden"
assert banner._text == "", "Expected warning indicator text to be cleared"
@then("the prompt should be marked as dangerous")
def step_prompt_marked_dangerous(context):
from cleveragents.tui.widgets.prompt import PromptInput
prompt = context._tui_app.query_one("#prompt", PromptInput)
assert prompt.has_class("dangerous"), "Expected prompt to have dangerous class"
@then("the prompt should not be marked as dangerous")
def step_prompt_not_dangerous(context):
from cleveragents.tui.widgets.prompt import PromptInput
prompt = context._tui_app.query_one("#prompt", PromptInput)
assert not prompt.has_class("dangerous"), "Expected prompt to be safe"
@given("shell danger warnings are disabled in settings")
@when("shell danger warnings are disabled in settings")
def step_disable_shell_warnings(context):
stub = SimpleNamespace(shell_warn_dangerous=False)
app_patcher = patch("cleveragents.tui.app.get_settings", return_value=stub)
app_patcher.start()
context.add_cleanup(app_patcher.stop)
+1 -1
View File
@@ -38,7 +38,7 @@ Feature: TUI Shell Exec Coverage
Given a confirm_dangerous callback that returns False
When I run a dangerous command "rm -rf /" with the callback
Then the shell result exit code should be 1
And the shell result stderr should be "blocked dangerous shell command"
And the shell result stderr should be "blocked by shell safety policy"
Scenario: Command that exceeds timeout returns timeout result
Given subprocess run is mocked to raise TimeoutExpired
+62
View File
@@ -0,0 +1,62 @@
*** Settings ***
Documentation Integration coverage for TUI shell safety wiring and safeguards.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment With Database Isolation
Suite Teardown Cleanup Test Environment
*** Variables ***
${TIMEOUT} 45s
*** Test Cases ***
Shell Safety Service Blocks Denied Command
[Documentation] ShellSafetyService verdict should block commands even when heuristics allow them.
[Tags] tui shell_safety regression
${script}= Catenate SEPARATOR=\n
... import os
... from cleveragents.tui.input.modes import InputModeRouter
... from cleveragents.tui.shell_safety import ShellSafetyService
... from cleveragents.tui.shell_safety.warning import DangerousCommandWarning
...
... warnings: list[DangerousCommandWarning] = []
...
... def warn_callback(warning: DangerousCommandWarning) -> bool:
... warnings.append(warning)
... return False
...
... router = InputModeRouter(lambda cmd: "handled", shell_safety=ShellSafetyService(warn_callback=warn_callback))
... os.environ.pop("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", None)
... result = router.process("!chmod -R 777 /tmp/test-shell-safety")
... assert result.shell_warning is not None, "Shell safety warning should be surfaced"
... assert warnings and warnings[0].command == "chmod -R 777 /tmp/test-shell-safety"
... assert result.shell_result is not None, "Shell result should be populated"
... assert result.shell_result.exit_code == 1, f"Expected blocked exit code, got {result.shell_result.exit_code}"
... assert "blocked" in result.shell_result.stderr.lower(), result.shell_result.stderr
... print("blocked-ok")
${result}= Run Process ${PYTHON} -c ${script}
... timeout=${TIMEOUT} on_timeout=kill
... env:PYTHONPATH=${CURDIR}/../src
Should Be Equal As Integers ${result.rc} 0 Shell safety blocking script failed: ${result.stderr}
Should Contain ${result.stdout} blocked-ok
Shell Confirm Callback Gates All Commands
[Documentation] run_shell_command must respect confirm callback regardless of built-in heuristics.
[Tags] tui shell_safety regression
${script}= Catenate SEPARATOR=\n
... from cleveragents.tui.input.shell_exec import run_shell_command
...
... counter = {"count": 0}
...
... def deny(command: str) -> bool:
... counter["count"] += 1
... return False
...
... result = run_shell_command("chmod -R 777 /tmp/test-shell-safety", confirm_dangerous=deny)
... assert counter["count"] == 1, f"Expected confirm to be invoked once, got {counter['count']}"
... assert result.exit_code == 1, f"Expected blocked exit code, got {result.exit_code}"
... assert "blocked" in result.stderr.lower(), result.stderr
... print("confirm-gate-ok")
${result}= Run Process ${PYTHON} -c ${script}
... timeout=${TIMEOUT} on_timeout=kill
... env:PYTHONPATH=${CURDIR}/../src
Should Be Equal As Integers ${result.rc} 0 Shell confirm gate script failed: ${result.stderr}
Should Contain ${result.stdout} confirm-gate-ok
+8 -10
View File
@@ -21,6 +21,7 @@ from cleveragents.domain.models.core.session import (
from cleveragents.tui.first_run import create_default_persona_for_actor, is_first_run
from cleveragents.tui.input.modes import InputMode, InputModeRouter
from cleveragents.tui.input.reference_parser import suggestions
from cleveragents.tui.input.shell_exec import looks_dangerous
from cleveragents.tui.persona.state import PersonaState
from cleveragents.tui.shell_safety import DangerousCommandWarning, ShellSafetyService
from cleveragents.tui.slash_catalog import slash_command_specs
@@ -299,9 +300,7 @@ if _TEXTUAL_AVAILABLE:
# Cached after mount to avoid repeated query_one() on every submit.
self._conversation: Any = None
self._settings = get_settings()
self._shell_warn_enabled = getattr(
self._settings, "shell_warn_dangerous", True
)
self._shell_warn_enabled = self._settings.shell_warn_dangerous
self._allow_dangerous_shell = self._resolve_allow_dangerous_shell()
self._shell_warning_active = False
self._last_shell_warning: DangerousCommandWarning | None = None
@@ -599,7 +598,10 @@ if _TEXTUAL_AVAILABLE:
def _confirm_dangerous_shell(self, command: str) -> bool:
if self._shell_safety is not None:
return self._allow_dangerous_shell
# The ShellSafetyService already provided the execution verdict.
return True
if not looks_dangerous(command):
return True
return self._allow_dangerous_shell
def _handle_shell_warning(self, warning: DangerousCommandWarning) -> bool:
@@ -615,10 +617,7 @@ if _TEXTUAL_AVAILABLE:
except Exception: # pragma: no cover - defensive
return
level = warning.danger_level.name.capitalize()
shell_warning.update(
f"⚠ Potentially destructive command detected ({level})"
)
shell_warning.update("⚠ Potentially destructive command detected")
shell_warning.display = True
if hasattr(prompt, "add_class"):
prompt.add_class("dangerous")
@@ -642,8 +641,7 @@ if _TEXTUAL_AVAILABLE:
@staticmethod
def _resolve_allow_dangerous_shell() -> bool:
raw = os.environ.get("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", "").strip()
if not raw:
return True
# Default to disallowing dangerous commands unless explicitly enabled.
return raw.lower() in {"1", "true", "yes", "on"}
_ResolvedTuiApp = _TextualCleverAgentsTuiApp
+5 -1
View File
@@ -84,7 +84,11 @@ class InputModeRouter:
safety_result = self._shell_safety.check_command(command)
warning = safety_result.warning
allowed = safety_result.allowed
confirm = lambda _cmd, allow=allowed: allow
def _safety_gate(_cmd: str, *, allow: bool = allowed) -> bool:
return allow
confirm = _safety_gate
shell_result = run_shell_command(
command,
confirm_dangerous=confirm,
+14 -11
View File
@@ -55,17 +55,20 @@ def run_shell_command(
stdout="",
stderr="shell mode is disabled",
)
if looks_dangerous(command):
confirmed = False
if confirm_dangerous is not None:
confirmed = confirm_dangerous(command)
if not confirmed:
return ShellResult(
command=command,
exit_code=1,
stdout="",
stderr="blocked dangerous shell command",
)
if confirm_dangerous is not None and not confirm_dangerous(command):
return ShellResult(
command=command,
exit_code=1,
stdout="",
stderr="blocked by shell safety policy",
)
if looks_dangerous(command) and confirm_dangerous is None:
return ShellResult(
command=command,
exit_code=1,
stdout="",
stderr="blocked dangerous shell command",
)
try:
proc = subprocess.run(
command,