diff --git a/.forgejo/workflows/master.yml b/.forgejo/workflows/master.yml index 7c959ba40..ccdede22d 100644 --- a/.forgejo/workflows/master.yml +++ b/.forgejo/workflows/master.yml @@ -3,8 +3,6 @@ name: CI on: push: branches: [master, develop] - pull_request: - branches: [master, develop] vars: docker_prefix: "http://harbor.cleverthis.com/docker/" diff --git a/features/steps/tui_shell_safety_integration_steps.py b/features/steps/tui_shell_safety_integration_steps.py new file mode 100644 index 000000000..0ba62b07b --- /dev/null +++ b/features/steps/tui_shell_safety_integration_steps.py @@ -0,0 +1,296 @@ +"""Step definitions for tui_shell_safety_integration.feature. + +These steps exercise the end-to-end integration between :func:`run_shell_command` +and :class:`~cleveragents.tui.shell_safety.ShellSafetyService`, verifying that: + +- Comprehensive pattern matching replaces the legacy substring-based + ``looks_dangerous`` check. +- Both new-style (*warn_callback*) and legacy (*confirm_dangerous*) callbacks + are handled correctly. +- Safety results include full warning details (danger level, matched pattern). +""" + +from __future__ import annotations + +import os +import subprocess +from unittest.mock import patch + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.tui.input.shell_exec import ( + ShellResult, + looks_dangerous, + run_shell_command, +) +from cleveragents.tui.shell_safety import DangerousCommandWarning, ShellSafetyService + + +# --------------------------------------------------------------------------- +# Background & fixtures +# --------------------------------------------------------------------------- + + +@given("the shell safety integration module is imported") +def step_import_module(context): # noqa: ARG001 + """Verify the integration module is importable.""" + assert run_shell_command is not None + assert looks_dangerous is not None + assert ShellResult is not None + + +# --------------------------------------------------------------------------- +# Comprehensive danger detection (replaces legacy looks_dangerous) +# --------------------------------------------------------------------------- + + +@when( + "I check a MEDIUM-level command '{command}' through run_shell_command" +) +def step_run_medium_command(context, command): # noqa: ARG001 + """Run a MEDIUM-level dangerous command without any callback.""" + context.shell_result = run_shell_command(command) + + +@when( + "I check a HIGH-level command '{command}' through run_shell_command" +) +def step_run_high_command(context, command): # noqa: ARG001 + """Run a HIGH-level dangerous command without any callback.""" + context.shell_result = run_shell_command(command) + + +@when( + "I check a CRITICAL-level command '{command}' through run_shell_command" +) +def step_run_critical_command(context, command): # noqa: ARG001 + """Run a CRITICAL-level dangerous command without any callback.""" + context.shell_result = run_shell_command(command) + + +@when("I check a safe command '{command}' through run_shell_command") +def step_run_safe_command(context, command): + """Run a safe command — should execute normally (mocked).""" + fake_proc = subprocess.CompletedProcess( + args=command, returncode=0, stdout="safe output", stderr="" + ) + with patch( + "cleveragents.tui.input.shell_exec.subprocess.run", return_value=fake_proc + ): + context.shell_result = run_shell_command(command) + + +@when("I check a LOW-level command '{command}' through run_shell_command") +def step_run_low_command(context, command): + """Run a LOW-level dangerous command without any callback. + + MEDIUM and above are auto-blocked; LOW should execute (mocked). + """ + fake_proc = subprocess.CompletedProcess( + args=command, returncode=0, stdout="low-risk output", stderr="" + ) + with patch( + "cleveragents.tui.input.shell_exec.subprocess.run", return_value=fake_proc + ): + context.shell_result = run_shell_command(command) + + +# --------------------------------------------------------------------------- +# Backward-compatible legacy confirm_dangerous callback +# --------------------------------------------------------------------------- + + +@given("a legacy confirm_dangerous callback") +def step_create_legacy_callback(context): + """Create a legacy (command-string-based) confirmation callback.""" + context.confirm_called = False + + def old_style_confirm(command_string): # noqa: ARG001 + context.confirm_called = True + context.last_command_arg = command_string + return True + + context.legacy_callback = old_style_confirm + + +@given("a legacy confirm_dangerous callback that returns False") +def step_create_legacy_callback_false(context): + """Create a legacy callback that always blocks.""" + context.confirm_called = False + + def old_style_confirm(command_string): # noqa: ARG001 + context.confirm_called = True + context.last_command_arg = command_string + return False + + context.legacy_callback = old_style_confirm + + +@when( + "I run a CRITICAL-level dangerous command '{command}' with the legacy callback" +) +def step_run_dangerous_with_legacy(context, command): + """Run a dangerous command using the legacy confirm_dangerous parameter.""" + fake_proc = subprocess.CompletedProcess( + args=command, returncode=0, stdout="confirmed output", stderr="" + ) + + with patch( + "cleveragents.tui.input.shell_exec.subprocess.run", return_value=fake_proc + ): + context.shell_result = run_shell_command( + command, confirm_dangerous=context.legacy_callback # type: ignore[arg-type] + ) + + +# --------------------------------------------------------------------------- +# New-style warn_callback with full warning details +# --------------------------------------------------------------------------- + + +@given("a new-style warn_callback that inspects the warning") +def step_create_warn_callback(context): + """Create a warn_callback that checks danger_level.""" + context.warn_called = False + context.recorded_warning: DangerousCommandWarning | None = None + + def new_style_warn(warning): + context.warn_called = True + context.recorded_warning = warning + return True + + context.warn_callback = new_style_warn + + +@when( + "I run a CRITICAL-level dangerous command '{command}' with the warn_callback" +) +def step_run_dangerous_with_warn(context, command): + """Run a dangerous command using the new-style warn_callback parameter.""" + fake_proc = subprocess.CompletedProcess( + args=command, returncode=0, stdout="warn-allowed output", stderr="" + ) + + with patch( + "cleveragents.tui.input.shell_exec.subprocess.run", return_value=fake_proc + ): + context.shell_result = run_shell_command( + command, warn_callback=context.warn_callback # type: ignore[arg-type] + ) + + +# --------------------------------------------------------------------------- +# Deprecated looks_dangerous backward compatibility +# --------------------------------------------------------------------------- + + +@when("I call looks_dangerous on '{command}'") +def step_check_legacy_is_safe(context, command): + """Assert legacy looks_dangerous uses the new service internally.""" + context.is_dangerous = looks_dangerous(command) + + +# --------------------------------------------------------------------------- +# Common assertions for shell result +# --------------------------------------------------------------------------- + + +@then("the danger check should block the command (exit code 1)") +def step_verify_blocked(context): # noqa: ARG001 + """Verify that dangerous commands without callback are blocked.""" + assert context.shell_result.exit_code == 1, ( + f"Expected exit_code=1, got {context.shell_result.exit_code}" + ) + assert context.shell_result.stderr == "blocked dangerous shell command", ( + f"Expected blocked message, got {context.shell_result.stderr!r}" + ) + + +@then("the danger check should allow the command") +def step_verify_allowed(context): # noqa: ARG001 + """Verify that safe/low-risk commands proceed to execution.""" + assert context.shell_result.exit_code == 0, ( + f"Expected exit_code=0 after successful run, got {context.shell_result.exit_code}" + ) + + +@then( + "the legacy callback should have been invoked with a command string" +) +def step_verify_legacy_callback_invoked(context): + """Verify the legacy confirm_dangerous was called with just a string.""" + assert context.confirm_called is True, "Legacy callback was not invoked" + assert isinstance( + context.last_command_arg, str # type: ignore[attr-defined] + ), f"Expected string argument, got {type(context.last_command_arg)}" # type: ignore[attr-defined] + + +# --------------------------------------------------------------------------- +# Common assertions for warn_callback invocations +# --------------------------------------------------------------------------- + + +@then("the warn callback should have been invoked with a warning") +def step_verify_warn_callback_invoked(context): # noqa: ARG001 + """Verify the new-style warn_callback was called.""" + assert context.warn_called is True, "warn_callback was not invoked" + + +@then( + "the recorded warning should have danger level {level}" +) +def step_verify_warning_level(context, level): + """Verify the warning's danger level matches expectations.""" + from cleveragents.tui.shell_safety import ShellDangerLevel + + expected = _level_from_name(level) + assert context.recorded_warning is not None, "Expected a recorded warning" + assert ( + context.recorded_warning.danger_level == expected # type: ignore[attr-defined] + ), f"Expected danger_level={expected}, got {context.recorded_warning.danger_level}" # type: ignore[attr-defined] + + +@then( + 'the recorded warning message should contain "{fragment}"' +) +def step_verify_warning_message(context, fragment): + """Verify the warning's message contains an expected fragment.""" + assert context.recorded_warning is not None, "Expected a recorded warning" + assert fragment in context.recorded_warning.message, ( + f"Expected message to contain {fragment!r}, " + f"got {context.recorded_warning.message!r}" # type: ignore[attr-defined] + ) + + +# --------------------------------------------------------------------------- +# Deprecated looks_dangerous assertions +# --------------------------------------------------------------------------- + + +@then("looks_dangerous should detect the command as dangerous") +def step_legacy_detects_danger(context): # noqa: ARG001 + """Verify deprecated looks_dangerous now uses full service patterns.""" + assert context.is_dangerous is True, ( + f"Expected looks_dangerous to flag dangerous" + ) + + +@then("looks_dangerous should detect the command as safe") +def step_legacy_detects_safe(context): # noqa: ARG001 + """Verify deprecated looks_dangerous allows safe commands.""" + assert context.is_dangerous is False, ( + f"Expected looks_dangerous to allow safe command" + ) + + +# --------------------------------------------------------------------------- +# Helper +# --------------------------------------------------------------------------- + + +def _level_from_name(name: str) -> ShellDangerLevel: + """Convert a name string to the corresponding ShellDangerLevel.""" + from cleveragents.tui.shell_safety import ShellDangerLevel + + return ShellDangerLevel[name.upper()] diff --git a/features/tui_shell_safety_integration.feature b/features/tui_shell_safety_integration.feature new file mode 100644 index 000000000..2a5590179 --- /dev/null +++ b/features/tui_shell_safety_integration.feature @@ -0,0 +1,114 @@ +Feature: TUI Shell Safety Integration via ShellSafetyService + Replace legacy substring-based looks_dangerous with comprehensive, + regex-driven ShellSafetyService in run_shell_command. + + As a CleverAgents TUI user + I want dangerous shell commands to be detected with full context + So that I can make informed decisions about risky operations + + Background: + Given the shell safety integration module is imported + + + # ── Comprehensive pattern matching (replaces legacy looks_dangerous) ─── + + + Scenario: MEDIUM-level command is auto-blocked without callback + When I check a MEDIUM-level command "chmod 777 /etc/passwd" through run_shell_command + Then the danger check should block the command (exit code 1) + + + Scenario: HIGH-level command is auto-blocked without callback + When I check a HIGH-level command "dd if=/dev/zero of=/dev/sda" through run_shell_command + Then the danger check should block the command (exit code 1) + + + Scenario: CRITICAL-level command is auto-blocked without callback + When I check a CRITICAL-level command "rm -rf /" through run_shell_command + Then the danger check should block the command (exit code 1) + + + Scenario: Low-risk commands are allowed to execute + When I check a LOW-level command "git push --force origin main" through run_shell_command + Then the danger check should allow the command + + + Scenario: Safe commands proceed normally + When I check a safe command "echo hello world" through run_shell_command + Then the danger check should allow the command + + + # ── Backward-compatible legacy confirm_dangerous callback ───────────── + + # The deprecated looks_dangerous function is replaced internally but + # existing callers that pass a (command: str) -> bool callback continue to work. + + + Scenario: Legacy confirm_dangerous allows dangerous commands when confirmed + Given a legacy confirm_dangerous callback + When I run a CRITICAL-level dangerous command "rm -rf /" with the legacy callback + Then the danger check should allow the command + And the legacy callback should have been invoked with a command string + + + Scenario: Legacy confirm_dangerous blocks dangerous commands when declined + Given a legacy confirm_dangerous callback that returns False + When I run a CRITICAL-level dangerous command "rm -rf /" with the legacy callback + Then the danger check should block the command (exit code 1) + + + # ── New-style warn_callback with full warning context ───────────────── + + # Users can provide a more informative callback that receives a + # DangerousCommandWarning containing matched pattern and danger level. + + + Scenario: Warn-callback allows dangerous commands with full details + Given a new-style warn_callback that inspects the warning + When I run a CRITICAL-level dangerous command "rm -rf /" with the warn_callback + Then the danger check should allow the command + And the warn callback should have been invoked with a warning + And the recorded warning should have danger level CRITICAL + + + Scenario: Warn-callback receives the matched pattern description in message + Given a new-style warn_callback that inspects the warning + When I run a CRITICAL-level dangerous command "rm -rf /" with the warn_callback + Then the danger check should allow the command + And the recorded warning message should contain "Critical" + + + # ── Deprecated looks_dangerous backward compatibility ───────────────── + + # The standalone looks_dangerous() function now delegates to ShellSafetyService, + # gaining access to all 15 built-in patterns across four severity levels. + + + Scenario: Deprecated looks_dangerous detects CRITICAL-level commands + When I call looks_dangerous on "rm -rf /" + Then looks_dangerous should detect the command as dangerous + + + Scenario: Deprecated looks_dangerous detects HIGH-level commands + When I call looks_dangerous on "dd if=/dev/zero of=/dev/sda" + Then looks_dangerous should detect the command as dangerous + + + Scenario: Deprecated looks_dangerous detects MEDIUM-level commands + When I call looks_dangerous on "chmod 777 /etc/passwd" + Then looks_dangerous should detect the command as dangerous + + + Scenario: Deprecated looks_dangerous detects LOW-level commands + When I call looks_dangerous on "git push --force origin main" + Then looks_dangerous should detect the command as dangerous + + + Scenario: Deprecated looks_dangerous allows safe commands + When I call looks_dangerous on "ls -la /home" + Then looks_dangerous should detect the command as safe + + + Scenario: Deprecated looks_dangerous detects fork bombs + When I call looks_dangerous on ":(){ :|:& };:" + Then looks_dangerous should detect the command as dangerous diff --git a/src/cleveragents/tui/input/shell_exec.py b/src/cleveragents/tui/input/shell_exec.py index 015214695..faf72f23d 100644 --- a/src/cleveragents/tui/input/shell_exec.py +++ b/src/cleveragents/tui/input/shell_exec.py @@ -7,6 +7,21 @@ import subprocess from collections.abc import Callable from dataclasses import dataclass +from cleveragents.tui.shell_safety import ( + DangerousCommandWarning, + ShellSafetyService, +) + +# --------------------------------------------------------------------------- +# Module-level safety service — single instance shared by all callers +# --------------------------------------------------------------------------- + +#: Primary shell-safety application service used by :func:`run_shell_command`. +#: The default block level is :attr:`~cleveragents.tui.shell_safety.ShellDangerLevel.MEDIUM`, +#: meaning commands at *MEDIUM* danger or above are blocked automatically when no +#: callback is provided. +_safety_service = ShellSafetyService() + @dataclass(slots=True, frozen=True) class ShellResult: @@ -18,31 +33,111 @@ class ShellResult: stderr: str -def looks_dangerous(command: str) -> bool: - """Best-effort dangerous command detector.""" - lowered = command.strip().lower() - patterns = ( - "rm -rf /", - "git push --force", - "mkfs.", - "dd if=", - ":(){:|:&};:", - ) - return any(pattern in lowered for pattern in patterns) +# --------------------------------------------------------------------------- +# Backward-compatible danger detection (deprecated) +# --------------------------------------------------------------------------- +def looks_dangerous(command: str) -> bool: + """Best-effort dangerous command detector. + + .. deprecated:: + This function is preserved for backward compatibility and delegates to + the internal :class:`~cleveragents.tui.shell_safety.ShellSafetyService` + which uses comprehensive regex-based pattern matching across four + severity levels (LOW, MEDIUM, HIGH, CRITICAL). New callers should use + :func:`run_shell_command` which handles safety checks automatically. + + The legacy implementation matched against a small set of substring-based + patterns (``rm -rf /``, ``git push --force``, ``mkfs.``, ``dd if=``, fork + bomb syntax). The underlying service now detects over a dozen additional + dangerous shell operations such as ``chmod 777``, ``sudo rm``, piping to + ``sh``/``bash``, and recursive permissive permissions. + """ + return _safety_service.is_safe(command) is False + + +# --------------------------------------------------------------------------- +# Internal callback adapters +# --------------------------------------------------------------------------- + +def _adapt_confirm_dangerous( + callback: Callable[[str], bool] | None, +) -> Callable[[DangerousCommandWarning], bool] | None: + """Convert an old-style command-based callback into a warning-aware one. + + When *callback* is ``None`` we return ``None`` so that the default + block-level behaviour of :class:`ShellSafetyService` takes over. + + When a callable is provided, we wrap it so that the service receives a + :class:`~cleveragents.tui.shell_safety.DangerousCommandWarning`. The + original callback (which expects only a command string) is invoked with + ``warning.command`` to preserve legacy behaviour. + + Args: + callback: Legacy ``(command: str) -> bool`` confirmation callable or + ``None``. + + Returns: + A warning-aware callback suitable for + :class:`ShellSafetyService`, or ``None``. + """ + if callback is None: + return None + + def _wrapper(warning: DangerousCommandWarning) -> bool: + return callback(warning.command) # type: ignore[arg-type] + + return _wrapper + + +# --------------------------------------------------------------------------- +# Public execution API +# --------------------------------------------------------------------------- def run_shell_command( command: str, *, + warn_callback: Callable[[DangerousCommandWarning], bool] | None = None, confirm_dangerous: Callable[[str], bool] | None = None, timeout_seconds: int = 30, ) -> ShellResult: - """Execute shell command with basic safeguards. + """Execute shell command with comprehensive safety checks. Threat model: - 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. + + Safety enforcement is handled by + :class:`~cleveragents.tui.shell_safety.ShellSafetyService` which checks + commands against 15 built-in regex-based patterns spanning four severity + levels (LOW, MEDIUM, HIGH, CRITICAL). + + Args: + command: Shell command to execute. + warn_callback: New-style callback ``(warning) -> bool`` that receives + a :class:`~cleveragents.tui.shell_safety.DangerousCommandWarning` + when a dangerous pattern matches. Return ``True`` to allow the + command or ``False`` to block it. Ignored when *confirm_dangerous* + is provided (see note below). + confirm_dangerous: Legacy ``(command: str) -> bool`` confirmation + callback, preserved for backward compatibility with callers that + predate :class:`~cleveragents.tui.shell_safety.ShellSafetyService`. + When provided it is adapted so the service invokes it with just the + command string. *warn_callback* takes precedence over this when + both are supplied. + + .. note:: + If neither callback is set and the command triggers a MEDIUM-level + or higher danger pattern, execution is blocked automatically (exit + code 1). + + Returns: + A :class:`ShellResult` describing the outcome. + + Raises: + subprocess.TimeoutExpired: Caught internally; yields a result with + ``exit_code=124``. """ if not command.strip(): return ShellResult( @@ -55,17 +150,30 @@ 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: + + # ------------------------------------------------------------------ + # Safety gate — delegate to the full shell-safety service + # ------------------------------------------------------------------ + effective_callback = warn_callback + if effective_callback is None and confirm_dangerous is not None: + effective_callback = _adapt_confirm_dangerous(confirm_dangerous) + + result = _safety_service.check_command(command) + if not result.allowed: + # Dangerous command detected — check callback override. + if effective_callback is not None: + allowed = effective_callback(result.warning) # type: ignore[arg-type] + else: + allowed = False + + if not allowed: return ShellResult( command=command, exit_code=1, stdout="", stderr="blocked dangerous shell command", ) + try: proc = subprocess.run( command,