fix(tui): use ShellSafetyService regex patterns in looks_dangerous() #10642

Open
HAL9000 wants to merge 4 commits from fix/v370/shell-safety-regex into master
5 changed files with 96 additions and 30 deletions
@@ -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
+36 -1
View File
@@ -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"
+8 -1
View File
@@ -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
+4 -2
View File
@@ -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
+44 -24
View File
@@ -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,