feat: add QDRANT_ALLOW_COLLECTIONS allowlist for least-privilege access

README.md

@@ -53,6 +53,7 @@ Configuration is done via environment variables. The only command-line argument
 | `TOOL_FIND_DESCRIPTION`  | Custom description for the find tool                                | See default in [`settings.py`](src/mcp_server_qdrant/settings.py) |
 | `QDRANT_SEARCH_LIMIT`    | Maximum number of results to return from search                     | `10`                                                              |
 | `QDRANT_READ_ONLY`       | Enable read-only mode (disables `qdrant-store` tool)                | `false`                                                           |
+| `QDRANT_ALLOW_COLLECTIONS`| Comma-separated allowlist of collections the tools may access. Unset means all collections are allowed. | `None` |
 
 ### FastMCP Environment Variables
 

src/mcp_server_qdrant/mcp_server.py

@@ -72,6 +72,7 @@ def __init__(
             self.embedding_provider,
             qdrant_settings.local_path,
             make_indexes(qdrant_settings.filterable_fields_dict()),
+            allow_collections=qdrant_settings.allowed_collections(),
         )
 
         super().__init__(name=name, instructions=instructions, **settings)

src/mcp_server_qdrant/qdrant.py

@@ -42,6 +42,7 @@ def __init__(
         embedding_provider: EmbeddingProvider,
         qdrant_local_path: str | None = None,
         field_indexes: dict[str, models.PayloadSchemaType] | None = None,
+        allow_collections: set[str] | None = None,
     ):
         self._qdrant_url = qdrant_url.rstrip("/") if qdrant_url else None
         self._qdrant_api_key = qdrant_api_key
@@ -51,6 +52,23 @@ def __init__(
             location=qdrant_url, api_key=qdrant_api_key, path=qdrant_local_path
         )
         self._field_indexes = field_indexes
+        self._allow_collections = allow_collections
+
+    def _check_collection_allowed(self, collection_name: str | None) -> None:
+        """Reject access to any collection outside the configured allowlist.
+
+        When no allowlist is configured, all collections are permitted, so
+        existing deployments are unaffected.
+        """
+        if (
+            self._allow_collections is not None
+            and collection_name is not None
+            and collection_name not in self._allow_collections
+        ):
+            raise ValueError(
+                f"Access to collection '{collection_name}' is not allowed. "
+                f"Allowed collections: {sorted(self._allow_collections)}."
+            )
 
     async def get_collection_names(self) -> list[str]:
         """
@@ -69,6 +87,7 @@ async def store(self, entry: Entry, *, collection_name: str | None = None):
         """
         collection_name = collection_name or self._default_collection_name
         assert collection_name is not None
+        self._check_collection_allowed(collection_name)
         await self._ensure_collection_exists(collection_name)
 
         # Embed the document
@@ -109,6 +128,7 @@ async def search(
         :return: A list of entries found.
         """
         collection_name = collection_name or self._default_collection_name
+        self._check_collection_allowed(collection_name)
         collection_exists = await self._client.collection_exists(collection_name)
         if not collection_exists:
             return []

src/mcp_server_qdrant/settings.py

@@ -90,6 +90,9 @@ class QdrantSettings(BaseSettings):
     allow_arbitrary_filter: bool = Field(
         default=False, validation_alias="QDRANT_ALLOW_ARBITRARY_FILTER"
     )
+    allow_collections: str | None = Field(
+        default=None, validation_alias="QDRANT_ALLOW_COLLECTIONS"
+    )
 
     def filterable_fields_dict(self) -> dict[str, FilterableField]:
         if self.filterable_fields is None:
@@ -105,6 +108,15 @@ def filterable_fields_dict_with_conditions(self) -> dict[str, FilterableField]:
             if field.condition is not None
         }
 
+    def allowed_collections(self) -> set[str] | None:
+        """Parse QDRANT_ALLOW_COLLECTIONS (comma-separated) into an allowlist set.
+
+        Returns None when unset, which means no restriction (backwards compatible).
+        """
+        if not self.allow_collections:
+            return None
+        return {c.strip() for c in self.allow_collections.split(",") if c.strip()}
+
     @model_validator(mode="after")
     def check_local_path_conflict(self) -> "QdrantSettings":
         if self.local_path:

tests/test_collection_allowlist.py

@@ -0,0 +1,55 @@
+import pytest
+
+from mcp_server_qdrant.embeddings.fastembed import FastEmbedProvider
+from mcp_server_qdrant.qdrant import Entry, QdrantConnector
+from mcp_server_qdrant.settings import QdrantSettings
+
+
+def test_allowed_collections_parses_csv(monkeypatch):
+    monkeypatch.setenv("QDRANT_ALLOW_COLLECTIONS", "notes, docs ,memories")
+    assert QdrantSettings().allowed_collections() == {"notes", "docs", "memories"}
+
+
+def test_allowed_collections_none_when_unset(monkeypatch):
+    monkeypatch.delenv("QDRANT_ALLOW_COLLECTIONS", raising=False)
+    assert QdrantSettings().allowed_collections() is None
+
+
+@pytest.fixture
+async def provider():
+    return FastEmbedProvider(model_name="sentence-transformers/all-MiniLM-L6-v2")
+
+
+def _connector(provider, allow):
+    return QdrantConnector(
+        qdrant_url=":memory:",
+        qdrant_api_key=None,
+        collection_name=None,
+        embedding_provider=provider,
+        allow_collections=allow,
+    )
+
+
+@pytest.mark.asyncio
+async def test_disallowed_collection_is_rejected(provider):
+    connector = _connector(provider, {"allowed"})
+    with pytest.raises(ValueError, match="not allowed"):
+        await connector.store(Entry(content="x"), collection_name="secret")
+    with pytest.raises(ValueError, match="not allowed"):
+        await connector.search("x", collection_name="secret")
+
+
+@pytest.mark.asyncio
+async def test_allowed_collection_works(provider):
+    connector = _connector(provider, {"allowed"})
+    await connector.store(Entry(content="hello world"), collection_name="allowed")
+    results = await connector.search("hello", collection_name="allowed")
+    assert any("hello" in e.content for e in results)
+
+
+@pytest.mark.asyncio
+async def test_no_allowlist_permits_any_collection(provider):
+    connector = _connector(provider, None)
+    await connector.store(Entry(content="anything"), collection_name="whatever")
+    results = await connector.search("anything", collection_name="whatever")
+    assert len(results) >= 1