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_helpers.py b/features/steps/_tui_helpers.py new file mode 100644 index 000000000..862ef48e3 --- /dev/null +++ b/features/steps/_tui_helpers.py @@ -0,0 +1,39 @@ +"""Shared TUI step-definition helpers.""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any +from unittest.mock import patch + +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. + + The patch target is the use site in ``modes`` rather than the + definition site in ``shell_exec``: ``modes`` did + ``from cleveragents.tui.input.shell_exec import run_shell_command``, + binding the symbol into its own namespace; patching the source + module would leave that binding (and therefore the real + ``subprocess.run`` call) untouched. + """ + + def fake_run(command: str, **_: Any) -> ShellResult: + return ShellResult(command=command, exit_code=0, stdout=stdout, stderr="") + + with patch( + "cleveragents.tui.input.modes.run_shell_command", + side_effect=fake_run, + ): + _submit_text(context, text) diff --git a/features/steps/_tui_mock_helpers.py b/features/steps/_tui_mock_helpers.py new file mode 100644 index 000000000..de71f8a26 --- /dev/null +++ b/features/steps/_tui_mock_helpers.py @@ -0,0 +1,191 @@ +"""Shared TUI mock-Textual infrastructure helpers for step-definition files.""" + +from __future__ import annotations + +import importlib +import shutil +import sys +import tempfile +from pathlib import Path +from types import ModuleType +from typing import Any, cast +from unittest.mock import MagicMock + + +_MOCK_TEXTUAL_KEYS = [ + "textual", + "textual.app", + "textual.containers", + "textual.widgets", +] + + +def _build_mock_textual(): + """Build mock textual modules that satisfy the app import gate.""" + mock_textual = ModuleType("textual") + mock_textual_app = ModuleType("textual.app") + mock_textual_containers = ModuleType("textual.containers") + mock_textual_widgets = ModuleType("textual.widgets") + + class MockApp: + """Minimal App stand-in for the Textual base class.""" + + def __init__(self, *args, **kwargs): + self._widgets = {} + + def query_one(self, selector, widget_type=None): + if selector in self._widgets: + return self._widgets[selector] + if widget_type is not None: + widget = widget_type(id=selector.lstrip("#")) + self._widgets[selector] = widget + return widget + return MagicMock() + + class MockVertical: + def __init__(self, *args, **kwargs): + pass + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + class MockHeader: + def __init__(self, *args, **kwargs): + pass + + class MockFooter: + def __init__(self, *args, **kwargs): + pass + + 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.""" + + value = "" + + def __init__(self, *args, **kwargs): + self.value = "" + self._classes: set[str] = set() + + 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, + "textual.app": mock_textual_app, + "textual.containers": mock_textual_containers, + "textual.widgets": mock_textual_widgets, + } + + +def _install_mock_textual(context): + """Inject mock textual into sys.modules and reload the app module.""" + mocks = _build_mock_textual() + context._tui_saved_modules = {} + for key in _MOCK_TEXTUAL_KEYS: + context._tui_saved_modules[key] = sys.modules.pop(key, None) + for key, mod in mocks.items(): + sys.modules[key] = mod + + # Reload widget modules so they pick up the mock Static/Input base class + import cleveragents.tui.widgets.help_panel_overlay as hp_mod + import cleveragents.tui.widgets.persona_bar as pb_mod + import cleveragents.tui.widgets.prompt as prompt_mod + import cleveragents.tui.widgets.reference_picker as rp_mod + import cleveragents.tui.widgets.slash_command_overlay as sco_mod + + importlib.reload(hp_mod) + importlib.reload(pb_mod) + importlib.reload(prompt_mod) + importlib.reload(rp_mod) + importlib.reload(sco_mod) + + import cleveragents.tui.app as app_mod + + importlib.reload(app_mod) + context._tui_app_mod = app_mod + context._tui_mock_static = mocks["textual.widgets"].Static + + +def _restore_modules(context): + """Restore original sys.modules and reload the app module.""" + for key, val in getattr(context, "_tui_saved_modules", {}).items(): + if val is None: + sys.modules.pop(key, None) + else: + sys.modules[key] = val + + # Reload widget modules so they pick up the real Static/Input base class again + import cleveragents.tui.widgets.help_panel_overlay as hp_mod + import cleveragents.tui.widgets.persona_bar as pb_mod + import cleveragents.tui.widgets.prompt as prompt_mod + import cleveragents.tui.widgets.reference_picker as rp_mod + import cleveragents.tui.widgets.slash_command_overlay as sco_mod + + importlib.reload(hp_mod) + importlib.reload(pb_mod) + importlib.reload(prompt_mod) + importlib.reload(rp_mod) + importlib.reload(sco_mod) + + import cleveragents.tui.app as app_mod + + importlib.reload(app_mod) + + +def _make_persona_state(context): + """Create a real PersonaState backed by a temp directory.""" + from cleveragents.tui.persona.registry import PersonaRegistry + from cleveragents.tui.persona.state import PersonaState + + tmp = tempfile.mkdtemp() + context._tui_tmpdir = tmp + registry = PersonaRegistry(config_dir=Path(tmp)) + registry.ensure_default() + return PersonaState(registry=registry) + + +def _cleanup_tmpdir(context): + tmp = getattr(context, "_tui_tmpdir", None) + if tmp: + shutil.rmtree(tmp, ignore_errors=True) + + +class _FakeCommandRouter: + """Test command router that returns predictable responses.""" + + def handle(self, raw, *, session_id): + return f"handled:{raw}" diff --git a/features/steps/tui_app_coverage_steps.py b/features/steps/tui_app_coverage_steps.py index 560564656..b69a728f5 100644 --- a/features/steps/tui_app_coverage_steps.py +++ b/features/steps/tui_app_coverage_steps.py @@ -1,181 +1,17 @@ -"""Step definitions for tui_app_coverage.feature. +"""Step definitions for tui_app_coverage.feature targeting cleveragents.tui.app.""" -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 -""" - -import importlib -import os -import shutil -import sys -import tempfile -from pathlib import Path -from types import ModuleType, SimpleNamespace -from unittest.mock import MagicMock, patch +from types import SimpleNamespace from behave import given, then, when -# --------------------------------------------------------------------------- -# Mock Textual infrastructure -# --------------------------------------------------------------------------- - -_MOCK_TEXTUAL_KEYS = [ - "textual", - "textual.app", - "textual.containers", - "textual.widgets", -] - - -def _build_mock_textual(): - """Build mock textual modules that satisfy the app import gate.""" - mock_textual = ModuleType("textual") - mock_textual_app = ModuleType("textual.app") - mock_textual_containers = ModuleType("textual.containers") - mock_textual_widgets = ModuleType("textual.widgets") - - class MockApp: - """Minimal App stand-in for the Textual base class.""" - - def __init__(self, *args, **kwargs): - self._widgets = {} - - def query_one(self, selector, widget_type=None): - if selector in self._widgets: - return self._widgets[selector] - if widget_type is not None: - widget = widget_type(id=selector.lstrip("#")) - self._widgets[selector] = widget - return widget - return MagicMock() - - class MockVertical: - def __init__(self, *args, **kwargs): - pass - - def __enter__(self): - return self - - def __exit__(self, *args): - pass - - class MockHeader: - def __init__(self, *args, **kwargs): - pass - - class MockFooter: - def __init__(self, *args, **kwargs): - pass - - class MockStatic: - def __init__(self, *args, **kwargs): - self._text = "" - - def update(self, text): - self._text = text - - class MockInput: - """Minimal Input stand-in for the Textual base class.""" - - value = "" - - def __init__(self, *args, **kwargs): - self.value = "" - - 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 - - return { - "textual": mock_textual, - "textual.app": mock_textual_app, - "textual.containers": mock_textual_containers, - "textual.widgets": mock_textual_widgets, - } - - -def _install_mock_textual(context): - """Inject mock textual into sys.modules and reload the app module.""" - mocks = _build_mock_textual() - context._tui_saved_modules = {} - for key in _MOCK_TEXTUAL_KEYS: - context._tui_saved_modules[key] = sys.modules.pop(key, None) - for key, mod in mocks.items(): - sys.modules[key] = mod - - # Reload widget modules so they pick up the mock Static/Input base class - import cleveragents.tui.widgets.help_panel_overlay as hp_mod - import cleveragents.tui.widgets.persona_bar as pb_mod - import cleveragents.tui.widgets.prompt as prompt_mod - import cleveragents.tui.widgets.reference_picker as rp_mod - import cleveragents.tui.widgets.slash_command_overlay as sco_mod - - importlib.reload(hp_mod) - importlib.reload(pb_mod) - importlib.reload(prompt_mod) - importlib.reload(rp_mod) - importlib.reload(sco_mod) - - import cleveragents.tui.app as app_mod - - importlib.reload(app_mod) - context._tui_app_mod = app_mod - context._tui_mock_static = mocks["textual.widgets"].Static - - -def _restore_modules(context): - """Restore original sys.modules and reload the app module.""" - for key, val in getattr(context, "_tui_saved_modules", {}).items(): - if val is None: - sys.modules.pop(key, None) - else: - sys.modules[key] = val - - # Reload widget modules so they pick up the real Static/Input base class again - import cleveragents.tui.widgets.help_panel_overlay as hp_mod - import cleveragents.tui.widgets.persona_bar as pb_mod - import cleveragents.tui.widgets.prompt as prompt_mod - import cleveragents.tui.widgets.reference_picker as rp_mod - import cleveragents.tui.widgets.slash_command_overlay as sco_mod - - importlib.reload(hp_mod) - importlib.reload(pb_mod) - importlib.reload(prompt_mod) - importlib.reload(rp_mod) - importlib.reload(sco_mod) - - import cleveragents.tui.app as app_mod - - importlib.reload(app_mod) - - -def _make_persona_state(context): - """Create a real PersonaState backed by a temp directory.""" - from cleveragents.tui.persona.registry import PersonaRegistry - from cleveragents.tui.persona.state import PersonaState - - tmp = tempfile.mkdtemp() - context._tui_tmpdir = tmp - registry = PersonaRegistry(config_dir=Path(tmp)) - registry.ensure_default() - return PersonaState(registry=registry) - - -def _cleanup_tmpdir(context): - tmp = getattr(context, "_tui_tmpdir", None) - if tmp: - shutil.rmtree(tmp, ignore_errors=True) +from features.steps._tui_helpers import _submit_text +from features.steps._tui_mock_helpers import ( + _FakeCommandRouter, + _cleanup_tmpdir, + _install_mock_textual, + _make_persona_state, + _restore_modules, +) # --------------------------------------------------------------------------- @@ -421,19 +257,6 @@ def step_persona_bar_shows_name(context): assert "scope refs" in bar._text -# --------------------------------------------------------------------------- -# on_input_submitted helpers -# --------------------------------------------------------------------------- -def _submit_text(context, text): - """Set prompt value and fire on_input_submitted.""" - 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) - - # --------------------------------------------------------------------------- # on_input_submitted: empty text (lines 144-150) # --------------------------------------------------------------------------- @@ -455,38 +278,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) - - -# --------------------------------------------------------------------------- -# 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, - ), - ): - _submit_text(context, "!nothing") - - # --------------------------------------------------------------------------- # 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..c81de7689 --- /dev/null +++ b/features/steps/tui_shell_safety_steps.py @@ -0,0 +1,130 @@ +"""Shell safety-related step definitions for the TUI app coverage suite.""" + +from __future__ import annotations + +import os +from types import SimpleNamespace +from unittest.mock import patch + +from behave import given, then, when + +from cleveragents.tui.input.modes import InputMode, ModeResult +from cleveragents.tui.shell_safety import ( + DangerousCommandWarning, + DangerousPattern, + ShellDangerLevel, +) + +from features.steps._tui_helpers import _submit_text, _submit_text_with_mocked_shell + + +@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) + + +@when('I ask the app to confirm shell command "{command}"') +def step_confirm_shell_command(context, command): + context._tui_shell_confirmation = context._tui_app._confirm_dangerous_shell(command) + + +@then("the shell confirmation result should be allowed") +def step_shell_confirmation_allowed(context): + assert context._tui_shell_confirmation is True + + +@then("the shell confirmation result should be blocked") +def step_shell_confirmation_blocked(context): + assert context._tui_shell_confirmation is False + + +@when("I ask the app to handle a shell warning") +def step_handle_shell_warning(context): + pattern = DangerousPattern( + name="test_warning", + pattern=r"rm -rf", + level=ShellDangerLevel.CRITICAL, + description="test warning", + ) + warning = DangerousCommandWarning.from_pattern("rm -rf /tmp", pattern) + context._tui_shell_warning_result = context._tui_app._handle_shell_warning(warning) + + +@then("the shell warning callback result should be allowed") +def step_shell_warning_callback_allowed(context): + assert context._tui_shell_warning_result is True diff --git a/features/tui_app_coverage.feature b/features/tui_app_coverage.feature index f9ce5f258..13fa262de 100644 --- a/features/tui_app_coverage.feature +++ b/features/tui_app_coverage.feature @@ -146,6 +146,59 @@ 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 + + Scenario: ShellSafetyService owns dangerous command confirmation + Given a mock command router and persona state + When I instantiate the Textual TUI app + And I ask the app to confirm shell command "rm -rf /tmp" + Then the shell confirmation result should be allowed + + Scenario: shell confirmation allows safe commands without ShellSafetyService + 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 ask the app to confirm shell command "echo safe" + Then the shell confirmation result should be allowed + + Scenario: shell confirmation uses the dangerous shell flag without ShellSafetyService + 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 ask the app to confirm shell command "rm -rf /tmp" + Then the shell confirmation result should be blocked + + Scenario: shell warning callback allows execution when warnings are disabled + 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 ask the app to handle a shell warning + Then the shell warning callback result should be allowed + # --- on_input_submitted with shell returning None (lines 170-171) --- Scenario: on_input_submitted handles None shell result 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..b77c1ff40 --- /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: + ... ${SPACE * 4}warnings.append(warning) + ... ${SPACE * 4}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: + ... ${SPACE * 4}counter["count"] += 1 + ... ${SPACE * 4}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/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..2b98f60a4 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, @@ -20,7 +21,9 @@ 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 from cleveragents.tui.widgets.actor_selection_overlay import ActorSelectionOverlay from cleveragents.tui.widgets.help_panel_overlay import ( @@ -296,6 +299,14 @@ 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 = self._settings.shell_warn_dangerous + self._allow_dangerous_shell = self._resolve_allow_dangerous_shell() + 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 +319,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 +347,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 +485,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 +594,50 @@ if _TEXTUAL_AVAILABLE: ) worker.done_callback = _on_llm_done + def _confirm_dangerous_shell(self, command: str) -> bool: + if self._shell_safety is not None: + # 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: + 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 + + shell_warning.update("⚠ Potentially destructive command detected") + shell_warning.display = True + if hasattr(prompt, "add_class"): + prompt.add_class("dangerous") + + 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") + + @staticmethod + def _resolve_allow_dangerous_shell() -> bool: + raw = os.environ.get("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", "").strip() + # Default to disallowing dangerous commands unless explicitly enabled. + 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..d1d79641d 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,25 @@ 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 + + def _safety_gate(_cmd: str, *, allow: bool = allowed) -> bool: + return allow + + confirm = _safety_gate shell_result = run_shell_command( command, - confirm_dangerous=self._shell_confirm, + confirm_dangerous=confirm, timeout_seconds=self._shell_timeout_seconds, ) return ModeResult( @@ -83,6 +100,7 @@ class InputModeRouter: references=[], shell_result=shell_result, command_result=None, + shell_warning=warning, ) parse_result: ReferenceParseResult = parse_references(text) @@ -92,4 +110,5 @@ class InputModeRouter: references=[match.canonical for match in parse_result.matches], shell_result=None, command_result=None, + shell_warning=None, ) 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, diff --git a/src/cleveragents/tui/widgets/prompt.py b/src/cleveragents/tui/widgets/prompt.py index 3f31272b3..86a8909c2 100644 --- a/src/cleveragents/tui/widgets/prompt.py +++ b/src/cleveragents/tui/widgets/prompt.py @@ -248,12 +248,20 @@ class _TextualPromptInput(_PromptSymbolMixin, _HorizontalBase): class _FallbackPromptInput(_PromptSymbolMixin): - """Fallback prompt input used when Textual is unavailable.""" + """Fallback prompt input used when Textual is unavailable. + + Mirrors the Textual Widget CSS-class API (``add_class`` / + ``remove_class`` / ``has_class``) so callers that toggle styling + classes on the outer prompt (e.g. ``add_class("dangerous")`` from + the shell-safety surfacing path) work uniformly across production + and fallback/mock paths. + """ def __init__(self, placeholder: str = "", **_: object) -> None: self.placeholder = placeholder self._input = cast(_MutableValueInput, _InputBase()) self._current_symbol = _PROMPT_SYMBOLS[InputMode.NORMAL] + self._classes: set[str] = set() self._update_symbol(self._input.value) @property @@ -270,6 +278,15 @@ class _FallbackPromptInput(_PromptSymbolMixin): if callable(focus): focus() + def add_class(self, name: str) -> None: + self._classes.add(name) + + def remove_class(self, name: str) -> None: + self._classes.discard(name) + + def has_class(self, name: str) -> bool: + return name in self._classes + def _apply_symbol(self, symbol: str) -> None: self._current_symbol = symbol