From ebf0749cecba289ba05e54df7dcbdfa7989e2dbd Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Thu, 2 Apr 2026 08:45:32 +0000 Subject: [PATCH] feat(tui): implement shell danger detection patterns Implement dangerous shell pattern detection with configurable pattern registry, danger level classification, and user warning system. - Domain: ShellDangerLevel enum (LOW, MEDIUM, HIGH, CRITICAL) - Domain: DangerousPattern value object with regex matching - Domain: DangerousCommandWarning value object - Domain: DangerousPatternDetector with configurable pattern registry - Application: ShellSafetyService with warn_callback and block_level - Default patterns: rm -rf, fork bomb, dd if=, mkfs, chmod 777, sudo rm, wget/curl piped to sh/bash, git push --force - BDD tests covering all pattern categories, safe commands, registry management, and service behavior ISSUES CLOSED: #1003 --- .../steps/tui_shell_danger_detection_steps.py | 392 ++++++++++++++++++ features/tui_shell_danger_detection.feature | 236 +++++++++++ src/cleveragents/tui/shell_safety/__init__.py | 17 + .../tui/shell_safety/danger_level.py | 29 ++ .../tui/shell_safety/dangerous_pattern.py | 42 ++ .../tui/shell_safety/pattern_detector.py | 106 +++++ .../tui/shell_safety/pattern_registry.py | 146 +++++++ .../tui/shell_safety/safety_service.py | 138 ++++++ src/cleveragents/tui/shell_safety/warning.py | 48 +++ 9 files changed, 1154 insertions(+) create mode 100644 features/steps/tui_shell_danger_detection_steps.py create mode 100644 features/tui_shell_danger_detection.feature create mode 100644 src/cleveragents/tui/shell_safety/__init__.py create mode 100644 src/cleveragents/tui/shell_safety/danger_level.py create mode 100644 src/cleveragents/tui/shell_safety/dangerous_pattern.py create mode 100644 src/cleveragents/tui/shell_safety/pattern_detector.py create mode 100644 src/cleveragents/tui/shell_safety/pattern_registry.py create mode 100644 src/cleveragents/tui/shell_safety/safety_service.py create mode 100644 src/cleveragents/tui/shell_safety/warning.py diff --git a/features/steps/tui_shell_danger_detection_steps.py b/features/steps/tui_shell_danger_detection_steps.py new file mode 100644 index 000000000..d14a056fc --- /dev/null +++ b/features/steps/tui_shell_danger_detection_steps.py @@ -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}" + ) diff --git a/features/tui_shell_danger_detection.feature b/features/tui_shell_danger_detection.feature new file mode 100644 index 000000000..75d5b78b5 --- /dev/null +++ b/features/tui_shell_danger_detection.feature @@ -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" diff --git a/src/cleveragents/tui/shell_safety/__init__.py b/src/cleveragents/tui/shell_safety/__init__.py new file mode 100644 index 000000000..92bc2cd7d --- /dev/null +++ b/src/cleveragents/tui/shell_safety/__init__.py @@ -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", +] diff --git a/src/cleveragents/tui/shell_safety/danger_level.py b/src/cleveragents/tui/shell_safety/danger_level.py new file mode 100644 index 000000000..5ecf01278 --- /dev/null +++ b/src/cleveragents/tui/shell_safety/danger_level.py @@ -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 /``, ``:(){ :|:& };:``).""" diff --git a/src/cleveragents/tui/shell_safety/dangerous_pattern.py b/src/cleveragents/tui/shell_safety/dangerous_pattern.py new file mode 100644 index 000000000..b2d5c8174 --- /dev/null +++ b/src/cleveragents/tui/shell_safety/dangerous_pattern.py @@ -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)) diff --git a/src/cleveragents/tui/shell_safety/pattern_detector.py b/src/cleveragents/tui/shell_safety/pattern_detector.py new file mode 100644 index 000000000..bcce448c5 --- /dev/null +++ b/src/cleveragents/tui/shell_safety/pattern_detector.py @@ -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 diff --git a/src/cleveragents/tui/shell_safety/pattern_registry.py b/src/cleveragents/tui/shell_safety/pattern_registry.py new file mode 100644 index 000000000..5b57235d4 --- /dev/null +++ b/src/cleveragents/tui/shell_safety/pattern_registry.py @@ -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." + ), + ), +) diff --git a/src/cleveragents/tui/shell_safety/safety_service.py b/src/cleveragents/tui/shell_safety/safety_service.py new file mode 100644 index 000000000..2b6d42950 --- /dev/null +++ b/src/cleveragents/tui/shell_safety/safety_service.py @@ -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})" + ) diff --git a/src/cleveragents/tui/shell_safety/warning.py b/src/cleveragents/tui/shell_safety/warning.py new file mode 100644 index 000000000..40ef42e1d --- /dev/null +++ b/src/cleveragents/tui/shell_safety/warning.py @@ -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, + ) -- 2.52.0