Merge branch 'refactor/shell-executor' into develop

Author: tumf

.github/workflows/publish.yml

@@ -1,15 +1,17 @@
 name: Publish
+
 on:
   push:
     tags:
       - "v*"
-    strategy:
-      matrix:
-        python-version: ["3.11"]
 
 jobs:
   test:
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.11"]
+
     steps:
       - uses: actions/checkout@v4
 
@@ -23,14 +25,13 @@ jobs:
           python -m pip install --upgrade pip
           pip install uv
 
-      - name: Install dev/test dependencies
+      - name: Install dependencies
         run: |
-          pip install -e ".[dev]"
-          pip install -e ".[test]"
+          uv pip install -e ".[dev,test]"
 
-      - name: Run tests
+      - name: Run tests and checks
         run: |
-          make check
+          make all
 
   publish:
     needs: test
@@ -44,14 +45,13 @@ jobs:
 
       - name: Update version from tag
         run: |
-          # Strip 'v' prefix from tag and update version.py
           VERSION=${GITHUB_REF#refs/tags/v}
           echo "__version__ = \"${VERSION}\"" > src/mcp_shell_server/version.py
 
-      - name: Set up Python ${{ matrix.python-version }}
+      - name: Set up Python 3.11
         uses: actions/setup-python@v5
         with:
-          python-version: ${{ matrix.python-version }}
+          python-version: "3.11"
 
       - name: Install uv
         run: |
@@ -60,10 +60,11 @@ jobs:
 
       - name: Build package
         run: |
-          uv build
+          uv pip install build
+          python -m build
 
       - name: Publish to PyPI
-        env:
-          PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}
-        run: |
-          uv publish --token $PYPI_TOKEN
+        uses: pypa/gh-action-pypi-publish@release/v1
+        with:
+          password: ${{ secrets.PYPI_TOKEN }}
+          password: ${{ secrets.PYPI_TOKEN }}

.github/workflows/test.yml

@@ -26,18 +26,17 @@ jobs:
           python -m pip install --upgrade pip
           pip install uv
 
-      - name: Install dev/test dependencies
+      - name: Install dependencies
         run: |
-          pip install -e ".[dev]"
-          pip install -e ".[test]"
+          uv pip install -e ".[dev,test]"
 
-      - name: Run lint and typecheck
+      - name: Run format checks and typecheck
         run: |
-          make lint typecheck
+          make check
 
       - name: Run tests with coverage
         run: |
-          pytest --cov --cov-report=xml
+          make coverage
 
       - name: Upload coverage to Codecov
         uses: codecov/codecov-action@v5

pyproject.toml

@@ -41,10 +41,15 @@ asyncio_mode = "strict"
 testpaths = "tests"
 # Set default event loop scope for async tests 
 asyncio_default_fixture_loop_scope = "function"
+markers = [
+    "macos: marks tests that should only run on macOS",
+    "slow: marks tests as slow running",
+]
 filterwarnings = [
     "ignore::RuntimeWarning:selectors:",
     "ignore::pytest.PytestUnhandledCoroutineWarning:",
     "ignore::pytest.PytestUnraisableExceptionWarning:",
+    "ignore::DeprecationWarning:pytest_asyncio.plugin:",
 ]
 
 [tool.ruff]

src/mcp_shell_server/command_preprocessor.py

@@ -0,0 +1,131 @@
+import shlex
+from typing import Dict, List, Tuple, Union
+
+
+class CommandPreProcessor:
+    """
+    Pre-processes and validates shell commands before execution
+    """
+
+    def preprocess_command(self, command: List[str]) -> List[str]:
+        """
+        Preprocess the command to handle cases where '|' is attached to a command.
+        """
+        preprocessed_command = []
+        for token in command:
+            if token in ["||", "&&", ";"]:  # Special shell operators
+                preprocessed_command.append(token)
+            elif "|" in token and token != "|":
+                parts = token.split("|")
+                preprocessed_command.extend(
+                    [part.strip() for part in parts if part.strip()]
+                )
+                preprocessed_command.append("|")
+            else:
+                preprocessed_command.append(token)
+        return preprocessed_command
+
+    def clean_command(self, command: List[str]) -> List[str]:
+        """
+        Clean command by trimming whitespace from each part.
+        Removes empty strings but preserves arguments that are meant to be spaces.
+
+        Args:
+            command (List[str]): Original command and its arguments
+
+        Returns:
+            List[str]: Cleaned command
+        """
+        return [arg for arg in command if arg]  # Remove empty strings
+
+    def create_shell_command(self, command: List[str]) -> str:
+        """
+        Create a shell command string from a list of arguments.
+        Handles wildcards and arguments properly.
+        """
+        if not command:
+            return ""
+
+        escaped_args = []
+        for arg in command:
+            if arg.isspace():
+                # Wrap space-only arguments in single quotes
+                escaped_args.append(f"'{arg}'")
+            else:
+                # Properly escape all arguments including those with wildcards
+                escaped_args.append(shlex.quote(arg.strip()))
+
+        return " ".join(escaped_args)
+
+    def split_pipe_commands(self, command: List[str]) -> List[List[str]]:
+        """
+        Split commands by pipe operator into separate commands.
+
+        Args:
+            command (List[str]): Command and its arguments with pipe operators
+
+        Returns:
+            List[List[str]]: List of commands split by pipe operator
+        """
+        commands: List[List[str]] = []
+        current_command: List[str] = []
+
+        for arg in command:
+            if arg.strip() == "|":
+                if current_command:
+                    commands.append(current_command)
+                    current_command = []
+            else:
+                current_command.append(arg)
+
+        if current_command:
+            commands.append(current_command)
+
+        return commands
+
+    def parse_command(
+        self, command: List[str]
+    ) -> Tuple[List[str], Dict[str, Union[None, str, bool]]]:
+        """
+        Parse command and extract redirections.
+        """
+        cmd = []
+        redirects: Dict[str, Union[None, str, bool]] = {
+            "stdin": None,
+            "stdout": None,
+            "stdout_append": False,
+        }
+
+        i = 0
+        while i < len(command):
+            token = command[i]
+
+            # Shell operators check
+            if token in ["|", ";", "&&", "||"]:
+                raise ValueError(f"Unexpected shell operator: {token}")
+
+            # Output redirection
+            if token in [">", ">>"]:
+                if i + 1 >= len(command):
+                    raise ValueError("Missing path for output redirection")
+                if i + 1 < len(command) and command[i + 1] in [">", ">>", "<"]:
+                    raise ValueError("Invalid redirection target: operator found")
+                path = command[i + 1]
+                redirects["stdout"] = path
+                redirects["stdout_append"] = token == ">>"
+                i += 2
+                continue
+
+            # Input redirection
+            if token == "<":
+                if i + 1 >= len(command):
+                    raise ValueError("Missing path for input redirection")
+                path = command[i + 1]
+                redirects["stdin"] = path
+                i += 2
+                continue
+
+            cmd.append(token)
+            i += 1
+
+        return cmd, redirects

