From 22ed9e99a0a812881df39b0946561949d55fd781 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 6 May 2026 12:12:54 +0200 Subject: [PATCH 1/7] feat(webdav): add tag-based file exclusion (#710) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hide sensitive files/folders from the WebDAV MCP tool surface by tagging them with a configured Nextcloud system tag. Defence-in-depth control for users who connect LLMs to accounts holding contracts, medical records, credentials, etc. A new EXCLUDED_TAGS env var (comma-separated tag names, empty by default) gates an exclusion layer that runs at the start of every WebDAV tool call: tag names are resolved to tag IDs, those IDs are expanded to the set of tagged paths, then listings/searches are filtered and read/write/delete/move/copy operations on excluded paths raise ToolError. Tagged folders exclude their descendants via prefix match. Empty EXCLUDED_TAGS disables the feature entirely. The threat model is preventing accidental data exfiltration via the LLM tool surface — not hiding files from a determined operator. The docs explicitly recommend creating exclusion tags with user_assignable=false so the credentials the MCP server uses cannot remove the tag. Implementation: - config.py: add `excluded_tags` to _DEFAULTS, Settings, and the _field_map alongside other comma-separated env vars. - client/webdav.py: get_files_by_tag now requests and surfaces is_directory so tagged directories can recursively exclude descendants. - server/tag_exclusion.py (new): get_excluded_tag_names, get_excluded_file_paths, is_path_excluded. - server/webdav.py: exclusion guards in all 11 WebDAV tools; read/write/create/delete/move/copy raise ToolError, list/search tools silently filter excluded entries. Existing f-string log calls converted to lazy %-style. - tests: 17 new unit tests covering path-matching edge cases (shared-prefix non-match, descendants of excluded dirs), tag-name parsing, and get_excluded_file_paths with mocked WebDAV; 1 new client test asserting -> is_directory parsing. - docs/configuration.md: new "Tag-Based File Exclusion" section with per-tool effect table, security guidance, and per-call cost note. - README.md: feature mention under Key Features. Closes #710. Co-Authored-By: Claude Opus 4.7 (1M context) --- README.md | 1 + docs/configuration.md | 81 ++++++++++ nextcloud_mcp_server/client/webdav.py | 15 +- nextcloud_mcp_server/config.py | 10 ++ nextcloud_mcp_server/server/tag_exclusion.py | 98 ++++++++++++ nextcloud_mcp_server/server/webdav.py | 115 +++++++++++++- tests/unit/client/test_webdav.py | 60 ++++++++ tests/unit/test_tag_exclusion.py | 152 +++++++++++++++++++ 8 files changed, 528 insertions(+), 4 deletions(-) create mode 100644 nextcloud_mcp_server/server/tag_exclusion.py create mode 100644 tests/unit/test_tag_exclusion.py 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..f72baf44 100644 --- a/nextcloud_mcp_server/client/webdav.py +++ b/nextcloud_mcp_server/client/webdav.py @@ -1197,7 +1197,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,6 +1209,7 @@ class WebDAVClient(BaseNextcloudClient): + {tag_id} @@ -1249,10 +1252,17 @@ 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 + # 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 file path href_path = unquote(href_elem.text) # Remove WebDAV prefix to get user-relative path @@ -1285,10 +1295,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..6f5252b5 --- /dev/null +++ b/nextcloud_mcp_server/server/tag_exclusion.py @@ -0,0 +1,98 @@ +"""Tag-based file exclusion for MCP file operations (issue #710). + +Resolves the configured ``EXCLUDED_TAGS`` to a set of file paths that +should be hidden from WebDAV MCP tools (list, read, search) and rejected +by mutating tools (write, delete, move, copy). + +The flow per call: + +1. Parse ``EXCLUDED_TAGS`` (comma-separated tag names) from config. +2. For each tag name, resolve to a tag ID via ``get_tag_by_name``. +3. For each tag ID, fetch all tagged file/folder paths via + ``get_files_by_tag``. +4. Collect normalised paths into a single ``set[str]``. + +Tagging a *folder* excludes the folder itself and every descendant via +prefix match in :func:`is_path_excluded`. + +Threat model: this is a defence-in-depth control to prevent accidental +exfiltration via the LLM tool surface. A user controlling the Nextcloud +account whose credentials the server uses can untag files unless the tag +is created with ``user_assignable=false``. +""" + +import logging + +from nextcloud_mcp_server.client.webdav import WebDAVClient +from nextcloud_mcp_server.config import get_settings + +logger = logging.getLogger(__name__) + + +def get_excluded_tag_names() -> list[str]: + """Return the configured excluded tag names (empty list if disabled).""" + raw = get_settings().excluded_tags + if not raw: + return [] + return [t.strip() for t in raw.split(",") if t.strip()] + + +async def get_excluded_file_paths(webdav: WebDAVClient) -> set[str]: + """Resolve excluded tags to the set of paths they cover. + + Tagged directories are added as their own normalised path; descendants + are blocked via prefix match in :func:`is_path_excluded`. + """ + tag_names = get_excluded_tag_names() + if not tag_names: + return set() + + excluded: set[str] = set() + for tag_name in tag_names: + tag = await webdav.get_tag_by_name(tag_name) + if tag is None: + logger.debug("Excluded tag %r does not exist — skipping", tag_name) + continue + + files = await webdav.get_files_by_tag(tag["id"]) + for f in files: + path = _normalise_path(f["path"]) + excluded.add(path) + if f.get("is_directory"): + logger.debug( + "Excluding directory %r (tag %r) — descendants will be hidden", + path, + tag_name, + ) + + if excluded: + logger.info( + "Tag-based exclusion: hiding %d path(s) matching tags: %s", + len(excluded), + ", ".join(tag_names), + ) + + return excluded + + +def is_path_excluded(path: str, excluded_paths: set[str]) -> bool: + """Return True if *path* (or any of its parents) is excluded. + + A path is excluded when it matches an entry exactly, or when an + excluded entry is one of its directory ancestors (prefix match on + ``/``). + """ + 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..c5f3ef1c 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,10 @@ def configure_webdav_tools(mcp: FastMCP): ) -> DirectoryListing: """List files and directories in the specified NextCloud path. + When ``EXCLUDED_TAGS`` is configured, entries tagged (or whose + ancestor folders are tagged) with an excluded system tag are + omitted from the result. + Args: path: Directory path to list (empty string for root directory) @@ -41,6 +50,13 @@ def configure_webdav_tools(mcp: FastMCP): client = await get_client(ctx) items = await client.webdav.list_directory(path) + # Filter out files/folders carrying an excluded tag. + excluded = await get_excluded_file_paths(client.webdav) + if excluded: + items = [ + i for i in items if not is_path_excluded(i.get("path", ""), excluded) + ] + # Convert to FileInfo models file_infos = [FileInfo(**item) for item in items] @@ -70,6 +86,9 @@ def configure_webdav_tools(mcp: FastMCP): async def nc_webdav_read_file(path: str, ctx: Context): """Read the content of a file from NextCloud. + Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and the + file (or an ancestor folder) carries an excluded system tag. + Args: path: Full path to the file to read @@ -80,13 +99,19 @@ def configure_webdav_tools(mcp: FastMCP): - Other binary files are base64 encoded """ client = await get_client(ctx) + + # Block reads of paths carrying an excluded tag. + excluded = await get_excluded_file_paths(client.webdav) + if is_path_excluded(path, excluded): + raise ToolError(f"Access denied: {path!r} is tagged with an excluded tag") + content, content_type = await client.webdav.read_file(path) # Check if this is a parseable document (PDF, DOCX, etc.) # is_parseable_document() checks if document processing is enabled if is_parseable_document(content_type): try: - logger.info(f"Parsing document '{path}' of type '{content_type}'") + logger.info("Parsing document %r of type %r", path, content_type) parsed_text, metadata = await parse_document( content, content_type, @@ -103,7 +128,9 @@ def configure_webdav_tools(mcp: FastMCP): } except Exception as e: logger.warning( - f"Failed to parse document '{path}', falling back to base64: {e}" + "Failed to parse document %r, falling back to base64: %s", + path, + e, ) # Fall through to base64 encoding on parse failure @@ -144,6 +171,9 @@ def configure_webdav_tools(mcp: FastMCP): ): """Write content to a file in NextCloud. + Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and the + target path (or an ancestor folder) carries an excluded system tag. + Args: path: Full path where to write the file content: File content (text or base64 for binary) @@ -154,6 +184,11 @@ def configure_webdav_tools(mcp: FastMCP): """ client = await get_client(ctx) + # Block writes to excluded paths. + excluded = await get_excluded_file_paths(client.webdav) + if is_path_excluded(path, excluded): + raise ToolError(f"Access denied: {path!r} is tagged with an excluded tag") + # Handle base64 encoded content if content_type and "base64" in content_type.lower(): content_bytes = base64.b64decode(content) @@ -175,6 +210,9 @@ def configure_webdav_tools(mcp: FastMCP): async def nc_webdav_create_directory(path: str, ctx: Context): """Create a directory in NextCloud. + Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and the + target path lies inside a folder carrying an excluded system tag. + Args: path: Full path of the directory to create @@ -182,6 +220,14 @@ def configure_webdav_tools(mcp: FastMCP): Dict with status_code (201 for created, 405 if already exists) """ client = await get_client(ctx) + + # Block directory creation inside excluded paths. + excluded = await get_excluded_file_paths(client.webdav) + if is_path_excluded(path, excluded): + raise ToolError( + f"Access denied: {path!r} is inside a path tagged with an excluded tag" + ) + return await client.webdav.create_directory(path) @mcp.tool( @@ -197,6 +243,9 @@ def configure_webdav_tools(mcp: FastMCP): async def nc_webdav_delete_resource(path: str, ctx: Context): """Delete a file or directory in NextCloud. + Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and the + target path (or an ancestor folder) carries an excluded system tag. + Args: path: Full path of the file or directory to delete @@ -204,6 +253,12 @@ def configure_webdav_tools(mcp: FastMCP): Dict with status_code indicating result (404 if not found) """ client = await get_client(ctx) + + # Block deletion of excluded files/directories. + excluded = await get_excluded_file_paths(client.webdav) + if is_path_excluded(path, excluded): + raise ToolError(f"Access denied: {path!r} is tagged with an excluded tag") + return await client.webdav.delete_resource(path) @mcp.tool( @@ -220,6 +275,10 @@ def configure_webdav_tools(mcp: FastMCP): ): """Move or rename a file or directory in NextCloud. + Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and either + the source or destination path (or one of their ancestor folders) + carries an excluded system tag. + Args: source_path: Full path of the file or directory to move destination_path: New path for the file or directory @@ -229,6 +288,19 @@ def configure_webdav_tools(mcp: FastMCP): Dict with status_code indicating result (404 if source not found, 412 if destination exists and overwrite is False) """ client = await get_client(ctx) + + # Block moves involving excluded paths on either side. + excluded = await get_excluded_file_paths(client.webdav) + if is_path_excluded(source_path, excluded): + raise ToolError( + f"Access denied: source {source_path!r} is tagged with an excluded tag" + ) + if is_path_excluded(destination_path, excluded): + raise ToolError( + f"Access denied: destination {destination_path!r} is inside a " + "path tagged with an excluded tag" + ) + return await client.webdav.move_resource( source_path, destination_path, overwrite ) @@ -247,6 +319,10 @@ def configure_webdav_tools(mcp: FastMCP): ): """Copy a file or directory in NextCloud. + Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and either + the source or destination path (or one of their ancestor folders) + carries an excluded system tag. + Args: source_path: Full path of the file or directory to copy destination_path: Destination path for the copy @@ -256,6 +332,19 @@ def configure_webdav_tools(mcp: FastMCP): Dict with status_code indicating result (404 if source not found, 412 if destination exists and overwrite is False) """ client = await get_client(ctx) + + # Block copies involving excluded paths on either side. + excluded = await get_excluded_file_paths(client.webdav) + if is_path_excluded(source_path, excluded): + raise ToolError( + f"Access denied: source {source_path!r} is tagged with an excluded tag" + ) + if is_path_excluded(destination_path, excluded): + raise ToolError( + f"Access denied: destination {destination_path!r} is inside a " + "path tagged with an excluded tag" + ) + return await client.webdav.copy_resource( source_path, destination_path, overwrite ) @@ -364,6 +453,13 @@ def configure_webdav_tools(mcp: FastMCP): limit=limit, ) + # Filter out tagged-excluded paths. + excluded = await get_excluded_file_paths(client.webdav) + if excluded: + results = [ + r for r in results if not is_path_excluded(r.get("path", ""), excluded) + ] + # Convert to FileInfo models file_infos = [FileInfo(**result) for result in results] @@ -409,6 +505,11 @@ def configure_webdav_tools(mcp: FastMCP): results = await client.webdav.find_by_name( pattern=pattern, scope=scope, limit=limit ) + excluded = await get_excluded_file_paths(client.webdav) + if excluded: + results = [ + r for r in results if not is_path_excluded(r.get("path", ""), excluded) + ] file_infos = [FileInfo(**result) for result in results] return SearchFilesResponse( results=file_infos, @@ -443,6 +544,11 @@ def configure_webdav_tools(mcp: FastMCP): results = await client.webdav.find_by_type( mime_type=mime_type, scope=scope, limit=limit ) + excluded = await get_excluded_file_paths(client.webdav) + if excluded: + results = [ + r for r in results if not is_path_excluded(r.get("path", ""), excluded) + ] file_infos = [FileInfo(**result) for result in results] return SearchFilesResponse( results=file_infos, @@ -474,6 +580,11 @@ def configure_webdav_tools(mcp: FastMCP): """ client = await get_client(ctx) results = await client.webdav.list_favorites(scope=scope, limit=limit) + excluded = await get_excluded_file_paths(client.webdav) + if excluded: + results = [ + r for r in results if not is_path_excluded(r.get("path", ""), excluded) + ] file_infos = [FileInfo(**result) for result in results] return SearchFilesResponse( results=file_infos, 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..1e31e96b --- /dev/null +++ b/tests/unit/test_tag_exclusion.py @@ -0,0 +1,152 @@ +"""Unit tests for tag-based file exclusion (issue #710).""" + +import os +from unittest.mock import AsyncMock, patch + +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_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 From a6c188abbbdf9ed1bd10035293b5b5a7748dbade Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 6 May 2026 12:26:25 +0200 Subject: [PATCH 2/7] fix(webdav): address PR #764 review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- nextcloud_mcp_server/server/tag_exclusion.py | 37 ++- nextcloud_mcp_server/server/webdav.py | 31 +- tests/unit/test_tag_exclusion.py | 70 ++++- tests/unit/test_webdav_tools_exclusion.py | 290 +++++++++++++++++++ 4 files changed, 413 insertions(+), 15 deletions(-) create mode 100644 tests/unit/test_webdav_tools_exclusion.py diff --git a/nextcloud_mcp_server/server/tag_exclusion.py b/nextcloud_mcp_server/server/tag_exclusion.py index 6f5252b5..ad5323e5 100644 --- a/nextcloud_mcp_server/server/tag_exclusion.py +++ b/nextcloud_mcp_server/server/tag_exclusion.py @@ -42,6 +42,14 @@ async def get_excluded_file_paths(webdav: WebDAVClient) -> set[str]: Tagged directories are added as their own normalised path; descendants are blocked via prefix match in :func:`is_path_excluded`. + + **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: @@ -49,12 +57,32 @@ async def get_excluded_file_paths(webdav: WebDAVClient) -> set[str]: excluded: set[str] = set() for tag_name in tag_names: - tag = await webdav.get_tag_by_name(tag_name) + 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, + ) + continue + if tag is None: logger.debug("Excluded tag %r does not exist — skipping", tag_name) continue - files = await webdav.get_files_by_tag(tag["id"]) + 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, + ) + continue + for f in files: path = _normalise_path(f["path"]) excluded.add(path) @@ -66,8 +94,11 @@ async def get_excluded_file_paths(webdav: WebDAVClient) -> set[str]: ) 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: hiding %d path(s) matching tags: %s", + "Tag-based exclusion resolved to %d directly-tagged path(s) " + "for tags: %s (descendants of tagged directories also hidden)", len(excluded), ", ".join(tag_names), ) diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index c5f3ef1c..7220613b 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -37,9 +37,11 @@ def configure_webdav_tools(mcp: FastMCP): ) -> DirectoryListing: """List files and directories in the specified NextCloud path. - When ``EXCLUDED_TAGS`` is configured, entries tagged (or whose - ancestor folders are tagged) with an excluded system tag are - omitted from the result. + 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) @@ -48,10 +50,16 @@ 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 files/folders carrying an excluded tag. - excluded = await get_excluded_file_paths(client.webdav) + # 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) @@ -221,11 +229,12 @@ def configure_webdav_tools(mcp: FastMCP): """ client = await get_client(ctx) - # Block directory creation inside excluded paths. + # 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 inside a path tagged with an excluded tag" + f"Access denied: {path!r} is or is inside a path tagged " + "with an excluded tag" ) return await client.webdav.create_directory(path) @@ -297,8 +306,8 @@ def configure_webdav_tools(mcp: FastMCP): ) if is_path_excluded(destination_path, excluded): raise ToolError( - f"Access denied: destination {destination_path!r} is inside a " - "path tagged with an excluded tag" + f"Access denied: destination {destination_path!r} is or is " + "inside a path tagged with an excluded tag" ) return await client.webdav.move_resource( @@ -341,8 +350,8 @@ def configure_webdav_tools(mcp: FastMCP): ) if is_path_excluded(destination_path, excluded): raise ToolError( - f"Access denied: destination {destination_path!r} is inside a " - "path tagged with an excluded tag" + f"Access denied: destination {destination_path!r} is or is " + "inside a path tagged with an excluded tag" ) return await client.webdav.copy_resource( diff --git a/tests/unit/test_tag_exclusion.py b/tests/unit/test_tag_exclusion.py index 1e31e96b..fc591480 100644 --- a/tests/unit/test_tag_exclusion.py +++ b/tests/unit/test_tag_exclusion.py @@ -1,5 +1,12 @@ -"""Unit tests for tag-based file exclusion (issue #710).""" +"""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 @@ -119,6 +126,67 @@ class TestGetExcludedFilePaths: 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_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( diff --git a/tests/unit/test_webdav_tools_exclusion.py b/tests/unit/test_webdav_tools_exclusion.py new file mode 100644 index 00000000..3a229672 --- /dev/null +++ b/tests/unit/test_webdav_tools_exclusion.py @@ -0,0 +1,290 @@ +"""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_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"] From d179ca8c8b90200106cc50ae82a2655fe74a9716 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 6 May 2026 19:15:39 +0200 Subject: [PATCH 3/7] fix(webdav): address PR #764 review round 2 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) --- nextcloud_mcp_server/client/webdav.py | 6 +- nextcloud_mcp_server/server/tag_exclusion.py | 95 ++++++++++++-------- nextcloud_mcp_server/server/webdav.py | 28 ++++-- tests/unit/test_tag_exclusion.py | 80 +++++++++++++++++ tests/unit/test_webdav_tools_exclusion.py | 93 +++++++++++++++++++ 5 files changed, 257 insertions(+), 45 deletions(-) diff --git a/nextcloud_mcp_server/client/webdav.py b/nextcloud_mcp_server/client/webdav.py index f72baf44..6911c1bc 100644 --- a/nextcloud_mcp_server/client/webdav.py +++ b/nextcloud_mcp_server/client/webdav.py @@ -1132,13 +1132,12 @@ class WebDAVClient(BaseNextcloudClient): """ - response = await self._client.request( + response = await self._make_request( "PROPFIND", "/remote.php/dav/systemtags/", headers={"Depth": "1"}, content=propfind_body, ) - response.raise_for_status() # Parse XML response root = ET.fromstring(response.content) @@ -1216,12 +1215,11 @@ class WebDAVClient(BaseNextcloudClient): """ - response = await self._client.request( + response = await self._make_request( "REPORT", f"{self._get_webdav_base_path()}/", content=report_body, ) - response.raise_for_status() # Parse XML response root = ET.fromstring(response.content) diff --git a/nextcloud_mcp_server/server/tag_exclusion.py b/nextcloud_mcp_server/server/tag_exclusion.py index ad5323e5..5abb55bc 100644 --- a/nextcloud_mcp_server/server/tag_exclusion.py +++ b/nextcloud_mcp_server/server/tag_exclusion.py @@ -23,6 +23,8 @@ 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 @@ -37,12 +39,65 @@ def get_excluded_tag_names() -> list[str]: return [t.strip() for t in raw.split(",") if t.strip()] +async def _resolve_one_tag( + tag_name: str, + webdav: WebDAVClient, + excluded: set[str], + lock: anyio.Lock, +) -> None: + """Resolve a single tag's paths and merge into *excluded* under *lock*. + + 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 + + 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 + + async with lock: + for f in files: + path = _normalise_path(f["path"]) + excluded.add(path) + if f.get("is_directory"): + logger.debug( + "Excluding directory %r (tag %r) — descendants will be hidden", + path, + tag_name, + ) + + 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. + **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 @@ -56,42 +111,10 @@ async def get_excluded_file_paths(webdav: WebDAVClient) -> set[str]: return set() excluded: set[str] = set() - for tag_name in tag_names: - 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, - ) - continue - - if tag is None: - logger.debug("Excluded tag %r does not exist — skipping", tag_name) - continue - - 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, - ) - continue - - for f in files: - path = _normalise_path(f["path"]) - excluded.add(path) - if f.get("is_directory"): - logger.debug( - "Excluding directory %r (tag %r) — descendants will be hidden", - path, - tag_name, - ) + lock = anyio.Lock() + async with anyio.create_task_group() as tg: + for tag_name in tag_names: + tg.start_soon(_resolve_one_tag, tag_name, webdav, excluded, lock) if excluded: # `len(excluded)` counts directly-tagged entries — descendants of diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index 7220613b..39ce775b 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -392,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 = [] @@ -462,8 +469,7 @@ def configure_webdav_tools(mcp: FastMCP): limit=limit, ) - # Filter out tagged-excluded paths. - excluded = await get_excluded_file_paths(client.webdav) + # 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) @@ -511,10 +517,14 @@ 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 ) - 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) @@ -550,10 +560,14 @@ 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 ) - 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) @@ -588,8 +602,12 @@ def configure_webdav_tools(mcp: FastMCP): 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 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) diff --git a/tests/unit/test_tag_exclusion.py b/tests/unit/test_tag_exclusion.py index fc591480..9e071ded 100644 --- a/tests/unit/test_tag_exclusion.py +++ b/tests/unit/test_tag_exclusion.py @@ -10,6 +10,7 @@ import logging import os from unittest.mock import AsyncMock, patch +import anyio import pytest from nextcloud_mcp_server.config import _reload_config @@ -218,3 +219,82 @@ class TestGetExcludedFilePaths: 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 index 3a229672..c1a116c9 100644 --- a/tests/unit/test_webdav_tools_exclusion.py +++ b/tests/unit/test_webdav_tools_exclusion.py @@ -288,3 +288,96 @@ async def test_find_by_name_filters_excluded( 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() From 35abfb2e3af426ac441ced879faa46256f8769f0 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 6 May 2026 19:33:19 +0200 Subject: [PATCH 4/7] fix(webdav): drop anyio.Lock and add integration tests for tag exclusion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses two points from the latest PR #764 review: 1. The anyio.Lock in get_excluded_file_paths bought nothing under anyio's cooperative multitasking model (single-threaded between awaits, raw set mutations are already safe). _resolve_one_tag now builds a local set of paths and appends it to a shared list — list append between awaits is safe without a lock — and the caller merges via set().union(*results) after the task group completes. This removes the cognitive overhead the reviewer flagged without changing the public API. 2. Adds tests/integration/test_tag_exclusion.py exercising the resolution pipeline end-to-end against a real Nextcloud instance: creates a system tag, tags a real file and a real directory, verifies get_excluded_file_paths resolves both via real PROPFIND + REPORT calls, and verifies is_path_excluded correctly classifies exact matches, descendants of tagged directories, and unrelated paths. Includes the disabled-feature short-circuit case. Cleanup runs in reverse order (untag, delete files); per-run uuid suffix avoids cross-run interference. Co-Authored-By: Claude Opus 4.7 (1M context) --- nextcloud_mcp_server/server/tag_exclusion.py | 43 +++-- tests/integration/test_tag_exclusion.py | 191 +++++++++++++++++++ 2 files changed, 215 insertions(+), 19 deletions(-) create mode 100644 tests/integration/test_tag_exclusion.py diff --git a/nextcloud_mcp_server/server/tag_exclusion.py b/nextcloud_mcp_server/server/tag_exclusion.py index 5abb55bc..65f256e0 100644 --- a/nextcloud_mcp_server/server/tag_exclusion.py +++ b/nextcloud_mcp_server/server/tag_exclusion.py @@ -42,13 +42,15 @@ def get_excluded_tag_names() -> list[str]: async def _resolve_one_tag( tag_name: str, webdav: WebDAVClient, - excluded: set[str], - lock: anyio.Lock, + results: list[set[str]], ) -> None: - """Resolve a single tag's paths and merge into *excluded* under *lock*. + """Resolve a single tag's paths and append them as a set to *results*. - Swallows its own exceptions so a failure for one tag does not abort - the surrounding task group (preserves fail-open per-tag semantics). + Each task writes to a distinct slot in the shared list — append from + cooperative tasks is safe under anyio (single-threaded between + awaits) without an explicit lock. 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) @@ -76,16 +78,17 @@ async def _resolve_one_tag( ) return - async with lock: - for f in files: - path = _normalise_path(f["path"]) - excluded.add(path) - if f.get("is_directory"): - logger.debug( - "Excluding directory %r (tag %r) — descendants will be hidden", - path, - tag_name, - ) + 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]: @@ -96,7 +99,9 @@ async def get_excluded_file_paths(webdav: WebDAVClient) -> set[str]: 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. + 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 @@ -110,12 +115,12 @@ async def get_excluded_file_paths(webdav: WebDAVClient) -> set[str]: if not tag_names: return set() - excluded: set[str] = set() - lock = anyio.Lock() + 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, excluded, lock) + 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. 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() From 2ee4d03e3f5799ef7b85ca0ada8376e0de2992aa Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 6 May 2026 19:53:07 +0200 Subject: [PATCH 5/7] fix(webdav): address PR #764 review round 3 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three important issues raised by the latest review on the get_tag_by_name and get_files_by_tag methods: 1. Add explicit response.raise_for_status() after _make_request in both methods. _make_request already raises HTTPStatusError on non-2xx so the calls are redundant in practice, but keeping them visible at the call site makes the contract self-documenting and prevents a future refactor from silently feeding an error body into ET.fromstring. 2. Replace href_path.replace(webdav_prefix, "/") with a startswith + slice. str.replace strips every occurrence of the prefix; while no real Nextcloud path embeds the prefix mid-string, the fix removes the theoretical exposure and matches the pattern used elsewhere in the file. 3. Add Content-Type: text/xml to the systemtags PROPFIND headers. Other PROPFIND-with-body calls in this file (list_directory line 240, list_attachments line 1041) include it; the systemtags PROPFIND was the only outlier. Same header added to the systemtag REPORT for symmetry. No test changes — the existing get_files_by_tag mock test continues to pass (the mock response yields valid XML so raise_for_status is a no-op, and the user-relative path comparison is unaffected by the prefix-strip swap on a non-adversarial path). Co-Authored-By: Claude Opus 4.7 (1M context) --- nextcloud_mcp_server/client/webdav.py | 23 +++++++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/nextcloud_mcp_server/client/webdav.py b/nextcloud_mcp_server/client/webdav.py index 6911c1bc..2e1a11ed 100644 --- a/nextcloud_mcp_server/client/webdav.py +++ b/nextcloud_mcp_server/client/webdav.py @@ -1135,9 +1135,14 @@ class WebDAVClient(BaseNextcloudClient): response = await self._make_request( "PROPFIND", "/remote.php/dav/systemtags/", - headers={"Depth": "1"}, + headers={"Depth": "1", "Content-Type": "text/xml"}, 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 root = ET.fromstring(response.content) @@ -1218,8 +1223,13 @@ class WebDAVClient(BaseNextcloudClient): response = await self._make_request( "REPORT", f"{self._get_webdav_base_path()}/", + headers={"Content-Type": "text/xml"}, 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 root = ET.fromstring(response.content) @@ -1261,11 +1271,16 @@ class WebDAVClient(BaseNextcloudClient): and resourcetype_elem.find("d:collection", ns) is not None ) - # Decode href path and extract the file path + # 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 From 56f01b3499164b1b8e4d9f2b2e596512f87d9c42 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 6 May 2026 20:24:11 +0200 Subject: [PATCH 6/7] fix(webdav): address PR #764 review round 4 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Guard against malformed PROPFIND responses where tag["id"] is None before calling get_files_by_tag (prevents None dispatch). - Add OCS-APIRequest: true header to get_tag_by_name and get_files_by_tag to match every other PROPFIND/REPORT in the file — fixes a latent reverse-proxy compatibility hazard. - Add test_copy_resource_blocks_excluded_source to mirror the existing move-source coverage; closes the asymmetric test gap. - Add test_skips_tag_with_missing_id covering the new fail-open branch in _resolve_one_tag. - Reword _resolve_one_tag docstring: "distinct slot" was misleading (tasks append rather than pre-allocate). Co-Authored-By: Claude Opus 4.7 (1M context) --- nextcloud_mcp_server/client/webdav.py | 8 ++++-- nextcloud_mcp_server/server/tag_exclusion.py | 19 ++++++++++---- tests/unit/test_tag_exclusion.py | 26 ++++++++++++++++++++ tests/unit/test_webdav_tools_exclusion.py | 17 +++++++++++++ 4 files changed, 63 insertions(+), 7 deletions(-) diff --git a/nextcloud_mcp_server/client/webdav.py b/nextcloud_mcp_server/client/webdav.py index 2e1a11ed..cff493a5 100644 --- a/nextcloud_mcp_server/client/webdav.py +++ b/nextcloud_mcp_server/client/webdav.py @@ -1135,7 +1135,11 @@ class WebDAVClient(BaseNextcloudClient): response = await self._make_request( "PROPFIND", "/remote.php/dav/systemtags/", - headers={"Depth": "1", "Content-Type": "text/xml"}, + headers={ + "Depth": "1", + "Content-Type": "text/xml", + "OCS-APIRequest": "true", + }, content=propfind_body, ) # Redundant after _make_request (which raises on non-2xx) but @@ -1223,7 +1227,7 @@ class WebDAVClient(BaseNextcloudClient): response = await self._make_request( "REPORT", f"{self._get_webdav_base_path()}/", - headers={"Content-Type": "text/xml"}, + headers={"Content-Type": "text/xml", "OCS-APIRequest": "true"}, content=report_body, ) # Redundant after _make_request (which raises on non-2xx) but diff --git a/nextcloud_mcp_server/server/tag_exclusion.py b/nextcloud_mcp_server/server/tag_exclusion.py index 65f256e0..8ced477b 100644 --- a/nextcloud_mcp_server/server/tag_exclusion.py +++ b/nextcloud_mcp_server/server/tag_exclusion.py @@ -46,11 +46,11 @@ async def _resolve_one_tag( ) -> None: """Resolve a single tag's paths and append them as a set to *results*. - Each task writes to a distinct slot in the shared list — append from - cooperative tasks is safe under anyio (single-threaded between - awaits) without an explicit lock. Swallows its own exceptions so a - failure for one tag does not abort the surrounding task group - (preserves fail-open per-tag semantics). + 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) @@ -67,6 +67,15 @@ async def _resolve_one_tag( 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: diff --git a/tests/unit/test_tag_exclusion.py b/tests/unit/test_tag_exclusion.py index 9e071ded..9ed3e318 100644 --- a/tests/unit/test_tag_exclusion.py +++ b/tests/unit/test_tag_exclusion.py @@ -127,6 +127,32 @@ class TestGetExcludedFilePaths: 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), diff --git a/tests/unit/test_webdav_tools_exclusion.py b/tests/unit/test_webdav_tools_exclusion.py index c1a116c9..d2cb6cf9 100644 --- a/tests/unit/test_webdav_tools_exclusion.py +++ b/tests/unit/test_webdav_tools_exclusion.py @@ -194,6 +194,23 @@ async def test_move_resource_blocks_excluded_destination_exact_match( ) +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 ): From 81c190c9c58abd498e63629e173506b8839a0e83 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 6 May 2026 23:40:53 +0200 Subject: [PATCH 7/7] fix(webdav): finish lazy-logging conversion in get_tag_by_name The two debug calls in get_tag_by_name were left as f-strings when the method was migrated to _make_request in round 1. Convert to lazy %-style formatting per repo convention (PR #764 review round 5). Co-Authored-By: Claude Opus 4.7 (1M context) --- nextcloud_mcp_server/client/webdav.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/nextcloud_mcp_server/client/webdav.py b/nextcloud_mcp_server/client/webdav.py index cff493a5..25d004e2 100644 --- a/nextcloud_mcp_server/client/webdav.py +++ b/nextcloud_mcp_server/client/webdav.py @@ -1190,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]]: