From 6fe9b86b60340a26fa7fbfa898ec908fbe3da441 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Thu, 9 Apr 2026 21:38:30 +0000 Subject: [PATCH] fix(tui): integrate ShellSafetyService properly in TUI app (#6361) - route shell submissions through ShellSafetyService and surface warnings in the UI - add shell warning banner, prompt styling, and configurable shell.warn_dangerous flag - extend TUI coverage scenarios for shell safety and document the fix ISSUES CLOSED: #6361 --- CHANGELOG.md | 8 ++ features/steps/tui_app_coverage_steps.py | 108 +++++++++++++++++++++-- features/tui_app_coverage.feature | 26 ++++++ src/cleveragents/config/settings.py | 9 ++ src/cleveragents/tui/app.py | 74 +++++++++++++++- src/cleveragents/tui/cleveragents.tcss | 15 ++++ src/cleveragents/tui/input/modes.py | 17 +++- 7 files changed, 246 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 71833a437..b3cd2bfd4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1255,6 +1255,14 @@ iteration` and data corruption under concurrent plan execution. All public `ResourceEdgeModel`, so the child-link check correctly blocks deletion. +### Fixed + +- **TUI — Shell safety integration** (#6361): The Textual prompt now uses + `ShellSafetyService` to analyse shell commands, highlight dangerous input + with `$error` styling, and display an advisory warning banner. Removed the + inline environment-variable gate, added `shell.warn_dangerous` configuration, + and wired the warning indicator into the TUI layout. + --- ### Fixed diff --git a/features/steps/tui_app_coverage_steps.py b/features/steps/tui_app_coverage_steps.py index 560564656..9934dc9d7 100644 --- a/features/steps/tui_app_coverage_steps.py +++ b/features/steps/tui_app_coverage_steps.py @@ -19,6 +19,7 @@ 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 behave import given, then, when @@ -78,10 +79,21 @@ def _build_mock_textual(): class MockStatic: def __init__(self, *args, **kwargs): self._text = "" + self.display = False + self._classes: set[str] = set() def update(self, text): self._text = text + def add_class(self, name): + self._classes.add(name) + + def remove_class(self, name): + self._classes.discard(name) + + def has_class(self, name): + return name in self._classes + class MockInput: """Minimal Input stand-in for the Textual base class.""" @@ -89,13 +101,23 @@ def _build_mock_textual(): def __init__(self, *args, **kwargs): self.value = "" + self._classes: set[str] = set() - mock_textual_app.App = MockApp - mock_textual_containers.Vertical = MockVertical - mock_textual_widgets.Header = MockHeader - mock_textual_widgets.Footer = MockFooter - mock_textual_widgets.Static = MockStatic - mock_textual_widgets.Input = MockInput + def add_class(self, name): + self._classes.add(name) + + def remove_class(self, name): + self._classes.discard(name) + + def has_class(self, name): + return name in self._classes + + cast(Any, mock_textual_app).App = MockApp + cast(Any, mock_textual_containers).Vertical = MockVertical + cast(Any, mock_textual_widgets).Header = MockHeader + cast(Any, mock_textual_widgets).Footer = MockFooter + cast(Any, mock_textual_widgets).Static = MockStatic + cast(Any, mock_textual_widgets).Input = MockInput return { "textual": mock_textual, @@ -189,6 +211,20 @@ 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) # --------------------------------------------------------------------------- @@ -434,6 +470,19 @@ 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) # --------------------------------------------------------------------------- @@ -466,6 +515,17 @@ def step_submit_text(context, text): _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) # --------------------------------------------------------------------------- @@ -482,11 +542,47 @@ def step_submit_shell_none(context): 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/tui_app_coverage.feature b/features/tui_app_coverage.feature index f9ce5f258..2ef057b31 100644 --- a/features/tui_app_coverage.feature +++ b/features/tui_app_coverage.feature @@ -146,6 +146,32 @@ Feature: TUI App Coverage And I submit "!echo tui_shell_test" to the app Then the conversation widget should contain "tui_shell_test" + Scenario: on_input_submitted surfaces shell safety warnings + Given a mock command router and persona state + When I instantiate the Textual TUI app + And I call on_mount on the app + And I submit "!rm -rf /tmp" to the app with shell execution mocked + Then the shell warning indicator should be visible + And the prompt should be marked as dangerous + + Scenario: shell warning indicator is cleared after safe command + Given a mock command router and persona state + When I instantiate the Textual TUI app + And I call on_mount on the app + And I submit "!rm -rf /tmp" to the app with shell execution mocked + And I submit "!echo cleared" to the app with shell execution mocked + Then the shell warning indicator should not be visible + And the prompt should not be marked as dangerous + + Scenario: shell danger warnings can be disabled via settings + Given shell danger warnings are disabled in settings + And a mock command router and persona state + When I instantiate the Textual TUI app + And I call on_mount on the app + And I submit "!rm -rf /tmp" to the app with shell execution mocked + Then the shell warning indicator should not be visible + And the prompt should not be marked as dangerous + # --- on_input_submitted with shell returning None (lines 170-171) --- Scenario: on_input_submitted handles None shell result diff --git a/src/cleveragents/config/settings.py b/src/cleveragents/config/settings.py index 87e52b802..0fb24ab45 100644 --- a/src/cleveragents/config/settings.py +++ b/src/cleveragents/config/settings.py @@ -180,6 +180,15 @@ class Settings(BaseSettings): validation_alias=AliasChoices("CLEVERAGENTS_DEBUG_ENABLED"), ) + shell_warn_dangerous: bool = Field( + default=True, + validation_alias=AliasChoices("CLEVERAGENTS_SHELL_WARN_DANGEROUS"), + description=( + "When true, highlight potentially destructive shell commands in the TUI " + "and surface advisory warnings." + ), + ) + # Core output and estimation configuration format: str | None = Field( default=None, diff --git a/src/cleveragents/tui/app.py b/src/cleveragents/tui/app.py index 7b4cfdfcb..8bcbabd07 100644 --- a/src/cleveragents/tui/app.py +++ b/src/cleveragents/tui/app.py @@ -12,6 +12,7 @@ import structlog from rich.markup import escape as _escape from cleveragents.a2a.models import A2aRequest +from cleveragents.config.settings import get_settings from cleveragents.core.exceptions import DatabaseError from cleveragents.domain.models.core.session import ( SessionActorNotConfiguredError, @@ -21,6 +22,7 @@ from cleveragents.tui.first_run import create_default_persona_for_actor, is_firs from cleveragents.tui.input.modes import InputMode, InputModeRouter from cleveragents.tui.input.reference_parser import suggestions from cleveragents.tui.persona.state import PersonaState +from cleveragents.tui.shell_safety import DangerousCommandWarning, ShellSafetyService from cleveragents.tui.slash_catalog import slash_command_specs from cleveragents.tui.widgets.actor_selection_overlay import ActorSelectionOverlay from cleveragents.tui.widgets.help_panel_overlay import ( @@ -296,6 +298,18 @@ if _TEXTUAL_AVAILABLE: self._dispatch_gen: int = 0 # 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._allow_dangerous_shell = self._resolve_allow_dangerous_shell() + self._shell_warning_active = False + self._last_shell_warning: DangerousCommandWarning | None = None + self._shell_safety: ShellSafetyService | None = ( + ShellSafetyService(warn_callback=self._handle_shell_warning) + if self._shell_warn_enabled + else None + ) def compose(self) -> Any: yield _Header(show_clock=True) @@ -308,6 +322,7 @@ if _TEXTUAL_AVAILABLE: yield PromptInput( placeholder="Type message, /command, or !shell ...", id="prompt" ) + yield _Static("", id="shell-warning") yield PersonaBar(id="persona-bar") yield _Footer() @@ -335,6 +350,7 @@ if _TEXTUAL_AVAILABLE: # the inner Input never gets focus and typing appears to do nothing. prompt = self.query_one("#prompt", PromptInput) prompt.focus() + self._clear_shell_warning() def _complete_first_run(self, actor: str) -> None: """Persist the chosen actor as the default persona and refresh the bar.""" @@ -472,14 +488,15 @@ if _TEXTUAL_AVAILABLE: if not text: return + self._allow_dangerous_shell = self._resolve_allow_dangerous_shell() + self._clear_shell_warning() + mode_router = InputModeRouter( command_handler=lambda raw: self._command_router.handle( raw, session_id=self._session.session_id ), - shell_confirm=lambda _cmd: ( - os.environ.get("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", "").strip() - in {"1", "true"} - ), + shell_confirm=self._confirm_dangerous_shell, + shell_safety=self._shell_safety, ) result = mode_router.process(text) # Use the cached widget (set in on_mount) to avoid repeated @@ -580,6 +597,55 @@ if _TEXTUAL_AVAILABLE: ) worker.done_callback = _on_llm_done + def _confirm_dangerous_shell(self, command: str) -> bool: + if self._shell_safety is not None: + return self._allow_dangerous_shell + return self._allow_dangerous_shell + + def _handle_shell_warning(self, warning: DangerousCommandWarning) -> bool: + if not self._shell_warn_enabled: + return True + self._show_shell_warning(warning) + return self._allow_dangerous_shell + + def _show_shell_warning(self, warning: DangerousCommandWarning) -> None: + try: + shell_warning = self.query_one("#shell-warning", _Static) + prompt = self.query_one("#prompt", PromptInput) + except Exception: # pragma: no cover - defensive + return + + level = warning.danger_level.name.capitalize() + shell_warning.update( + f"⚠ Potentially destructive command detected ({level})" + ) + shell_warning.display = True + if hasattr(prompt, "add_class"): + prompt.add_class("dangerous") + self._last_shell_warning = warning + self._shell_warning_active = True + + def _clear_shell_warning(self) -> None: + try: + shell_warning = self.query_one("#shell-warning", _Static) + prompt = self.query_one("#prompt", PromptInput) + except Exception: # pragma: no cover - defensive + return + + shell_warning.update("") + shell_warning.display = False + if hasattr(prompt, "remove_class"): + prompt.remove_class("dangerous") + self._last_shell_warning = None + self._shell_warning_active = False + + @staticmethod + def _resolve_allow_dangerous_shell() -> bool: + raw = os.environ.get("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", "").strip() + if not raw: + return True + return raw.lower() in {"1", "true", "yes", "on"} + _ResolvedTuiApp = _TextualCleverAgentsTuiApp CleverAgentsTuiApp = _ResolvedTuiApp diff --git a/src/cleveragents/tui/cleveragents.tcss b/src/cleveragents/tui/cleveragents.tcss index 0d0344357..be098ed9d 100644 --- a/src/cleveragents/tui/cleveragents.tcss +++ b/src/cleveragents/tui/cleveragents.tcss @@ -59,6 +59,21 @@ Screen { height: 1; } +#prompt.dangerous { + border: round $error; + color: $error; +} + +#shell-warning { + height: auto; + margin: 0 0 1 0; + padding: 0 1; + border: round $warning; + color: $warning; + background: $warning 12%; + display: none; +} + #persona-bar { height: auto; padding: 0 1; diff --git a/src/cleveragents/tui/input/modes.py b/src/cleveragents/tui/input/modes.py index 2049e0ad6..0f072f71d 100644 --- a/src/cleveragents/tui/input/modes.py +++ b/src/cleveragents/tui/input/modes.py @@ -11,6 +11,7 @@ from cleveragents.tui.input.reference_parser import ( parse_references, ) from cleveragents.tui.input.shell_exec import ShellResult, run_shell_command +from cleveragents.tui.shell_safety import DangerousCommandWarning, ShellSafetyService class InputMode(StrEnum): @@ -31,6 +32,7 @@ class ModeResult: references: list[str] shell_result: ShellResult | None command_result: str | None + shell_warning: DangerousCommandWarning | None class InputModeRouter: @@ -42,10 +44,12 @@ class InputModeRouter: *, shell_confirm: Callable[[str], bool] | None = None, shell_timeout_seconds: int = 30, + shell_safety: ShellSafetyService | None = None, ) -> None: self._command_handler = command_handler self._shell_confirm = shell_confirm self._shell_timeout_seconds = shell_timeout_seconds + self._shell_safety = shell_safety @staticmethod def detect_mode(text: str) -> InputMode: @@ -69,12 +73,21 @@ class InputModeRouter: references=[], shell_result=None, command_result=result, + shell_warning=None, ) if mode == InputMode.SHELL: command = text.lstrip()[1:].strip() + warning: DangerousCommandWarning | None = None + allowed = True + confirm = self._shell_confirm + if self._shell_safety is not None: + safety_result = self._shell_safety.check_command(command) + warning = safety_result.warning + allowed = safety_result.allowed + confirm = lambda _cmd, allow=allowed: allow shell_result = run_shell_command( command, - confirm_dangerous=self._shell_confirm, + confirm_dangerous=confirm, timeout_seconds=self._shell_timeout_seconds, ) return ModeResult( @@ -83,6 +96,7 @@ class InputModeRouter: references=[], shell_result=shell_result, command_result=None, + shell_warning=warning, ) parse_result: ReferenceParseResult = parse_references(text) @@ -92,4 +106,5 @@ class InputModeRouter: references=[match.canonical for match in parse_result.matches], shell_result=None, command_result=None, + shell_warning=None, )