Chris CoutinhoandClaude Opus 4.7
a6c188abbb
Six findings raised in the PR review: 🔴 Blocking - Fail-open on tag-resolution errors. get_excluded_file_paths now wraps each tag's get_tag_by_name and get_files_by_tag call in try/except; failures log a warning and the tag is skipped, rather than propagating to the caller and disabling all WebDAV tools when the systemtags endpoint is degraded. Documented in the docstring as the intended fail-open behaviour (threat model is preventing accidental exfiltration, not surviving server compromise). 🟡 Important - nc_webdav_list_directory now raises ToolError when the listed path itself is tagged, instead of silently returning an empty listing after a wasted PROPFIND. Behaviour now mirrors the mutating tools. - Destination error messages in move/copy/create_directory said "is inside" but is_path_excluded matches exact paths too. Reworded to "is or is inside". 🟢 Nits - get_excluded_file_paths log message clarified: N counts directly-tagged paths, not total descendants. - Test isolation: tests/unit/conftest.py already has an autouse _reload_dynaconf_after_test fixture that handles teardown. Removed the redundant module-local fixture I had drafted; documented the reliance in the module docstring instead. - Added tests/unit/test_webdav_tools_exclusion.py: 12 server-layer tests that register the WebDAV tools on a fresh FastMCP and invoke each tool's underlying function with a mocked excluded set, asserting ToolError is raised / results filtered as expected. Catches future guard-integration regressions (e.g. wrong argument order). Also added two unit tests for the new fail-open behaviour. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
604 lines
21 KiB
Python
604 lines
21 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)
|
|
|
|
# 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.
|
|
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]
|
|
|
|
# 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)
|
|
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,
|
|
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)
|
|
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,
|
|
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)
|
|
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,
|
|
total_found=len(file_infos),
|
|
scope=scope,
|
|
filters_applied={"only_favorites": True},
|
|
)
|