Simplify command pattern generation and parsing logic

- Modified _generate_single_command_pattern() to return list of prefix-based patterns (1-3 words) instead of complex file extension logic
- Added word boundary logic to prevent partial word matches (e.g., 'ls' won't match 'lsof')
- Simplified _parse_piped_command() to use shlex.split() with proper fallback handling
- Updated _generate_command_patterns() to handle both single and piped commands
- Enhanced read_confirmation_input() with new pattern selection workflow
- Added _handle_pattern_selection() function providing multiple pattern options (exact, prefix-based, custom)
- Created comprehensive unit tests covering all pattern matching algorithms
- All 31 tests pass, validating the new simplified logic
- Fixed type annotation for mypy compliance

config.template.toml

@@ -360,13 +360,28 @@ classpath = "my_package.my_module.MyCustomAgent"
 [security]
 
 # Enable confirmation mode (For Headless / CLI only -  In Web this is overridden by Session Init)
-#confirmation_mode = false
+# When using command_approval analyzer, this should be enabled
+confirmation_mode = true
 
 # The security analyzer to use (For Headless / CLI only -  In Web this is overridden by Session Init)
-#security_analyzer = ""
+# Available options: "invariant", "command_approval"
+# For CLI with confirmation mode, "command_approval" is recommended
+security_analyzer = "command_approval"
 
 # Whether to enable security analyzer
-#enable_security_analyzer = false
+# When using command_approval analyzer, this should be enabled
+enable_security_analyzer = true
+
+# Dictionary of approved commands that don't require confirmation
+# The key is the command, and the value is a boolean (true to approve)
+#approved_commands = { "ls -la" = true, "git status" = true }
+
+# List of approved command patterns (regex) that don't require confirmation
+#approved_command_patterns = [
+#  { pattern = "^ls( -[a-zA-Z]+)?( \\S+)?$", description = "List directory contents" },
+#  { pattern = "^cd \\S+$", description = "Change directory" },
+#  { pattern = "^git (status|log|diff)$", description = "Basic git commands" }
+#]
 
 #################################### Condenser #################################
 # Condensers control how conversation history is managed and compressed when

openhands/cli/main.py

@@ -236,8 +236,43 @@ async def run_session(
                     )
                     return
 
-                confirmation_status = await read_confirmation_input(config)
-                if confirmation_status in ('yes', 'always'):
+                # Get the pending action from the agent controller
+                pending_action = controller._pending_action
+                command = ''
+                if pending_action:
+                    if hasattr(pending_action, 'command'):
+                        command = pending_action.command
+                    elif hasattr(pending_action, 'code'):
+                        command = pending_action.code
+
+                confirmation_status = await read_confirmation_input(
+                    config, command, pending_action
+                )
+
+                # Handle different confirmation responses
+                if confirmation_status == 'always':
+                    # Set always confirm mode to skip future confirmations
+                    always_confirm_mode = True
+                    event_stream.add_event(
+                        ChangeAgentStateAction(AgentState.USER_CONFIRMED),
+                        EventSource.USER,
+                    )
+                elif confirmation_status.startswith('remember:'):
+                    # Parse the remember response: remember:pattern:description
+                    parts = confirmation_status.split(':', 2)
+                    if len(parts) == 3:
+                        _, pattern, description = parts
+                        # Save the command pattern to config
+                        from openhands.cli.utils import save_approved_command_to_config
+
+                        save_approved_command_to_config(
+                            command, pattern=pattern, description=description
+                        )
+                    event_stream.add_event(
+                        ChangeAgentStateAction(AgentState.USER_CONFIRMED),
+                        EventSource.USER,
+                    )
+                elif confirmation_status == 'yes':
                     event_stream.add_event(
                         ChangeAgentStateAction(AgentState.USER_CONFIRMED),
                         EventSource.USER,

openhands/cli/tui.py

@@ -700,12 +700,140 @@ async def read_prompt_input(
         return '/exit'
 
 
-async def read_confirmation_input(config: OpenHandsConfig) -> str:
+def _generate_single_command_pattern(command: str) -> list[str]:
+    """Generate candidate regex patterns for a single command based on prefixes.
+
+    Args:
+        command: The single command to generate patterns for.
+
+    Returns:
+        list[str]: List of candidate regex patterns that match similar commands.
+    """
+    import re
+
+    # Split command into parts
+    parts = command.split()
+    if not parts:
+        return [f'^{re.escape(command)}$']
+
+    patterns = []
+
+    # Generate patterns for first word, first two words, first three words
+    for i in range(1, min(4, len(parts) + 1)):
+        prefix_parts = parts[:i]
+        escaped_prefix = '\\s+'.join(re.escape(part) for part in prefix_parts)
+
+        # Create pattern: prefix followed by word boundary and anything until end of line
+        # This prevents matching partial words (e.g., "ls" won't match "lsof")
+        if i == 1:
+            # For single word, add word boundary to prevent partial matches
+            pattern = f'^{escaped_prefix}(\\s.*|$)'
+        else:
+            # For multiple words, the space already acts as a boundary
+            pattern = f'^{escaped_prefix}.*$'
+        patterns.append(pattern)
+
+    return patterns
+
+
+def _parse_piped_command(command: str) -> list[str]:
+    """Parse a command that may contain pipes into individual commands.
+
+    Args:
+        command: The command string to parse.
+
+    Returns:
+        list[str]: List of individual commands split by pipes.
+    """
+    import shlex
+
+    # Handle edge cases first
+    if not command or not command.strip():
+        return []
+
+    # Use shlex to split the command, which handles quotes and escapes properly
+    try:
+        # Split the entire command into tokens
+        tokens = shlex.split(command, posix=True)
+    except ValueError:
+        # If shlex fails (e.g., unmatched quotes), fall back to simple split
+        return [part.strip() for part in command.split('|') if part.strip()]
+
+    # Find pipe tokens and split the command accordingly
+    parts = []
+    current_part: list[str] = []
+
+    for token in tokens:
+        if token == '|':
+            if current_part:
+                parts.append(' '.join(current_part))
+                current_part = []
+        else:
+            current_part.append(token)
+
+    # Add the last part if it exists
+    if current_part:
+        parts.append(' '.join(current_part))
+
+    # If no pipes were found in tokens, check if the original command has pipes
+    # This handles cases where pipes are not separated by spaces
+    if len(parts) <= 1 and '|' in command:
+        return [part.strip() for part in command.split('|') if part.strip()]
+
+    return parts
+
+
+def _generate_command_patterns(command: str) -> list[str]:
+    """Generate candidate regex patterns for a command that can match similar commands.
+
+    This function handles both simple commands and piped commands.
+    For piped commands, it generates patterns for each sub-command.
+
+    Args:
+        command: The command to generate patterns for (may contain pipes).
+
+    Returns:
+        list[str]: List of candidate regex patterns that match similar commands.
+    """
+    # Parse the command to handle pipes
+    sub_commands = _parse_piped_command(command)
+
+    if len(sub_commands) <= 1:
+        # Single command or empty, return all candidate patterns
+        return _generate_single_command_pattern(command)
+    else:
+        # Piped command, generate pattern for each sub-command
+        # For simplicity, we'll just use the first pattern from each sub-command
+        sub_patterns = []
+        for sub_command in sub_commands:
+            patterns = _generate_single_command_pattern(sub_command.strip())
+            if patterns:
+                # Use the most specific pattern (first one)
+                sub_pattern = patterns[0]
+                # Remove the ^ and $ anchors from sub-patterns
+                if sub_pattern.startswith('^'):
+                    sub_pattern = sub_pattern[1:]
+                if sub_pattern.endswith('$'):
+                    sub_pattern = sub_pattern[:-1]
+                sub_patterns.append(sub_pattern)
+
+        if sub_patterns:
+            # Join sub-patterns with pipe separator (allowing flexible whitespace around pipes)
+            pattern = '\\s*\\|\\s*'.join(sub_patterns)
+            return [f'^{pattern}$']
+        else:
+            return []
+
+
+async def read_confirmation_input(
+    config: OpenHandsConfig, command: str = '', pending_action=None
+) -> str:
     try:
         choices = [
             'Yes, proceed',
             'No (and allow to enter instructions)',
-            "Always proceed (don't ask again)",
+            'Always proceed (skip all confirmations)',
+            "Yes, and don't ask again for similar commands",
         ]
 
         # keep the outer coroutine responsive by using asyncio.to_thread which puts the blocking call app.run() of cli_confirm() in a separate thread
@@ -713,7 +841,89 @@ async def read_confirmation_input(config: OpenHandsConfig) -> str:
             cli_confirm, config, 'Choose an option:', choices
         )
 
-        return {0: 'yes', 1: 'no', 2: 'always'}.get(index, 'no')
+        result = {0: 'yes', 1: 'no', 2: 'always', 3: 'remember'}.get(index, 'no')
+
+        # If the user chose "remember", show pattern selection options
+        if result == 'remember' and command:
+            return await _handle_pattern_selection(config, command)
+
+        return result
+
+    except (KeyboardInterrupt, EOFError):
+        return 'no'
+
+
+async def _handle_pattern_selection(config: OpenHandsConfig, command: str) -> str:
+    """Handle pattern selection when user chooses to remember similar commands."""
+    try:
+        # Generate candidate patterns
+        patterns = _generate_command_patterns(command)
+
+        # Create pattern descriptions
+        pattern_choices = []
+        pattern_data = []
+
+        # Add exact command option
+        import re
+
+        exact_pattern = f'^{re.escape(command)}$'
+        pattern_choices.append(f'Exact command: {command}')
+        pattern_data.append(('exact', exact_pattern, f'Exact command: {command}'))
+
+        # Add prefix-based patterns
+        parts = command.split()
+        if parts:
+            for i, pattern in enumerate(patterns):
+                if i < len(parts):
+                    prefix = ' '.join(parts[: i + 1])
+                    description = f'Commands starting with: {prefix}'
+                    pattern_choices.append(description)
+                    pattern_data.append(('pattern', pattern, description))
+
+        # Add custom pattern option
+        pattern_choices.append('Enter custom pattern')
+        pattern_data.append(('custom', '', 'Custom pattern'))
+
+        # Show pattern selection menu
+        pattern_index = await asyncio.to_thread(
+            cli_confirm, config, 'Choose which commands to remember:', pattern_choices
+        )
+
+        if pattern_index < len(pattern_data):
+            pattern_type, pattern_value, description = pattern_data[pattern_index]
+
+            if pattern_type == 'custom':
+                # Get custom pattern from user
+                prompt_session = create_prompt_session(config)
+                print_formatted_text(
+                    HTML(
+                        '<gold>Enter a custom regex pattern (example: ^ls.*$ for all ls commands):</gold>'
+                    )
+                )
+                try:
+                    custom_pattern = await prompt_session.prompt_async('Pattern: ')
+                    if custom_pattern.strip():
+                        # Validate the regex pattern
+                        import re
+
+                        try:
+                            re.compile(custom_pattern.strip())
+                            return f'remember:{custom_pattern.strip()}:Custom pattern: {custom_pattern.strip()}'
+                        except re.error:
+                            print_formatted_text(
+                                HTML(
+                                    '<ansired>Invalid regex pattern. Using exact command instead.</ansired>'
+                                )
+                            )
+                            return f'remember:^{re.escape(command)}$:Exact command: {command}'
+                    else:
+                        return 'no'
+                except (KeyboardInterrupt, EOFError):
+                    return 'no'
+            else:
+                return f'remember:{pattern_value}:{description}'
+
+        return 'no'
 
     except (KeyboardInterrupt, EOFError):
         return 'no'

openhands/cli/utils.py

@@ -243,3 +243,82 @@ def read_file(file_path: str | Path) -> str:
 def write_to_file(file_path: str | Path, content: str) -> None:
     with open(file_path, 'w') as f:
         f.write(content)
+
+
+def save_approved_command_to_config(
+    command: str, pattern: str | None = None, description: str | None = None
+) -> None:
+    """Save an approved command or pattern to the config file.
+
+    Args:
+        command: The command to save as approved.
+        pattern: Optional regex pattern to save instead of exact command.
+        description: Optional description for the pattern.
+    """
+    config_path = _LOCAL_CONFIG_FILE_PATH
+
+    # Load existing config or create a new one
+    if config_path.exists():
+        try:
+            with open(config_path, 'r') as f:
+                config_data = toml.load(f)
+        except Exception as e:
+            from openhands.core.logger import openhands_logger
+
+            openhands_logger.warning(f'Error loading config file: {e}')
+            config_data = {}
+    else:
+        config_data = {}
+        config_path.parent.mkdir(parents=True, exist_ok=True)
+
+    # Ensure security section exists
+    if 'security' not in config_data:
+        config_data['security'] = {}
+
+    if pattern and description:
+        # Save as a pattern
+        if 'approved_command_patterns' not in config_data['security']:
+            config_data['security']['approved_command_patterns'] = []
+
+        # Check if pattern already exists
+        pattern_exists = any(
+            p == pattern
+            for p in config_data['security']['approved_command_patterns']
+            if isinstance(p, str)
+        ) or any(
+            p.get('pattern') == pattern
+            for p in config_data['security']['approved_command_patterns']
+            if isinstance(p, dict)
+        )
+
+        if not pattern_exists:
+            # Just save the pattern string directly
+            config_data['security']['approved_command_patterns'].append(pattern)
+
+        from openhands.core.logger import openhands_logger
+
+        openhands_logger.info(
+            f"Pattern '{pattern}' saved to approved command patterns in {config_path}"
+        )
+    else:
+        # Save as exact command
+        if 'approved_commands' not in config_data['security']:
+            config_data['security']['approved_commands'] = {}
+
+        # Add the command to the approved commands
+        config_data['security']['approved_commands'][command] = True
+
+        from openhands.core.logger import openhands_logger
+
+        openhands_logger.info(
+            f"Command '{command}' saved to approved commands in {config_path}"
+        )
+
+    # Write the updated config back to the file
+    try:
+        with open(config_path, 'w') as f:
+            toml.dump(config_data, f)
+    except Exception as e:
+        from openhands.core.logger import openhands_logger
+
+        openhands_logger.error(f'Error saving approved command to config: {e}')

openhands/controller/agent_controller.py

@@ -875,9 +875,40 @@ class AgentController:
             if self.state.confirmation_mode and (
                 type(action) is CmdRunAction or type(action) is IPythonRunCellAction
             ):
-                action.confirmation_state = (
-                    ActionConfirmationStatus.AWAITING_CONFIRMATION
-                )
+                # Check if the command is already approved
+                command = ''
+                if type(action) is CmdRunAction:
+                    command = action.command
+                elif type(action) is IPythonRunCellAction:
+                    command = action.code
+
+                # Get the security config
+                import toml
+
+                from openhands.core.config import SecurityConfig
+
+                # Load security config from the config file
+                security_config = SecurityConfig()
+                try:
+                    with open('config.toml', 'r', encoding='utf-8') as f:
+                        config_data = toml.load(f)
+                        if 'security' in config_data:
+                            security_config = SecurityConfig.model_validate(
+                                config_data['security']
+                            )
+                except Exception:
+                    # If loading fails, use default config
+                    pass
+
+                # Check if the command is approved
+                if security_config.is_command_approved(command):
+                    # Command is already approved, no need for confirmation
+                    action.confirmation_state = ActionConfirmationStatus.CONFIRMED
+                else:
+                    # Command needs confirmation
+                    action.confirmation_state = (
+                        ActionConfirmationStatus.AWAITING_CONFIRMATION
+                    )
             self._pending_action = action
 
         if not isinstance(action, NullAction):

openhands/core/config/security_config.py

@@ -1,5 +1,8 @@
 from pydantic import BaseModel, ConfigDict, Field, ValidationError
 
+# ApprovedCommandPattern is now just a string containing the regex pattern
+ApprovedCommandPattern = str
+
 
 class SecurityConfig(BaseModel):
     """Configuration for security related functionalities.
@@ -7,13 +10,33 @@ class SecurityConfig(BaseModel):
     Attributes:
         confirmation_mode: Whether to enable confirmation mode.
         security_analyzer: The security analyzer to use.
+        approved_command_patterns: List of regex patterns for commands that don't require confirmation.
+        approved_commands: Dictionary of exact commands that have been approved.
     """
 
     confirmation_mode: bool = Field(default=False)
     security_analyzer: str | None = Field(default=None)
+    approved_command_patterns: list[ApprovedCommandPattern] = Field(
+        default_factory=list
+    )
+    approved_commands: dict[str, bool] = Field(default_factory=dict)
 
     model_config = ConfigDict(extra='forbid')
 
+    def is_command_approved(self, command: str) -> bool:
+        """Check if a command is approved.
+
+        This is a stub method that always returns False.
+        The actual implementation is in CommandApprovalAnalyzer.
+
+        Args:
+            command: The command to check.
+
+        Returns:
+            bool: Always False in this stub implementation.
+        """
+        return False
+
     @classmethod
     def from_toml_section(cls, data: dict) -> dict[str, 'SecurityConfig']:
         """
@@ -28,9 +51,32 @@ class SecurityConfig(BaseModel):
         # Initialize the result mapping
         security_mapping: dict[str, SecurityConfig] = {}
 
+        # Extract approved command patterns if present
+        approved_patterns = []
+        if 'approved_command_patterns' in data:
+            patterns_data = data.pop('approved_command_patterns')
+            if isinstance(patterns_data, list):
+                for pattern_data in patterns_data:
+                    # Handle the new format (just a string pattern)
+                    if isinstance(pattern_data, str):
+                        approved_patterns.append(pattern_data)
+                    # Handle the old format (dict with pattern and description)
+                    elif isinstance(pattern_data, dict) and 'pattern' in pattern_data:
+                        approved_patterns.append(pattern_data['pattern'])
+
+        # Extract approved commands if present
+        approved_commands = {}
+        if 'approved_commands' in data:
+            commands_data = data.pop('approved_commands')
+            if isinstance(commands_data, dict):
+                approved_commands = commands_data
+
         # Try to create the configuration instance
         try:
-            security_mapping['security'] = cls.model_validate(data)
+            config = cls.model_validate(data)
+            config.approved_command_patterns = approved_patterns
+            config.approved_commands = approved_commands
+            security_mapping['security'] = config
         except ValidationError as e:
             raise ValueError(f'Invalid security configuration: {e}')
 

openhands/security/__init__.py

@@ -1,7 +1,9 @@
 from openhands.security.analyzer import SecurityAnalyzer
+from openhands.security.command_approval.analyzer import CommandApprovalAnalyzer
 from openhands.security.invariant.analyzer import InvariantAnalyzer
 
 __all__ = [
     'SecurityAnalyzer',
     'InvariantAnalyzer',
+    'CommandApprovalAnalyzer',
 ]

openhands/security/command_approval/__init__.py

@@ -0,0 +1,3 @@
+from openhands.security.command_approval.analyzer import CommandApprovalAnalyzer
+
+__all__ = ['CommandApprovalAnalyzer']

openhands/security/command_approval/analyzer.py

@@ -0,0 +1,305 @@
+"""Command approval analyzer for security."""
+
+import re
+from typing import Any
+
+import bashlex
+from fastapi import Request
+
+from openhands.core.logger import openhands_logger as logger
+from openhands.events.action.action import (
+    Action,
+    ActionConfirmationStatus,
+    ActionSecurityRisk,
+)
+from openhands.events.action.commands import CmdRunAction, IPythonRunCellAction
+from openhands.events.event import Event
+from openhands.events.stream import EventStream
+from openhands.security.analyzer import SecurityAnalyzer
+
+
+class CommandPattern:
+    """A pattern for matching commands."""
+
+    def __init__(self, pattern: str, description: str):
+        """Initialize a new command pattern.
+
+        Args:
+            pattern: The regex pattern to match commands against.
+            description: A human-readable description of what this pattern matches.
+        """
+        self.pattern = pattern
+        self.description = description
+        self._compiled_pattern = re.compile(pattern)
+
+    def matches(self, command: str) -> bool:
+        """Check if the command matches this pattern.
+
+        Args:
+            command: The command to check.
+
+        Returns:
+            bool: True if the command matches, False otherwise.
+        """
+        return bool(self._compiled_pattern.match(command))
+
+
+class CommandParser:
+    """Parser for bash commands using bashlex."""
+
+    def is_piped_command(self, command: str) -> bool:
+        """Check if a command contains pipes.
+
+        Args:
+            command: The command to check.
+
+        Returns:
+            bool: True if the command contains pipes, False otherwise.
+        """
+        if not command or not command.strip():
+            return False
+
+        try:
+            parts = bashlex.parse(command)
+            for part in parts:
+                if part.kind == 'pipeline':
+                    return True
+            return False
+        except Exception as e:
+            logger.warning(f'Error parsing command with bashlex: {e}')
+            # Fallback: check for pipe character not in quotes
+            # This is a simple heuristic and not as accurate as bashlex parsing
+            in_single_quote = False
+            in_double_quote = False
+            for char in command:
+                if char == "'" and not in_double_quote:
+                    in_single_quote = not in_single_quote
+                elif char == '"' and not in_single_quote:
+                    in_double_quote = not in_double_quote
+                elif char == '|' and not in_single_quote and not in_double_quote:
+                    return True
+            return False
+
+    def parse_command(self, command: str) -> list[str]:
+        """Parse a command into individual parts, handling pipes.
+
+        Args:
+            command: The command to parse.
+
+        Returns:
+            List[str]: List of individual commands.
+        """
+        if not command or not command.strip():
+            return []
+
+        try:
+            parts = bashlex.parse(command)
+            commands = []
+
+            # Helper function to extract command from a node
+            def extract_command(node):
+                if node.kind == 'command':
+                    cmd_parts = []
+                    for part in node.parts:
+                        if hasattr(part, 'word'):
+                            cmd_parts.append(part.word)
+                    if cmd_parts:
+                        return ' '.join(cmd_parts)
+                return None
+
+            # Process the AST
+            for part in parts:
+                if part.kind == 'pipeline':
+                    # A pipeline has multiple commands
+                    for subpart in part.parts:
+                        if subpart.kind == 'command':
+                            cmd = extract_command(subpart)
+                            if cmd:
+                                commands.append(cmd)
+                elif part.kind == 'command':
+                    # A single command
+                    cmd = extract_command(part)
+                    if cmd:
+                        commands.append(cmd)
+                elif part.kind == 'list':
+                    # A list of commands (e.g., with && or ||)
+                    # We only take the first command for approval purposes
+                    for subpart in part.parts:
+                        if subpart.kind == 'command':
+                            cmd = extract_command(subpart)
+                            if cmd:
+                                commands.append(cmd)
+                                break
+                        elif subpart.kind == 'operator':
+                            # Stop at the first operator
+                            break
+
+            return commands
+        except Exception as e:
+            logger.warning(f'Error parsing command with bashlex: {e}')
+            # Fallback: simple split by pipe
+            # This is a simple heuristic and not as accurate as bashlex parsing
+            if '|' in command:
+                return [part.strip() for part in command.split('|') if part.strip()]
+            else:
+                return [command.strip()] if command.strip() else []
+
+
+class CommandApprovalAnalyzer(SecurityAnalyzer):
+    """Security analyzer that automatically approves commands based on patterns and previously approved commands."""
+
+    def __init__(
+        self,
+        event_stream: EventStream,
+        policy: str | None = None,
+        sid: str | None = None,
+    ) -> None:
+        """Initializes a new instance of the CommandApprovalAnalyzer class."""
+        super().__init__(event_stream)
+        self.parser = CommandParser()
+        self.approved_commands: dict[
+            str, bool
+        ] = {}  # Dict of exact commands that have been approved
+        self.approved_patterns: list[
+            CommandPattern
+        ] = []  # List of regex patterns for approved commands
+        self.compiled_patterns: dict[
+            str, re.Pattern
+        ] = {}  # Cache of compiled regex patterns
+
+        # Add some default patterns
+        self._add_default_patterns()
+
+    def _add_default_patterns(self) -> None:
+        """Add default command patterns that are always approved."""
+        # Simple, safe commands
+        self.approved_patterns.append(
+            CommandPattern(
+                pattern=r'^ls(\s+-[a-zA-Z]+)*(\s+\S+)*$',
+                description='List directory contents',
+            )
+        )
+        self.approved_patterns.append(
+            CommandPattern(pattern=r'^cd(\s+\S+)?$', description='Change directory')
+        )
+        self.approved_patterns.append(
+            CommandPattern(pattern=r'^pwd$', description='Print working directory')
+        )
+        self.approved_patterns.append(
+            CommandPattern(pattern=r'^echo\s+.*$', description='Echo text')
+        )
+
+    def is_command_approved(self, command: str) -> bool:
+        """Check if a command is approved and doesn't need confirmation.
+
+        Args:
+            command: The command to check.
+
+        Returns:
+            bool: True if the command is approved, False otherwise.
+        """
+        if not command or not command.strip():
+            return False
+
+        # Check if this is a piped command
+        if self.parser.is_piped_command(command):
+            # For piped commands, all parts must be approved
+            sub_commands = self.parser.parse_command(command)
+            return all(self._is_single_command_approved(cmd) for cmd in sub_commands)
+        else:
+            # For single commands, just check directly
+            return self._is_single_command_approved(command)
+
+    def _is_single_command_approved(self, command: str) -> bool:
+        """Check if a single (non-piped) command is approved.
+
+        Args:
+            command: The command to check.
+
+        Returns:
+            bool: True if the command is approved, False otherwise.
+        """
+        command = command.strip()
+
+        # Check exact matches first
+        if command in self.approved_commands:
+            return self.approved_commands[command]
+
+        # Then check patterns from CommandPattern objects
+        for pattern in self.approved_patterns:
+            if pattern.matches(command):
+                return True
+
+        # Then check string patterns from the config
+        from openhands.core.config import load_openhands_config
+
+        try:
+            config = load_openhands_config()
+            if hasattr(config, 'security') and hasattr(
+                config.security, 'approved_command_patterns'
+            ):
+                for pattern_str in config.security.approved_command_patterns:
+                    # Compile the pattern if not already compiled
+                    if pattern_str not in self.compiled_patterns:
+                        try:
+                            self.compiled_patterns[pattern_str] = re.compile(
+                                pattern_str
+                            )
+                        except re.error:
+                            # Skip invalid patterns
+                            continue
+
+                    # Check if the command matches the pattern
+                    if self.compiled_patterns[pattern_str].match(command):
+                        return True
+        except Exception:
+            # If there's any error loading the config, just continue without checking patterns
+            pass
+
+        return False
+
+    def approve_command(self, command: str) -> None:
+        """Add a command to the approved commands list.
+
+        Args:
+            command: The command to approve.
+        """
+        self.approved_commands[command] = True
+
+        # In a real implementation, we would save this to config.toml
+        logger.info(f"Command '{command}' approved for future use")
+
+    async def handle_api_request(self, request: Request) -> Any:
+        """Handles the incoming API request."""
+        # This analyzer doesn't need to handle API requests
+        return {'message': "Command approval analyzer doesn't support API requests"}
+
+    async def security_risk(self, event: Action) -> ActionSecurityRisk:
+        """Evaluates the Action for security risks and returns the risk level.
+
+        For command approval analyzer, we always return LOW risk level,
+        but we set the confirmation_state based on whether the command is approved.
+        """
+        # Only process CmdRunAction and IPythonRunCellAction
+        if isinstance(event, CmdRunAction):
+            command = event.command
+            if self.is_command_approved(command):
+                event.confirmation_state = ActionConfirmationStatus.CONFIRMED
+                logger.info(f'Command automatically approved: {command}')
+
+        elif isinstance(event, IPythonRunCellAction):
+            code = event.code
+            if self.is_command_approved(code):
+                event.confirmation_state = ActionConfirmationStatus.CONFIRMED
+                logger.info(f'Python code automatically approved: {code}')
+
+        # Always return LOW risk level - we're not evaluating risk, just auto-approving
+        return ActionSecurityRisk.LOW
+
+    async def act(self, event: Event) -> None:
+        """Performs an action based on the analyzed event.
+
+        This analyzer doesn't need to perform any actions since command approval
+        is handled directly in the CLI interface.
+        """
+        pass

openhands/security/options.py

@@ -1,6 +1,8 @@
 from openhands.security.analyzer import SecurityAnalyzer
+from openhands.security.command_approval.analyzer import CommandApprovalAnalyzer
 from openhands.security.invariant.analyzer import InvariantAnalyzer
 
 SecurityAnalyzers: dict[str, type[SecurityAnalyzer]] = {
     'invariant': InvariantAnalyzer,
+    'command_approval': CommandApprovalAnalyzer,
 }

tests/unit/test_command_patterns.py

@@ -0,0 +1,373 @@
+"""Tests for command pattern generation and parsing."""
+
+import re
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from openhands.cli.tui import (
+    _generate_command_patterns,
+    _generate_single_command_pattern,
+    _handle_pattern_selection,
+    _parse_piped_command,
+    read_confirmation_input,
+)
+
+
+class TestGenerateSingleCommandPattern:
+    """Test the _generate_single_command_pattern function."""
+
+    def test_empty_command(self):
+        """Test pattern generation for empty command."""
+        patterns = _generate_single_command_pattern('')
+        assert len(patterns) == 1
+        assert patterns[0] == '^$'
+
+    def test_single_word_command(self):
+        """Test pattern generation for single word command."""
+        patterns = _generate_single_command_pattern('ls')
+        assert len(patterns) == 1
+        assert patterns[0] == '^ls(\\s.*|$)'
+
+    def test_two_word_command(self):
+        """Test pattern generation for two word command."""
+        patterns = _generate_single_command_pattern('ls -la')
+        assert len(patterns) == 2
+        assert patterns[0] == '^ls(\\s.*|$)'
+        assert patterns[1] == '^ls\\s+\\-la.*$'
+
+    def test_three_word_command(self):
+        """Test pattern generation for three word command."""
+        patterns = _generate_single_command_pattern('git commit -m')
+        assert len(patterns) == 3
+        assert patterns[0] == '^git(\\s.*|$)'
+        assert patterns[1] == '^git\\s+commit.*$'
+        assert patterns[2] == '^git\\s+commit\\s+\\-m.*$'
+
+    def test_four_word_command(self):
+        """Test pattern generation for four word command (should only generate 3 patterns)."""
+        patterns = _generate_single_command_pattern('git commit -m message')
+        assert len(patterns) == 3  # Only first 3 prefixes
+        assert patterns[0] == '^git(\\s.*|$)'
+        assert patterns[1] == '^git\\s+commit.*$'
+        assert patterns[2] == '^git\\s+commit\\s+\\-m.*$'
+
+    def test_pattern_matching(self):
+        """Test that generated patterns actually match similar commands."""
+        patterns = _generate_single_command_pattern('ls -la')
+
+        # Test first pattern (ls.*)
+        pattern1 = re.compile(patterns[0])
+        assert pattern1.match('ls')
+        assert pattern1.match('ls -la')
+        assert pattern1.match('ls -alh /home')
+        assert not pattern1.match('cat file.txt')
+
+        # Test second pattern (ls -la.*)
+        pattern2 = re.compile(patterns[1])
+        assert pattern2.match('ls -la')
+        assert pattern2.match('ls -la /home')
+        assert not pattern2.match('ls')
+        assert not pattern2.match('ls -alh')
+
+    def test_special_characters_escaped(self):
+        """Test that special regex characters are properly escaped."""
+        patterns = _generate_single_command_pattern('echo $HOME')
+        assert len(patterns) == 2
+        assert patterns[0] == '^echo(\\s.*|$)'
+        assert patterns[1] == '^echo\\s+\\$HOME.*$'
+
+        # Test that the pattern works
+        pattern = re.compile(patterns[1])
+        assert pattern.match('echo $HOME')
+        assert pattern.match('echo $HOME/test')
+
+
+class TestParsePipedCommand:
+    """Test the _parse_piped_command function."""
+
+    def test_empty_command(self):
+        """Test parsing empty command."""
+        result = _parse_piped_command('')
+        assert result == []
+
+    def test_whitespace_only_command(self):
+        """Test parsing whitespace-only command."""
+        result = _parse_piped_command('   ')
+        assert result == []
+
+    def test_single_command(self):
+        """Test parsing single command without pipes."""
+        result = _parse_piped_command('ls -la')
+        assert result == ['ls -la']
+
+    def test_simple_piped_command(self):
+        """Test parsing simple piped command."""
+        result = _parse_piped_command('ls -la | grep test')
+        assert result == ['ls -la', 'grep test']
+
+    def test_three_command_pipe(self):
+        """Test parsing three-command pipe."""
+        result = _parse_piped_command('cat file.txt | grep pattern | wc -l')
+        assert result == ['cat file.txt', 'grep pattern', 'wc -l']
+
+    def test_pipe_with_quotes(self):
+        """Test parsing piped command with quoted arguments."""
+        result = _parse_piped_command('echo "hello world" | grep "hello"')
+        # shlex removes quotes, so we get the unquoted content
+        assert result == ['echo hello world', 'grep hello']
+
+    def test_pipe_without_spaces(self):
+        """Test parsing piped command without spaces around pipes."""
+        result = _parse_piped_command('ls|grep test')
+        assert result == ['ls', 'grep test']
+
+    def test_complex_command_with_options(self):
+        """Test parsing complex command with various options."""
+        result = _parse_piped_command(
+            "find /home -name '*.py' | xargs grep -l 'import os'"
+        )
+        # shlex removes quotes, so we get the unquoted content
+        assert result == ['find /home -name *.py', 'xargs grep -l import os']
+
+    def test_invalid_quotes(self):
+        """Test parsing command with invalid quotes falls back gracefully."""
+        result = _parse_piped_command('echo "unclosed quote | grep test')
+        assert result == ['echo "unclosed quote', 'grep test']
+
+
+class TestGenerateCommandPatterns:
+    """Test the _generate_command_patterns function."""
+
+    def test_single_command(self):
+        """Test pattern generation for single command."""
+        patterns = _generate_command_patterns('ls -la')
+        assert len(patterns) == 2
+        assert patterns[0] == '^ls(\\s.*|$)'
+        assert patterns[1] == '^ls\\s+\\-la.*$'
+
+    def test_piped_command(self):
+        """Test pattern generation for piped command."""
+        patterns = _generate_command_patterns('ls -la | grep test')
+        assert len(patterns) == 1
+        # Should combine the first pattern from each sub-command
+        assert patterns[0] == '^ls(\\s.*|$)\\s*\\|\\s*grep(\\s.*|$)$'
+
+    def test_empty_command(self):
+        """Test pattern generation for empty command."""
+        patterns = _generate_command_patterns('')
+        assert len(patterns) == 1
+        assert patterns[0] == '^$'
+
+
+class TestReadConfirmationInput:
+    """Test the read_confirmation_input function."""
+
+    @pytest.mark.asyncio
+    @patch('openhands.cli.tui.cli_confirm')
+    async def test_yes_option(self, mock_confirm):
+        """Test selecting 'yes' option."""
+        mock_confirm.return_value = 0  # First option (Yes, proceed)
+
+        config = MagicMock()
+        config.cli = MagicMock(vi_mode=False)
+
+        result = await read_confirmation_input(config=config, command='ls -la')
+        assert result == 'yes'
+
+    @pytest.mark.asyncio
+    @patch('openhands.cli.tui.cli_confirm')
+    async def test_no_option(self, mock_confirm):
+        """Test selecting 'no' option."""
+        mock_confirm.return_value = 1  # Second option (No)
+
+        config = MagicMock()
+        config.cli = MagicMock(vi_mode=False)
+
+        result = await read_confirmation_input(config=config, command='ls -la')
+        assert result == 'no'
+
+    @pytest.mark.asyncio
+    @patch('openhands.cli.tui.cli_confirm')
+    async def test_always_option(self, mock_confirm):
+        """Test selecting 'always' option."""
+        mock_confirm.return_value = 2  # Third option (Always proceed)
+
+        config = MagicMock()
+        config.cli = MagicMock(vi_mode=False)
+
+        result = await read_confirmation_input(config=config, command='ls -la')
+        assert result == 'always'
+
+    @pytest.mark.asyncio
+    @patch('openhands.cli.tui._handle_pattern_selection')
+    @patch('openhands.cli.tui.cli_confirm')
+    async def test_remember_option(self, mock_confirm, mock_pattern_selection):
+        """Test selecting 'remember' option."""
+        mock_confirm.return_value = 3  # Fourth option (Remember)
+        mock_pattern_selection.return_value = (
+            'remember:^ls.*$:Commands starting with: ls'
+        )
+
+        config = MagicMock()
+        config.cli = MagicMock(vi_mode=False)
+
+        result = await read_confirmation_input(config=config, command='ls -la')
+        assert result == 'remember:^ls.*$:Commands starting with: ls'
+        mock_pattern_selection.assert_called_once_with(config, 'ls -la')
+
+
+class TestHandlePatternSelection:
+    """Test the _handle_pattern_selection function."""
+
+    @pytest.mark.asyncio
+    @patch('openhands.cli.tui.cli_confirm')
+    async def test_exact_command_selection(self, mock_confirm):
+        """Test selecting exact command pattern."""
+        mock_confirm.return_value = 0  # First option (exact command)
+
+        config = MagicMock()
+        config.cli = MagicMock(vi_mode=False)
+
+        result = await _handle_pattern_selection(config, 'ls -la')
+        assert result.startswith('remember:^ls\\ \\-la$:Exact command: ls -la')
+
+    @pytest.mark.asyncio
+    @patch('openhands.cli.tui.cli_confirm')
+    async def test_prefix_pattern_selection(self, mock_confirm):
+        """Test selecting prefix pattern."""
+        mock_confirm.return_value = 1  # Second option (first prefix pattern)
+
+        config = MagicMock()
+        config.cli = MagicMock(vi_mode=False)
+
+        result = await _handle_pattern_selection(config, 'ls -la')
+        assert result.startswith('remember:^ls(\\s.*|$):Commands starting with: ls')
+
+    @pytest.mark.asyncio
+    @patch('openhands.cli.tui.create_prompt_session')
+    @patch('openhands.cli.tui.cli_confirm')
+    async def test_custom_pattern_selection_valid(
+        self, mock_confirm, mock_create_session
+    ):
+        """Test selecting custom pattern with valid regex."""
+        mock_confirm.return_value = (
+            3  # Custom pattern option (assuming 3 total options)
+        )
+
+        # Mock the prompt session
+        mock_session = MagicMock()
+
+        # Create a proper async mock
+        async def mock_prompt_async(prompt):
+            return '^git.*$'
+
+        mock_session.prompt_async = mock_prompt_async
+        mock_create_session.return_value = mock_session
+
+        config = MagicMock()
+        config.cli = MagicMock(vi_mode=False)
+
+        result = await _handle_pattern_selection(config, 'ls -la')
+        assert result == 'remember:^git.*$:Custom pattern: ^git.*$'
+
+    @pytest.mark.asyncio
+    @patch('openhands.cli.tui.create_prompt_session')
+    @patch('openhands.cli.tui.cli_confirm')
+    @patch('openhands.cli.tui.print_formatted_text')
+    async def test_custom_pattern_selection_invalid(
+        self, mock_print, mock_confirm, mock_create_session
+    ):
+        """Test selecting custom pattern with invalid regex."""
+        mock_confirm.return_value = 3  # Custom pattern option
+
+        # Mock the prompt session to return invalid regex
+        mock_session = MagicMock()
+
+        # Create a proper async mock
+        async def mock_prompt_async(prompt):
+            return '[invalid regex'
+
+        mock_session.prompt_async = mock_prompt_async
+        mock_create_session.return_value = mock_session
+
+        config = MagicMock()
+        config.cli = MagicMock(vi_mode=False)
+
+        result = await _handle_pattern_selection(config, 'ls -la')
+        # Should fall back to exact command
+        assert result.startswith('remember:^ls\\ \\-la$:Exact command: ls -la')
+        # Should print error message
+        mock_print.assert_called()
+
+
+class TestPatternMatching:
+    """Test that the generated patterns work correctly for matching commands."""
+
+    def test_ls_patterns(self):
+        """Test patterns generated for ls command."""
+        patterns = _generate_single_command_pattern('ls -alh')
+
+        # Test first pattern (ls.*)
+        pattern1 = re.compile(patterns[0])
+        assert pattern1.match('ls')
+        assert pattern1.match('ls -la')
+        assert pattern1.match('ls -alh /home')
+        assert pattern1.match('ls --help')
+        assert not pattern1.match('cat file.txt')
+        assert not pattern1.match('lsof')  # Should not match partial word
+
+        # Test second pattern (ls -alh.*)
+        pattern2 = re.compile(patterns[1])
+        assert pattern2.match('ls -alh')
+        assert pattern2.match('ls -alh /home')
+        assert not pattern2.match('ls')
+        assert not pattern2.match('ls -la')
+
+    def test_git_patterns(self):
+        """Test patterns generated for git command."""
+        patterns = _generate_single_command_pattern('git commit -m')
+
+        # Test git(\s.*|$)
+        pattern1 = re.compile(patterns[0])
+        assert pattern1.match('git status')
+        assert pattern1.match('git commit')
+        assert pattern1.match('git push origin main')
+        assert not pattern1.match('github')  # Should not match partial word
+
+        # Test git commit.*
+        pattern2 = re.compile(patterns[1])
+        assert pattern2.match('git commit')
+        assert pattern2.match("git commit -m 'message'")
+        assert pattern2.match('git commit --amend')
+        assert not pattern2.match('git status')
+        assert not pattern2.match('git push')
+
+    def test_piped_command_patterns(self):
+        """Test patterns generated for piped commands."""
+        patterns = _generate_command_patterns('cat file.txt | grep pattern')
+
+        pattern = re.compile(patterns[0])
+        assert pattern.match('cat file.txt | grep pattern')
+        assert pattern.match('cat another.txt | grep something')
+        assert pattern.match('cat /path/to/file | grep test')
+        assert not pattern.match('cat file.txt')
+        assert not pattern.match('grep pattern')
+
+    def test_complex_command_patterns(self):
+        """Test patterns for complex commands with special characters."""
+        patterns = _generate_single_command_pattern("find /home -name '*.py'")
+
+        # Test find(\s.*|$)
+        pattern1 = re.compile(patterns[0])
+        assert pattern1.match("find /home -name '*.py'")
+        assert pattern1.match('find . -type f')
+        assert pattern1.match('find /usr/bin -executable')
+        assert not pattern1.match('finder')  # Should not match partial word
+
+        # Test find /home.*
+        pattern2 = re.compile(patterns[1])
+        assert pattern2.match("find /home -name '*.py'")
+        assert pattern2.match('find /home -type d')
+        assert not pattern2.match("find . -name '*.py'")
+        assert not pattern2.match("find /usr -name '*.py'")

tests/unit/test_security_config_pipe_parsing.py

@@ -0,0 +1,48 @@
+"""Tests for the security config pipe parsing functionality."""
+
+from openhands.security.command_approval.analyzer import CommandParser
+
+
+class TestBashlexParsing:
+    """Test bashlex parsing functionality."""
+
+    def test_bashlex_pipe_detection(self):
+        """Test detection of piped commands using bashlex."""
+        parser = CommandParser()
+
+        # Commands with pipes
+        assert parser.is_piped_command('ls -la | grep .py')
+        assert parser.is_piped_command('cat file.txt | head -10 | tail -5')
+        assert parser.is_piped_command("find . -name '*.py' | xargs grep 'import'")
+
+        # Commands without pipes
+        assert not parser.is_piped_command('ls -la')
+        assert not parser.is_piped_command("echo 'hello world'")
+        assert not parser.is_piped_command('ls -la > output.txt')
+
+        # Edge cases
+        assert not parser.is_piped_command('')
+        # Pipe in quotes is not a real pipe
+        assert not parser.is_piped_command("echo 'hello | world'")
+
+    def test_bashlex_command_extraction(self):
+        """Test extraction of commands from pipelines using bashlex."""
+        parser = CommandParser()
+
+        # Simple command
+        assert parser.parse_command('ls -la') == ['ls -la']
+
+        # Piped commands
+        assert parser.parse_command('ls -la | grep .py') == ['ls -la', 'grep .py']
+        assert parser.parse_command('cat file.txt | head -10 | tail -5') == [
+            'cat file.txt',
+            'head -10',
+            'tail -5',
+        ]
+
+        # Commands with redirections
+        assert parser.parse_command('ls -la > output.txt') == ['ls -la']
+
+        # Edge cases
+        assert parser.parse_command('') == []
+        assert parser.parse_command("echo 'hello | world'") == ['echo hello | world']