Improve field validation for path parameters (#220)

* Return a Path object from path validation function

A Path object is much more robust and useful for internal processing.

* Use field validation instead of calling functions directly

* Raise ToolError instead of returning strings

* Allow Path or string objects

Author: samdoran

src/linux_mcp_server/formatters.py

@@ -5,6 +5,7 @@
 """
 
 from datetime import datetime
+from pathlib import Path
 
 from linux_mcp_server.utils import format_bytes
 from linux_mcp_server.utils.types import CpuInfo
@@ -364,7 +365,7 @@ def format_journal_logs(
     return "\n".join(lines)
 
 
-def format_log_file(stdout: str, log_path: str, lines_count: int) -> str:
+def format_log_file(stdout: str, log_path: Path, lines_count: int) -> str:
     """Format log file output.
 
     Args:
@@ -456,7 +457,7 @@ def format_hardware_info(results: dict[str, str]) -> str:
 
 def format_directory_listing(
     entries: list[NodeEntry],
-    path: str,
+    path: str | Path,
     sort_by: str,
     reverse: bool = False,
 ) -> str:
@@ -496,7 +497,7 @@ def format_directory_listing(
 
 def format_file_listing(
     entries: list[NodeEntry],
-    path: str,
+    path: str | Path,
     sort_by: str,
     reverse: bool = False,
 ) -> str:

src/linux_mcp_server/tools/logs.py

@@ -4,8 +4,10 @@
 
 from pathlib import Path
 
+from fastmcp.exceptions import ToolError
 from mcp.types import ToolAnnotations
 from pydantic import Field
+from pydantic.functional_validators import BeforeValidator
 
 from linux_mcp_server.audit import log_tool_call
 from linux_mcp_server.commands import get_command
@@ -17,7 +19,6 @@
 from linux_mcp_server.utils.decorators import disallow_local_execution_in_containers
 from linux_mcp_server.utils.types import Host
 from linux_mcp_server.utils.validation import is_empty_output
-from linux_mcp_server.utils.validation import PathValidationError
 from linux_mcp_server.utils.validation import validate_path
 
 
@@ -143,9 +144,10 @@ async def get_journal_logs(
 )
 @log_tool_call
 @disallow_local_execution_in_containers
-async def read_log_file(  # noqa: C901
+async def read_log_file(
     log_path: t.Annotated[
-        str,
+        Path,
+        BeforeValidator(validate_path),
         Field(
             description="Absolute path to the log file (must be in allowed list)",
             examples=["/var/log/messages", "/var/log/secure", "/var/log/audit/audit.log", "/var/log/dnf.log"],
@@ -163,22 +165,16 @@ async def read_log_file(  # noqa: C901
     allowed_paths_env = CONFIG.allowed_log_paths
 
     if not allowed_paths_env:
-        return (
+        raise ToolError(
             "No log files are allowed. Set LINUX_MCP_ALLOWED_LOG_PATHS environment variable "
             "with comma-separated list of allowed log file paths."
         )
 
-    allowed_paths = [p.strip() for p in allowed_paths_env.split(",") if p.strip()]
-
-    # Validate path for injection attacks (applies to both local and remote)
-    try:
-        validated_path = validate_path(log_path)
-    except PathValidationError as e:
-        return f"Invalid log file path: {e}"
+    allowed_paths = [Path(p.strip()) for p in allowed_paths_env.split(",") if p.strip()]
 
     if not host:
         # For local execution, resolve and check against allowlist
-        requested_path = Path(validated_path).resolve()
+        requested_path = log_path.resolve()
 
         is_allowed = False
         for allowed_path in allowed_paths:
@@ -188,36 +184,32 @@ async def read_log_file(  # noqa: C901
                 break
 
         if not is_allowed:
-            return (
-                f"Access to log file '{log_path}' is not allowed.\n"
-                f"Allowed log files: {', '.join(allowed_paths)}"
-            )  # nofmt
+            raise ToolError(f"Access to log file '{log_path}' is not allowed.")
 
         if not requested_path.exists():
-            return f"Log file not found: {log_path}"
+            raise ToolError(f"Log file not found: {log_path}")
 
         if not requested_path.is_file():
-            return f"Path is not a file: {log_path}"
+            raise ToolError(f"Path is not a file: {log_path}")
 
         log_path_str = str(requested_path)
     else:
         # For remote execution, check against allowlist without resolving
-        if validated_path not in allowed_paths:
-            return (
-                f"Access to log file '{log_path}' is not allowed.\n"
-                f"Allowed log files: {', '.join(allowed_paths)}"
-            )  # nofmt
-        log_path_str = validated_path
+        if log_path not in allowed_paths:
+            raise ToolError(f"Access to log file '{log_path}' is not allowed.")
+
+        log_path_str = str(log_path)
 
     cmd = get_command("read_log_file")
     returncode, stdout, stderr = await cmd.run(host=host, lines=lines, log_path=log_path_str)
 
     if returncode != 0:
         if "Permission denied" in stderr:
-            return f"Permission denied reading log file: {log_path}"
-        return f"Error reading log file: {stderr}"
+            raise ToolError(f"Permission denied reading log file: {log_path}")
+
+        raise ToolError(f"Error reading log file: {stderr}")
 
     if is_empty_output(stdout):
-        return f"Log file is empty: {log_path}"
+        raise ToolError(f"Log file is empty: {log_path}")
 
     return format_log_file(stdout, log_path, lines)

src/linux_mcp_server/tools/storage.py

@@ -3,9 +3,12 @@
 import os
 import typing as t
 
+from pathlib import Path
+
 from fastmcp.exceptions import ToolError
 from mcp.types import ToolAnnotations
 from pydantic import Field
+from pydantic.functional_validators import BeforeValidator
 
 from linux_mcp_server.audit import log_tool_call
 from linux_mcp_server.commands import get_command
@@ -19,7 +22,6 @@
 from linux_mcp_server.utils.decorators import disallow_local_execution_in_containers
 from linux_mcp_server.utils.types import Host
 from linux_mcp_server.utils.validation import is_successful_output
-from linux_mcp_server.utils.validation import PathValidationError
 from linux_mcp_server.utils.validation import validate_path
 
 
@@ -47,14 +49,6 @@ class SortBy(StrEnum):
 }
 
 
-def _validate_path(path: str) -> str:
-    """Validate path, converting PathValidationError to ToolError for MCP compatibility."""
-    try:
-        return validate_path(path)
-    except PathValidationError as e:
-        raise ToolError(str(e)) from e
-
-
 @mcp.tool(
     title="List block devices",
     description="List block devices on the system",
@@ -90,7 +84,8 @@ async def list_block_devices(
 @disallow_local_execution_in_containers
 async def list_directories(
     path: t.Annotated[
-        str,
+        Path,
+        BeforeValidator(validate_path),
         Field(
             description="Absolute path to the directory to analyze",
             examples=["/var/log", "/etc", "/home", "/opt", "/tmp"],
@@ -113,8 +108,6 @@ async def list_directories(
     Retrieves subdirectories with their size (when ordered by size) or
     modification time, supporting flexible sorting and result limiting.
     """
-    path = _validate_path(path)
-
     # Get the appropriate command for the order_by field
     cmd_name = DIRECTORY_COMMANDS[order_by]
     cmd = get_command(cmd_name)
@@ -154,7 +147,8 @@ async def list_directories(
 @disallow_local_execution_in_containers
 async def list_files(
     path: t.Annotated[
-        str,
+        Path,
+        BeforeValidator(validate_path),
         Field(
             description="Absolute path to the directory to analyze",
             examples=["/var/log", "/etc", "/home", "/opt", "/tmp"],
@@ -177,10 +171,6 @@ async def list_files(
     Retrieves files with their size or modification time, supporting flexible
     sorting and result limiting. Useful for finding large or recently modified files.
     """
-    # For local execution, validate path
-    if not host:
-        path = _validate_path(path)
-
     # Get the appropriate command for the order_by field
     cmd_name = FILE_COMMANDS[order_by]
     cmd = get_command(cmd_name)
@@ -218,7 +208,8 @@ async def list_files(
 @disallow_local_execution_in_containers
 async def read_file(
     path: t.Annotated[
-        str,
+        Path,
+        BeforeValidator(validate_path),
         Field(
             description="Absolute path to the file to read",
             examples=["/etc/hosts", "/etc/resolv.conf", "/etc/os-release", "/proc/cpuinfo"],
@@ -231,10 +222,6 @@ async def read_file(
     Retrieves the full contents of a text file. The path must be absolute
     and the file must exist. Binary files may not display correctly.
     """
-
-    # Validate path
-    path = _validate_path(path)
-
     if not host:
         # For local execution, check early if file exists
         if not os.path.isfile(path):

src/linux_mcp_server/utils/validation.py

@@ -11,7 +11,7 @@ class PathValidationError(ValueError):
     pass
 
 
-def validate_path(path: str) -> str:
+def validate_path(path: str) -> Path:
     """Validate a filesystem path for security and correctness.
 
     Performs security checks to prevent command injection and path traversal attacks:
@@ -53,7 +53,7 @@ def validate_path(path: str) -> str:
     if not Path(path).is_absolute():
         raise PathValidationError(f"Path must be absolute: {path}")
 
-    return Path(path).as_posix()
+    return Path(path)
 
 
 def is_empty_output(stdout: str | None) -> bool:

tests/test_formatters.py

@@ -1,5 +1,7 @@
 """Tests for formatters module."""
 
+from pathlib import Path
+
 from linux_mcp_server.formatters import format_block_devices
 from linux_mcp_server.formatters import format_cpu_info
 from linux_mcp_server.formatters import format_directory_listing
@@ -416,7 +418,7 @@ class TestFormatLogFile:
     def test_format_log_file(self):
         """Test formatting log file."""
         stdout = "2024-01-01 10:00:00 INFO Application started"
-        result = format_log_file(stdout, "/var/log/app.log", 100)
+        result = format_log_file(stdout, Path("/var/log/app.log"), 100)
         assert "=== Log File: /var/log/app.log (last 100 lines) ===" in result
         assert "Application started" in result