diff --git a/README.md b/README.md index c0625999..8a99a60b 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,7 @@ For Kubernetes, see [cbcoutinho/helm-charts](https://github.com/cbcoutinho/helm- - **Document Processing** - OCR and text extraction from PDFs, DOCX, images with progress notifications - **Flexible Deployment** - Docker, Kubernetes ([Helm chart](https://github.com/cbcoutinho/helm-charts)), VM, or local installation - **Production-Ready Auth** - Basic Auth with app passwords; multi-user via Login Flow v2 — MCP clients authenticate via OAuth, the server handles Nextcloud app passwords transparently +- **Tag-Based File Exclusion** - Hide sensitive files/folders from MCP file tools by tagging them with a configured Nextcloud system tag (`EXCLUDED_TAGS`). See [docs/configuration.md](docs/configuration.md#tag-based-file-exclusion-optional) - **Multiple Transports** - streamable-http (default) and stdio ## Supported Apps diff --git a/docs/configuration.md b/docs/configuration.md index d31f40f8..f6f5d8e0 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -576,6 +576,87 @@ docker-compose up --- +## Tag-Based File Exclusion (Optional) + +Some files (contracts, medical records, credentials, private notes) should +never be exposed to an LLM, even when the assistant has valid credentials +for the account. The MCP server can hide such files from all WebDAV tools +based on **Nextcloud system tags** (the same collaborative tags users +manage from the Nextcloud UI). + +### Setup + +Set `EXCLUDED_TAGS` to a comma-separated list of system tag names: + +```bash +EXCLUDED_TAGS=confidential,no-ai,private +``` + +Then create the tags in Nextcloud (one-time, as admin): + +```bash +docker compose exec app php occ tag:add 'no-ai' --user-visible=true --user-assignable=false +``` + +`--user-assignable=false` is **strongly recommended** for the threat model +this feature is designed to address — see *Security considerations* below. +Tag any file or folder with one of these tags from the Nextcloud UI to +hide it from the MCP tools. + +Empty (`EXCLUDED_TAGS=""`, the default) disables the feature entirely. + +### Behaviour + +When `EXCLUDED_TAGS` is set, every WebDAV MCP tool resolves the configured +tag names to file paths and applies the following: + +| Tool | Effect on tagged paths | +|------|------------------------| +| `nc_webdav_list_directory` | Excluded files/folders are omitted from listings | +| `nc_webdav_read_file` | Raises `ToolError` (access denied) | +| `nc_webdav_write_file` | Raises `ToolError` (access denied) | +| `nc_webdav_create_directory` | Blocked inside excluded paths | +| `nc_webdav_delete_resource` | Raises `ToolError` (access denied) | +| `nc_webdav_move_resource` | Blocked when source **or** destination is excluded | +| `nc_webdav_copy_resource` | Blocked when source **or** destination is excluded | +| `nc_webdav_search_files` | Excluded files are filtered from results | +| `nc_webdav_find_by_name` | Excluded files are filtered from results | +| `nc_webdav_find_by_type` | Excluded files are filtered from results | +| `nc_webdav_list_favorites` | Excluded files are filtered from results | + +Tagging a **folder** hides the folder itself **and** every descendant +recursively, via path-prefix match. + +### Security considerations + +The threat model is **preventing accidental data exfiltration via the LLM +tool surface**, not hiding files from a determined operator. Specifically: + +- Create exclusion tags with `user_assignable=false` so the credentials + the MCP server uses cannot remove the tag from a file (and thereby + bypass the exclusion). With `user_assignable=true`, any user — including + the one whose credentials the MCP server uses — can untag a file. +- Optionally set `user_visible=false` if the exclusion tag itself is + sensitive metadata. +- The exclusion is enforced at the MCP tool layer only. Direct WebDAV / + Nextcloud client access still sees the files; this feature does not + alter Nextcloud's underlying access control. + +### Performance note + +The excluded path set is resolved per WebDAV tool call (1 PROPFIND for +each tag name + 1 REPORT per tag). For typical setups (a handful of +tagged files under one or two tag names) the overhead is negligible. +Caching may be added in a future release. + +### Scope + +This feature only covers WebDAV file operations. Notes, Calendar, +Contacts, Deck, etc. are not filtered, because they use ID-based APIs +rather than file paths. + +--- + ## Loading Environment Variables After creating your `.env` file, load the environment variables: diff --git a/nextcloud_mcp_server/client/webdav.py b/nextcloud_mcp_server/client/webdav.py index 6c267927..25d004e2 100644 --- a/nextcloud_mcp_server/client/webdav.py +++ b/nextcloud_mcp_server/client/webdav.py @@ -1132,12 +1132,20 @@ class WebDAVClient(BaseNextcloudClient): """ - response = await self._client.request( + response = await self._make_request( "PROPFIND", "/remote.php/dav/systemtags/", - headers={"Depth": "1"}, + headers={ + "Depth": "1", + "Content-Type": "text/xml", + "OCS-APIRequest": "true", + }, content=propfind_body, ) + # Redundant after _make_request (which raises on non-2xx) but + # makes the contract explicit at the call site so a future + # refactor of _make_request cannot silently feed an error body + # into ET.fromstring below. response.raise_for_status() # Parse XML response @@ -1182,10 +1190,10 @@ class WebDAVClient(BaseNextcloudClient): and user_assignable_elem.text is not None else True, } - logger.debug(f"Found tag '{tag_name}' with ID {tag_info['id']}") + logger.debug("Found tag %r with ID %s", tag_name, tag_info["id"]) return tag_info - logger.debug(f"Tag '{tag_name}' not found") + logger.debug("Tag %r not found", tag_name) return None async def get_files_by_tag(self, tag_id: int) -> list[dict[str, Any]]: @@ -1197,7 +1205,9 @@ class WebDAVClient(BaseNextcloudClient): Returns: List of file info dictionaries with path, size, content_type, etc. """ - # Use WebDAV REPORT method with systemtag filter, requesting all properties + # Use WebDAV REPORT method with systemtag filter. resourcetype is + # included so callers can distinguish folders from files (needed for + # recursive exclusion of tagged directories — see issue #710). report_body = f""" @@ -1207,17 +1217,22 @@ class WebDAVClient(BaseNextcloudClient): + {tag_id} """ - response = await self._client.request( + response = await self._make_request( "REPORT", f"{self._get_webdav_base_path()}/", + headers={"Content-Type": "text/xml", "OCS-APIRequest": "true"}, content=report_body, ) + # Redundant after _make_request (which raises on non-2xx) but + # makes the contract explicit at the call site — see the same + # rationale in get_tag_by_name. response.raise_for_status() # Parse XML response @@ -1249,15 +1264,27 @@ class WebDAVClient(BaseNextcloudClient): contenttype_elem = prop.find("d:getcontenttype", ns) lastmodified_elem = prop.find("d:getlastmodified", ns) etag_elem = prop.find("d:getetag", ns) + resourcetype_elem = prop.find("d:resourcetype", ns) if fileid_elem is None or not fileid_elem.text: continue - # Decode href path and extract the file path + # A resourcetype with a child indicates a folder. + is_directory = ( + resourcetype_elem is not None + and resourcetype_elem.find("d:collection", ns) is not None + ) + + # Decode href path and extract the user-relative file path. + # str.replace() would strip every occurrence of the prefix, + # so an adversarially-named directory could collide; strip + # only the leading occurrence via startswith + slice. href_path = unquote(href_elem.text) - # Remove WebDAV prefix to get user-relative path webdav_prefix = f"/remote.php/dav/files/{self.username}/" - file_path = href_path.replace(webdav_prefix, "/") + if href_path.startswith(webdav_prefix): + file_path = "/" + href_path[len(webdav_prefix) :] + else: + file_path = href_path # Parse last modified timestamp last_modified_timestamp = None @@ -1285,10 +1312,11 @@ class WebDAVClient(BaseNextcloudClient): else None, "last_modified_timestamp": last_modified_timestamp, "etag": etag_elem.text if etag_elem is not None else None, + "is_directory": is_directory, } files.append(file_info) - logger.debug(f"Found {len(files)} files with tag ID {tag_id}") + logger.debug("Found %d files with tag ID %s", len(files), tag_id) return files async def get_file_info(self, path: str) -> dict[str, Any] | None: diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index eb2cb59f..60a0a714 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -117,6 +117,10 @@ _DEFAULTS: dict[str, Any] = { "custom_processor_name": "custom", "custom_processor_api_key": None, "custom_processor_timeout": 60, + # Tag-based file exclusion (issue #710): comma-separated list of + # Nextcloud system tag names. Files/folders carrying any of these tags + # are hidden from WebDAV MCP tools. Empty = feature off. + "excluded_tags": "", } @@ -508,6 +512,11 @@ class Settings: log_level: str = "INFO" log_include_trace_context: bool = True + # Tag-based file exclusion (issue #710): comma-separated list of + # Nextcloud system tag names. Files/folders carrying any of these tags + # are hidden from WebDAV MCP tools. + excluded_tags: str = "" + def __post_init__(self): """Validate configuration and set defaults.""" logger = logging.getLogger(__name__) @@ -845,6 +854,7 @@ def get_settings() -> Settings: "log_format": "LOG_FORMAT", "log_level": "LOG_LEVEL", "log_include_trace_context": "LOG_INCLUDE_TRACE_CONTEXT", + "excluded_tags": "EXCLUDED_TAGS", } # Only pass values that dynaconf actually has; omit unset keys so diff --git a/nextcloud_mcp_server/server/tag_exclusion.py b/nextcloud_mcp_server/server/tag_exclusion.py new file mode 100644 index 00000000..8ced477b --- /dev/null +++ b/nextcloud_mcp_server/server/tag_exclusion.py @@ -0,0 +1,166 @@ +"""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 + +import anyio + +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 _resolve_one_tag( + tag_name: str, + webdav: WebDAVClient, + results: list[set[str]], +) -> None: + """Resolve a single tag's paths and append them as a set to *results*. + + Each task appends its own set to the shared list; ``list.append`` is + atomic between cooperative yields under anyio (single-threaded + between awaits) so no explicit lock is needed. Swallows its own + exceptions so a failure for one tag does not abort the surrounding + task group (preserves fail-open per-tag semantics). + """ + try: + tag = await webdav.get_tag_by_name(tag_name) + except Exception as e: + logger.warning( + "Tag exclusion lookup failed for tag %r (%s); " + "skipping — files tagged with this tag will be visible", + tag_name, + e, + ) + return + + if tag is None: + logger.debug("Excluded tag %r does not exist — skipping", tag_name) + return + + if tag.get("id") is None: + # Malformed PROPFIND response: entry without + # . Skip rather than dispatch None. + logger.debug( + "Excluded tag %r has no id in PROPFIND response — skipping", + tag_name, + ) + return + + try: + files = await webdav.get_files_by_tag(tag["id"]) + except Exception as e: + logger.warning( + "Tag exclusion file enumeration failed for tag %r (%s); " + "skipping — files tagged with this tag will be visible", + tag_name, + e, + ) + return + + paths: set[str] = set() + for f in files: + path = _normalise_path(f["path"]) + paths.add(path) + if f.get("is_directory"): + logger.debug( + "Excluding directory %r (tag %r) — descendants will be hidden", + path, + tag_name, + ) + results.append(paths) + + +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`. + + Per-tag resolution is fanned out via ``anyio.create_task_group`` so + that the 2N network calls (1 PROPFIND + 1 REPORT per tag) run + concurrently rather than serially. No lock is needed: each task + appends its own ``set`` to a shared list, and append is atomic + between awaits under anyio's cooperative single-threaded model. + + **Failure mode is fail-open per tag**: if the systemtags endpoint is + unreachable or returns an error for a given tag, that tag is skipped + with a warning rather than propagating the error. Reasoning: the + threat model is preventing *accidental* exfiltration via the LLM tool + surface; a Nextcloud-side outage of the systemtags API should not + take down all WebDAV tools. Operators relying on this for stronger + guarantees should monitor the warning logs. + """ + tag_names = get_excluded_tag_names() + if not tag_names: + return set() + + results: list[set[str]] = [] + async with anyio.create_task_group() as tg: + for tag_name in tag_names: + tg.start_soon(_resolve_one_tag, tag_name, webdav, results) + + excluded: set[str] = set().union(*results) + if excluded: + # `len(excluded)` counts directly-tagged entries — descendants of + # tagged directories are hidden too but resolved at check time. + logger.info( + "Tag-based exclusion resolved to %d directly-tagged path(s) " + "for tags: %s (descendants of tagged directories also hidden)", + 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 + ``/``). + """ + 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("/") diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index c89f3896..39ce775b 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -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,12 @@ def configure_webdav_tools(mcp: FastMCP): ) -> 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) @@ -39,8 +50,21 @@ def configure_webdav_tools(mcp: FastMCP): 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] @@ -70,6 +94,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 +107,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 +136,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 +179,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 +192,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 +218,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 +228,15 @@ 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 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( @@ -197,6 +252,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 +262,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 +284,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 +297,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 or is " + "inside a path tagged with an excluded tag" + ) + return await client.webdav.move_resource( source_path, destination_path, overwrite ) @@ -247,6 +328,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 +341,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 or is " + "inside a path tagged with an excluded tag" + ) + return await client.webdav.copy_resource( source_path, destination_path, overwrite ) @@ -294,6 +392,13 @@ def configure_webdav_tools(mcp: FastMCP): """ 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 = [] @@ -364,6 +469,12 @@ def configure_webdav_tools(mcp: FastMCP): 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] @@ -406,9 +517,18 @@ def configure_webdav_tools(mcp: FastMCP): 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, @@ -440,9 +560,18 @@ def configure_webdav_tools(mcp: FastMCP): 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, @@ -473,7 +602,16 @@ def configure_webdav_tools(mcp: FastMCP): 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, diff --git a/tests/integration/test_tag_exclusion.py b/tests/integration/test_tag_exclusion.py new file mode 100644 index 00000000..f86014a8 --- /dev/null +++ b/tests/integration/test_tag_exclusion.py @@ -0,0 +1,191 @@ +"""End-to-end integration tests for tag-based file exclusion (issue #710). + +These exercise the full resolution pipeline against a real Nextcloud +instance: + +1. Create a system tag via the WebDAV ``/systemtags`` API. +2. Create real files / a real directory and tag them. +3. Resolve the configured ``EXCLUDED_TAGS`` to paths via + ``get_excluded_file_paths`` — this issues a real PROPFIND against + ``/systemtags/`` and a real REPORT against the user's WebDAV root. +4. Verify ``is_path_excluded`` correctly classifies tagged files, + descendants of tagged directories, and unrelated paths. + +Unlike the unit tests in ``tests/unit/test_tag_exclusion.py``, which +mock the WebDAV layer, this test catches integration-level issues: +malformed XML responses, namespace mismatches, missing fields, and +PROPFIND/REPORT semantics that diverge from what the unit-test mocks +assume. + +The MCP server layer is exercised by the unit tests (where +``EXCLUDED_TAGS`` is patched at the config layer); spinning up a fresh +MCP container with a custom env var per integration test would not +add coverage proportional to the cost. +""" + +import logging +import uuid + +import pytest + +from nextcloud_mcp_server.client import NextcloudClient +from nextcloud_mcp_server.server.tag_exclusion import ( + get_excluded_file_paths, + is_path_excluded, +) + +logger = logging.getLogger(__name__) +pytestmark = pytest.mark.integration + + +@pytest.fixture +async def excluded_tag_environment(nc_client: NextcloudClient): + """Provision a tag, a tagged file, a tagged directory, and an + untagged sibling — all in a unique per-run namespace. + + Yields a dict with the layout. Cleanup runs in reverse order: + untag, delete files, leave the tag (no public delete API on the + client today; tags are cheap and unique-per-run). + """ + suffix = uuid.uuid4().hex[:8] + tag_name = f"mcp-no-ai-{suffix}" + test_dir = f"mcp_tag_excl_{suffix}" + tagged_file = f"{test_dir}/SECRET.txt" + tagged_dir = f"{test_dir}/private" + tagged_dir_child = f"{tagged_dir}/inside.txt" + untagged_file = f"{test_dir}/visible.txt" + + # Layout + await nc_client.webdav.create_directory(test_dir) + await nc_client.webdav.create_directory(tagged_dir) + await nc_client.webdav.write_file(tagged_file, b"top secret", "text/plain") + await nc_client.webdav.write_file(tagged_dir_child, b"inside private", "text/plain") + await nc_client.webdav.write_file(untagged_file, b"public", "text/plain") + + # Tag definition + tag = await nc_client.webdav.get_or_create_tag( + name=tag_name, + user_visible=True, + # In production we recommend user_assignable=False; for tests we + # keep it True so cleanup via remove_tag_from_file works under + # the same credentials. + user_assignable=True, + ) + assert tag["id"] is not None, "tag creation did not return an id" + + # Tag assignments — needs file IDs + secret_info = await nc_client.webdav.get_file_info(tagged_file) + assert secret_info is not None + private_info = await nc_client.webdav.get_file_info(tagged_dir) + assert private_info is not None + + await nc_client.webdav.assign_tag_to_file(secret_info["id"], tag["id"]) + await nc_client.webdav.assign_tag_to_file(private_info["id"], tag["id"]) + + yield { + "tag_name": tag_name, + "tag_id": tag["id"], + "test_dir": test_dir, + "tagged_file": tagged_file, + "tagged_dir": tagged_dir, + "tagged_dir_child": tagged_dir_child, + "untagged_file": untagged_file, + "tagged_file_id": secret_info["id"], + "tagged_dir_id": private_info["id"], + } + + # Cleanup + for file_id, tag_id in ( + (secret_info["id"], tag["id"]), + (private_info["id"], tag["id"]), + ): + try: + await nc_client.webdav.remove_tag_from_file(file_id, tag_id) + except Exception as e: + logger.warning("failed to untag file %s: %s", file_id, e) + try: + await nc_client.webdav.delete_resource(test_dir) + except Exception as e: + logger.warning("failed to delete %s: %s", test_dir, e) + + +async def test_get_excluded_file_paths_resolves_real_systemtags( + excluded_tag_environment, nc_client: NextcloudClient, mocker +): + """``get_excluded_file_paths`` resolves a real Nextcloud system tag + to real WebDAV paths via PROPFIND + REPORT. + + Patches ``get_excluded_tag_names`` at the module level so we can + target our per-run tag without restarting the MCP server with a + custom ``EXCLUDED_TAGS`` env var. + """ + env = excluded_tag_environment + mocker.patch( + "nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names", + return_value=[env["tag_name"]], + ) + + excluded = await get_excluded_file_paths(nc_client.webdav) + + # Both directly-tagged entries appear (paths are normalised — no + # leading slashes). + assert env["tagged_file"].lstrip("/") in excluded + assert env["tagged_dir"].lstrip("/") in excluded + + # Descendants of the tagged directory are NOT in the resolved set + # by themselves — they are blocked at check time via prefix match. + assert env["tagged_dir_child"].lstrip("/") not in excluded + + # Untagged sibling is not in the set. + assert env["untagged_file"].lstrip("/") not in excluded + + +async def test_is_path_excluded_against_real_resolved_set( + excluded_tag_environment, nc_client: NextcloudClient, mocker +): + """End-to-end: real tag → real PROPFIND/REPORT → ``is_path_excluded`` + classifies real paths correctly. Covers exact match, descendant of + tagged directory, and unrelated path against an untagged sibling. + """ + env = excluded_tag_environment + mocker.patch( + "nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names", + return_value=[env["tag_name"]], + ) + + excluded = await get_excluded_file_paths(nc_client.webdav) + + # Exact match on the tagged file. + assert is_path_excluded(env["tagged_file"], excluded) is True + + # Exact match on the tagged directory. + assert is_path_excluded(env["tagged_dir"], excluded) is True + + # A child of the tagged directory is excluded by prefix match (this + # is the descendant-of-tagged-dir case that ``get_excluded_file_paths`` + # alone would NOT cover; the prefix matching in ``is_path_excluded`` + # is what makes recursive exclusion work). + assert is_path_excluded(env["tagged_dir_child"], excluded) is True + + # The untagged sibling file is NOT excluded. + assert is_path_excluded(env["untagged_file"], excluded) is False + + # A path outside the test directory is NOT excluded. + assert is_path_excluded("/this-path-was-never-created", excluded) is False + + +async def test_feature_disabled_returns_empty_set_against_real_server( + excluded_tag_environment, nc_client: NextcloudClient, mocker +): + """With ``EXCLUDED_TAGS`` empty, ``get_excluded_file_paths`` is a + no-op even when tags exist on the server. Verifies the early-exit + short-circuit still holds against a real instance. + """ + mocker.patch( + "nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names", + return_value=[], + ) + + excluded = await get_excluded_file_paths(nc_client.webdav) + + assert excluded == set() diff --git a/tests/unit/client/test_webdav.py b/tests/unit/client/test_webdav.py index 0218d336..03b0b728 100644 --- a/tests/unit/client/test_webdav.py +++ b/tests/unit/client/test_webdav.py @@ -354,3 +354,63 @@ async def test_remove_tag_from_file_not_assigned(mocker): # Verify result (should succeed even with 404) assert result is True + + +@pytest.mark.unit +async def test_get_files_by_tag_detects_directories(mocker): + """get_files_by_tag must flag tagged folders via . + + Tagged folders need ``is_directory=True`` so the tag-exclusion layer + (issue #710) can hide their descendants. + """ + mock_http_client = AsyncMock() + client = WebDAVClient(mock_http_client, "testuser") + + # Two-entry response: one regular file, one collection (folder). + xml_content = b""" + + + /remote.php/dav/files/testuser/Secret.txt + + + 101 + Secret.txt + 42 + text/plain + Wed, 01 Jan 2025 00:00:00 GMT + "abc" + + + + + + /remote.php/dav/files/testuser/Private/ + + + 102 + Private + Wed, 01 Jan 2025 00:00:00 GMT + "def" + + + + + """ + + mock_response = AsyncMock() + mock_response.content = xml_content + mock_response.raise_for_status = mocker.Mock() + mock_http_client.request = AsyncMock(return_value=mock_response) + + files = await client.get_files_by_tag(42) + + assert len(files) == 2 + by_path = {f["path"]: f for f in files} + + assert by_path["/Secret.txt"]["is_directory"] is False + assert by_path["/Private/"]["is_directory"] is True + + # Sanity-check the REPORT body asks for resourcetype. + call_args = mock_http_client.request.call_args + assert "" in call_args.kwargs["content"] + assert "42" in call_args.kwargs["content"] diff --git a/tests/unit/test_tag_exclusion.py b/tests/unit/test_tag_exclusion.py new file mode 100644 index 00000000..9ed3e318 --- /dev/null +++ b/tests/unit/test_tag_exclusion.py @@ -0,0 +1,326 @@ +"""Unit tests for tag-based file exclusion (issue #710). + +Tests in :class:`TestGetExcludedTagNames` patch ``os.environ`` and call +``_reload_config()`` to make dynaconf observe the patched value. Cleanup +is handled by the autouse ``_reload_dynaconf_after_test`` fixture in +``tests/unit/conftest.py``, which reloads dynaconf after every test. +""" + +import logging +import os +from unittest.mock import AsyncMock, patch + +import anyio +import pytest + +from nextcloud_mcp_server.config import _reload_config +from nextcloud_mcp_server.server.tag_exclusion import ( + _normalise_path, + get_excluded_file_paths, + get_excluded_tag_names, + is_path_excluded, +) + + +class TestNormalisePath: + @pytest.mark.unit + def test_strips_leading_and_trailing_slash(self): + assert _normalise_path("/foo/bar/") == "foo/bar" + + @pytest.mark.unit + def test_unchanged_when_already_clean(self): + assert _normalise_path("foo/bar") == "foo/bar" + + @pytest.mark.unit + def test_empty_string(self): + assert _normalise_path("") == "" + + +class TestIsPathExcluded: + @pytest.mark.unit + def test_empty_set_excludes_nothing(self): + assert is_path_excluded("/anything", set()) is False + + @pytest.mark.unit + def test_direct_match(self): + assert is_path_excluded("/Secret.txt", {"Secret.txt"}) is True + + @pytest.mark.unit + def test_path_argument_is_normalised(self): + # The path argument is normalised before comparison; the excluded + # set is expected to already contain normalised entries (it always + # is, in practice, because get_excluded_file_paths builds it). + assert is_path_excluded("/Secret.txt/", {"Secret.txt"}) is True + + @pytest.mark.unit + def test_descendant_of_excluded_directory(self): + assert is_path_excluded("/Private/notes.md", {"Private"}) is True + assert is_path_excluded("/Private/sub/file.txt", {"Private"}) is True + + @pytest.mark.unit + def test_unrelated_path_not_excluded(self): + assert is_path_excluded("/Public/notes.md", {"Private"}) is False + + @pytest.mark.unit + def test_shared_prefix_is_not_a_match(self): + # 'foobar' must NOT be excluded just because 'foo' is. + # This is the bug a naive `startswith(exc)` would have. + assert is_path_excluded("/foobar/x", {"foo"}) is False + assert is_path_excluded("/foobar", {"foo"}) is False + + @pytest.mark.unit + def test_excluded_path_itself(self): + # The excluded entry itself is excluded (not just its descendants). + assert is_path_excluded("/Private", {"Private"}) is True + + +class TestGetExcludedTagNames: + @pytest.mark.unit + @patch.dict(os.environ, {"EXCLUDED_TAGS": ""}, clear=False) + def test_empty_returns_empty_list(self): + _reload_config() + assert get_excluded_tag_names() == [] + + @pytest.mark.unit + @patch.dict(os.environ, {"EXCLUDED_TAGS": "secret"}, clear=False) + def test_single_tag(self): + _reload_config() + assert get_excluded_tag_names() == ["secret"] + + @pytest.mark.unit + @patch.dict(os.environ, {"EXCLUDED_TAGS": " a , b , c "}, clear=False) + def test_strips_whitespace_around_each_tag(self): + _reload_config() + assert get_excluded_tag_names() == ["a", "b", "c"] + + @pytest.mark.unit + @patch.dict(os.environ, {"EXCLUDED_TAGS": "a,,b,"}, clear=False) + def test_skips_empty_entries(self): + _reload_config() + assert get_excluded_tag_names() == ["a", "b"] + + +class TestGetExcludedFilePaths: + @pytest.mark.unit + async def test_returns_empty_set_when_feature_disabled(self, mocker): + mocker.patch( + "nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names", + return_value=[], + ) + webdav = AsyncMock() + result = await get_excluded_file_paths(webdav) + assert result == set() + webdav.get_tag_by_name.assert_not_called() + + @pytest.mark.unit + async def test_skips_unknown_tag(self, mocker): + mocker.patch( + "nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names", + return_value=["does-not-exist"], + ) + webdav = AsyncMock() + webdav.get_tag_by_name = AsyncMock(return_value=None) + + result = await get_excluded_file_paths(webdav) + + assert result == set() + webdav.get_tag_by_name.assert_awaited_once_with("does-not-exist") + webdav.get_files_by_tag.assert_not_called() + + @pytest.mark.unit + async def test_skips_tag_with_missing_id(self, mocker): + """If get_tag_by_name returns a dict with id=None (malformed + PROPFIND response — entry without ), skip + the tag rather than dispatching None + to get_files_by_tag (PR #764 review round 4).""" + mocker.patch( + "nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names", + return_value=["malformed"], + ) + webdav = AsyncMock() + webdav.get_tag_by_name = AsyncMock( + return_value={ + "id": None, + "name": "malformed", + "userVisible": True, + "userAssignable": True, + } + ) + + result = await get_excluded_file_paths(webdav) + + assert result == set() + webdav.get_tag_by_name.assert_awaited_once_with("malformed") + webdav.get_files_by_tag.assert_not_called() + + @pytest.mark.unit + async def test_fail_open_when_tag_lookup_raises(self, mocker, caplog): + """If get_tag_by_name raises (e.g. 5xx from systemtags endpoint), + the offending tag is skipped with a warning rather than the error + propagating and disabling all WebDAV tools (PR review #764).""" + mocker.patch( + "nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names", + return_value=["broken", "ok"], + ) + webdav = AsyncMock() + webdav.get_tag_by_name = AsyncMock( + side_effect=[ + RuntimeError("upstream 503"), + {"id": 7, "name": "ok"}, + ] + ) + webdav.get_files_by_tag = AsyncMock( + return_value=[{"path": "/ok.txt", "is_directory": False}] + ) + + caplog.set_level( + logging.WARNING, logger="nextcloud_mcp_server.server.tag_exclusion" + ) + result = await get_excluded_file_paths(webdav) + + assert result == {"ok.txt"} + assert "Tag exclusion lookup failed" in caplog.text + assert "broken" in caplog.text + + @pytest.mark.unit + async def test_fail_open_when_file_enumeration_raises(self, mocker, caplog): + """If get_files_by_tag raises (e.g. REPORT timeout), the + offending tag is skipped with a warning. Other tags are still + resolved (PR review #764).""" + mocker.patch( + "nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names", + return_value=["broken", "ok"], + ) + webdav = AsyncMock() + webdav.get_tag_by_name = AsyncMock( + side_effect=[ + {"id": 1, "name": "broken"}, + {"id": 2, "name": "ok"}, + ] + ) + webdav.get_files_by_tag = AsyncMock( + side_effect=[ + RuntimeError("REPORT timeout"), + [{"path": "/ok.txt", "is_directory": False}], + ] + ) + + caplog.set_level( + logging.WARNING, logger="nextcloud_mcp_server.server.tag_exclusion" + ) + result = await get_excluded_file_paths(webdav) + + assert result == {"ok.txt"} + assert "Tag exclusion file enumeration failed" in caplog.text + assert "broken" in caplog.text + + @pytest.mark.unit + async def test_collects_paths_from_multiple_tags(self, mocker): + mocker.patch( + "nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names", + return_value=["secret", "no-ai"], + ) + webdav = AsyncMock() + webdav.get_tag_by_name = AsyncMock( + side_effect=[ + {"id": 1, "name": "secret"}, + {"id": 2, "name": "no-ai"}, + ] + ) + webdav.get_files_by_tag = AsyncMock( + side_effect=[ + [ + {"path": "/Secret.txt", "is_directory": False}, + {"path": "/Private/", "is_directory": True}, + ], + [ + # Same dir under a second tag — set dedupes it. + {"path": "/Private", "is_directory": True}, + {"path": "/Other/notes.md", "is_directory": False}, + ], + ] + ) + + result = await get_excluded_file_paths(webdav) + + assert result == {"Secret.txt", "Private", "Other/notes.md"} + assert webdav.get_files_by_tag.await_count == 2 + + @pytest.mark.unit + async def test_resolves_tags_concurrently(self, mocker): + """Per-tag resolution must run under a task group so the 2N + network calls overlap (PR review #764). The barrier event below + only completes if all three tags' ``get_tag_by_name`` invocations + are in flight simultaneously — sequential resolution would block + the first task forever and trip ``fail_after``. + """ + tags = ["a", "b", "c"] + mocker.patch( + "nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names", + return_value=tags, + ) + + all_started = anyio.Event() + started_count = 0 + + async def fake_get_tag(tag_name: str): + nonlocal started_count + started_count += 1 + if started_count == len(tags): + all_started.set() + with anyio.fail_after(2.0): + await all_started.wait() + return {"id": ord(tag_name), "name": tag_name} + + async def fake_get_files(tag_id: int): + return [{"path": f"/tag-{tag_id}.txt", "is_directory": False}] + + webdav = AsyncMock() + webdav.get_tag_by_name = AsyncMock(side_effect=fake_get_tag) + webdav.get_files_by_tag = AsyncMock(side_effect=fake_get_files) + + with anyio.fail_after(5.0): + result = await get_excluded_file_paths(webdav) + + assert result == {f"tag-{ord(t)}.txt" for t in tags} + + @pytest.mark.unit + async def test_fail_open_under_task_group(self, mocker, caplog): + """One tag failing must not abort sibling tasks in the task group + (PR review #764). Uses callable side_effects keyed by tag name so + the assertion is independent of the order in which the task group + schedules the per-tag coroutines. + """ + mocker.patch( + "nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names", + return_value=["good-1", "broken", "good-2"], + ) + + async def fake_get_tag(tag_name: str): + if tag_name == "broken": + raise RuntimeError("upstream 503") + return {"id": hash(tag_name) & 0xFFFF, "name": tag_name} + + async def fake_get_files(tag_id: int): + # Map tag_id back to a deterministic path. + return [{"path": f"/tagged-by-{tag_id}.txt", "is_directory": False}] + + webdav = AsyncMock() + webdav.get_tag_by_name = AsyncMock(side_effect=fake_get_tag) + webdav.get_files_by_tag = AsyncMock(side_effect=fake_get_files) + + caplog.set_level( + logging.WARNING, logger="nextcloud_mcp_server.server.tag_exclusion" + ) + result = await get_excluded_file_paths(webdav) + + # Both healthy tags must contribute one path each; the broken + # tag is silently skipped. + good_1_id = hash("good-1") & 0xFFFF + good_2_id = hash("good-2") & 0xFFFF + assert result == { + f"tagged-by-{good_1_id}.txt", + f"tagged-by-{good_2_id}.txt", + } + assert "Tag exclusion lookup failed" in caplog.text + assert "broken" in caplog.text diff --git a/tests/unit/test_webdav_tools_exclusion.py b/tests/unit/test_webdav_tools_exclusion.py new file mode 100644 index 00000000..d2cb6cf9 --- /dev/null +++ b/tests/unit/test_webdav_tools_exclusion.py @@ -0,0 +1,400 @@ +"""Server-layer regression tests for tag-based file exclusion (issue #710). + +These tests register the WebDAV tools on a fresh ``FastMCP`` instance and +invoke each tool's underlying function directly via the tool registry. +Their purpose is **not** to re-test the path-matching logic (covered in +``test_tag_exclusion.py``) but to catch wiring regressions: that each +tool actually consults ``get_excluded_file_paths`` / ``is_path_excluded`` +at the right point and raises / filters as expected. + +The decorators on each tool (``@require_scopes``, ``@instrument_tool``) +are transparent under our mocked ``Context`` (no ``access_token`` set → +BasicAuth pass-through path). +""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from mcp.server.fastmcp import FastMCP +from mcp.server.fastmcp.exceptions import ToolError + +from nextcloud_mcp_server.server.webdav import configure_webdav_tools + +pytestmark = pytest.mark.unit + + +@pytest.fixture +def webdav_tools() -> dict: + """Register the WebDAV tools on a fresh FastMCP and return them by name.""" + mcp = FastMCP(name="test-webdav-tools") + configure_webdav_tools(mcp) + return {t.name: t for t in mcp._tool_manager.list_tools()} + + +def _mock_ctx(client) -> SimpleNamespace: + """Build a minimal Context-shaped object for the tool decorators. + + Setting ``request_context.access_token = None`` causes ``require_scopes`` + to take the BasicAuth pass-through branch (see scope_authorization.py). + """ + ctx = SimpleNamespace() + ctx.request_context = SimpleNamespace(access_token=None) + ctx._client = client # only used by tools that fetch via get_client(ctx) + return ctx + + +@pytest.fixture +def patch_get_client(mocker): + """Replace ``get_client`` in the webdav server module with a mock.""" + + def _install(client): + async def fake_get_client(ctx): + return client + + mocker.patch( + "nextcloud_mcp_server.server.webdav.get_client", + side_effect=fake_get_client, + ) + + return _install + + +@pytest.fixture +def patch_excluded(mocker): + """Replace ``get_excluded_file_paths`` with a fixed return value.""" + + def _install(excluded: set[str]): + async def fake(*_, **__): + return excluded + + mocker.patch( + "nextcloud_mcp_server.server.webdav.get_excluded_file_paths", + side_effect=fake, + ) + + return _install + + +@pytest.fixture +def fake_client(): + """A NextcloudClient-shaped mock with an AsyncMock webdav attribute.""" + client = SimpleNamespace() + client.webdav = AsyncMock() + return client + + +# ── Read / mutate guards ──────────────────────────────────────────────── + + +async def test_read_file_raises_when_path_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Secret.txt"}) + + fn = webdav_tools["nc_webdav_read_file"].fn + with pytest.raises(ToolError, match="excluded tag"): + await fn(path="/Secret.txt", ctx=_mock_ctx(fake_client)) + + fake_client.webdav.read_file.assert_not_called() + + +async def test_read_file_passes_through_when_not_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Secret.txt"}) + fake_client.webdav.read_file = AsyncMock(return_value=(b"hello", "text/plain")) + + fn = webdav_tools["nc_webdav_read_file"].fn + result = await fn(path="/Public/notes.md", ctx=_mock_ctx(fake_client)) + + assert result["content"] == "hello" + fake_client.webdav.read_file.assert_awaited_once_with("/Public/notes.md") + + +async def test_write_file_raises_when_path_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Private"}) + + fn = webdav_tools["nc_webdav_write_file"].fn + with pytest.raises(ToolError, match="excluded tag"): + await fn( + path="/Private/note.md", + content="hi", + ctx=_mock_ctx(fake_client), + ) + + fake_client.webdav.write_file.assert_not_called() + + +async def test_delete_resource_raises_when_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Secret.txt"}) + + fn = webdav_tools["nc_webdav_delete_resource"].fn + with pytest.raises(ToolError, match="excluded tag"): + await fn(path="/Secret.txt", ctx=_mock_ctx(fake_client)) + + fake_client.webdav.delete_resource.assert_not_called() + + +async def test_create_directory_raises_when_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Private"}) + + fn = webdav_tools["nc_webdav_create_directory"].fn + with pytest.raises(ToolError, match="is or is inside"): + await fn(path="/Private/sub", ctx=_mock_ctx(fake_client)) + + fake_client.webdav.create_directory.assert_not_called() + + +async def test_move_resource_blocks_excluded_source( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Secret.txt"}) + + fn = webdav_tools["nc_webdav_move_resource"].fn + with pytest.raises(ToolError, match="source"): + await fn( + source_path="/Secret.txt", + destination_path="/Public/x.txt", + ctx=_mock_ctx(fake_client), + ) + + fake_client.webdav.move_resource.assert_not_called() + + +async def test_move_resource_blocks_excluded_destination_exact_match( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + """Destination check must trip on an *exact* match, not just a prefix. + + Regression guard for review #764: previously the message said "is + inside" but is_path_excluded also matches exact paths. + """ + patch_get_client(fake_client) + patch_excluded({"Private"}) + + fn = webdav_tools["nc_webdav_move_resource"].fn + with pytest.raises(ToolError, match="is or is inside"): + await fn( + source_path="/Public/x.txt", + destination_path="/Private", + ctx=_mock_ctx(fake_client), + ) + + +async def test_copy_resource_blocks_excluded_source( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Secret.txt"}) + + fn = webdav_tools["nc_webdav_copy_resource"].fn + with pytest.raises(ToolError, match="source"): + await fn( + source_path="/Secret.txt", + destination_path="/Public/copy.txt", + ctx=_mock_ctx(fake_client), + ) + + fake_client.webdav.copy_resource.assert_not_called() + + +async def test_copy_resource_blocks_excluded_destination_descendant( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Private"}) + + fn = webdav_tools["nc_webdav_copy_resource"].fn + with pytest.raises(ToolError, match="is or is inside"): + await fn( + source_path="/Public/x.txt", + destination_path="/Private/copy.txt", + ctx=_mock_ctx(fake_client), + ) + + +# ── Listing / search filtering ────────────────────────────────────────── + + +async def test_list_directory_filters_excluded_children( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Public/Secret.txt"}) + fake_client.webdav.list_directory = AsyncMock( + return_value=[ + { + "path": "/Public/Secret.txt", + "name": "Secret.txt", + "is_directory": False, + }, + { + "path": "/Public/visible.md", + "name": "visible.md", + "is_directory": False, + }, + ] + ) + + fn = webdav_tools["nc_webdav_list_directory"].fn + result = await fn(path="/Public", ctx=_mock_ctx(fake_client)) + + assert [f.path for f in result.files] == ["/Public/visible.md"] + + +async def test_list_directory_raises_when_listed_path_itself_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + """The early guard prevents the round-trip to Nextcloud and signals + the access denial, instead of silently returning an empty listing + (review #764).""" + patch_get_client(fake_client) + patch_excluded({"Private"}) + + fn = webdav_tools["nc_webdav_list_directory"].fn + with pytest.raises(ToolError, match="excluded tag"): + await fn(path="/Private", ctx=_mock_ctx(fake_client)) + + fake_client.webdav.list_directory.assert_not_called() + + +async def test_search_files_filters_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Secret.txt"}) + fake_client.webdav.search_files = AsyncMock( + return_value=[ + {"path": "/Secret.txt", "name": "Secret.txt", "is_directory": False}, + {"path": "/notes.md", "name": "notes.md", "is_directory": False}, + ] + ) + + fn = webdav_tools["nc_webdav_search_files"].fn + result = await fn(ctx=_mock_ctx(fake_client), name_pattern="%.%") + + assert [r.path for r in result.results] == ["/notes.md"] + + +async def test_find_by_name_filters_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Secret.txt"}) + fake_client.webdav.find_by_name = AsyncMock( + return_value=[ + {"path": "/Secret.txt", "name": "Secret.txt", "is_directory": False}, + {"path": "/visible.txt", "name": "visible.txt", "is_directory": False}, + ] + ) + + fn = webdav_tools["nc_webdav_find_by_name"].fn + result = await fn(pattern="%.txt", ctx=_mock_ctx(fake_client)) + + assert [r.path for r in result.results] == ["/visible.txt"] + + +async def test_find_by_type_filters_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Secret.txt"}) + fake_client.webdav.find_by_type = AsyncMock( + return_value=[ + {"path": "/Secret.txt", "name": "Secret.txt", "is_directory": False}, + {"path": "/visible.txt", "name": "visible.txt", "is_directory": False}, + ] + ) + + fn = webdav_tools["nc_webdav_find_by_type"].fn + result = await fn(mime_type="text/plain", ctx=_mock_ctx(fake_client)) + + assert [r.path for r in result.results] == ["/visible.txt"] + + +async def test_list_favorites_filters_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Secret.txt"}) + fake_client.webdav.list_favorites = AsyncMock( + return_value=[ + {"path": "/Secret.txt", "name": "Secret.txt", "is_directory": False}, + {"path": "/visible.txt", "name": "visible.txt", "is_directory": False}, + ] + ) + + fn = webdav_tools["nc_webdav_list_favorites"].fn + result = await fn(ctx=_mock_ctx(fake_client)) + + assert [r.path for r in result.results] == ["/visible.txt"] + + +# ── Search-tool scope guards (review #764) ────────────────────────────── + + +async def test_search_files_raises_when_scope_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + """Mirror the ``list_directory`` early guard so the four search tools + cannot silently return an empty result for an excluded ``scope``.""" + patch_get_client(fake_client) + patch_excluded({"Private"}) + + fn = webdav_tools["nc_webdav_search_files"].fn + with pytest.raises(ToolError, match="excluded tag"): + await fn(ctx=_mock_ctx(fake_client), scope="/Private", name_pattern="%.txt") + + fake_client.webdav.search_files.assert_not_called() + + +async def test_find_by_name_raises_when_scope_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Private"}) + + fn = webdav_tools["nc_webdav_find_by_name"].fn + with pytest.raises(ToolError, match="excluded tag"): + await fn(pattern="%.txt", scope="/Private", ctx=_mock_ctx(fake_client)) + + fake_client.webdav.find_by_name.assert_not_called() + + +async def test_find_by_type_raises_when_scope_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Private"}) + + fn = webdav_tools["nc_webdav_find_by_type"].fn + with pytest.raises(ToolError, match="excluded tag"): + await fn(mime_type="text/plain", scope="/Private", ctx=_mock_ctx(fake_client)) + + fake_client.webdav.find_by_type.assert_not_called() + + +async def test_list_favorites_raises_when_scope_excluded( + webdav_tools, fake_client, patch_get_client, patch_excluded +): + patch_get_client(fake_client) + patch_excluded({"Private"}) + + fn = webdav_tools["nc_webdav_list_favorites"].fn + with pytest.raises(ToolError, match="excluded tag"): + await fn(ctx=_mock_ctx(fake_client), scope="/Private") + + fake_client.webdav.list_favorites.assert_not_called()