fix: path traversal security vulnerability in mcp tools (#223)

Signed-off-by: Joe P <joe@basicmemory.com>

Author: jope-bm

src/basic_memory/mcp/async_client.py

@@ -4,26 +4,25 @@
 from basic_memory.api.app import app as fastapi_app
 from basic_memory.config import ConfigManager
 
+
 def create_client() -> AsyncClient:
     """Create an HTTP client based on configuration.
-    
+
     Returns:
         AsyncClient configured for either local ASGI or remote HTTP transport
     """
     config_manager = ConfigManager()
     config = config_manager.load_config()
-    
+
     if config.api_url:
         # Use HTTP transport for remote API
         logger.info(f"Creating HTTP client for remote Basic Memory API: {config.api_url}")
         return AsyncClient(base_url=config.api_url)
     else:
         # Use ASGI transport for local API
         logger.debug("Creating ASGI client for local Basic Memory API")
-        return AsyncClient(
-            transport=ASGITransport(app=fastapi_app), 
-            base_url="http://test"
-        )
+        return AsyncClient(transport=ASGITransport(app=fastapi_app), base_url="http://test")
+
 
 # Create shared async client
 client = create_client()

src/basic_memory/mcp/tools/move_note.py

@@ -11,6 +11,7 @@
 from basic_memory.mcp.project_session import get_active_project
 from basic_memory.schemas import EntityResponse
 from basic_memory.schemas.project_info import ProjectList
+from basic_memory.utils import validate_project_path
 
 
 async def _detect_cross_project_move_attempt(
@@ -393,6 +394,28 @@ async def move_note(
     active_project = get_active_project(project)
     project_url = active_project.project_url
 
+    # Validate destination path to prevent path traversal attacks
+    project_path = active_project.home
+    if not validate_project_path(destination_path, project_path):
+        logger.warning(
+            "Attempted path traversal attack blocked",
+            destination_path=destination_path,
+            project=active_project.name,
+        )
+        return f"""# Move Failed - Security Validation Error
+
+The destination path '{destination_path}' is not allowed - paths must stay within project boundaries.
+
+## Valid path examples:
+- `notes/my-file.md`
+- `projects/2025/meeting-notes.md`
+- `archive/old-notes.md`
+
+## Try again with a safe path:
+```
+move_note("{identifier}", "notes/{destination_path.split("/")[-1] if "/" in destination_path else destination_path}")
+```"""
+
     # Check for potential cross-project move attempts
     cross_project_error = await _detect_cross_project_move_attempt(
         identifier, destination_path, active_project.name

src/basic_memory/mcp/tools/read_content.py

@@ -17,6 +17,7 @@
 from basic_memory.mcp.tools.utils import call_get
 from basic_memory.mcp.project_session import get_active_project
 from basic_memory.schemas.memory import memory_url_path
+from basic_memory.utils import validate_project_path
 
 
 def calculate_target_params(content_length):
@@ -188,6 +189,21 @@ async def read_content(path: str, project: Optional[str] = None) -> dict:
     project_url = active_project.project_url
 
     url = memory_url_path(path)
+
+    # Validate path to prevent path traversal attacks
+    project_path = active_project.home
+    if not validate_project_path(url, project_path):
+        logger.warning(
+            "Attempted path traversal attack blocked",
+            path=path,
+            url=url,
+            project=active_project.name,
+        )
+        return {
+            "type": "error",
+            "error": f"Path '{path}' is not allowed - paths must stay within project boundaries",
+        }
+
     response = await call_get(client, f"{project_url}/resource/{url}")
     content_type = response.headers.get("content-type", "application/octet-stream")
     content_length = int(response.headers.get("content-length", 0))

src/basic_memory/mcp/tools/read_note.py

@@ -11,6 +11,7 @@
 from basic_memory.mcp.tools.utils import call_get
 from basic_memory.mcp.project_session import get_active_project
 from basic_memory.schemas.memory import memory_url_path
+from basic_memory.utils import validate_project_path
 
 
 @mcp.tool(
@@ -67,6 +68,17 @@ async def read_note(
 
     # Get the file via REST API - first try direct permalink lookup
     entity_path = memory_url_path(identifier)
+    
+    # Validate path to prevent path traversal attacks
+    project_path = active_project.home
+    if not validate_project_path(entity_path, project_path):
+        logger.warning(
+            "Attempted path traversal attack blocked",
+            identifier=identifier,
+            entity_path=entity_path,
+            project=active_project.name,
+        )
+        return f"# Error\n\nPath '{identifier}' is not allowed - paths must stay within project boundaries"
     path = f"{project_url}/resource/{entity_path}"
     logger.info(f"Attempting to read note from URL: {path}")
 

src/basic_memory/mcp/tools/write_note.py

@@ -10,7 +10,7 @@
 from basic_memory.mcp.project_session import get_active_project
 from basic_memory.schemas import EntityResponse
 from basic_memory.schemas.base import Entity
-from basic_memory.utils import parse_tags
+from basic_memory.utils import parse_tags, validate_project_path
 
 # Define TagType as a Union that can accept either a string or a list of strings or None
 TagType = Union[List[str], str, None]
@@ -75,6 +75,14 @@ async def write_note(
     # Get the active project first to check project-specific sync status
     active_project = get_active_project(project)
 
+    # Validate folder path to prevent path traversal attacks
+    project_path = active_project.home
+    if folder and not validate_project_path(folder, project_path):
+        logger.warning(
+            "Attempted path traversal attack blocked", folder=folder, project=active_project.name
+        )
+        return f"# Error\n\nFolder path '{folder}' is not allowed - paths must stay within project boundaries"
+
     # Check migration status and wait briefly if needed
     from basic_memory.mcp.tools.utils import wait_for_migration_or_return_status
 

src/basic_memory/utils.py

@@ -213,3 +213,32 @@ def parse_tags(tags: Union[List[str], str, None]) -> List[str]:
     except (ValueError, TypeError):  # pragma: no cover
         logger.warning(f"Couldn't parse tags from input of type {type(tags)}: {tags}")
         return []
+
+
+def validate_project_path(path: str, project_path: Path) -> bool:
+    """Ensure path stays within project boundaries."""
+    # Allow empty strings as they resolve to the project root
+    if not path:
+        return True
+    
+    # Check for obvious path traversal patterns first
+    if ".." in path or "~" in path:
+        return False
+    
+    # Check for Windows-style path traversal (even on Unix systems)
+    if "\\.." in path or path.startswith("\\"):
+        return False
+    
+    # Block absolute paths (Unix-style starting with / or Windows-style with drive letters)
+    if path.startswith("/") or (len(path) >= 2 and path[1] == ":"):
+        return False
+    
+    # Block paths with control characters (but allow whitespace that will be stripped)
+    if path.strip() and any(ord(c) < 32 and c not in [' ', '\t'] for c in path):
+        return False
+    
+    try:
+        resolved = (project_path / path).resolve()
+        return resolved.is_relative_to(project_path.resolve())
+    except (ValueError, OSError):
+        return False

tests/api/test_async_client.py

@@ -1,6 +1,5 @@
 """Tests for async_client configuration."""
 
-import pytest
 from unittest.mock import patch
 from httpx import AsyncClient, ASGITransport
 
@@ -11,12 +10,12 @@
 def test_create_client_uses_asgi_when_no_api_url():
     """Test that create_client uses ASGI transport when api_url is None."""
     mock_config = BasicMemoryConfig(api_url=None)
-    
-    with patch('basic_memory.mcp.async_client.ConfigManager') as mock_config_manager:
+
+    with patch("basic_memory.mcp.async_client.ConfigManager") as mock_config_manager:
         mock_config_manager.return_value.load_config.return_value = mock_config
-        
+
         client = create_client()
-        
+
         assert isinstance(client, AsyncClient)
         assert isinstance(client._transport, ASGITransport)
         assert str(client.base_url) == "http://test"
@@ -26,12 +25,12 @@ def test_create_client_uses_http_when_api_url_set():
     """Test that create_client uses HTTP transport when api_url is configured."""
     remote_url = "https://api.basicmemory.example.com"
     mock_config = BasicMemoryConfig(api_url=remote_url)
-    
-    with patch('basic_memory.mcp.async_client.ConfigManager') as mock_config_manager:
+
+    with patch("basic_memory.mcp.async_client.ConfigManager") as mock_config_manager:
         mock_config_manager.return_value.load_config.return_value = mock_config
-        
+
         client = create_client()
-        
+
         assert isinstance(client, AsyncClient)
         assert not isinstance(client._transport, ASGITransport)
-        assert str(client.base_url) == remote_url
+        assert str(client.base_url) == remote_url