fix(mcp): harden workspace permalink routing

Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/api/app.py

@@ -33,6 +33,7 @@
 from basic_memory.workspace_context import (
     WORKSPACE_SLUG_HEADER,
     WORKSPACE_TYPE_HEADER,
+    validate_workspace_permalink_context_values,
     workspace_permalink_context,
 )
 
@@ -100,19 +101,19 @@ async def workspace_permalink_context_middleware(request: Request, call_next):
     workspace_slug = request.headers.get(WORKSPACE_SLUG_HEADER)
     workspace_type = request.headers.get(WORKSPACE_TYPE_HEADER)
 
-    if bool(workspace_slug) != bool(workspace_type):
+    try:
+        validate_workspace_permalink_context_values(workspace_slug, workspace_type)
+    except ValueError as exc:
         return JSONResponse(
             status_code=400,
-            content={
-                "detail": (
-                    f"{WORKSPACE_SLUG_HEADER} and {WORKSPACE_TYPE_HEADER} must be provided together"
-                )
-            },
+            content={"detail": str(exc)},
         )
 
     if not workspace_slug:
         return await call_next(request)
 
+    # ContextVar state remains active across the awaited downstream handler while
+    # this context manager is open, so entity creation can see request metadata.
     with workspace_permalink_context(
         workspace_slug=workspace_slug,
         workspace_type=workspace_type,

src/basic_memory/mcp/project_context.py

@@ -483,6 +483,15 @@ async def resolve_workspace_qualified_memory_url(
             f"Project '{project_identifier}' was not found in workspace "
             f"'{workspace.name}' ({workspace.slug}). Available projects: {available}"
         )
+    if len(matches) > 1:
+        details = ", ".join(
+            f"{entry.qualified_name} ({entry.project.external_id})" for entry in matches
+        )
+        raise ValueError(
+            f"Project '{project_identifier}' matched multiple projects in workspace "
+            f"'{workspace.name}' ({workspace.slug}). Project permalinks must be unique. "
+            f"Matches: {details}"
+        )
 
     entry = matches[0]
     canonical_path = f"{entry.workspace.slug}/{entry.project.permalink}/{remainder}"
@@ -707,6 +716,15 @@ async def resolve_workspace_project_identifier(
                 f"Project '{project_identifier}' was not found in workspace "
                 f"'{workspace.name}' ({workspace.slug}). Available projects: {available}"
             )
+        if len(matches) > 1:
+            details = ", ".join(
+                f"{entry.qualified_name} ({entry.project.external_id})" for entry in matches
+            )
+            raise ValueError(
+                f"Project '{project_identifier}' matched multiple projects in workspace "
+                f"'{workspace.name}' ({workspace.slug}). Project permalinks must be unique. "
+                f"Matches: {details}"
+            )
         return matches[0]
 
     matches = index.entries_by_permalink.get(project_permalink, ())

src/basic_memory/utils.py

@@ -245,6 +245,14 @@ def build_canonical_permalink(
         return normalized_path
 
     normalized_project = generate_permalink(project_permalink)
+    normalized_workspace = generate_permalink(workspace_permalink) if workspace_permalink else None
+    if normalized_workspace:
+        workspace_project_prefix = f"{normalized_workspace}/{normalized_project}"
+        if normalized_path == workspace_project_prefix or normalized_path.startswith(
+            f"{workspace_project_prefix}/"
+        ):
+            return normalized_path
+
     if normalized_path == normalized_project or normalized_path.startswith(
         f"{normalized_project}/"
     ):
@@ -255,10 +263,6 @@ def build_canonical_permalink(
     if not workspace_permalink:
         return project_path
 
-    normalized_workspace = generate_permalink(workspace_permalink)
-    if project_path == normalized_workspace or project_path.startswith(f"{normalized_workspace}/"):
-        return project_path
-
     return f"{normalized_workspace}/{project_path}"
 
 

src/basic_memory/workspace_context.py

@@ -1,12 +1,15 @@
 """Request-local workspace context for canonical permalink generation."""
 
+import re
 from contextlib import contextmanager
 from contextvars import ContextVar
 from dataclasses import dataclass
 from typing import Iterator
 
 WORKSPACE_SLUG_HEADER = "X-Basic-Memory-Workspace-Slug"
 WORKSPACE_TYPE_HEADER = "X-Basic-Memory-Workspace-Type"
+_WORKSPACE_SLUG_PATTERN = re.compile(r"^[a-z0-9_-]+$")
+_WORKSPACE_TYPES = {"personal", "organization"}
 
 
 @dataclass(frozen=True)
@@ -32,6 +35,25 @@ def current_workspace_permalink_context() -> WorkspacePermalinkContext | None:
     return _workspace_permalink_context.get()
 
 
+def validate_workspace_permalink_context_values(
+    workspace_slug: str | None,
+    workspace_type: str | None,
+) -> None:
+    """Validate workspace permalink metadata before it can affect stored permalinks."""
+    if bool(workspace_slug) != bool(workspace_type):
+        raise ValueError("workspace_slug and workspace_type must be provided together")
+
+    if not workspace_slug or not workspace_type:
+        return
+
+    if _WORKSPACE_SLUG_PATTERN.fullmatch(workspace_slug) is None:
+        raise ValueError(f"{WORKSPACE_SLUG_HEADER} must match [a-z0-9_-]+")
+
+    if workspace_type not in _WORKSPACE_TYPES:
+        allowed = ", ".join(sorted(_WORKSPACE_TYPES))
+        raise ValueError(f"{WORKSPACE_TYPE_HEADER} must be one of: {allowed}")
+
+
 @contextmanager
 def workspace_permalink_context(
     workspace_slug: str | None,
@@ -42,8 +64,7 @@ def workspace_permalink_context(
     Cloud can populate this per request without storing workspace metadata in
     local project config. The slug/type pair is all permalink generation needs.
     """
-    if bool(workspace_slug) != bool(workspace_type):
-        raise ValueError("workspace_slug and workspace_type must be provided together")
+    validate_workspace_permalink_context_values(workspace_slug, workspace_type)
 
     if not workspace_slug or not workspace_type:
         yield

tests/api/v2/test_workspace_permalink_headers.py

@@ -0,0 +1,41 @@
+"""Tests for workspace permalink context headers."""
+
+import pytest
+from httpx import AsyncClient
+
+from basic_memory.workspace_context import WORKSPACE_SLUG_HEADER, WORKSPACE_TYPE_HEADER
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "headers, expected_detail",
+    [
+        (
+            {WORKSPACE_SLUG_HEADER: "team-paul"},
+            "workspace_slug and workspace_type must be provided together",
+        ),
+        (
+            {
+                WORKSPACE_SLUG_HEADER: "../team-paul",
+                WORKSPACE_TYPE_HEADER: "organization",
+            },
+            f"{WORKSPACE_SLUG_HEADER} must match [a-z0-9_-]+",
+        ),
+        (
+            {
+                WORKSPACE_SLUG_HEADER: "team-paul",
+                WORKSPACE_TYPE_HEADER: "enterprise",
+            },
+            f"{WORKSPACE_TYPE_HEADER} must be one of: organization, personal",
+        ),
+    ],
+)
+async def test_workspace_permalink_headers_fail_fast(
+    client: AsyncClient,
+    headers: dict[str, str],
+    expected_detail: str,
+):
+    response = await client.get("/v2/projects/", headers=headers)
+
+    assert response.status_code == 400
+    assert response.json()["detail"] == expected_detail

tests/mcp/test_project_context.py

@@ -663,6 +663,45 @@ async def fake_index(context=None):
     assert resolved == "team-paul/main"
 
 
+@pytest.mark.asyncio
+async def test_resolve_workspace_qualified_memory_url_fails_on_duplicate_project_permalink(
+    monkeypatch,
+):
+    import basic_memory.mcp.project_context as project_context
+    from basic_memory.mcp.project_context import (
+        WorkspaceProjectEntry,
+        _build_workspace_project_index,
+        resolve_workspace_qualified_memory_url,
+    )
+
+    team = _workspace(
+        tenant_id="team-tenant",
+        workspace_type="organization",
+        slug="team-paul",
+        name="Team Paul",
+        role="editor",
+    )
+    entries = (
+        WorkspaceProjectEntry(
+            workspace=team,
+            project=_project("main", id=1, external_id="team-main-id-1"),
+        ),
+        WorkspaceProjectEntry(
+            workspace=team,
+            project=_project("Main", id=2, external_id="team-main-id-2"),
+        ),
+    )
+    index = _build_workspace_project_index((team,), entries)
+
+    async def fake_index(context=None):
+        return index
+
+    monkeypatch.setattr(project_context, "_ensure_workspace_project_index", fake_index)
+
+    with pytest.raises(ValueError, match="matched multiple projects"):
+        await resolve_workspace_qualified_memory_url("memory://team-paul/main/notes/foo")
+
+
 @pytest.mark.asyncio
 async def test_get_project_client_routes_duplicate_project_through_workspace_slug(
     config_manager,

tests/utils/test_permalink_formatting.py

@@ -7,7 +7,7 @@
 from basic_memory.config import ProjectConfig
 from basic_memory.services import EntityService
 from basic_memory.sync.sync_service import SyncService
-from basic_memory.utils import generate_permalink
+from basic_memory.utils import build_canonical_permalink, generate_permalink
 
 
 async def create_test_file(path: Path, content: str = "test content") -> None:
@@ -124,3 +124,27 @@ def test_chinese_character_preservation(input_path, expected):
 def test_mixed_character_sets(input_path, expected):
     """Test handling of mixed character sets and edge cases."""
     assert generate_permalink(input_path) == expected
+
+
+def test_build_canonical_permalink_prefixes_same_workspace_and_project_slug():
+    """Workspace and project slugs may be equal but remain distinct permalink segments."""
+    assert (
+        build_canonical_permalink(
+            "acme",
+            "notes/foo.md",
+            workspace_permalink="acme",
+        )
+        == "acme/acme/notes/foo"
+    )
+
+
+def test_build_canonical_permalink_preserves_complete_workspace_prefix():
+    """Already workspace-qualified canonical paths should not gain duplicate prefixes."""
+    assert (
+        build_canonical_permalink(
+            "main",
+            "team-paul/main/notes/foo.md",
+            workspace_permalink="team-paul",
+        )
+        == "team-paul/main/notes/foo"
+    )

tests/utils/test_workspace_context.py

@@ -0,0 +1,44 @@
+"""Tests for request-local workspace permalink context."""
+
+import pytest
+
+from basic_memory.workspace_context import (
+    WORKSPACE_SLUG_HEADER,
+    WORKSPACE_TYPE_HEADER,
+    current_workspace_permalink_context,
+    workspace_permalink_context,
+    workspace_permalink_headers,
+)
+
+
+def test_workspace_permalink_context_requires_slug_and_type_together():
+    with pytest.raises(ValueError, match="provided together"):
+        with workspace_permalink_context("team-paul", None):
+            pass
+
+
+def test_workspace_permalink_context_rejects_unsafe_slug():
+    with pytest.raises(ValueError, match=WORKSPACE_SLUG_HEADER):
+        with workspace_permalink_context("../team-paul", "organization"):
+            pass
+
+
+def test_workspace_permalink_context_rejects_unknown_type():
+    with pytest.raises(ValueError, match=WORKSPACE_TYPE_HEADER):
+        with workspace_permalink_context("team-paul", "enterprise"):
+            pass
+
+
+def test_workspace_permalink_headers_reflect_active_context():
+    assert workspace_permalink_headers() == {}
+
+    with workspace_permalink_context("team-paul", "organization"):
+        context = current_workspace_permalink_context()
+        assert context is not None
+        assert context.workspace_slug == "team-paul"
+        assert workspace_permalink_headers() == {
+            WORKSPACE_SLUG_HEADER: "team-paul",
+            WORKSPACE_TYPE_HEADER: "organization",
+        }
+
+    assert current_workspace_permalink_context() is None