Chris CoutinhoandClaude Opus 4.7
d179ca8c8b
Addresses the four points raised in the automated review on PR #764: 1. Scope guards on the four search tools (search_files, find_by_name, find_by_type, list_favorites) so an excluded `scope` raises ToolError instead of silently returning an empty result. Previously an LLM could probe the asymmetry between list_directory (raises) and the search tools (silent) to infer that an excluded directory exists. The 4 search tools now mirror the early-guard pattern from list_directory and avoid an unnecessary upstream query for known- excluded scopes. 2. Concurrent per-tag resolution in get_excluded_file_paths via anyio.create_task_group(). Previously the 2N network calls (1 PROPFIND + 1 REPORT per tag) ran serially. Per-tag fail-open behaviour is preserved by extracting _resolve_one_tag, which swallows its own exceptions so a single tag failure does not abort the surrounding task group. 3. WebDAVClient.get_tag_by_name and get_files_by_tag now route through _make_request, inheriting the @retry_on_429 decorator. Previously they bypassed it; with tag exclusion invoked on every WebDAV tool call, a transient 429 from the systemtags endpoint was hitting the fail-open path instead of being transparently retried. 4. Test coverage: 6 new tests in test_webdav_tools_exclusion.py (4 scope-guard, 2 missing filter tests for find_by_type and list_favorites) and 2 new tests in test_tag_exclusion.py (a concurrency proof using an event-barrier that would deadlock under sequential execution, and a fail-open-under-task-group test with order-independent side_effect callables). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
622 lines
22 KiB
Python
622 lines
22 KiB
Python
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,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def configure_webdav_tools(mcp: FastMCP):
|
|
# WebDAV file system tools
|
|
@mcp.tool(
|
|
title="List Files and Directories",
|
|
annotations=ToolAnnotations(
|
|
readOnlyHint=True,
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("files.read")
|
|
@instrument_tool
|
|
async def nc_webdav_list_directory(
|
|
ctx: Context, path: str = ""
|
|
) -> DirectoryListing:
|
|
"""List files and directories in the specified NextCloud path.
|
|
|
|
When ``EXCLUDED_TAGS`` is configured: raises ``ToolError`` if the
|
|
listed path itself is tagged (or sits inside a tagged folder),
|
|
and otherwise omits any tagged children from the listing. The
|
|
early guard is consistent with the mutating tools and avoids a
|
|
round-trip to Nextcloud for a known-excluded path.
|
|
|
|
Args:
|
|
path: Directory path to list (empty string for root directory)
|
|
|
|
Returns:
|
|
DirectoryListing with files, total_count, directories_count, files_count, and total_size
|
|
"""
|
|
client = await get_client(ctx)
|
|
|
|
# Resolve once and use for both the path-itself guard and the
|
|
# children filter below.
|
|
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")
|
|
|
|
items = await client.webdav.list_directory(path)
|
|
|
|
# Filter out child files/folders carrying an excluded tag.
|
|
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]
|
|
|
|
# Calculate metadata
|
|
directories_count = sum(1 for f in file_infos if f.is_directory)
|
|
files_count = sum(1 for f in file_infos if not f.is_directory)
|
|
total_size = sum(f.size or 0 for f in file_infos if not f.is_directory)
|
|
|
|
return DirectoryListing(
|
|
path=path,
|
|
files=file_infos,
|
|
total_count=len(file_infos),
|
|
directories_count=directories_count,
|
|
files_count=files_count,
|
|
total_size=total_size,
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="Read File",
|
|
annotations=ToolAnnotations(
|
|
readOnlyHint=True,
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("files.read")
|
|
@instrument_tool
|
|
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
|
|
|
|
Returns:
|
|
Dict with path, content, content_type, size, and optional parsing metadata
|
|
- Text files are decoded to UTF-8
|
|
- Documents (PDF, DOCX, etc.) are parsed and text is extracted
|
|
- 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("Parsing document %r of type %r", path, content_type)
|
|
parsed_text, metadata = await parse_document(
|
|
content,
|
|
content_type,
|
|
filename=path,
|
|
progress_callback=ctx.report_progress,
|
|
)
|
|
return {
|
|
"path": path,
|
|
"content": parsed_text,
|
|
"content_type": content_type,
|
|
"size": len(content),
|
|
"parsed": True,
|
|
"parsing_metadata": metadata,
|
|
}
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Failed to parse document %r, falling back to base64: %s",
|
|
path,
|
|
e,
|
|
)
|
|
# Fall through to base64 encoding on parse failure
|
|
|
|
# For text files, decode content for easier viewing
|
|
if content_type and content_type.startswith("text/"):
|
|
try:
|
|
decoded_content = content.decode("utf-8")
|
|
return {
|
|
"path": path,
|
|
"content": decoded_content,
|
|
"content_type": content_type,
|
|
"size": len(content),
|
|
}
|
|
except UnicodeDecodeError:
|
|
pass
|
|
|
|
# For binary files, return metadata and base64 encoded content
|
|
|
|
return {
|
|
"path": path,
|
|
"content": base64.b64encode(content).decode("ascii"),
|
|
"content_type": content_type,
|
|
"size": len(content),
|
|
"encoding": "base64",
|
|
}
|
|
|
|
@mcp.tool(
|
|
title="Write File",
|
|
annotations=ToolAnnotations(
|
|
idempotentHint=True, # HTTP PUT without version control is idempotent
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("files.write")
|
|
@instrument_tool
|
|
async def nc_webdav_write_file(
|
|
path: str, content: str, ctx: Context, content_type: str | None = None
|
|
):
|
|
"""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)
|
|
content_type: MIME type (auto-detected if not provided, use 'type;base64' for binary)
|
|
|
|
Returns:
|
|
Dict with status_code indicating success
|
|
"""
|
|
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)
|
|
content_type = content_type.replace(";base64", "")
|
|
else:
|
|
content_bytes = content.encode("utf-8")
|
|
|
|
return await client.webdav.write_file(path, content_bytes, content_type)
|
|
|
|
@mcp.tool(
|
|
title="Create Directory",
|
|
annotations=ToolAnnotations(
|
|
idempotentHint=True, # Creating existing dir returns 405 = same end state
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("files.write")
|
|
@instrument_tool
|
|
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
|
|
|
|
Returns:
|
|
Dict with status_code (201 for created, 405 if already exists)
|
|
"""
|
|
client = await get_client(ctx)
|
|
|
|
# Block directory creation at or 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 or is inside a path tagged "
|
|
"with an excluded tag"
|
|
)
|
|
|
|
return await client.webdav.create_directory(path)
|
|
|
|
@mcp.tool(
|
|
title="Delete File or Directory",
|
|
annotations=ToolAnnotations(
|
|
destructiveHint=True, # Permanently deletes data
|
|
idempotentHint=True, # Deleting deleted resource = same end state
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("files.write")
|
|
@instrument_tool
|
|
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
|
|
|
|
Returns:
|
|
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(
|
|
title="Move or Rename File",
|
|
annotations=ToolAnnotations(
|
|
idempotentHint=False, # Moving changes source and dest
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("files.write")
|
|
@instrument_tool
|
|
async def nc_webdav_move_resource(
|
|
source_path: str, destination_path: str, ctx: Context, overwrite: bool = False
|
|
):
|
|
"""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
|
|
overwrite: Whether to overwrite the destination if it exists (default: False)
|
|
|
|
Returns:
|
|
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 or is "
|
|
"inside a path tagged with an excluded tag"
|
|
)
|
|
|
|
return await client.webdav.move_resource(
|
|
source_path, destination_path, overwrite
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="Copy File or Directory",
|
|
annotations=ToolAnnotations(
|
|
idempotentHint=False, # Creates new resource each time
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("files.write")
|
|
@instrument_tool
|
|
async def nc_webdav_copy_resource(
|
|
source_path: str, destination_path: str, ctx: Context, overwrite: bool = False
|
|
):
|
|
"""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
|
|
overwrite: Whether to overwrite the destination if it exists (default: False)
|
|
|
|
Returns:
|
|
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 or is "
|
|
"inside a path tagged with an excluded tag"
|
|
)
|
|
|
|
return await client.webdav.copy_resource(
|
|
source_path, destination_path, overwrite
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="Search Files",
|
|
annotations=ToolAnnotations(
|
|
readOnlyHint=True,
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("files.read")
|
|
@instrument_tool
|
|
async def nc_webdav_search_files(
|
|
ctx: Context,
|
|
scope: str = "",
|
|
name_pattern: str | None = None,
|
|
mime_type: str | None = None,
|
|
only_favorites: bool = False,
|
|
limit: int | None = None,
|
|
) -> SearchFilesResponse:
|
|
"""Search for files in NextCloud using WebDAV SEARCH.
|
|
|
|
This is a high-level search tool that supports common search patterns.
|
|
For more complex queries, use the specific search tools.
|
|
|
|
Args:
|
|
scope: Directory path to search in (empty string for user root)
|
|
name_pattern: File name pattern (supports % wildcard, e.g., "%.txt" for all text files)
|
|
mime_type: MIME type to filter by (supports % wildcard, e.g., "image/%" for all images)
|
|
only_favorites: If True, only return favorited files
|
|
limit: Maximum number of results to return
|
|
|
|
Returns:
|
|
SearchFilesResponse with list of matching files
|
|
"""
|
|
client = await get_client(ctx)
|
|
|
|
# Resolve once and use for both the scope guard and the result filter.
|
|
excluded = await get_excluded_file_paths(client.webdav)
|
|
if scope and is_path_excluded(scope, excluded):
|
|
raise ToolError(
|
|
f"Access denied: scope {scope!r} is tagged with an excluded tag"
|
|
)
|
|
|
|
# Build where conditions based on filters
|
|
conditions = []
|
|
|
|
if name_pattern:
|
|
conditions.append(
|
|
f"""
|
|
<d:like>
|
|
<d:prop>
|
|
<d:displayname/>
|
|
</d:prop>
|
|
<d:literal>{name_pattern}</d:literal>
|
|
</d:like>
|
|
"""
|
|
)
|
|
|
|
if mime_type:
|
|
conditions.append(
|
|
f"""
|
|
<d:like>
|
|
<d:prop>
|
|
<d:getcontenttype/>
|
|
</d:prop>
|
|
<d:literal>{mime_type}</d:literal>
|
|
</d:like>
|
|
"""
|
|
)
|
|
|
|
if only_favorites:
|
|
conditions.append(
|
|
"""
|
|
<d:eq>
|
|
<d:prop>
|
|
<oc:favorite/>
|
|
</d:prop>
|
|
<d:literal>1</d:literal>
|
|
</d:eq>
|
|
"""
|
|
)
|
|
|
|
# Combine conditions with AND if multiple
|
|
if len(conditions) > 1:
|
|
where_conditions = f"""
|
|
<d:and>
|
|
{"".join(conditions)}
|
|
</d:and>
|
|
"""
|
|
elif len(conditions) == 1:
|
|
where_conditions = conditions[0]
|
|
else:
|
|
where_conditions = None
|
|
|
|
# Include extended properties
|
|
properties = [
|
|
"displayname",
|
|
"getcontentlength",
|
|
"getcontenttype",
|
|
"getlastmodified",
|
|
"resourcetype",
|
|
"getetag",
|
|
"fileid",
|
|
"favorite",
|
|
]
|
|
|
|
results = await client.webdav.search_files(
|
|
scope=scope,
|
|
where_conditions=where_conditions,
|
|
properties=properties,
|
|
limit=limit,
|
|
)
|
|
|
|
# Filter out tagged-excluded paths from the result set.
|
|
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]
|
|
|
|
# Build filters applied dict
|
|
filters = {}
|
|
if name_pattern:
|
|
filters["name_pattern"] = name_pattern
|
|
if mime_type:
|
|
filters["mime_type"] = mime_type
|
|
if only_favorites:
|
|
filters["only_favorites"] = True
|
|
|
|
return SearchFilesResponse(
|
|
results=file_infos,
|
|
total_found=len(file_infos),
|
|
scope=scope,
|
|
filters_applied=filters if filters else None,
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="Find Files by Name",
|
|
annotations=ToolAnnotations(
|
|
readOnlyHint=True,
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("files.read")
|
|
@instrument_tool
|
|
async def nc_webdav_find_by_name(
|
|
pattern: str, ctx: Context, scope: str = "", limit: int | None = None
|
|
) -> SearchFilesResponse:
|
|
"""Find files by name pattern in NextCloud.
|
|
|
|
Args:
|
|
pattern: Name pattern to search for (supports % wildcard)
|
|
scope: Directory path to search in (empty string for user root)
|
|
limit: Maximum number of results to return
|
|
|
|
Returns:
|
|
SearchFilesResponse with list of matching files
|
|
"""
|
|
client = await get_client(ctx)
|
|
excluded = await get_excluded_file_paths(client.webdav)
|
|
if scope and is_path_excluded(scope, excluded):
|
|
raise ToolError(
|
|
f"Access denied: scope {scope!r} is tagged with an excluded tag"
|
|
)
|
|
results = await client.webdav.find_by_name(
|
|
pattern=pattern, scope=scope, limit=limit
|
|
)
|
|
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,
|
|
total_found=len(file_infos),
|
|
scope=scope,
|
|
filters_applied={"name_pattern": pattern},
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="Find Files by Type",
|
|
annotations=ToolAnnotations(
|
|
readOnlyHint=True,
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("files.read")
|
|
@instrument_tool
|
|
async def nc_webdav_find_by_type(
|
|
mime_type: str, ctx: Context, scope: str = "", limit: int | None = None
|
|
) -> SearchFilesResponse:
|
|
"""Find files by MIME type in NextCloud.
|
|
|
|
Args:
|
|
mime_type: MIME type to search for (supports % wildcard)
|
|
scope: Directory path to search in (empty string for user root)
|
|
limit: Maximum number of results to return
|
|
|
|
Returns:
|
|
SearchFilesResponse with list of matching files
|
|
"""
|
|
client = await get_client(ctx)
|
|
excluded = await get_excluded_file_paths(client.webdav)
|
|
if scope and is_path_excluded(scope, excluded):
|
|
raise ToolError(
|
|
f"Access denied: scope {scope!r} is tagged with an excluded tag"
|
|
)
|
|
results = await client.webdav.find_by_type(
|
|
mime_type=mime_type, scope=scope, limit=limit
|
|
)
|
|
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,
|
|
total_found=len(file_infos),
|
|
scope=scope,
|
|
filters_applied={"mime_type": mime_type},
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="List Favorite Files",
|
|
annotations=ToolAnnotations(
|
|
readOnlyHint=True,
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("files.read")
|
|
@instrument_tool
|
|
async def nc_webdav_list_favorites(
|
|
ctx: Context, scope: str = "", limit: int | None = None
|
|
) -> SearchFilesResponse:
|
|
"""List all favorite files in NextCloud.
|
|
|
|
Args:
|
|
scope: Directory path to search in (empty string for all favorites)
|
|
limit: Maximum number of results to return
|
|
|
|
Returns:
|
|
SearchFilesResponse with list of favorite files
|
|
"""
|
|
client = await get_client(ctx)
|
|
excluded = await get_excluded_file_paths(client.webdav)
|
|
if scope and is_path_excluded(scope, excluded):
|
|
raise ToolError(
|
|
f"Access denied: scope {scope!r} is tagged with an excluded tag"
|
|
)
|
|
results = await client.webdav.list_favorites(scope=scope, limit=limit)
|
|
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,
|
|
total_found=len(file_infos),
|
|
scope=scope,
|
|
filters_applied={"only_favorites": True},
|
|
)
|