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"]