feat(webdav): add tag-based file exclusion (#710)
Hide sensitive files/folders from the WebDAV MCP tool surface by tagging them with a configured Nextcloud system tag. Defence-in-depth control for users who connect LLMs to accounts holding contracts, medical records, credentials, etc. A new EXCLUDED_TAGS env var (comma-separated tag names, empty by default) gates an exclusion layer that runs at the start of every WebDAV tool call: tag names are resolved to tag IDs, those IDs are expanded to the set of tagged paths, then listings/searches are filtered and read/write/delete/move/copy operations on excluded paths raise ToolError. Tagged folders exclude their descendants via prefix match. Empty EXCLUDED_TAGS disables the feature entirely. The threat model is preventing accidental data exfiltration via the LLM tool surface — not hiding files from a determined operator. The docs explicitly recommend creating exclusion tags with user_assignable=false so the credentials the MCP server uses cannot remove the tag. Implementation: - config.py: add `excluded_tags` to _DEFAULTS, Settings, and the _field_map alongside other comma-separated env vars. - client/webdav.py: get_files_by_tag now requests <d:resourcetype/> and surfaces is_directory so tagged directories can recursively exclude descendants. - server/tag_exclusion.py (new): get_excluded_tag_names, get_excluded_file_paths, is_path_excluded. - server/webdav.py: exclusion guards in all 11 WebDAV tools; read/write/create/delete/move/copy raise ToolError, list/search tools silently filter excluded entries. Existing f-string log calls converted to lazy %-style. - tests: 17 new unit tests covering path-matching edge cases (shared-prefix non-match, descendants of excluded dirs), tag-name parsing, and get_excluded_file_paths with mocked WebDAV; 1 new client test asserting <d:resourcetype/> -> is_directory parsing. - docs/configuration.md: new "Tag-Based File Exclusion" section with per-tool effect table, security guidance, and per-call cost note. - README.md: feature mention under Key Features. Closes #710. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
b23f7d9534
commit
22ed9e99a0
@@ -0,0 +1,98 @@
|
||||
"""Tag-based file exclusion for MCP file operations (issue #710).
|
||||
|
||||
Resolves the configured ``EXCLUDED_TAGS`` to a set of file paths that
|
||||
should be hidden from WebDAV MCP tools (list, read, search) and rejected
|
||||
by mutating tools (write, delete, move, copy).
|
||||
|
||||
The flow per call:
|
||||
|
||||
1. Parse ``EXCLUDED_TAGS`` (comma-separated tag names) from config.
|
||||
2. For each tag name, resolve to a tag ID via ``get_tag_by_name``.
|
||||
3. For each tag ID, fetch all tagged file/folder paths via
|
||||
``get_files_by_tag``.
|
||||
4. Collect normalised paths into a single ``set[str]``.
|
||||
|
||||
Tagging a *folder* excludes the folder itself and every descendant via
|
||||
prefix match in :func:`is_path_excluded`.
|
||||
|
||||
Threat model: this is a defence-in-depth control to prevent accidental
|
||||
exfiltration via the LLM tool surface. A user controlling the Nextcloud
|
||||
account whose credentials the server uses can untag files unless the tag
|
||||
is created with ``user_assignable=false``.
|
||||
"""
|
||||
|
||||
import logging
|
||||
|
||||
from nextcloud_mcp_server.client.webdav import WebDAVClient
|
||||
from nextcloud_mcp_server.config import get_settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def get_excluded_tag_names() -> list[str]:
|
||||
"""Return the configured excluded tag names (empty list if disabled)."""
|
||||
raw = get_settings().excluded_tags
|
||||
if not raw:
|
||||
return []
|
||||
return [t.strip() for t in raw.split(",") if t.strip()]
|
||||
|
||||
|
||||
async def get_excluded_file_paths(webdav: WebDAVClient) -> set[str]:
|
||||
"""Resolve excluded tags to the set of paths they cover.
|
||||
|
||||
Tagged directories are added as their own normalised path; descendants
|
||||
are blocked via prefix match in :func:`is_path_excluded`.
|
||||
"""
|
||||
tag_names = get_excluded_tag_names()
|
||||
if not tag_names:
|
||||
return set()
|
||||
|
||||
excluded: set[str] = set()
|
||||
for tag_name in tag_names:
|
||||
tag = await webdav.get_tag_by_name(tag_name)
|
||||
if tag is None:
|
||||
logger.debug("Excluded tag %r does not exist — skipping", tag_name)
|
||||
continue
|
||||
|
||||
files = await webdav.get_files_by_tag(tag["id"])
|
||||
for f in files:
|
||||
path = _normalise_path(f["path"])
|
||||
excluded.add(path)
|
||||
if f.get("is_directory"):
|
||||
logger.debug(
|
||||
"Excluding directory %r (tag %r) — descendants will be hidden",
|
||||
path,
|
||||
tag_name,
|
||||
)
|
||||
|
||||
if excluded:
|
||||
logger.info(
|
||||
"Tag-based exclusion: hiding %d path(s) matching tags: %s",
|
||||
len(excluded),
|
||||
", ".join(tag_names),
|
||||
)
|
||||
|
||||
return excluded
|
||||
|
||||
|
||||
def is_path_excluded(path: str, excluded_paths: set[str]) -> bool:
|
||||
"""Return True if *path* (or any of its parents) is excluded.
|
||||
|
||||
A path is excluded when it matches an entry exactly, or when an
|
||||
excluded entry is one of its directory ancestors (prefix match on
|
||||
``<dir>/``).
|
||||
"""
|
||||
if not excluded_paths:
|
||||
return False
|
||||
normalised = _normalise_path(path)
|
||||
if normalised in excluded_paths:
|
||||
return True
|
||||
for exc in excluded_paths:
|
||||
if normalised.startswith(exc + "/"):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _normalise_path(path: str) -> str:
|
||||
"""Strip leading/trailing slashes for consistent comparison."""
|
||||
return path.strip("/")
|
||||
@@ -2,12 +2,17 @@ import base64
|
||||
import logging
|
||||
|
||||
from mcp.server.fastmcp import Context, FastMCP
|
||||
from mcp.server.fastmcp.exceptions import ToolError
|
||||
from mcp.types import ToolAnnotations
|
||||
|
||||
from nextcloud_mcp_server.auth import require_scopes
|
||||
from nextcloud_mcp_server.context import get_client
|
||||
from nextcloud_mcp_server.models import DirectoryListing, FileInfo, SearchFilesResponse
|
||||
from nextcloud_mcp_server.observability.metrics import instrument_tool
|
||||
from nextcloud_mcp_server.server.tag_exclusion import (
|
||||
get_excluded_file_paths,
|
||||
is_path_excluded,
|
||||
)
|
||||
from nextcloud_mcp_server.utils.document_parser import (
|
||||
is_parseable_document,
|
||||
parse_document,
|
||||
@@ -32,6 +37,10 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
) -> DirectoryListing:
|
||||
"""List files and directories in the specified NextCloud path.
|
||||
|
||||
When ``EXCLUDED_TAGS`` is configured, entries tagged (or whose
|
||||
ancestor folders are tagged) with an excluded system tag are
|
||||
omitted from the result.
|
||||
|
||||
Args:
|
||||
path: Directory path to list (empty string for root directory)
|
||||
|
||||
@@ -41,6 +50,13 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
client = await get_client(ctx)
|
||||
items = await client.webdav.list_directory(path)
|
||||
|
||||
# Filter out files/folders carrying an excluded tag.
|
||||
excluded = await get_excluded_file_paths(client.webdav)
|
||||
if excluded:
|
||||
items = [
|
||||
i for i in items if not is_path_excluded(i.get("path", ""), excluded)
|
||||
]
|
||||
|
||||
# Convert to FileInfo models
|
||||
file_infos = [FileInfo(**item) for item in items]
|
||||
|
||||
@@ -70,6 +86,9 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
async def nc_webdav_read_file(path: str, ctx: Context):
|
||||
"""Read the content of a file from NextCloud.
|
||||
|
||||
Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and the
|
||||
file (or an ancestor folder) carries an excluded system tag.
|
||||
|
||||
Args:
|
||||
path: Full path to the file to read
|
||||
|
||||
@@ -80,13 +99,19 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
- Other binary files are base64 encoded
|
||||
"""
|
||||
client = await get_client(ctx)
|
||||
|
||||
# Block reads of paths carrying an excluded tag.
|
||||
excluded = await get_excluded_file_paths(client.webdav)
|
||||
if is_path_excluded(path, excluded):
|
||||
raise ToolError(f"Access denied: {path!r} is tagged with an excluded tag")
|
||||
|
||||
content, content_type = await client.webdav.read_file(path)
|
||||
|
||||
# Check if this is a parseable document (PDF, DOCX, etc.)
|
||||
# is_parseable_document() checks if document processing is enabled
|
||||
if is_parseable_document(content_type):
|
||||
try:
|
||||
logger.info(f"Parsing document '{path}' of type '{content_type}'")
|
||||
logger.info("Parsing document %r of type %r", path, content_type)
|
||||
parsed_text, metadata = await parse_document(
|
||||
content,
|
||||
content_type,
|
||||
@@ -103,7 +128,9 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
}
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Failed to parse document '{path}', falling back to base64: {e}"
|
||||
"Failed to parse document %r, falling back to base64: %s",
|
||||
path,
|
||||
e,
|
||||
)
|
||||
# Fall through to base64 encoding on parse failure
|
||||
|
||||
@@ -144,6 +171,9 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
):
|
||||
"""Write content to a file in NextCloud.
|
||||
|
||||
Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and the
|
||||
target path (or an ancestor folder) carries an excluded system tag.
|
||||
|
||||
Args:
|
||||
path: Full path where to write the file
|
||||
content: File content (text or base64 for binary)
|
||||
@@ -154,6 +184,11 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
"""
|
||||
client = await get_client(ctx)
|
||||
|
||||
# Block writes to excluded paths.
|
||||
excluded = await get_excluded_file_paths(client.webdav)
|
||||
if is_path_excluded(path, excluded):
|
||||
raise ToolError(f"Access denied: {path!r} is tagged with an excluded tag")
|
||||
|
||||
# Handle base64 encoded content
|
||||
if content_type and "base64" in content_type.lower():
|
||||
content_bytes = base64.b64decode(content)
|
||||
@@ -175,6 +210,9 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
async def nc_webdav_create_directory(path: str, ctx: Context):
|
||||
"""Create a directory in NextCloud.
|
||||
|
||||
Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and the
|
||||
target path lies inside a folder carrying an excluded system tag.
|
||||
|
||||
Args:
|
||||
path: Full path of the directory to create
|
||||
|
||||
@@ -182,6 +220,14 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
Dict with status_code (201 for created, 405 if already exists)
|
||||
"""
|
||||
client = await get_client(ctx)
|
||||
|
||||
# Block directory creation inside excluded paths.
|
||||
excluded = await get_excluded_file_paths(client.webdav)
|
||||
if is_path_excluded(path, excluded):
|
||||
raise ToolError(
|
||||
f"Access denied: {path!r} is inside a path tagged with an excluded tag"
|
||||
)
|
||||
|
||||
return await client.webdav.create_directory(path)
|
||||
|
||||
@mcp.tool(
|
||||
@@ -197,6 +243,9 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
async def nc_webdav_delete_resource(path: str, ctx: Context):
|
||||
"""Delete a file or directory in NextCloud.
|
||||
|
||||
Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and the
|
||||
target path (or an ancestor folder) carries an excluded system tag.
|
||||
|
||||
Args:
|
||||
path: Full path of the file or directory to delete
|
||||
|
||||
@@ -204,6 +253,12 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
Dict with status_code indicating result (404 if not found)
|
||||
"""
|
||||
client = await get_client(ctx)
|
||||
|
||||
# Block deletion of excluded files/directories.
|
||||
excluded = await get_excluded_file_paths(client.webdav)
|
||||
if is_path_excluded(path, excluded):
|
||||
raise ToolError(f"Access denied: {path!r} is tagged with an excluded tag")
|
||||
|
||||
return await client.webdav.delete_resource(path)
|
||||
|
||||
@mcp.tool(
|
||||
@@ -220,6 +275,10 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
):
|
||||
"""Move or rename a file or directory in NextCloud.
|
||||
|
||||
Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and either
|
||||
the source or destination path (or one of their ancestor folders)
|
||||
carries an excluded system tag.
|
||||
|
||||
Args:
|
||||
source_path: Full path of the file or directory to move
|
||||
destination_path: New path for the file or directory
|
||||
@@ -229,6 +288,19 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
Dict with status_code indicating result (404 if source not found, 412 if destination exists and overwrite is False)
|
||||
"""
|
||||
client = await get_client(ctx)
|
||||
|
||||
# Block moves involving excluded paths on either side.
|
||||
excluded = await get_excluded_file_paths(client.webdav)
|
||||
if is_path_excluded(source_path, excluded):
|
||||
raise ToolError(
|
||||
f"Access denied: source {source_path!r} is tagged with an excluded tag"
|
||||
)
|
||||
if is_path_excluded(destination_path, excluded):
|
||||
raise ToolError(
|
||||
f"Access denied: destination {destination_path!r} is inside a "
|
||||
"path tagged with an excluded tag"
|
||||
)
|
||||
|
||||
return await client.webdav.move_resource(
|
||||
source_path, destination_path, overwrite
|
||||
)
|
||||
@@ -247,6 +319,10 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
):
|
||||
"""Copy a file or directory in NextCloud.
|
||||
|
||||
Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and either
|
||||
the source or destination path (or one of their ancestor folders)
|
||||
carries an excluded system tag.
|
||||
|
||||
Args:
|
||||
source_path: Full path of the file or directory to copy
|
||||
destination_path: Destination path for the copy
|
||||
@@ -256,6 +332,19 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
Dict with status_code indicating result (404 if source not found, 412 if destination exists and overwrite is False)
|
||||
"""
|
||||
client = await get_client(ctx)
|
||||
|
||||
# Block copies involving excluded paths on either side.
|
||||
excluded = await get_excluded_file_paths(client.webdav)
|
||||
if is_path_excluded(source_path, excluded):
|
||||
raise ToolError(
|
||||
f"Access denied: source {source_path!r} is tagged with an excluded tag"
|
||||
)
|
||||
if is_path_excluded(destination_path, excluded):
|
||||
raise ToolError(
|
||||
f"Access denied: destination {destination_path!r} is inside a "
|
||||
"path tagged with an excluded tag"
|
||||
)
|
||||
|
||||
return await client.webdav.copy_resource(
|
||||
source_path, destination_path, overwrite
|
||||
)
|
||||
@@ -364,6 +453,13 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
limit=limit,
|
||||
)
|
||||
|
||||
# Filter out tagged-excluded paths.
|
||||
excluded = await get_excluded_file_paths(client.webdav)
|
||||
if excluded:
|
||||
results = [
|
||||
r for r in results if not is_path_excluded(r.get("path", ""), excluded)
|
||||
]
|
||||
|
||||
# Convert to FileInfo models
|
||||
file_infos = [FileInfo(**result) for result in results]
|
||||
|
||||
@@ -409,6 +505,11 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
results = await client.webdav.find_by_name(
|
||||
pattern=pattern, scope=scope, limit=limit
|
||||
)
|
||||
excluded = await get_excluded_file_paths(client.webdav)
|
||||
if excluded:
|
||||
results = [
|
||||
r for r in results if not is_path_excluded(r.get("path", ""), excluded)
|
||||
]
|
||||
file_infos = [FileInfo(**result) for result in results]
|
||||
return SearchFilesResponse(
|
||||
results=file_infos,
|
||||
@@ -443,6 +544,11 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
results = await client.webdav.find_by_type(
|
||||
mime_type=mime_type, scope=scope, limit=limit
|
||||
)
|
||||
excluded = await get_excluded_file_paths(client.webdav)
|
||||
if excluded:
|
||||
results = [
|
||||
r for r in results if not is_path_excluded(r.get("path", ""), excluded)
|
||||
]
|
||||
file_infos = [FileInfo(**result) for result in results]
|
||||
return SearchFilesResponse(
|
||||
results=file_infos,
|
||||
@@ -474,6 +580,11 @@ def configure_webdav_tools(mcp: FastMCP):
|
||||
"""
|
||||
client = await get_client(ctx)
|
||||
results = await client.webdav.list_favorites(scope=scope, limit=limit)
|
||||
excluded = await get_excluded_file_paths(client.webdav)
|
||||
if excluded:
|
||||
results = [
|
||||
r for r in results if not is_path_excluded(r.get("path", ""), excluded)
|
||||
]
|
||||
file_infos = [FileInfo(**result) for result in results]
|
||||
return SearchFilesResponse(
|
||||
results=file_infos,
|
||||
|
||||
Reference in New Issue
Block a user