src/mcp_shell_server/command_validator.py

@@ -0,0 +1,104 @@
+"""
+Provides validation for shell commands and ensures they are allowed to be executed.
+"""
+
+import os
+from typing import Dict, List
+
+
+class CommandValidator:
+    """
+    Validates shell commands against a whitelist and checks for unsafe operators.
+    """
+
+    def __init__(self):
+        """
+        Initialize the validator.
+        """
+        pass
+
+    def _get_allowed_commands(self) -> set[str]:
+        """Get the set of allowed commands from environment variables"""
+        allow_commands = os.environ.get("ALLOW_COMMANDS", "")
+        allowed_commands = os.environ.get("ALLOWED_COMMANDS", "")
+        commands = allow_commands + "," + allowed_commands
+        return {cmd.strip() for cmd in commands.split(",") if cmd.strip()}
+
+    def get_allowed_commands(self) -> list[str]:
+        """Get the list of allowed commands from environment variables"""
+        return list(self._get_allowed_commands())
+
+    def is_command_allowed(self, command: str) -> bool:
+        """Check if a command is in the allowed list"""
+        cmd = command.strip()
+        return cmd in self._get_allowed_commands()
+
+    def validate_no_shell_operators(self, cmd: str) -> None:
+        """
+        Validate that the command does not contain shell operators.
+
+        Args:
+            cmd (str): Command to validate
+
+        Raises:
+            ValueError: If the command contains shell operators
+        """
+        if cmd in [";", "&&", "||", "|"]:
+            raise ValueError(f"Unexpected shell operator: {cmd}")
+
+    def validate_pipeline(self, commands: List[str]) -> Dict[str, str]:
+        """
+        Validate pipeline command and ensure all parts are allowed.
+
+        Args:
+            commands (List[str]): List of commands to validate
+
+        Returns:
+            Dict[str, str]: Error message if validation fails, empty dict if success
+
+        Raises:
+            ValueError: If validation fails
+        """
+        current_cmd: List[str] = []
+
+        for token in commands:
+            if token == "|":
+                if not current_cmd:
+                    raise ValueError("Empty command before pipe operator")
+                if not self.is_command_allowed(current_cmd[0]):
+                    raise ValueError(f"Command not allowed: {current_cmd[0]}")
+                current_cmd = []
+            elif token in [";", "&&", "||"]:
+                raise ValueError(f"Unexpected shell operator in pipeline: {token}")
+            else:
+                current_cmd.append(token)
+
+        if current_cmd:
+            if not self.is_command_allowed(current_cmd[0]):
+                raise ValueError(f"Command not allowed: {current_cmd[0]}")
+
+        return {}
+
+    def validate_command(self, command: List[str]) -> None:
+        """
+        Validate if the command is allowed to be executed.
+
+        Args:
+            command (List[str]): Command and its arguments
+
+        Raises:
+            ValueError: If the command is empty, not allowed, or contains invalid shell operators
+        """
+        if not command:
+            raise ValueError("Empty command")
+
+        allowed_commands = self._get_allowed_commands()
+        if not allowed_commands:
+            raise ValueError(
+                "No commands are allowed. Please set ALLOW_COMMANDS environment variable."
+            )
+
+        # Clean and check the first command
+        cleaned_cmd = command[0].strip()
+        if cleaned_cmd not in allowed_commands:
+            raise ValueError(f"Command not allowed: {cleaned_cmd}")