feat: make directory required and absolute path

Author: Yoshihiro Takahara

src/mcp_shell_server/server.py

@@ -64,7 +64,7 @@ async def run_tool(self, arguments: dict) -> Sequence[TextContent]:
         """Execute the shell command with the given arguments"""
         command = arguments.get("command", [])
         stdin = arguments.get("stdin")
-        directory = arguments.get("directory")
+        directory = arguments.get("directory", "/tmp")  # default to /tmp for safety
         timeout = arguments.get("timeout")
 
         if not command:
@@ -73,7 +73,11 @@ async def run_tool(self, arguments: dict) -> Sequence[TextContent]:
         if not isinstance(command, list):
             raise ValueError("'command' must be an array")
 
-        result = await self.executor.execute(command, stdin, directory, timeout)
+        # Make sure directory exists
+        if not directory:
+            raise ValueError("Directory is required")
+
+        result = await self.executor.execute(command, directory, stdin, timeout)
 
         # Raise error if command execution failed
         if result.get("error"):

src/mcp_shell_server/shell_executor.py

@@ -238,11 +238,14 @@ def _validate_directory(self, directory: Optional[str]) -> None:
             directory (Optional[str]): Directory path to validate
 
         Raises:
-            ValueError: If the directory doesn't exist or is not accessible
+            ValueError: If the directory doesn't exist, not absolute or is not accessible
         """
+        # make directory required
         if directory is None:
-            return
-
+            raise ValueError("Directory is required")
+        # verify directory is absolute path
+        if not os.path.isabs(directory):
+            raise ValueError(f"Directory must be an absolute path: {directory}")
         if not os.path.exists(directory):
             raise ValueError(f"Directory does not exist: {directory}")
         if not os.path.isdir(directory):
@@ -371,8 +374,8 @@ def _preprocess_command(self, command: List[str]) -> List[str]:
     async def execute(
         self,
         command: List[str],
+        directory: str,
         stdin: Optional[str] = None,
-        directory: Optional[str] = None,
         timeout: Optional[int] = None,
     ) -> Dict[str, Any]:
         start_time = time.time()

tests/test_server.py

@@ -32,22 +32,26 @@ async def test_list_tools():
 
 
 @pytest.mark.asyncio
-async def test_call_tool_valid_command(monkeypatch):
+async def test_call_tool_valid_command(monkeypatch, temp_test_dir):
     """Test execution of a valid command"""
     monkeypatch.setenv("ALLOW_COMMANDS", "echo")
-    result = await call_tool("shell_execute", {"command": ["echo", "hello world"]})
+    result = await call_tool(
+        "shell_execute",
+        {"command": ["echo", "hello world"], "directory": temp_test_dir},
+    )
     assert len(result) == 1
     assert isinstance(result[0], TextContent)
     assert result[0].type == "text"
     assert result[0].text.strip() == "hello world"
 
 
 @pytest.mark.asyncio
-async def test_call_tool_with_stdin(monkeypatch):
+async def test_call_tool_with_stdin(monkeypatch, temp_test_dir):
     """Test command execution with stdin"""
     monkeypatch.setenv("ALLOW_COMMANDS", "cat")
     result = await call_tool(
-        "shell_execute", {"command": ["cat"], "stdin": "test input"}
+        "shell_execute",
+        {"command": ["cat"], "stdin": "test input", "directory": temp_test_dir},
     )
     assert len(result) == 1
     assert isinstance(result[0], TextContent)
@@ -56,11 +60,14 @@ async def test_call_tool_with_stdin(monkeypatch):
 
 
 @pytest.mark.asyncio
-async def test_call_tool_invalid_command(monkeypatch):
+async def test_call_tool_invalid_command(monkeypatch, temp_test_dir):
     """Test execution of an invalid command"""
     monkeypatch.setenv("ALLOW_COMMANDS", "echo")
     with pytest.raises(RuntimeError) as excinfo:
-        await call_tool("shell_execute", {"command": ["invalid_command"]})
+        await call_tool(
+            "shell_execute",
+            {"command": ["invalid_command"], "directory": temp_test_dir},
+        )
     assert "Command not allowed: invalid_command" in str(excinfo.value)
 
 

tests/test_shell_executor.py

@@ -21,67 +21,69 @@ def temp_test_dir():
 
 
 @pytest.mark.asyncio
-async def test_basic_command_execution(executor, monkeypatch):
+async def test_basic_command_execution(executor, temp_test_dir, monkeypatch):
     monkeypatch.setenv("ALLOW_COMMANDS", "echo")
-    result = await executor.execute(["echo", "hello"])
+    result = await executor.execute(["echo", "hello"], temp_test_dir)
     assert result["stdout"].strip() == "hello"
     assert result["status"] == 0
 
 
 @pytest.mark.asyncio
-async def test_stdin_input(executor, monkeypatch):
+async def test_stdin_input(executor, temp_test_dir, monkeypatch):
     monkeypatch.setenv("ALLOW_COMMANDS", "cat")
-    result = await executor.execute(["cat"], stdin="hello world")
+    result = await executor.execute(["cat"], temp_test_dir, stdin="hello world")
     assert result["stdout"].strip() == "hello world"
     assert result["status"] == 0
 
 
 @pytest.mark.asyncio
-async def test_command_with_space_allowed(executor, monkeypatch):
+async def test_command_with_space_allowed(executor, temp_test_dir, monkeypatch):
     monkeypatch.setenv("ALLOW_COMMANDS", "cat")
-    result = await executor.execute(["cat "], stdin="hello world")
+    result = await executor.execute(["cat "], temp_test_dir, stdin="hello world")
     assert result["error"] is None
     assert result["stdout"].strip() == "hello world"
     assert result["status"] == 0
 
 
 @pytest.mark.asyncio
-async def test_command_not_allowed(executor, monkeypatch):
+async def test_command_not_allowed(executor, temp_test_dir, monkeypatch):
     monkeypatch.setenv("ALLOW_COMMANDS", "ls")
-    result = await executor.execute(["rm", "-rf", "/"])
+    result = await executor.execute(["rm", "-rf", "/"], temp_test_dir)
     assert result["error"] == "Command not allowed: rm"
     assert result["status"] == 1
 
 
 @pytest.mark.asyncio
-async def test_empty_command(executor):
-    result = await executor.execute([])
+async def test_empty_command(executor, temp_test_dir):
+    result = await executor.execute([], temp_test_dir)
     assert result["error"] == "Empty command"
     assert result["status"] == 1
 
 
 @pytest.mark.asyncio
-async def test_command_with_space_in_allow_commands(executor, monkeypatch):
+async def test_command_with_space_in_allow_commands(
+    executor, temp_test_dir, monkeypatch
+):
     monkeypatch.setenv("ALLOW_COMMANDS", "ls, echo ,cat")
-    result = await executor.execute(["echo", "test"])
+    result = await executor.execute(["echo", "test"], temp_test_dir)
     assert result["stdout"].strip() == "test"
     assert result["status"] == 0
 
 
 @pytest.mark.asyncio
-async def test_multiple_commands_with_operator(executor, monkeypatch):
+async def test_multiple_commands_with_operator(executor, temp_test_dir, monkeypatch):
     monkeypatch.setenv("ALLOW_COMMANDS", "echo,ls")
-    result = await executor.execute(["echo", "hello", ";"])
+    result = await executor.execute(["echo", "hello", ";"], temp_test_dir)
     assert result["error"] == "Unexpected shell operator: ;"
     assert result["status"] == 1
 
 
 @pytest.mark.asyncio
-async def test_shell_operators_not_allowed(executor, monkeypatch):
+async def test_shell_operators_not_allowed(executor, temp_test_dir, monkeypatch):
     monkeypatch.setenv("ALLOW_COMMANDS", "echo,ls,true")
     operators = [";", "&&", "||"]
     for op in operators:
-        result = await executor.execute(["echo", "hello", op, "true"])
+        result = await executor.execute(["echo", "hello", op, "true"], temp_test_dir)
         assert result["error"] == f"Unexpected shell operator: {op}"
         assert result["status"] == 1
 
@@ -140,17 +142,6 @@ async def test_execute_with_file_as_directory(executor, temp_test_dir, monkeypat
     assert result["status"] == 1
 
 
-@pytest.mark.asyncio
-async def test_execute_with_no_directory_specified(executor, monkeypatch):
-    """Test command execution without specifying a directory"""
-    monkeypatch.setenv("ALLOW_COMMANDS", "pwd")
-    result = await executor.execute(["pwd"])
-    assert result["error"] is None
-    assert result["status"] == 0
-    assert os.path.exists(result["stdout"].strip())
-
-
-@pytest.mark.asyncio
 async def test_execute_with_nested_directory(executor, temp_test_dir, monkeypatch):
     """Test command execution in a nested directory"""
     monkeypatch.setenv("ALLOW_COMMANDS", "pwd,mkdir,ls")
@@ -167,54 +158,54 @@ async def test_execute_with_nested_directory(executor, temp_test_dir, monkeypatc
 
 
 @pytest.mark.asyncio
-async def test_command_timeout(executor, monkeypatch):
+async def test_command_timeout(executor, temp_test_dir, monkeypatch):
     """Test command timeout functionality"""
     monkeypatch.setenv("ALLOW_COMMANDS", "sleep")
-    result = await executor.execute(["sleep", "2"], timeout=1)
+    result = await executor.execute(["sleep", "2"], temp_test_dir, timeout=1)
     assert result["error"] == "Command timed out after 1 seconds"
     assert result["status"] == -1
     assert result["stdout"] == ""
     assert result["stderr"] == "Command timed out after 1 seconds"
 
 
 @pytest.mark.asyncio
-async def test_command_completes_within_timeout(executor, monkeypatch):
+async def test_command_completes_within_timeout(executor, temp_test_dir, monkeypatch):
     """Test command that completes within timeout period"""
     monkeypatch.setenv("ALLOW_COMMANDS", "sleep")
-    result = await executor.execute(["sleep", "1"], timeout=2)
+    result = await executor.execute(["sleep", "1"], temp_test_dir, timeout=2)
     assert result["error"] is None
     assert result["status"] == 0
     assert result["stdout"] == ""
 
 
 @pytest.mark.asyncio
-async def test_allowed_commands_alias(executor, monkeypatch):
+async def test_allowed_commands_alias(executor, temp_test_dir, monkeypatch):
     """Test ALLOWED_COMMANDS alias functionality"""
     monkeypatch.setenv("ALLOWED_COMMANDS", "echo")
-    result = await executor.execute(["echo", "hello"])
+    result = await executor.execute(["echo", "hello"], temp_test_dir)
     assert result["stdout"].strip() == "hello"
     assert result["status"] == 0
 
 
 @pytest.mark.asyncio
-async def test_both_allow_commands_vars(executor, monkeypatch):
+async def test_both_allow_commands_vars(executor, temp_test_dir, monkeypatch):
     """Test both ALLOW_COMMANDS and ALLOWED_COMMANDS working together"""
     monkeypatch.setenv("ALLOW_COMMANDS", "echo")
     monkeypatch.setenv("ALLOWED_COMMANDS", "cat")
 
     # Test command from ALLOW_COMMANDS
-    result1 = await executor.execute(["echo", "hello"])
+    result1 = await executor.execute(["echo", "hello"], temp_test_dir)
     assert result1["stdout"].strip() == "hello"
     assert result1["status"] == 0
 
     # Test command from ALLOWED_COMMANDS
-    result2 = await executor.execute(["cat"], stdin="world")
+    result2 = await executor.execute(["cat"], temp_test_dir, stdin="world")
     assert result2["stdout"].strip() == "world"
     assert result2["status"] == 0
 
 
 @pytest.mark.asyncio
-async def test_allow_commands_precedence(executor, monkeypatch):
+async def test_allow_commands_precedence(executor, temp_test_dir, monkeypatch):
     """Test that commands are combined from both environment variables"""
     monkeypatch.setenv("ALLOW_COMMANDS", "echo,ls")
     monkeypatch.setenv("ALLOWED_COMMANDS", "echo,cat")
@@ -224,10 +215,12 @@ async def test_allow_commands_precedence(executor, monkeypatch):
 
 
 @pytest.mark.asyncio
-async def test_pipe_operator(executor, monkeypatch):
+async def test_pipe_operator(executor, temp_test_dir, monkeypatch):
     """Test that pipe operator works correctly"""
     monkeypatch.setenv("ALLOW_COMMANDS", "echo,grep")
-    result = await executor.execute(["echo", "hello\nworld", "|", "grep", "world"])
+    result = await executor.execute(
+        ["echo", "hello\nworld", "|", "grep", "world"], temp_test_dir
+    )
     assert result["error"] is None
     assert result["status"] == 0
     assert result["stdout"].strip() == "world"
@@ -237,12 +230,15 @@ async def test_pipe_operator(executor, monkeypatch):
 async def test_pipe_commands(executor, temp_test_dir, monkeypatch):
     """Test piping commands together"""
     monkeypatch.setenv("ALLOW_COMMANDS", "echo,grep,cut,tr")
-    result = await executor.execute(["echo", "hello world", "|", "grep", "world"])
+    result = await executor.execute(
+        ["echo", "hello world", "|", "grep", "world"], temp_test_dir
+    )
     assert result["stdout"].strip() == "hello world"
 
     # Test multiple pipes
     result = await executor.execute(
-        ["echo", "hello world", "|", "cut", "-d", " ", "-f2", "|", "tr", "a-z", "A-Z"]
+        ["echo", "hello world", "|", "cut", "-d", " ", "-f2", "|", "tr", "a-z", "A-Z"],
+        temp_test_dir,
     )
     assert result["stdout"].strip() == "WORLD"
 

tests/test_shell_executor_error_cases.py

@@ -1,10 +1,20 @@
 import asyncio
+import os
+import tempfile
 
 import pytest
 
 from mcp_shell_server.shell_executor import ShellExecutor
 
 
+@pytest.fixture
+def temp_test_dir():
+    """Create a temporary directory for testing"""
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        # Return the real path to handle macOS /private/tmp symlink
+        yield os.path.realpath(tmpdirname)
+
+
 @pytest.mark.asyncio
 async def test_empty_command_validation():
     """Test validation of empty commands"""
@@ -43,23 +53,23 @@ async def test_shell_operator_validation():
 
 
 @pytest.mark.asyncio
-async def test_process_execution_timeout(monkeypatch):
+async def test_process_execution_timeout(monkeypatch, temp_test_dir):
     """Test process execution timeout handling"""
     monkeypatch.setenv("ALLOW_COMMANDS", "sleep")
     executor = ShellExecutor()
 
     # プロセスのタイムアウトをテスト
     command = ["sleep", "5"]
     with pytest.raises(asyncio.TimeoutError):
-        await asyncio.wait_for(executor.execute(command), timeout=0.1)
+        await asyncio.wait_for(executor.execute(command, temp_test_dir), timeout=0.1)
 
 
 @pytest.mark.asyncio
-async def test_process_failure(monkeypatch):
+async def test_process_failure(monkeypatch, temp_test_dir):
     """Test handling of process execution failure"""
     monkeypatch.setenv("ALLOW_COMMANDS", "false")
     executor = ShellExecutor()
 
     # falseコマンドは常に終了コード1を返す
-    result = await executor.execute(["false"])
+    result = await executor.execute(["false"], temp_test_dir)
     assert result["status"] == 1

tests/test_shell_executor_pipe.py

@@ -1,26 +1,41 @@
+import os
+import tempfile
+
 import pytest
 
 from mcp_shell_server.shell_executor import ShellExecutor
 
 
+@pytest.fixture
+def temp_test_dir():
+    """Create a temporary directory for testing"""
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        # Return the real path to handle macOS /private/tmp symlink
+        yield os.path.realpath(tmpdirname)
+
+
 @pytest.fixture
 def executor():
     return ShellExecutor()
 
 
 @pytest.mark.asyncio
-async def test_basic_pipe_command(executor, monkeypatch):
+async def test_basic_pipe_command(executor, temp_test_dir, monkeypatch):
     """Test basic pipe functionality with allowed commands"""
     monkeypatch.setenv("ALLOW_COMMANDS", "echo,grep")
-    result = await executor.execute(["echo", "hello world", "|", "grep", "world"])
+    result = await executor.execute(
+        ["echo", "hello world", "|", "grep", "world"], temp_test_dir
+    )
     assert result["status"] == 0
     assert result["stdout"].strip() == "hello world"
 
 
 @pytest.mark.asyncio
-async def test_invalid_pipe_command(executor, monkeypatch):
+async def test_invalid_pipe_command(executor, temp_test_dir, monkeypatch):
     """Test pipe command with non-allowed command"""
     monkeypatch.setenv("ALLOW_COMMANDS", "echo")
-    result = await executor.execute(["echo", "hello", "|", "grep", "hello"])
+    result = await executor.execute(
+        ["echo", "hello", "|", "grep", "hello"], temp_test_dir
+    )
     assert result["status"] == 1
     assert "Command not allowed: grep" in result["error"]