feat(tui): implement shell danger detection patterns #1284

Merged
freemo merged 1 commits from feature/m8-tui-shell-danger-detection into master 2026-04-03 01:15:23 +00:00
9 changed files with 1154 additions and 0 deletions
@@ -0,0 +1,392 @@
"""Step definitions for tui_shell_danger_detection.feature.
Tests for:
- ShellDangerLevel enum
- DangerousPattern value object
- DangerousCommandWarning value object
- DangerousPatternDetector domain service
- ShellSafetyService application service
"""
from __future__ import annotations
from behave import given, then, when
from cleveragents.tui.shell_safety import (
DangerousCommandWarning,
DangerousPattern,
DangerousPatternDetector,
ShellDangerLevel,
ShellSafetyService,
)
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("the shell safety module is imported")
def step_shell_safety_imported(context: object) -> None:
"""Verify the shell safety module is importable."""
assert ShellDangerLevel is not None
assert DangerousPattern is not None
assert DangerousCommandWarning is not None
assert DangerousPatternDetector is not None
assert ShellSafetyService is not None
# ---------------------------------------------------------------------------
# ShellDangerLevel
# ---------------------------------------------------------------------------
@then("the ShellDangerLevel enum should have levels LOW MEDIUM HIGH CRITICAL")
def step_danger_level_has_four_levels(context: object) -> None:
"""Verify all four danger levels exist."""
assert hasattr(ShellDangerLevel, "LOW")
assert hasattr(ShellDangerLevel, "MEDIUM")
assert hasattr(ShellDangerLevel, "HIGH")
assert hasattr(ShellDangerLevel, "CRITICAL")
@then("LOW should be less severe than MEDIUM")
def step_low_less_than_medium(context: object) -> None:
assert ShellDangerLevel.LOW < ShellDangerLevel.MEDIUM
@then("MEDIUM should be less severe than HIGH")
def step_medium_less_than_high(context: object) -> None:
assert ShellDangerLevel.MEDIUM < ShellDangerLevel.HIGH
@then("HIGH should be less severe than CRITICAL")
def step_high_less_than_critical(context: object) -> None:
assert ShellDangerLevel.HIGH < ShellDangerLevel.CRITICAL
# ---------------------------------------------------------------------------
# DangerousPattern
# ---------------------------------------------------------------------------
def _level_from_name(name: str) -> ShellDangerLevel:
return ShellDangerLevel[name.upper()]
@given('a DangerousPattern named "{name}" with pattern "{pattern}" at {level} level')
def step_create_dangerous_pattern(
context: object, name: str, pattern: str, level: str
) -> None:
context.pattern = DangerousPattern( # type: ignore[attr-defined]
name=name,
pattern=pattern,
level=_level_from_name(level),
description=f"Test pattern: {name}",
)
@given(
'a DangerousPattern named "{name}" with pattern "{pattern}" at {level} level and description "{description}"'
)
def step_create_dangerous_pattern_with_description(
context: object, name: str, pattern: str, level: str, description: str
) -> None:
context.pattern = DangerousPattern( # type: ignore[attr-defined]
name=name,
pattern=pattern,
level=_level_from_name(level),
description=description,
)
@when('I check if "{command}" matches the pattern')
def step_check_pattern_match(context: object, command: str) -> None:
context.pattern_matched = context.pattern.matches(command) # type: ignore[attr-defined]
@then("the pattern should match")
def step_pattern_should_match(context: object) -> None:
assert context.pattern_matched is True, "Expected pattern to match but it did not" # type: ignore[attr-defined]
@then("the pattern should not match")
def step_pattern_should_not_match(context: object) -> None:
assert context.pattern_matched is False, "Expected pattern not to match but it did" # type: ignore[attr-defined]
# ---------------------------------------------------------------------------
# DangerousCommandWarning
# ---------------------------------------------------------------------------
@when('I create a warning for command "{command}" from the pattern')
def step_create_warning_from_pattern(context: object, command: str) -> None:
context.warning = DangerousCommandWarning.from_pattern( # type: ignore[attr-defined]
command,
context.pattern, # type: ignore[attr-defined]
)
@then('the warning command should be "{expected}"')
def step_warning_command(context: object, expected: str) -> None:
assert context.warning.command == expected, ( # type: ignore[attr-defined]
f"Expected command={expected!r}, got {context.warning.command!r}" # type: ignore[attr-defined]
)
@then("the warning danger level should be {level}")
def step_warning_danger_level(context: object, level: str) -> None:
expected = _level_from_name(level)
assert context.warning.danger_level == expected, ( # type: ignore[attr-defined]
f"Expected danger_level={expected}, got {context.warning.danger_level}" # type: ignore[attr-defined]
)
@then('the warning message should contain "{fragment}"')
def step_warning_message_contains(context: object, fragment: str) -> None:
assert fragment in context.warning.message, ( # type: ignore[attr-defined]
f"Expected message to contain {fragment!r}, got {context.warning.message!r}" # type: ignore[attr-defined]
)
# ---------------------------------------------------------------------------
# DangerousPatternDetector
# ---------------------------------------------------------------------------
@given("a DangerousPatternDetector with default patterns")
def step_create_default_detector(context: object) -> None:
context.detector = DangerousPatternDetector() # type: ignore[attr-defined]
@then("the detector should have patterns registered")
def step_detector_has_patterns(context: object) -> None:
assert len(context.detector.patterns) > 0, "Expected detector to have patterns" # type: ignore[attr-defined]
@when('I check the command "{command}"')
def step_check_command(context: object, command: str) -> None:
context.last_command = command # type: ignore[attr-defined]
context.is_dangerous = context.detector.is_dangerous(command) # type: ignore[attr-defined]
context.max_level = context.detector.max_danger_level(command) # type: ignore[attr-defined]
@then("the command should be detected as dangerous")
def step_command_is_dangerous(context: object) -> None:
assert context.is_dangerous is True, ( # type: ignore[attr-defined]
f"Expected command {context.last_command!r} to be dangerous" # type: ignore[attr-defined]
)
@then("the command should not be detected as dangerous")
def step_command_is_not_dangerous(context: object) -> None:
assert context.is_dangerous is False, ( # type: ignore[attr-defined]
f"Expected command {context.last_command!r} to be safe but it was flagged as dangerous" # type: ignore[attr-defined]
)
@then("the max danger level should be {level}")
def step_max_danger_level(context: object, level: str) -> None:
expected = _level_from_name(level)
assert context.max_level == expected, ( # type: ignore[attr-defined]
f"Expected max danger level={expected}, got {context.max_level}" # type: ignore[attr-defined]
)
@when('I call check_first on "{command}"')
def step_call_check_first(context: object, command: str) -> None:
context.check_first_result = context.detector.check_first(command) # type: ignore[attr-defined]
@then("check_first should return a warning")
def step_check_first_returns_warning(context: object) -> None:
assert context.check_first_result is not None, (
"Expected check_first to return a warning"
) # type: ignore[attr-defined]
@then("check_first should return None")
def step_check_first_returns_none(context: object) -> None:
assert context.check_first_result is None, ( # type: ignore[attr-defined]
f"Expected check_first to return None, got {context.check_first_result!r}" # type: ignore[attr-defined]
)
@then("the first warning danger level should be {level}")
def step_first_warning_danger_level(context: object, level: str) -> None:
expected = _level_from_name(level)
assert context.check_first_result.danger_level == expected, ( # type: ignore[attr-defined]
f"Expected danger_level={expected}, got {context.check_first_result.danger_level}" # type: ignore[attr-defined]
)
@when('I call check on "{command}"')
def step_call_check(context: object, command: str) -> None:
context.check_result = context.detector.check(command) # type: ignore[attr-defined]
@then("check should return at least one warning")
def step_check_returns_warnings(context: object) -> None:
assert len(context.check_result) >= 1, ( # type: ignore[attr-defined]
f"Expected at least one warning, got {context.check_result!r}" # type: ignore[attr-defined]
)
@when('I add a custom pattern named "{name}" with pattern "{pattern}" at {level} level')
def step_add_custom_pattern(
context: object, name: str, pattern: str, level: str
) -> None:
custom = DangerousPattern(
name=name,
pattern=pattern,
level=_level_from_name(level),
description=f"Custom test pattern: {name}",
)
context.detector.add_pattern(custom) # type: ignore[attr-defined]
@when('I remove the pattern named "{name}"')
def step_remove_pattern(context: object, name: str) -> None:
context.detector.remove_pattern(name) # type: ignore[attr-defined]
@when(
'I replace all patterns with a single pattern named "{name}" matching "{pattern}" at {level} level'
)
def step_replace_patterns(context: object, name: str, pattern: str, level: str) -> None:
single = DangerousPattern(
name=name,
pattern=pattern,
level=_level_from_name(level),
description=f"Replacement pattern: {name}",
)
context.detector.replace_patterns([single]) # type: ignore[attr-defined]
@then("the detector patterns property should return a tuple")
def step_patterns_property_is_tuple(context: object) -> None:
result = context.detector.patterns # type: ignore[attr-defined]
assert isinstance(result, tuple), f"Expected tuple, got {type(result)}"
# ---------------------------------------------------------------------------
# ShellSafetyService
# ---------------------------------------------------------------------------
@given("a ShellSafetyService with default settings")
def step_create_default_safety_service(context: object) -> None:
context.service = ShellSafetyService() # type: ignore[attr-defined]
context.warn_callback_called = False # type: ignore[attr-defined]
@given("a ShellSafetyService with a warn_callback that returns True")
def step_create_service_with_allow_callback(context: object) -> None:
context.warn_callback_called = False # type: ignore[attr-defined]
def callback(warning: DangerousCommandWarning) -> bool:
context.warn_callback_called = True # type: ignore[attr-defined]
return True
context.service = ShellSafetyService(warn_callback=callback) # type: ignore[attr-defined]
@given("a ShellSafetyService with a warn_callback that returns False")
def step_create_service_with_block_callback(context: object) -> None:
context.warn_callback_called = False # type: ignore[attr-defined]
def callback(warning: DangerousCommandWarning) -> bool:
context.warn_callback_called = True # type: ignore[attr-defined]
return False
context.service = ShellSafetyService(warn_callback=callback) # type: ignore[attr-defined]
@given(
'a ShellSafetyService with an extra pattern "{name}" matching "{pattern}" at {level} level'
)
def step_create_service_with_extra_pattern(
context: object, name: str, pattern: str, level: str
) -> None:
extra = DangerousPattern(
name=name,
pattern=pattern,
level=_level_from_name(level),
description=f"Extra pattern: {name}",
)
context.service = ShellSafetyService(extra_patterns=[extra]) # type: ignore[attr-defined]
@when('I check the safety of command "{command}"')
def step_check_safety(context: object, command: str) -> None:
context.safety_result = context.service.check_command(command) # type: ignore[attr-defined]
@then("the safety check result should be allowed")
def step_safety_result_allowed(context: object) -> None:
assert context.safety_result.allowed is True, ( # type: ignore[attr-defined]
f"Expected result to be allowed, got blocked. Warning: {context.safety_result.warning}" # type: ignore[attr-defined]
)
@then("the safety check result should be blocked")
def step_safety_result_blocked(context: object) -> None:
assert context.safety_result.allowed is False, ( # type: ignore[attr-defined]
"Expected result to be blocked, got allowed"
)
@then("the safety check warning should be None")
def step_safety_warning_is_none(context: object) -> None:
assert context.safety_result.warning is None, ( # type: ignore[attr-defined]
f"Expected warning to be None, got {context.safety_result.warning!r}" # type: ignore[attr-defined]
)
@then("the safety check warning should not be None")
def step_safety_warning_is_not_none(context: object) -> None:
assert context.safety_result.warning is not None, "Expected warning to be set" # type: ignore[attr-defined]
@then("the warn_callback should have been called")
def step_warn_callback_called(context: object) -> None:
assert context.warn_callback_called is True, (
"Expected warn_callback to have been called"
) # type: ignore[attr-defined]
@when('I call is_safe on "{command}"')
def step_call_is_safe(context: object, command: str) -> None:
context.is_safe_result = context.service.is_safe(command) # type: ignore[attr-defined]
@then("is_safe should return True")
def step_is_safe_true(context: object) -> None:
assert context.is_safe_result is True, "Expected is_safe to return True" # type: ignore[attr-defined]
@then("is_safe should return False")
def step_is_safe_false(context: object) -> None:
assert context.is_safe_result is False, "Expected is_safe to return False" # type: ignore[attr-defined]
@then("the service detector property should be a DangerousPatternDetector")
def step_service_detector_property(context: object) -> None:
assert isinstance(context.service.detector, DangerousPatternDetector), ( # type: ignore[attr-defined]
f"Expected DangerousPatternDetector, got {type(context.service.detector)}" # type: ignore[attr-defined]
)
@then("the service block_level should be MEDIUM")
def step_service_block_level(context: object) -> None:
assert context.service.block_level == ShellDangerLevel.MEDIUM, ( # type: ignore[attr-defined]
f"Expected block_level=MEDIUM, got {context.service.block_level}" # type: ignore[attr-defined]
)
@then('the safety check result repr should contain "{fragment}"')
def step_safety_result_repr_contains(context: object, fragment: str) -> None:
result_repr = repr(context.safety_result) # type: ignore[attr-defined]
assert fragment in result_repr, (
f"Expected repr to contain {fragment!r}, got {result_repr!r}"
)
+236
View File
@@ -0,0 +1,236 @@
Feature: TUI Shell Danger Detection
As a TUI user
I want dangerous shell commands to be detected and warned about
So that I can avoid accidentally running destructive commands
Background:
Given the shell safety module is imported
# ── Danger level enum ────────────────────────────────────────────────────
Scenario: ShellDangerLevel enum has four levels
Then the ShellDangerLevel enum should have levels LOW MEDIUM HIGH CRITICAL
Scenario: ShellDangerLevel levels are ordered by severity
Then LOW should be less severe than MEDIUM
And MEDIUM should be less severe than HIGH
And HIGH should be less severe than CRITICAL
# ── DangerousPattern value object ────────────────────────────────────────
Scenario: DangerousPattern matches a dangerous command
Given a DangerousPattern named "test_rm" with pattern "rm -rf" at CRITICAL level
When I check if "rm -rf /" matches the pattern
Then the pattern should match
Scenario: DangerousPattern does not match a safe command
Given a DangerousPattern named "test_rm" with pattern "rm -rf" at CRITICAL level
When I check if "ls -la" matches the pattern
Then the pattern should not match
Scenario: DangerousPattern is case-insensitive by default
Given a DangerousPattern named "test_rm" with pattern "rm -rf" at CRITICAL level
When I check if "RM -RF /" matches the pattern
Then the pattern should match
# ── DangerousCommandWarning value object ─────────────────────────────────
Scenario: DangerousCommandWarning is created from a pattern
Given a DangerousPattern named "rm_root" with pattern "rm -rf /" at CRITICAL level
When I create a warning for command "rm -rf /" from the pattern
Then the warning command should be "rm -rf /"
And the warning danger level should be CRITICAL
And the warning message should contain "Critical"
And the warning message should contain "Dangerous command detected"
Scenario: DangerousCommandWarning message includes pattern description
Given a DangerousPattern named "fork_bomb" with pattern ":(){" at CRITICAL level and description "Fork bomb"
When I create a warning for command ":(){ :|:& };:" from the pattern
Then the warning message should contain "Fork bomb"
# ── DangerousPatternDetector ─────────────────────────────────────────────
Scenario: Detector uses default patterns when none provided
Given a DangerousPatternDetector with default patterns
Then the detector should have patterns registered
Scenario: Detector detects rm -rf root as CRITICAL
Given a DangerousPatternDetector with default patterns
When I check the command "rm -rf /"
Then the command should be detected as dangerous
And the max danger level should be CRITICAL
Scenario: Detector detects rm -rf wildcard as CRITICAL
Given a DangerousPatternDetector with default patterns
When I check the command "rm -rf /*"
Then the command should be detected as dangerous
And the max danger level should be CRITICAL
Scenario: Detector detects fork bomb as CRITICAL
Given a DangerousPatternDetector with default patterns
When I check the command ":(){ :|:& };:"
Then the command should be detected as dangerous
And the max danger level should be CRITICAL
Scenario: Detector detects dd if= as HIGH
Given a DangerousPatternDetector with default patterns
When I check the command "dd if=/dev/zero of=/dev/sda"
Then the command should be detected as dangerous
And the max danger level should be HIGH
Scenario: Detector detects mkfs as HIGH
Given a DangerousPatternDetector with default patterns
When I check the command "mkfs.ext4 /dev/sdb1"
Then the command should be detected as dangerous
And the max danger level should be HIGH
Scenario: Detector detects chmod 777 as MEDIUM
Given a DangerousPatternDetector with default patterns
When I check the command "chmod 777 /etc/passwd"
Then the command should be detected as dangerous
And the max danger level should be MEDIUM
Scenario: Detector detects sudo rm as MEDIUM
Given a DangerousPatternDetector with default patterns
When I check the command "sudo rm file.txt"
Then the command should be detected as dangerous
And the max danger level should be MEDIUM
Scenario: Detector detects wget piped to sh as MEDIUM
Given a DangerousPatternDetector with default patterns
When I check the command "wget http://example.com/install.sh | sh"
Then the command should be detected as dangerous
And the max danger level should be MEDIUM
Scenario: Detector detects curl piped to sh as MEDIUM
Given a DangerousPatternDetector with default patterns
When I check the command "curl http://example.com/install.sh | sh"
Then the command should be detected as dangerous
And the max danger level should be MEDIUM
Scenario: Detector detects curl piped to bash as MEDIUM
Given a DangerousPatternDetector with default patterns
When I check the command "curl https://get.example.com | bash"
Then the command should be detected as dangerous
And the max danger level should be MEDIUM
Scenario: Detector detects git push force as LOW
Given a DangerousPatternDetector with default patterns
When I check the command "git push --force origin main"
Then the command should be detected as dangerous
And the max danger level should be LOW
Scenario: Safe commands pass through without warnings
Given a DangerousPatternDetector with default patterns
When I check the command "ls -la"
Then the command should not be detected as dangerous
Scenario: Safe commands - echo passes through
Given a DangerousPatternDetector with default patterns
When I check the command "echo hello world"
Then the command should not be detected as dangerous
Scenario: Safe commands - git status passes through
Given a DangerousPatternDetector with default patterns
When I check the command "git status"
Then the command should not be detected as dangerous
Scenario: Safe commands - python script passes through
Given a DangerousPatternDetector with default patterns
When I check the command "python script.py"
Then the command should not be detected as dangerous
Scenario: check_first returns the first matching warning
Given a DangerousPatternDetector with default patterns
When I call check_first on "rm -rf /"
Then check_first should return a warning
And the first warning danger level should be CRITICAL
Scenario: check_first returns None for safe commands
Given a DangerousPatternDetector with default patterns
When I call check_first on "ls -la"
Then check_first should return None
Scenario: check returns all matching warnings
Given a DangerousPatternDetector with default patterns
When I call check on "rm -rf /"
Then check should return at least one warning
Scenario: Pattern registry is configurable - add pattern
Given a DangerousPatternDetector with default patterns
When I add a custom pattern named "custom_test" with pattern "evil_command" at HIGH level
And I check the command "evil_command --destroy"
Then the command should be detected as dangerous
Scenario: Pattern registry is configurable - remove pattern
Given a DangerousPatternDetector with default patterns
When I remove the pattern named "git_push_force"
And I check the command "git push --force origin main"
Then the command should not be detected as dangerous
Scenario: Pattern registry is configurable - replace patterns
Given a DangerousPatternDetector with default patterns
When I replace all patterns with a single pattern named "only_one" matching "special_cmd" at LOW level
And I check the command "rm -rf /"
Then the command should not be detected as dangerous
When I check the command "special_cmd"
Then the command should be detected as dangerous
Scenario: Detector patterns property returns current registry
Given a DangerousPatternDetector with default patterns
Then the detector patterns property should return a tuple
# ── ShellSafetyService ───────────────────────────────────────────────────
Scenario: ShellSafetyService allows safe commands
Given a ShellSafetyService with default settings
When I check the safety of command "ls -la"
Then the safety check result should be allowed
And the safety check warning should be None
Scenario: ShellSafetyService blocks dangerous commands above block level
Given a ShellSafetyService with default settings
When I check the safety of command "rm -rf /"
Then the safety check result should be blocked
And the safety check warning should not be None
Scenario: ShellSafetyService uses warn_callback when provided
Given a ShellSafetyService with a warn_callback that returns True
When I check the safety of command "rm -rf /"
Then the safety check result should be allowed
And the warn_callback should have been called
Scenario: ShellSafetyService blocks when warn_callback returns False
Given a ShellSafetyService with a warn_callback that returns False
When I check the safety of command "rm -rf /"
Then the safety check result should be blocked
And the warn_callback should have been called
Scenario: ShellSafetyService is_safe returns True for safe commands
Given a ShellSafetyService with default settings
When I call is_safe on "echo hello"
Then is_safe should return True
Scenario: ShellSafetyService is_safe returns False for dangerous commands
Given a ShellSafetyService with default settings
When I call is_safe on "rm -rf /"
Then is_safe should return False
Scenario: ShellSafetyService exposes detector property
Given a ShellSafetyService with default settings
Then the service detector property should be a DangerousPatternDetector
Scenario: ShellSafetyService exposes block_level property
Given a ShellSafetyService with default settings
Then the service block_level should be MEDIUM
Scenario: ShellSafetyService accepts extra_patterns
Given a ShellSafetyService with an extra pattern "custom_danger" matching "forbidden_cmd" at HIGH level
When I check the safety of command "forbidden_cmd --run"
Then the safety check result should be blocked
Scenario: SafetyCheckResult repr includes command and allowed status
Given a ShellSafetyService with default settings
When I check the safety of command "ls -la"
Then the safety check result repr should contain "ls -la"
And the safety check result repr should contain "allowed=True"
@@ -0,0 +1,17 @@
"""Shell safety domain models and application service for TUI."""
from __future__ import annotations
from cleveragents.tui.shell_safety.danger_level import ShellDangerLevel
from cleveragents.tui.shell_safety.dangerous_pattern import DangerousPattern
from cleveragents.tui.shell_safety.pattern_detector import DangerousPatternDetector
from cleveragents.tui.shell_safety.safety_service import ShellSafetyService
from cleveragents.tui.shell_safety.warning import DangerousCommandWarning
__all__ = [
"DangerousCommandWarning",
"DangerousPattern",
"DangerousPatternDetector",
"ShellDangerLevel",
"ShellSafetyService",
]
@@ -0,0 +1,29 @@
"""Shell danger level enumeration."""
from __future__ import annotations
from enum import IntEnum
class ShellDangerLevel(IntEnum):
"""Severity classification for dangerous shell commands.
Levels are ordered from least to most severe so that numeric
comparisons (``level >= ShellDangerLevel.HIGH``) work naturally.
"""
LOW = 1
"""Minor risk — command may have unintended side-effects but is
generally recoverable (e.g. ``chmod 777`` on a single file)."""
MEDIUM = 2
"""Moderate risk — command can cause data loss or security exposure
in common scenarios (e.g. ``sudo rm``, ``wget | sh``)."""
HIGH = 3
"""High risk — command is likely to cause significant, hard-to-reverse
damage (e.g. ``dd if=``, ``mkfs``)."""
CRITICAL = 4
"""Critical risk — command can destroy the entire system or create a
fork bomb (e.g. ``rm -rf /``, ``:(){ :|:& };:``)."""
@@ -0,0 +1,42 @@
"""Dangerous shell pattern value object."""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from cleveragents.tui.shell_safety.danger_level import ShellDangerLevel
@dataclass(slots=True, frozen=True)
class DangerousPattern:
"""Immutable descriptor for a single dangerous shell pattern.
Attributes:
name: Short human-readable identifier (e.g. ``"rm_rf_root"``).
pattern: Regular expression that matches the dangerous command text.
The pattern is compiled with ``re.IGNORECASE`` by default.
level: Danger severity classification.
description: Human-readable explanation of why this pattern is
dangerous and what damage it can cause.
case_sensitive: When ``True`` the regex is compiled without
``re.IGNORECASE``. Defaults to ``False``.
"""
name: str
pattern: str
level: ShellDangerLevel
description: str
case_sensitive: bool = False
# Compiled regex is stored in a non-frozen slot via __post_init__
_compiled: re.Pattern[str] = field(init=False, compare=False, repr=False)
def __post_init__(self) -> None:
flags = 0 if self.case_sensitive else re.IGNORECASE
# frozen=True prevents direct attribute assignment; use object.__setattr__
object.__setattr__(self, "_compiled", re.compile(self.pattern, flags))
def matches(self, command: str) -> bool:
"""Return ``True`` if *command* matches this pattern."""
return bool(self._compiled.search(command))
@@ -0,0 +1,106 @@
"""DangerousPatternDetector — domain service for shell danger detection."""
from __future__ import annotations
from cleveragents.tui.shell_safety.danger_level import ShellDangerLevel
from cleveragents.tui.shell_safety.dangerous_pattern import DangerousPattern
from cleveragents.tui.shell_safety.pattern_registry import DEFAULT_PATTERNS
from cleveragents.tui.shell_safety.warning import DangerousCommandWarning
class DangerousPatternDetector:
"""Checks shell commands against a configurable registry of dangerous patterns.
The detector ships with a set of built-in
:data:`~cleveragents.tui.shell_safety.pattern_registry.DEFAULT_PATTERNS`
but callers can supply a custom registry, add patterns at runtime, or
replace the registry entirely.
Usage::
detector = DangerousPatternDetector()
warnings = detector.check("rm -rf /")
if warnings:
print(warnings[0].message)
Attributes:
_patterns: Ordered list of :class:`DangerousPattern` objects that
are checked against each command.
"""
def __init__(
self,
patterns: list[DangerousPattern] | None = None,
) -> None:
"""Initialise the detector.
Args:
patterns: Custom pattern list. When ``None`` the built-in
:data:`DEFAULT_PATTERNS` are used.
"""
self._patterns: list[DangerousPattern] = (
list(patterns) if patterns is not None else list(DEFAULT_PATTERNS)
)
# ------------------------------------------------------------------
# Registry management
# ------------------------------------------------------------------
def add_pattern(self, pattern: DangerousPattern) -> None:
"""Append *pattern* to the registry.
Patterns are checked in insertion order; the first match wins when
:meth:`check_first` is used.
"""
self._patterns.append(pattern)
def remove_pattern(self, name: str) -> bool:
"""Remove the pattern with the given *name* from the registry.
Returns:
``True`` if a pattern was removed, ``False`` if no pattern with
that name existed.
"""
before = len(self._patterns)
self._patterns = [p for p in self._patterns if p.name != name]
return len(self._patterns) < before
def replace_patterns(self, patterns: list[DangerousPattern]) -> None:
"""Replace the entire pattern registry with *patterns*."""
self._patterns = list(patterns)
@property
def patterns(self) -> tuple[DangerousPattern, ...]:
"""Read-only view of the current pattern registry."""
return tuple(self._patterns)
# ------------------------------------------------------------------
# Detection
# ------------------------------------------------------------------
def check(self, command: str) -> list[DangerousCommandWarning]:
"""Return all warnings for *command* (may be empty).
All matching patterns are returned, ordered by insertion order.
"""
return [
DangerousCommandWarning.from_pattern(command, pattern)
for pattern in self._patterns
if pattern.matches(command)
]
def check_first(self, command: str) -> DangerousCommandWarning | None:
"""Return the first (highest-priority) warning for *command*, or ``None``."""
for pattern in self._patterns:
if pattern.matches(command):
return DangerousCommandWarning.from_pattern(command, pattern)
return None
def is_dangerous(self, command: str) -> bool:
"""Return ``True`` if *command* matches any registered pattern."""
return any(p.matches(command) for p in self._patterns)
def max_danger_level(self, command: str) -> ShellDangerLevel | None:
"""Return the highest :class:`ShellDangerLevel` matched, or ``None``."""
levels = [p.level for p in self._patterns if p.matches(command)]
return max(levels) if levels else None
@@ -0,0 +1,146 @@
"""Default dangerous shell pattern registry."""
from __future__ import annotations
from cleveragents.tui.shell_safety.danger_level import ShellDangerLevel
from cleveragents.tui.shell_safety.dangerous_pattern import DangerousPattern
# ---------------------------------------------------------------------------
# Default pattern definitions
# ---------------------------------------------------------------------------
#: Built-in dangerous patterns shipped with CleverAgents.
#: Callers may extend or replace this list via
#: :class:`~cleveragents.tui.shell_safety.pattern_detector.DangerousPatternDetector`.
DEFAULT_PATTERNS: tuple[DangerousPattern, ...] = (
# ── CRITICAL ────────────────────────────────────────────────────────────
DangerousPattern(
name="rm_rf_root",
pattern=r"rm\s+(-\w*r\w*f|-\w*f\w*r)\s+/\s*$",
level=ShellDangerLevel.CRITICAL,
description=(
"rm -rf / recursively deletes the entire filesystem root. "
"This will destroy the operating system and all data."
),
),
DangerousPattern(
name="rm_rf_wildcard",
pattern=r"rm\s+(-\w*r\w*f|-\w*f\w*r)\s+[/*]",
level=ShellDangerLevel.CRITICAL,
description=(
"rm -rf with a root or wildcard path can delete large portions "
"of the filesystem irreversibly."
),
),
DangerousPattern(
name="fork_bomb",
pattern=r":\s*\(\s*\)\s*\{.*:\s*\|.*:.*&.*\}",
level=ShellDangerLevel.CRITICAL,
description=(
"Fork bomb pattern :(){ :|:& };: exhausts system process table, "
"causing a denial-of-service that requires a hard reboot."
),
),
# ── HIGH ────────────────────────────────────────────────────────────────
DangerousPattern(
name="dd_if_device",
pattern=r"\bdd\b.*\bif\s*=",
level=ShellDangerLevel.HIGH,
description=(
"dd if= can overwrite disk devices or produce large files that "
"fill storage, potentially causing data loss."
),
),
DangerousPattern(
name="mkfs",
pattern=r"\bmkfs\b",
level=ShellDangerLevel.HIGH,
description=(
"mkfs formats a filesystem, permanently erasing all data on the "
"target partition or device."
),
),
DangerousPattern(
name="shred_device",
pattern=r"\bshred\b.*(/dev/|--remove)",
level=ShellDangerLevel.HIGH,
description=(
"shred on a device or with --remove overwrites data in a way "
"that makes recovery impossible."
),
),
# ── MEDIUM ──────────────────────────────────────────────────────────────
DangerousPattern(
name="chmod_777",
pattern=r"\bchmod\b.*\b777\b",
level=ShellDangerLevel.MEDIUM,
description=(
"chmod 777 grants world-readable, world-writable, and "
"world-executable permissions, creating a security vulnerability."
),
),
DangerousPattern(
name="sudo_rm",
pattern=r"\bsudo\b.*\brm\b",
level=ShellDangerLevel.MEDIUM,
description=(
"sudo rm runs rm with elevated privileges, bypassing normal "
"permission checks and increasing the risk of accidental deletion."
),
),
DangerousPattern(
name="wget_pipe_sh",
pattern=r"\bwget\b.*\|\s*(ba)?sh\b",
level=ShellDangerLevel.MEDIUM,
description=(
"Piping wget output directly to sh executes arbitrary remote "
"code without inspection, a common supply-chain attack vector."
),
),
DangerousPattern(
name="curl_pipe_sh",
pattern=r"\bcurl\b.*\|\s*(ba)?sh\b",
level=ShellDangerLevel.MEDIUM,
description=(
"Piping curl output directly to sh executes arbitrary remote "
"code without inspection, a common supply-chain attack vector."
),
),
DangerousPattern(
name="wget_pipe_bash",
pattern=r"\bwget\b.*\|\s*bash\b",
level=ShellDangerLevel.MEDIUM,
description=(
"Piping wget output directly to bash executes arbitrary remote "
"code without inspection."
),
),
DangerousPattern(
name="curl_pipe_bash",
pattern=r"\bcurl\b.*\|\s*bash\b",
level=ShellDangerLevel.MEDIUM,
description=(
"Piping curl output directly to bash executes arbitrary remote "
"code without inspection."
),
),
# ── LOW ─────────────────────────────────────────────────────────────────
DangerousPattern(
name="git_push_force",
pattern=r"\bgit\b.*\bpush\b.*--force\b",
level=ShellDangerLevel.LOW,
description=(
"git push --force overwrites remote history, which can cause "
"permanent data loss for collaborators."
),
),
DangerousPattern(
name="chmod_recursive_permissive",
pattern=r"\bchmod\b.*-R.*\b[67][0-7][0-7]\b",
level=ShellDangerLevel.LOW,
description=(
"Recursive chmod with permissive modes can expose sensitive "
"files to unintended access."
),
),
)
@@ -0,0 +1,138 @@
"""ShellSafetyService — application service for shell command safety checks."""
from __future__ import annotations
from collections.abc import Callable
from cleveragents.tui.shell_safety.danger_level import ShellDangerLevel
from cleveragents.tui.shell_safety.dangerous_pattern import DangerousPattern
from cleveragents.tui.shell_safety.pattern_detector import DangerousPatternDetector
from cleveragents.tui.shell_safety.warning import DangerousCommandWarning
class ShellSafetyService:
"""Application service that checks shell commands before execution.
Wraps
:class:`~cleveragents.tui.shell_safety.pattern_detector.DangerousPatternDetector`
with a higher-level API suitable for use by the TUI and other callers.
The service supports an optional *warn_callback* that is invoked whenever
a dangerous command is detected. The callback receives the
:class:`~cleveragents.tui.shell_safety.warning.DangerousCommandWarning`
and should return ``True`` to allow execution or ``False`` to block it.
When no callback is provided, commands at or above *block_level* are
blocked automatically.
Usage::
service = ShellSafetyService()
result = service.check_command("rm -rf /")
if not result.allowed:
print(result.warning.message)
Attributes:
_detector: The underlying :class:`DangerousPatternDetector`.
_block_level: Commands at or above this level are blocked when no
callback is provided.
_warn_callback: Optional callable invoked for dangerous commands.
"""
def __init__(
self,
*,
detector: DangerousPatternDetector | None = None,
block_level: ShellDangerLevel = ShellDangerLevel.MEDIUM,
warn_callback: Callable[[DangerousCommandWarning], bool] | None = None,
extra_patterns: list[DangerousPattern] | None = None,
) -> None:
"""Initialise the service.
Args:
detector: Custom detector. When ``None`` a default
:class:`DangerousPatternDetector` is created.
block_level: Minimum :class:`ShellDangerLevel` that triggers
automatic blocking when no *warn_callback* is provided.
Defaults to :attr:`~ShellDangerLevel.MEDIUM`.
warn_callback: Optional callable ``(warning) -> bool``. Return
``True`` to allow execution, ``False`` to block.
extra_patterns: Additional patterns to register on top of the
defaults.
"""
self._detector = (
detector if detector is not None else DangerousPatternDetector()
)
self._block_level = block_level
self._warn_callback = warn_callback
if extra_patterns:
for pattern in extra_patterns:
self._detector.add_pattern(pattern)
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def check_command(self, command: str) -> SafetyCheckResult:
"""Check *command* and return a :class:`SafetyCheckResult`.
The result indicates whether the command is allowed to proceed and
includes any warning that was generated.
"""
warning = self._detector.check_first(command)
if warning is None:
return SafetyCheckResult(command=command, warning=None, allowed=True)
if self._warn_callback is not None:
allowed = self._warn_callback(warning)
else:
allowed = warning.danger_level < self._block_level
return SafetyCheckResult(command=command, warning=warning, allowed=allowed)
def is_safe(self, command: str) -> bool:
"""Return ``True`` if *command* passes the safety check."""
return self.check_command(command).allowed
@property
def detector(self) -> DangerousPatternDetector:
"""The underlying :class:`DangerousPatternDetector`."""
return self._detector
@property
def block_level(self) -> ShellDangerLevel:
"""The minimum danger level that triggers automatic blocking."""
return self._block_level
class SafetyCheckResult:
"""Result of a :meth:`ShellSafetyService.check_command` call.
Attributes:
command: The command that was checked.
warning: The
:class:`~cleveragents.tui.shell_safety.warning.DangerousCommandWarning`
if a dangerous pattern was detected, otherwise ``None``.
allowed: ``True`` if the command is allowed to proceed.
"""
__slots__ = ("allowed", "command", "warning")
def __init__(
self,
command: str,
warning: DangerousCommandWarning | None,
allowed: bool,
) -> None:
self.command = command
self.warning = warning
self.allowed = allowed
def __repr__(self) -> str:
return (
f"SafetyCheckResult(command={self.command!r}, "
f"allowed={self.allowed}, "
f"warning={self.warning!r})"
)
@@ -0,0 +1,48 @@
"""DangerousCommandWarning value object."""
from __future__ import annotations
from dataclasses import dataclass
from cleveragents.tui.shell_safety.danger_level import ShellDangerLevel
from cleveragents.tui.shell_safety.dangerous_pattern import DangerousPattern
@dataclass(slots=True, frozen=True)
class DangerousCommandWarning:
"""Immutable value object describing a detected dangerous command.
Produced by
:class:`~cleveragents.tui.shell_safety.pattern_detector.DangerousPatternDetector`
and consumed by the TUI to present a confirmation dialog to the user.
Attributes:
command: The original command string that triggered the warning.
matched_pattern: The :class:`DangerousPattern` that matched.
danger_level: Convenience alias for ``matched_pattern.level``.
message: Human-readable warning message suitable for display.
"""
command: str
matched_pattern: DangerousPattern
danger_level: ShellDangerLevel
message: str
@classmethod
def from_pattern(
cls,
command: str,
pattern: DangerousPattern,
) -> DangerousCommandWarning:
"""Construct a warning from a command and the pattern that matched it.
The ``message`` is derived from the pattern's ``description``.
"""
level_name = pattern.level.name.capitalize()
message = f"[{level_name}] Dangerous command detected: {pattern.description}"
return cls(
command=command,
matched_pattern=pattern,
danger_level=pattern.level,
message=message,
)