From 8e8f2d5f8938fae3ffa49799ee7aa8761d452b5a Mon Sep 17 00:00:00 2001 From: CleverThis Date: Sun, 12 Apr 2026 19:16:40 +0000 Subject: [PATCH] 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 --- features/steps/tui_app_coverage_steps.py | 123 +---------------------- features/steps/tui_shell_safety_steps.py | 115 +++++++++++++++++++++ features/tui_shell_exec_coverage.feature | 2 +- robot/tui_shell_safety.robot | 62 ++++++++++++ src/cleveragents/tui/app.py | 18 ++-- src/cleveragents/tui/input/modes.py | 6 +- src/cleveragents/tui/input/shell_exec.py | 25 +++-- 7 files changed, 207 insertions(+), 144 deletions(-) create mode 100644 features/steps/tui_shell_safety_steps.py create mode 100644 robot/tui_shell_safety.robot diff --git a/features/steps/tui_app_coverage_steps.py b/features/steps/tui_app_coverage_steps.py index 9934dc9d7..52a152811 100644 --- a/features/steps/tui_app_coverage_steps.py +++ b/features/steps/tui_app_coverage_steps.py @@ -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) # --------------------------------------------------------------------------- diff --git a/features/steps/tui_shell_safety_steps.py b/features/steps/tui_shell_safety_steps.py new file mode 100644 index 000000000..dee241c25 --- /dev/null +++ b/features/steps/tui_shell_safety_steps.py @@ -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) diff --git a/features/tui_shell_exec_coverage.feature b/features/tui_shell_exec_coverage.feature index 20455dda1..742c5c183 100644 --- a/features/tui_shell_exec_coverage.feature +++ b/features/tui_shell_exec_coverage.feature @@ -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 diff --git a/robot/tui_shell_safety.robot b/robot/tui_shell_safety.robot new file mode 100644 index 000000000..ba394ce23 --- /dev/null +++ b/robot/tui_shell_safety.robot @@ -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 diff --git a/src/cleveragents/tui/app.py b/src/cleveragents/tui/app.py index 8bcbabd07..f70471ae3 100644 --- a/src/cleveragents/tui/app.py +++ b/src/cleveragents/tui/app.py @@ -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 diff --git a/src/cleveragents/tui/input/modes.py b/src/cleveragents/tui/input/modes.py index 0f072f71d..d1d79641d 100644 --- a/src/cleveragents/tui/input/modes.py +++ b/src/cleveragents/tui/input/modes.py @@ -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, diff --git a/src/cleveragents/tui/input/shell_exec.py b/src/cleveragents/tui/input/shell_exec.py index 015214695..9e57af77e 100644 --- a/src/cleveragents/tui/input/shell_exec.py +++ b/src/cleveragents/tui/input/shell_exec.py @@ -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,