diff --git a/features/steps/tui_shell_exec_coverage_steps.py b/features/steps/tui_shell_exec_coverage_steps.py index 24376af07..6f73f5771 100644 --- a/features/steps/tui_shell_exec_coverage_steps.py +++ b/features/steps/tui_shell_exec_coverage_steps.py @@ -78,8 +78,9 @@ def step_confirm_callback_true(context): """Create a confirm_dangerous callback that approves dangerous commands.""" context.confirm_called = False - def confirm_true(cmd): + def confirm_true(warning): context.confirm_called = True + context.confirm_warning = warning return True context.confirm_callback = confirm_true @@ -90,8 +91,9 @@ def step_confirm_callback_false(context): """Create a confirm_dangerous callback that rejects dangerous commands.""" context.confirm_called = False - def confirm_false(cmd): + def confirm_false(warning): context.confirm_called = True + context.confirm_warning = warning return False context.confirm_callback = confirm_false diff --git a/features/tui_shell_exec_coverage.feature b/features/tui_shell_exec_coverage.feature index 742c5c183..e312c9963 100644 --- a/features/tui_shell_exec_coverage.feature +++ b/features/tui_shell_exec_coverage.feature @@ -38,10 +38,45 @@ 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 by shell safety policy" + And the shell result stderr should be "blocked dangerous shell command" Scenario: Command that exceeds timeout returns timeout result Given subprocess run is mocked to raise TimeoutExpired When I run a shell command "sleep 999" with a 1 second timeout Then the shell result exit code should be 124 And the shell result stderr should contain "command timed out after" + + Scenario: Dangerous command with double space is blocked + When I run a shell command "rm -rf /" + Then the shell result exit code should be 1 + And the shell result stderr should be "blocked dangerous shell command" + + Scenario: Dangerous command with tab character is blocked + When I run a shell command "rm -rf /" + Then the shell result exit code should be 1 + And the shell result stderr should be "blocked dangerous shell command" + + Scenario: Dangerous command with sudo prefix is blocked + When I run a shell command "sudo rm -rf /" + Then the shell result exit code should be 1 + And the shell result stderr should be "blocked dangerous shell command" + + Scenario: Dangerous command with mixed spacing is blocked + When I run a shell command "rm -rf /" + Then the shell result exit code should be 1 + And the shell result stderr should be "blocked dangerous shell command" + + Scenario: Git push force with spacing variations is blocked + When I run a shell command "git push --force" + Then the shell result exit code should be 1 + And the shell result stderr should be "blocked dangerous shell command" + + Scenario: mkfs with spacing variations is blocked + When I run a shell command "mkfs.ext4 /dev/sdb1" + Then the shell result exit code should be 1 + And the shell result stderr should be "blocked dangerous shell command" + + Scenario: dd if= with spacing variations is blocked + When I run a shell command "dd if=/dev/zero" + Then the shell result exit code should be 1 + And the shell result stderr should be "blocked dangerous shell command" diff --git a/src/cleveragents/tui/app.py b/src/cleveragents/tui/app.py index 6dceed3d7..0690314e4 100644 --- a/src/cleveragents/tui/app.py +++ b/src/cleveragents/tui/app.py @@ -625,10 +625,17 @@ if _TEXTUAL_AVAILABLE: ) worker.done_callback = _on_llm_done - def _confirm_dangerous_shell(self, command: str) -> bool: + def _confirm_dangerous_shell( + self, warning_or_command: DangerousCommandWarning | str + ) -> bool: if self._shell_safety is not None: # The ShellSafetyService already provided the execution verdict. return True + command = ( + warning_or_command.command + if isinstance(warning_or_command, DangerousCommandWarning) + else warning_or_command + ) if not looks_dangerous(command): return True return self._allow_dangerous_shell diff --git a/src/cleveragents/tui/input/modes.py b/src/cleveragents/tui/input/modes.py index d1d79641d..2ebdf9fe0 100644 --- a/src/cleveragents/tui/input/modes.py +++ b/src/cleveragents/tui/input/modes.py @@ -42,7 +42,7 @@ class InputModeRouter: self, command_handler: Callable[[str], str], *, - shell_confirm: Callable[[str], bool] | None = None, + shell_confirm: Callable[[DangerousCommandWarning], bool] | None = None, shell_timeout_seconds: int = 30, shell_safety: ShellSafetyService | None = None, ) -> None: @@ -85,7 +85,9 @@ class InputModeRouter: warning = safety_result.warning allowed = safety_result.allowed - def _safety_gate(_cmd: str, *, allow: bool = allowed) -> bool: + def _safety_gate( + _warning: DangerousCommandWarning, *, allow: bool = allowed + ) -> bool: return allow confirm = _safety_gate diff --git a/src/cleveragents/tui/input/shell_exec.py b/src/cleveragents/tui/input/shell_exec.py index 9e57af77e..8567affd1 100644 --- a/src/cleveragents/tui/input/shell_exec.py +++ b/src/cleveragents/tui/input/shell_exec.py @@ -7,6 +7,12 @@ import subprocess from collections.abc import Callable from dataclasses import dataclass +from cleveragents.tui.shell_safety import ( + DangerousCommandWarning, + ShellDangerLevel, + ShellSafetyService, +) + @dataclass(slots=True, frozen=True) class ShellResult: @@ -19,22 +25,23 @@ class ShellResult: def looks_dangerous(command: str) -> bool: - """Best-effort dangerous command detector.""" - lowered = command.strip().lower() - patterns = ( - "rm -rf /", - "git push --force", - "mkfs.", - "dd if=", - ":(){:|:&};:", + """Best-effort dangerous command detector. + + .. deprecated:: + Use :class:`~cleveragents.tui.shell_safety.ShellSafetyService` + instead. This function delegates to the regex-based shell safety + service and is kept only for backward compatibility. + """ + service = ShellSafetyService( + block_level=ShellDangerLevel.LOW, ) - return any(pattern in lowered for pattern in patterns) + return not service.check_command(command).allowed def run_shell_command( command: str, *, - confirm_dangerous: Callable[[str], bool] | None = None, + confirm_dangerous: Callable[[DangerousCommandWarning], bool] | None = None, timeout_seconds: int = 30, ) -> ShellResult: """Execute shell command with basic safeguards. @@ -43,6 +50,16 @@ def run_shell_command( - Shell mode is a convenience feature for local development, not a sandbox. - We block obviously dangerous command patterns unless explicitly confirmed. - We apply a timeout to avoid hanging the UI event loop indefinitely. + + Args: + command: The shell command to execute. + confirm_dangerous: Optional callback invoked when a dangerous command + is detected. Receives a :class:`DangerousCommandWarning` and + should return ``True`` to allow execution or ``False`` to block. + timeout_seconds: Maximum time in seconds to allow the command to run. + + Returns: + A :class:`ShellResult` with the command execution outcome. """ if not command.strip(): return ShellResult( @@ -55,20 +72,23 @@ def run_shell_command( stdout="", stderr="shell mode is disabled", ) - 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", - ) + + service = ShellSafetyService( + block_level=ShellDangerLevel.LOW, + ) + check_result = service.check_command(command) + + if not check_result.allowed: + confirmed = False + if confirm_dangerous is not None and check_result.warning is not None: + confirmed = confirm_dangerous(check_result.warning) + if not confirmed: + return ShellResult( + command=command, + exit_code=1, + stdout="", + stderr="blocked dangerous shell command", + ) try: proc = subprocess.run( command,