Merge pull request #764 from cbcoutinho/feat/tag-based-file-exclusion

feat(webdav): tag-based file exclusion (#710)
This commit is contained in:
Chris Coutinho
2026-05-06 23:53:59 +02:00
committed by GitHub
10 changed files with 1413 additions and 12 deletions
+1
View File
@@ -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
+81
View File
@@ -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:
+38 -10
View File
@@ -1132,12 +1132,20 @@ class WebDAVClient(BaseNextcloudClient):
</d:prop>
</d:propfind>"""
response = await self._client.request(
response = await self._make_request(
"PROPFIND",
"/remote.php/dav/systemtags/",
headers={"Depth": "1"},
headers={
"Depth": "1",
"Content-Type": "text/xml",
"OCS-APIRequest": "true",
},
content=propfind_body,
)
# Redundant after _make_request (which raises on non-2xx) but
# makes the contract explicit at the call site so a future
# refactor of _make_request cannot silently feed an error body
# into ET.fromstring below.
response.raise_for_status()
# Parse XML response
@@ -1182,10 +1190,10 @@ class WebDAVClient(BaseNextcloudClient):
and user_assignable_elem.text is not None
else True,
}
logger.debug(f"Found tag '{tag_name}' with ID {tag_info['id']}")
logger.debug("Found tag %r with ID %s", tag_name, tag_info["id"])
return tag_info
logger.debug(f"Tag '{tag_name}' not found")
logger.debug("Tag %r not found", tag_name)
return None
async def get_files_by_tag(self, tag_id: int) -> list[dict[str, Any]]:
@@ -1197,7 +1205,9 @@ class WebDAVClient(BaseNextcloudClient):
Returns:
List of file info dictionaries with path, size, content_type, etc.
"""
# Use WebDAV REPORT method with systemtag filter, requesting all properties
# Use WebDAV REPORT method with systemtag filter. resourcetype is
# included so callers can distinguish folders from files (needed for
# recursive exclusion of tagged directories — see issue #710).
report_body = f"""<?xml version="1.0"?>
<oc:filter-files xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns" xmlns:nc="http://nextcloud.org/ns">
<d:prop>
@@ -1207,17 +1217,22 @@ class WebDAVClient(BaseNextcloudClient):
<d:getcontenttype/>
<d:getlastmodified/>
<d:getetag/>
<d:resourcetype/>
</d:prop>
<oc:filter-rules>
<oc:systemtag>{tag_id}</oc:systemtag>
</oc:filter-rules>
</oc:filter-files>"""
response = await self._client.request(
response = await self._make_request(
"REPORT",
f"{self._get_webdav_base_path()}/",
headers={"Content-Type": "text/xml", "OCS-APIRequest": "true"},
content=report_body,
)
# Redundant after _make_request (which raises on non-2xx) but
# makes the contract explicit at the call site — see the same
# rationale in get_tag_by_name.
response.raise_for_status()
# Parse XML response
@@ -1249,15 +1264,27 @@ class WebDAVClient(BaseNextcloudClient):
contenttype_elem = prop.find("d:getcontenttype", ns)
lastmodified_elem = prop.find("d:getlastmodified", ns)
etag_elem = prop.find("d:getetag", ns)
resourcetype_elem = prop.find("d:resourcetype", ns)
if fileid_elem is None or not fileid_elem.text:
continue
# Decode href path and extract the file path
# A resourcetype with a <d:collection/> child indicates a folder.
is_directory = (
resourcetype_elem is not None
and resourcetype_elem.find("d:collection", ns) is not None
)
# Decode href path and extract the user-relative file path.
# str.replace() would strip every occurrence of the prefix,
# so an adversarially-named directory could collide; strip
# only the leading occurrence via startswith + slice.
href_path = unquote(href_elem.text)
# Remove WebDAV prefix to get user-relative path
webdav_prefix = f"/remote.php/dav/files/{self.username}/"
file_path = href_path.replace(webdav_prefix, "/")
if href_path.startswith(webdav_prefix):
file_path = "/" + href_path[len(webdav_prefix) :]
else:
file_path = href_path
# Parse last modified timestamp
last_modified_timestamp = None
@@ -1285,10 +1312,11 @@ class WebDAVClient(BaseNextcloudClient):
else None,
"last_modified_timestamp": last_modified_timestamp,
"etag": etag_elem.text if etag_elem is not None else None,
"is_directory": is_directory,
}
files.append(file_info)
logger.debug(f"Found {len(files)} files with tag ID {tag_id}")
logger.debug("Found %d files with tag ID %s", len(files), tag_id)
return files
async def get_file_info(self, path: str) -> dict[str, Any] | None:
+10
View File
@@ -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
@@ -0,0 +1,166 @@
"""Tag-based file exclusion for MCP file operations (issue #710).
Resolves the configured ``EXCLUDED_TAGS`` to a set of file paths that
should be hidden from WebDAV MCP tools (list, read, search) and rejected
by mutating tools (write, delete, move, copy).
The flow per call:
1. Parse ``EXCLUDED_TAGS`` (comma-separated tag names) from config.
2. For each tag name, resolve to a tag ID via ``get_tag_by_name``.
3. For each tag ID, fetch all tagged file/folder paths via
``get_files_by_tag``.
4. Collect normalised paths into a single ``set[str]``.
Tagging a *folder* excludes the folder itself and every descendant via
prefix match in :func:`is_path_excluded`.
Threat model: this is a defence-in-depth control to prevent accidental
exfiltration via the LLM tool surface. A user controlling the Nextcloud
account whose credentials the server uses can untag files unless the tag
is created with ``user_assignable=false``.
"""
import logging
import anyio
from nextcloud_mcp_server.client.webdav import WebDAVClient
from nextcloud_mcp_server.config import get_settings
logger = logging.getLogger(__name__)
def get_excluded_tag_names() -> list[str]:
"""Return the configured excluded tag names (empty list if disabled)."""
raw = get_settings().excluded_tags
if not raw:
return []
return [t.strip() for t in raw.split(",") if t.strip()]
async def _resolve_one_tag(
tag_name: str,
webdav: WebDAVClient,
results: list[set[str]],
) -> None:
"""Resolve a single tag's paths and append them as a set to *results*.
Each task appends its own set to the shared list; ``list.append`` is
atomic between cooperative yields under anyio (single-threaded
between awaits) so no explicit lock is needed. Swallows its own
exceptions so a failure for one tag does not abort the surrounding
task group (preserves fail-open per-tag semantics).
"""
try:
tag = await webdav.get_tag_by_name(tag_name)
except Exception as e:
logger.warning(
"Tag exclusion lookup failed for tag %r (%s); "
"skipping — files tagged with this tag will be visible",
tag_name,
e,
)
return
if tag is None:
logger.debug("Excluded tag %r does not exist — skipping", tag_name)
return
if tag.get("id") is None:
# Malformed PROPFIND response: <oc:systemtag> entry without
# <oc:id/>. Skip rather than dispatch <oc:systemtag>None</oc:systemtag>.
logger.debug(
"Excluded tag %r has no id in PROPFIND response — skipping",
tag_name,
)
return
try:
files = await webdav.get_files_by_tag(tag["id"])
except Exception as e:
logger.warning(
"Tag exclusion file enumeration failed for tag %r (%s); "
"skipping — files tagged with this tag will be visible",
tag_name,
e,
)
return
paths: set[str] = set()
for f in files:
path = _normalise_path(f["path"])
paths.add(path)
if f.get("is_directory"):
logger.debug(
"Excluding directory %r (tag %r) — descendants will be hidden",
path,
tag_name,
)
results.append(paths)
async def get_excluded_file_paths(webdav: WebDAVClient) -> set[str]:
"""Resolve excluded tags to the set of paths they cover.
Tagged directories are added as their own normalised path; descendants
are blocked via prefix match in :func:`is_path_excluded`.
Per-tag resolution is fanned out via ``anyio.create_task_group`` so
that the 2N network calls (1 PROPFIND + 1 REPORT per tag) run
concurrently rather than serially. No lock is needed: each task
appends its own ``set`` to a shared list, and append is atomic
between awaits under anyio's cooperative single-threaded model.
**Failure mode is fail-open per tag**: if the systemtags endpoint is
unreachable or returns an error for a given tag, that tag is skipped
with a warning rather than propagating the error. Reasoning: the
threat model is preventing *accidental* exfiltration via the LLM tool
surface; a Nextcloud-side outage of the systemtags API should not
take down all WebDAV tools. Operators relying on this for stronger
guarantees should monitor the warning logs.
"""
tag_names = get_excluded_tag_names()
if not tag_names:
return set()
results: list[set[str]] = []
async with anyio.create_task_group() as tg:
for tag_name in tag_names:
tg.start_soon(_resolve_one_tag, tag_name, webdav, results)
excluded: set[str] = set().union(*results)
if excluded:
# `len(excluded)` counts directly-tagged entries — descendants of
# tagged directories are hidden too but resolved at check time.
logger.info(
"Tag-based exclusion resolved to %d directly-tagged path(s) "
"for tags: %s (descendants of tagged directories also hidden)",
len(excluded),
", ".join(tag_names),
)
return excluded
def is_path_excluded(path: str, excluded_paths: set[str]) -> bool:
"""Return True if *path* (or any of its parents) is excluded.
A path is excluded when it matches an entry exactly, or when an
excluded entry is one of its directory ancestors (prefix match on
``<dir>/``).
"""
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("/")
+140 -2
View File
@@ -2,12 +2,17 @@ import base64
import logging
from mcp.server.fastmcp import Context, FastMCP
from mcp.server.fastmcp.exceptions import ToolError
from mcp.types import ToolAnnotations
from nextcloud_mcp_server.auth import require_scopes
from nextcloud_mcp_server.context import get_client
from nextcloud_mcp_server.models import DirectoryListing, FileInfo, SearchFilesResponse
from nextcloud_mcp_server.observability.metrics import instrument_tool
from nextcloud_mcp_server.server.tag_exclusion import (
get_excluded_file_paths,
is_path_excluded,
)
from nextcloud_mcp_server.utils.document_parser import (
is_parseable_document,
parse_document,
@@ -32,6 +37,12 @@ def configure_webdav_tools(mcp: FastMCP):
) -> DirectoryListing:
"""List files and directories in the specified NextCloud path.
When ``EXCLUDED_TAGS`` is configured: raises ``ToolError`` if the
listed path itself is tagged (or sits inside a tagged folder),
and otherwise omits any tagged children from the listing. The
early guard is consistent with the mutating tools and avoids a
round-trip to Nextcloud for a known-excluded path.
Args:
path: Directory path to list (empty string for root directory)
@@ -39,8 +50,21 @@ def configure_webdav_tools(mcp: FastMCP):
DirectoryListing with files, total_count, directories_count, files_count, and total_size
"""
client = await get_client(ctx)
# Resolve once and use for both the path-itself guard and the
# children filter below.
excluded = await get_excluded_file_paths(client.webdav)
if is_path_excluded(path, excluded):
raise ToolError(f"Access denied: {path!r} is tagged with an excluded tag")
items = await client.webdav.list_directory(path)
# Filter out child files/folders carrying an excluded tag.
if excluded:
items = [
i for i in items if not is_path_excluded(i.get("path", ""), excluded)
]
# Convert to FileInfo models
file_infos = [FileInfo(**item) for item in items]
@@ -70,6 +94,9 @@ def configure_webdav_tools(mcp: FastMCP):
async def nc_webdav_read_file(path: str, ctx: Context):
"""Read the content of a file from NextCloud.
Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and the
file (or an ancestor folder) carries an excluded system tag.
Args:
path: Full path to the file to read
@@ -80,13 +107,19 @@ def configure_webdav_tools(mcp: FastMCP):
- Other binary files are base64 encoded
"""
client = await get_client(ctx)
# Block reads of paths carrying an excluded tag.
excluded = await get_excluded_file_paths(client.webdav)
if is_path_excluded(path, excluded):
raise ToolError(f"Access denied: {path!r} is tagged with an excluded tag")
content, content_type = await client.webdav.read_file(path)
# Check if this is a parseable document (PDF, DOCX, etc.)
# is_parseable_document() checks if document processing is enabled
if is_parseable_document(content_type):
try:
logger.info(f"Parsing document '{path}' of type '{content_type}'")
logger.info("Parsing document %r of type %r", path, content_type)
parsed_text, metadata = await parse_document(
content,
content_type,
@@ -103,7 +136,9 @@ def configure_webdav_tools(mcp: FastMCP):
}
except Exception as e:
logger.warning(
f"Failed to parse document '{path}', falling back to base64: {e}"
"Failed to parse document %r, falling back to base64: %s",
path,
e,
)
# Fall through to base64 encoding on parse failure
@@ -144,6 +179,9 @@ def configure_webdav_tools(mcp: FastMCP):
):
"""Write content to a file in NextCloud.
Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and the
target path (or an ancestor folder) carries an excluded system tag.
Args:
path: Full path where to write the file
content: File content (text or base64 for binary)
@@ -154,6 +192,11 @@ def configure_webdav_tools(mcp: FastMCP):
"""
client = await get_client(ctx)
# Block writes to excluded paths.
excluded = await get_excluded_file_paths(client.webdav)
if is_path_excluded(path, excluded):
raise ToolError(f"Access denied: {path!r} is tagged with an excluded tag")
# Handle base64 encoded content
if content_type and "base64" in content_type.lower():
content_bytes = base64.b64decode(content)
@@ -175,6 +218,9 @@ def configure_webdav_tools(mcp: FastMCP):
async def nc_webdav_create_directory(path: str, ctx: Context):
"""Create a directory in NextCloud.
Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and the
target path lies inside a folder carrying an excluded system tag.
Args:
path: Full path of the directory to create
@@ -182,6 +228,15 @@ def configure_webdav_tools(mcp: FastMCP):
Dict with status_code (201 for created, 405 if already exists)
"""
client = await get_client(ctx)
# Block directory creation at or inside excluded paths.
excluded = await get_excluded_file_paths(client.webdav)
if is_path_excluded(path, excluded):
raise ToolError(
f"Access denied: {path!r} is or is inside a path tagged "
"with an excluded tag"
)
return await client.webdav.create_directory(path)
@mcp.tool(
@@ -197,6 +252,9 @@ def configure_webdav_tools(mcp: FastMCP):
async def nc_webdav_delete_resource(path: str, ctx: Context):
"""Delete a file or directory in NextCloud.
Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and the
target path (or an ancestor folder) carries an excluded system tag.
Args:
path: Full path of the file or directory to delete
@@ -204,6 +262,12 @@ def configure_webdav_tools(mcp: FastMCP):
Dict with status_code indicating result (404 if not found)
"""
client = await get_client(ctx)
# Block deletion of excluded files/directories.
excluded = await get_excluded_file_paths(client.webdav)
if is_path_excluded(path, excluded):
raise ToolError(f"Access denied: {path!r} is tagged with an excluded tag")
return await client.webdav.delete_resource(path)
@mcp.tool(
@@ -220,6 +284,10 @@ def configure_webdav_tools(mcp: FastMCP):
):
"""Move or rename a file or directory in NextCloud.
Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and either
the source or destination path (or one of their ancestor folders)
carries an excluded system tag.
Args:
source_path: Full path of the file or directory to move
destination_path: New path for the file or directory
@@ -229,6 +297,19 @@ def configure_webdav_tools(mcp: FastMCP):
Dict with status_code indicating result (404 if source not found, 412 if destination exists and overwrite is False)
"""
client = await get_client(ctx)
# Block moves involving excluded paths on either side.
excluded = await get_excluded_file_paths(client.webdav)
if is_path_excluded(source_path, excluded):
raise ToolError(
f"Access denied: source {source_path!r} is tagged with an excluded tag"
)
if is_path_excluded(destination_path, excluded):
raise ToolError(
f"Access denied: destination {destination_path!r} is or is "
"inside a path tagged with an excluded tag"
)
return await client.webdav.move_resource(
source_path, destination_path, overwrite
)
@@ -247,6 +328,10 @@ def configure_webdav_tools(mcp: FastMCP):
):
"""Copy a file or directory in NextCloud.
Raises ``ToolError`` when ``EXCLUDED_TAGS`` is configured and either
the source or destination path (or one of their ancestor folders)
carries an excluded system tag.
Args:
source_path: Full path of the file or directory to copy
destination_path: Destination path for the copy
@@ -256,6 +341,19 @@ def configure_webdav_tools(mcp: FastMCP):
Dict with status_code indicating result (404 if source not found, 412 if destination exists and overwrite is False)
"""
client = await get_client(ctx)
# Block copies involving excluded paths on either side.
excluded = await get_excluded_file_paths(client.webdav)
if is_path_excluded(source_path, excluded):
raise ToolError(
f"Access denied: source {source_path!r} is tagged with an excluded tag"
)
if is_path_excluded(destination_path, excluded):
raise ToolError(
f"Access denied: destination {destination_path!r} is or is "
"inside a path tagged with an excluded tag"
)
return await client.webdav.copy_resource(
source_path, destination_path, overwrite
)
@@ -294,6 +392,13 @@ def configure_webdav_tools(mcp: FastMCP):
"""
client = await get_client(ctx)
# Resolve once and use for both the scope guard and the result filter.
excluded = await get_excluded_file_paths(client.webdav)
if scope and is_path_excluded(scope, excluded):
raise ToolError(
f"Access denied: scope {scope!r} is tagged with an excluded tag"
)
# Build where conditions based on filters
conditions = []
@@ -364,6 +469,12 @@ def configure_webdav_tools(mcp: FastMCP):
limit=limit,
)
# Filter out tagged-excluded paths from the result set.
if excluded:
results = [
r for r in results if not is_path_excluded(r.get("path", ""), excluded)
]
# Convert to FileInfo models
file_infos = [FileInfo(**result) for result in results]
@@ -406,9 +517,18 @@ def configure_webdav_tools(mcp: FastMCP):
SearchFilesResponse with list of matching files
"""
client = await get_client(ctx)
excluded = await get_excluded_file_paths(client.webdav)
if scope and is_path_excluded(scope, excluded):
raise ToolError(
f"Access denied: scope {scope!r} is tagged with an excluded tag"
)
results = await client.webdav.find_by_name(
pattern=pattern, scope=scope, limit=limit
)
if excluded:
results = [
r for r in results if not is_path_excluded(r.get("path", ""), excluded)
]
file_infos = [FileInfo(**result) for result in results]
return SearchFilesResponse(
results=file_infos,
@@ -440,9 +560,18 @@ def configure_webdav_tools(mcp: FastMCP):
SearchFilesResponse with list of matching files
"""
client = await get_client(ctx)
excluded = await get_excluded_file_paths(client.webdav)
if scope and is_path_excluded(scope, excluded):
raise ToolError(
f"Access denied: scope {scope!r} is tagged with an excluded tag"
)
results = await client.webdav.find_by_type(
mime_type=mime_type, scope=scope, limit=limit
)
if excluded:
results = [
r for r in results if not is_path_excluded(r.get("path", ""), excluded)
]
file_infos = [FileInfo(**result) for result in results]
return SearchFilesResponse(
results=file_infos,
@@ -473,7 +602,16 @@ def configure_webdav_tools(mcp: FastMCP):
SearchFilesResponse with list of favorite files
"""
client = await get_client(ctx)
excluded = await get_excluded_file_paths(client.webdav)
if scope and is_path_excluded(scope, excluded):
raise ToolError(
f"Access denied: scope {scope!r} is tagged with an excluded tag"
)
results = await client.webdav.list_favorites(scope=scope, limit=limit)
if excluded:
results = [
r for r in results if not is_path_excluded(r.get("path", ""), excluded)
]
file_infos = [FileInfo(**result) for result in results]
return SearchFilesResponse(
results=file_infos,
+191
View File
@@ -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()
+60
View File
@@ -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 <d:resourcetype/>.
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"""<?xml version="1.0"?>
<d:multistatus xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns">
<d:response>
<d:href>/remote.php/dav/files/testuser/Secret.txt</d:href>
<d:propstat>
<d:prop>
<oc:fileid>101</oc:fileid>
<d:displayname>Secret.txt</d:displayname>
<d:getcontentlength>42</d:getcontentlength>
<d:getcontenttype>text/plain</d:getcontenttype>
<d:getlastmodified>Wed, 01 Jan 2025 00:00:00 GMT</d:getlastmodified>
<d:getetag>"abc"</d:getetag>
<d:resourcetype/>
</d:prop>
</d:propstat>
</d:response>
<d:response>
<d:href>/remote.php/dav/files/testuser/Private/</d:href>
<d:propstat>
<d:prop>
<oc:fileid>102</oc:fileid>
<d:displayname>Private</d:displayname>
<d:getlastmodified>Wed, 01 Jan 2025 00:00:00 GMT</d:getlastmodified>
<d:getetag>"def"</d:getetag>
<d:resourcetype><d:collection/></d:resourcetype>
</d:prop>
</d:propstat>
</d:response>
</d:multistatus>"""
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 "<d:resourcetype/>" in call_args.kwargs["content"]
assert "<oc:systemtag>42</oc:systemtag>" in call_args.kwargs["content"]
+326
View File
@@ -0,0 +1,326 @@
"""Unit tests for tag-based file exclusion (issue #710).
Tests in :class:`TestGetExcludedTagNames` patch ``os.environ`` and call
``_reload_config()`` to make dynaconf observe the patched value. Cleanup
is handled by the autouse ``_reload_dynaconf_after_test`` fixture in
``tests/unit/conftest.py``, which reloads dynaconf after every test.
"""
import logging
import os
from unittest.mock import AsyncMock, patch
import anyio
import pytest
from nextcloud_mcp_server.config import _reload_config
from nextcloud_mcp_server.server.tag_exclusion import (
_normalise_path,
get_excluded_file_paths,
get_excluded_tag_names,
is_path_excluded,
)
class TestNormalisePath:
@pytest.mark.unit
def test_strips_leading_and_trailing_slash(self):
assert _normalise_path("/foo/bar/") == "foo/bar"
@pytest.mark.unit
def test_unchanged_when_already_clean(self):
assert _normalise_path("foo/bar") == "foo/bar"
@pytest.mark.unit
def test_empty_string(self):
assert _normalise_path("") == ""
class TestIsPathExcluded:
@pytest.mark.unit
def test_empty_set_excludes_nothing(self):
assert is_path_excluded("/anything", set()) is False
@pytest.mark.unit
def test_direct_match(self):
assert is_path_excluded("/Secret.txt", {"Secret.txt"}) is True
@pytest.mark.unit
def test_path_argument_is_normalised(self):
# The path argument is normalised before comparison; the excluded
# set is expected to already contain normalised entries (it always
# is, in practice, because get_excluded_file_paths builds it).
assert is_path_excluded("/Secret.txt/", {"Secret.txt"}) is True
@pytest.mark.unit
def test_descendant_of_excluded_directory(self):
assert is_path_excluded("/Private/notes.md", {"Private"}) is True
assert is_path_excluded("/Private/sub/file.txt", {"Private"}) is True
@pytest.mark.unit
def test_unrelated_path_not_excluded(self):
assert is_path_excluded("/Public/notes.md", {"Private"}) is False
@pytest.mark.unit
def test_shared_prefix_is_not_a_match(self):
# 'foobar' must NOT be excluded just because 'foo' is.
# This is the bug a naive `startswith(exc)` would have.
assert is_path_excluded("/foobar/x", {"foo"}) is False
assert is_path_excluded("/foobar", {"foo"}) is False
@pytest.mark.unit
def test_excluded_path_itself(self):
# The excluded entry itself is excluded (not just its descendants).
assert is_path_excluded("/Private", {"Private"}) is True
class TestGetExcludedTagNames:
@pytest.mark.unit
@patch.dict(os.environ, {"EXCLUDED_TAGS": ""}, clear=False)
def test_empty_returns_empty_list(self):
_reload_config()
assert get_excluded_tag_names() == []
@pytest.mark.unit
@patch.dict(os.environ, {"EXCLUDED_TAGS": "secret"}, clear=False)
def test_single_tag(self):
_reload_config()
assert get_excluded_tag_names() == ["secret"]
@pytest.mark.unit
@patch.dict(os.environ, {"EXCLUDED_TAGS": " a , b , c "}, clear=False)
def test_strips_whitespace_around_each_tag(self):
_reload_config()
assert get_excluded_tag_names() == ["a", "b", "c"]
@pytest.mark.unit
@patch.dict(os.environ, {"EXCLUDED_TAGS": "a,,b,"}, clear=False)
def test_skips_empty_entries(self):
_reload_config()
assert get_excluded_tag_names() == ["a", "b"]
class TestGetExcludedFilePaths:
@pytest.mark.unit
async def test_returns_empty_set_when_feature_disabled(self, mocker):
mocker.patch(
"nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names",
return_value=[],
)
webdav = AsyncMock()
result = await get_excluded_file_paths(webdav)
assert result == set()
webdav.get_tag_by_name.assert_not_called()
@pytest.mark.unit
async def test_skips_unknown_tag(self, mocker):
mocker.patch(
"nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names",
return_value=["does-not-exist"],
)
webdav = AsyncMock()
webdav.get_tag_by_name = AsyncMock(return_value=None)
result = await get_excluded_file_paths(webdav)
assert result == set()
webdav.get_tag_by_name.assert_awaited_once_with("does-not-exist")
webdav.get_files_by_tag.assert_not_called()
@pytest.mark.unit
async def test_skips_tag_with_missing_id(self, mocker):
"""If get_tag_by_name returns a dict with id=None (malformed
PROPFIND response — <oc:systemtag> entry without <oc:id/>), skip
the tag rather than dispatching <oc:systemtag>None</oc:systemtag>
to get_files_by_tag (PR #764 review round 4)."""
mocker.patch(
"nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names",
return_value=["malformed"],
)
webdav = AsyncMock()
webdav.get_tag_by_name = AsyncMock(
return_value={
"id": None,
"name": "malformed",
"userVisible": True,
"userAssignable": True,
}
)
result = await get_excluded_file_paths(webdav)
assert result == set()
webdav.get_tag_by_name.assert_awaited_once_with("malformed")
webdav.get_files_by_tag.assert_not_called()
@pytest.mark.unit
async def test_fail_open_when_tag_lookup_raises(self, mocker, caplog):
"""If get_tag_by_name raises (e.g. 5xx from systemtags endpoint),
the offending tag is skipped with a warning rather than the error
propagating and disabling all WebDAV tools (PR review #764)."""
mocker.patch(
"nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names",
return_value=["broken", "ok"],
)
webdav = AsyncMock()
webdav.get_tag_by_name = AsyncMock(
side_effect=[
RuntimeError("upstream 503"),
{"id": 7, "name": "ok"},
]
)
webdav.get_files_by_tag = AsyncMock(
return_value=[{"path": "/ok.txt", "is_directory": False}]
)
caplog.set_level(
logging.WARNING, logger="nextcloud_mcp_server.server.tag_exclusion"
)
result = await get_excluded_file_paths(webdav)
assert result == {"ok.txt"}
assert "Tag exclusion lookup failed" in caplog.text
assert "broken" in caplog.text
@pytest.mark.unit
async def test_fail_open_when_file_enumeration_raises(self, mocker, caplog):
"""If get_files_by_tag raises (e.g. REPORT timeout), the
offending tag is skipped with a warning. Other tags are still
resolved (PR review #764)."""
mocker.patch(
"nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names",
return_value=["broken", "ok"],
)
webdav = AsyncMock()
webdav.get_tag_by_name = AsyncMock(
side_effect=[
{"id": 1, "name": "broken"},
{"id": 2, "name": "ok"},
]
)
webdav.get_files_by_tag = AsyncMock(
side_effect=[
RuntimeError("REPORT timeout"),
[{"path": "/ok.txt", "is_directory": False}],
]
)
caplog.set_level(
logging.WARNING, logger="nextcloud_mcp_server.server.tag_exclusion"
)
result = await get_excluded_file_paths(webdav)
assert result == {"ok.txt"}
assert "Tag exclusion file enumeration failed" in caplog.text
assert "broken" in caplog.text
@pytest.mark.unit
async def test_collects_paths_from_multiple_tags(self, mocker):
mocker.patch(
"nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names",
return_value=["secret", "no-ai"],
)
webdav = AsyncMock()
webdav.get_tag_by_name = AsyncMock(
side_effect=[
{"id": 1, "name": "secret"},
{"id": 2, "name": "no-ai"},
]
)
webdav.get_files_by_tag = AsyncMock(
side_effect=[
[
{"path": "/Secret.txt", "is_directory": False},
{"path": "/Private/", "is_directory": True},
],
[
# Same dir under a second tag — set dedupes it.
{"path": "/Private", "is_directory": True},
{"path": "/Other/notes.md", "is_directory": False},
],
]
)
result = await get_excluded_file_paths(webdav)
assert result == {"Secret.txt", "Private", "Other/notes.md"}
assert webdav.get_files_by_tag.await_count == 2
@pytest.mark.unit
async def test_resolves_tags_concurrently(self, mocker):
"""Per-tag resolution must run under a task group so the 2N
network calls overlap (PR review #764). The barrier event below
only completes if all three tags' ``get_tag_by_name`` invocations
are in flight simultaneously — sequential resolution would block
the first task forever and trip ``fail_after``.
"""
tags = ["a", "b", "c"]
mocker.patch(
"nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names",
return_value=tags,
)
all_started = anyio.Event()
started_count = 0
async def fake_get_tag(tag_name: str):
nonlocal started_count
started_count += 1
if started_count == len(tags):
all_started.set()
with anyio.fail_after(2.0):
await all_started.wait()
return {"id": ord(tag_name), "name": tag_name}
async def fake_get_files(tag_id: int):
return [{"path": f"/tag-{tag_id}.txt", "is_directory": False}]
webdav = AsyncMock()
webdav.get_tag_by_name = AsyncMock(side_effect=fake_get_tag)
webdav.get_files_by_tag = AsyncMock(side_effect=fake_get_files)
with anyio.fail_after(5.0):
result = await get_excluded_file_paths(webdav)
assert result == {f"tag-{ord(t)}.txt" for t in tags}
@pytest.mark.unit
async def test_fail_open_under_task_group(self, mocker, caplog):
"""One tag failing must not abort sibling tasks in the task group
(PR review #764). Uses callable side_effects keyed by tag name so
the assertion is independent of the order in which the task group
schedules the per-tag coroutines.
"""
mocker.patch(
"nextcloud_mcp_server.server.tag_exclusion.get_excluded_tag_names",
return_value=["good-1", "broken", "good-2"],
)
async def fake_get_tag(tag_name: str):
if tag_name == "broken":
raise RuntimeError("upstream 503")
return {"id": hash(tag_name) & 0xFFFF, "name": tag_name}
async def fake_get_files(tag_id: int):
# Map tag_id back to a deterministic path.
return [{"path": f"/tagged-by-{tag_id}.txt", "is_directory": False}]
webdav = AsyncMock()
webdav.get_tag_by_name = AsyncMock(side_effect=fake_get_tag)
webdav.get_files_by_tag = AsyncMock(side_effect=fake_get_files)
caplog.set_level(
logging.WARNING, logger="nextcloud_mcp_server.server.tag_exclusion"
)
result = await get_excluded_file_paths(webdav)
# Both healthy tags must contribute one path each; the broken
# tag is silently skipped.
good_1_id = hash("good-1") & 0xFFFF
good_2_id = hash("good-2") & 0xFFFF
assert result == {
f"tagged-by-{good_1_id}.txt",
f"tagged-by-{good_2_id}.txt",
}
assert "Tag exclusion lookup failed" in caplog.text
assert "broken" in caplog.text
+400
View File
@@ -0,0 +1,400 @@
"""Server-layer regression tests for tag-based file exclusion (issue #710).
These tests register the WebDAV tools on a fresh ``FastMCP`` instance and
invoke each tool's underlying function directly via the tool registry.
Their purpose is **not** to re-test the path-matching logic (covered in
``test_tag_exclusion.py``) but to catch wiring regressions: that each
tool actually consults ``get_excluded_file_paths`` / ``is_path_excluded``
at the right point and raises / filters as expected.
The decorators on each tool (``@require_scopes``, ``@instrument_tool``)
are transparent under our mocked ``Context`` (no ``access_token`` set →
BasicAuth pass-through path).
"""
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.exceptions import ToolError
from nextcloud_mcp_server.server.webdav import configure_webdav_tools
pytestmark = pytest.mark.unit
@pytest.fixture
def webdav_tools() -> dict:
"""Register the WebDAV tools on a fresh FastMCP and return them by name."""
mcp = FastMCP(name="test-webdav-tools")
configure_webdav_tools(mcp)
return {t.name: t for t in mcp._tool_manager.list_tools()}
def _mock_ctx(client) -> SimpleNamespace:
"""Build a minimal Context-shaped object for the tool decorators.
Setting ``request_context.access_token = None`` causes ``require_scopes``
to take the BasicAuth pass-through branch (see scope_authorization.py).
"""
ctx = SimpleNamespace()
ctx.request_context = SimpleNamespace(access_token=None)
ctx._client = client # only used by tools that fetch via get_client(ctx)
return ctx
@pytest.fixture
def patch_get_client(mocker):
"""Replace ``get_client`` in the webdav server module with a mock."""
def _install(client):
async def fake_get_client(ctx):
return client
mocker.patch(
"nextcloud_mcp_server.server.webdav.get_client",
side_effect=fake_get_client,
)
return _install
@pytest.fixture
def patch_excluded(mocker):
"""Replace ``get_excluded_file_paths`` with a fixed return value."""
def _install(excluded: set[str]):
async def fake(*_, **__):
return excluded
mocker.patch(
"nextcloud_mcp_server.server.webdav.get_excluded_file_paths",
side_effect=fake,
)
return _install
@pytest.fixture
def fake_client():
"""A NextcloudClient-shaped mock with an AsyncMock webdav attribute."""
client = SimpleNamespace()
client.webdav = AsyncMock()
return client
# ── Read / mutate guards ────────────────────────────────────────────────
async def test_read_file_raises_when_path_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Secret.txt"})
fn = webdav_tools["nc_webdav_read_file"].fn
with pytest.raises(ToolError, match="excluded tag"):
await fn(path="/Secret.txt", ctx=_mock_ctx(fake_client))
fake_client.webdav.read_file.assert_not_called()
async def test_read_file_passes_through_when_not_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Secret.txt"})
fake_client.webdav.read_file = AsyncMock(return_value=(b"hello", "text/plain"))
fn = webdav_tools["nc_webdav_read_file"].fn
result = await fn(path="/Public/notes.md", ctx=_mock_ctx(fake_client))
assert result["content"] == "hello"
fake_client.webdav.read_file.assert_awaited_once_with("/Public/notes.md")
async def test_write_file_raises_when_path_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Private"})
fn = webdav_tools["nc_webdav_write_file"].fn
with pytest.raises(ToolError, match="excluded tag"):
await fn(
path="/Private/note.md",
content="hi",
ctx=_mock_ctx(fake_client),
)
fake_client.webdav.write_file.assert_not_called()
async def test_delete_resource_raises_when_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Secret.txt"})
fn = webdav_tools["nc_webdav_delete_resource"].fn
with pytest.raises(ToolError, match="excluded tag"):
await fn(path="/Secret.txt", ctx=_mock_ctx(fake_client))
fake_client.webdav.delete_resource.assert_not_called()
async def test_create_directory_raises_when_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Private"})
fn = webdav_tools["nc_webdav_create_directory"].fn
with pytest.raises(ToolError, match="is or is inside"):
await fn(path="/Private/sub", ctx=_mock_ctx(fake_client))
fake_client.webdav.create_directory.assert_not_called()
async def test_move_resource_blocks_excluded_source(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Secret.txt"})
fn = webdav_tools["nc_webdav_move_resource"].fn
with pytest.raises(ToolError, match="source"):
await fn(
source_path="/Secret.txt",
destination_path="/Public/x.txt",
ctx=_mock_ctx(fake_client),
)
fake_client.webdav.move_resource.assert_not_called()
async def test_move_resource_blocks_excluded_destination_exact_match(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
"""Destination check must trip on an *exact* match, not just a prefix.
Regression guard for review #764: previously the message said "is
inside" but is_path_excluded also matches exact paths.
"""
patch_get_client(fake_client)
patch_excluded({"Private"})
fn = webdav_tools["nc_webdav_move_resource"].fn
with pytest.raises(ToolError, match="is or is inside"):
await fn(
source_path="/Public/x.txt",
destination_path="/Private",
ctx=_mock_ctx(fake_client),
)
async def test_copy_resource_blocks_excluded_source(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Secret.txt"})
fn = webdav_tools["nc_webdav_copy_resource"].fn
with pytest.raises(ToolError, match="source"):
await fn(
source_path="/Secret.txt",
destination_path="/Public/copy.txt",
ctx=_mock_ctx(fake_client),
)
fake_client.webdav.copy_resource.assert_not_called()
async def test_copy_resource_blocks_excluded_destination_descendant(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Private"})
fn = webdav_tools["nc_webdav_copy_resource"].fn
with pytest.raises(ToolError, match="is or is inside"):
await fn(
source_path="/Public/x.txt",
destination_path="/Private/copy.txt",
ctx=_mock_ctx(fake_client),
)
# ── Listing / search filtering ──────────────────────────────────────────
async def test_list_directory_filters_excluded_children(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Public/Secret.txt"})
fake_client.webdav.list_directory = AsyncMock(
return_value=[
{
"path": "/Public/Secret.txt",
"name": "Secret.txt",
"is_directory": False,
},
{
"path": "/Public/visible.md",
"name": "visible.md",
"is_directory": False,
},
]
)
fn = webdav_tools["nc_webdav_list_directory"].fn
result = await fn(path="/Public", ctx=_mock_ctx(fake_client))
assert [f.path for f in result.files] == ["/Public/visible.md"]
async def test_list_directory_raises_when_listed_path_itself_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
"""The early guard prevents the round-trip to Nextcloud and signals
the access denial, instead of silently returning an empty listing
(review #764)."""
patch_get_client(fake_client)
patch_excluded({"Private"})
fn = webdav_tools["nc_webdav_list_directory"].fn
with pytest.raises(ToolError, match="excluded tag"):
await fn(path="/Private", ctx=_mock_ctx(fake_client))
fake_client.webdav.list_directory.assert_not_called()
async def test_search_files_filters_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Secret.txt"})
fake_client.webdav.search_files = AsyncMock(
return_value=[
{"path": "/Secret.txt", "name": "Secret.txt", "is_directory": False},
{"path": "/notes.md", "name": "notes.md", "is_directory": False},
]
)
fn = webdav_tools["nc_webdav_search_files"].fn
result = await fn(ctx=_mock_ctx(fake_client), name_pattern="%.%")
assert [r.path for r in result.results] == ["/notes.md"]
async def test_find_by_name_filters_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Secret.txt"})
fake_client.webdav.find_by_name = AsyncMock(
return_value=[
{"path": "/Secret.txt", "name": "Secret.txt", "is_directory": False},
{"path": "/visible.txt", "name": "visible.txt", "is_directory": False},
]
)
fn = webdav_tools["nc_webdav_find_by_name"].fn
result = await fn(pattern="%.txt", ctx=_mock_ctx(fake_client))
assert [r.path for r in result.results] == ["/visible.txt"]
async def test_find_by_type_filters_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Secret.txt"})
fake_client.webdav.find_by_type = AsyncMock(
return_value=[
{"path": "/Secret.txt", "name": "Secret.txt", "is_directory": False},
{"path": "/visible.txt", "name": "visible.txt", "is_directory": False},
]
)
fn = webdav_tools["nc_webdav_find_by_type"].fn
result = await fn(mime_type="text/plain", ctx=_mock_ctx(fake_client))
assert [r.path for r in result.results] == ["/visible.txt"]
async def test_list_favorites_filters_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Secret.txt"})
fake_client.webdav.list_favorites = AsyncMock(
return_value=[
{"path": "/Secret.txt", "name": "Secret.txt", "is_directory": False},
{"path": "/visible.txt", "name": "visible.txt", "is_directory": False},
]
)
fn = webdav_tools["nc_webdav_list_favorites"].fn
result = await fn(ctx=_mock_ctx(fake_client))
assert [r.path for r in result.results] == ["/visible.txt"]
# ── Search-tool scope guards (review #764) ──────────────────────────────
async def test_search_files_raises_when_scope_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
"""Mirror the ``list_directory`` early guard so the four search tools
cannot silently return an empty result for an excluded ``scope``."""
patch_get_client(fake_client)
patch_excluded({"Private"})
fn = webdav_tools["nc_webdav_search_files"].fn
with pytest.raises(ToolError, match="excluded tag"):
await fn(ctx=_mock_ctx(fake_client), scope="/Private", name_pattern="%.txt")
fake_client.webdav.search_files.assert_not_called()
async def test_find_by_name_raises_when_scope_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Private"})
fn = webdav_tools["nc_webdav_find_by_name"].fn
with pytest.raises(ToolError, match="excluded tag"):
await fn(pattern="%.txt", scope="/Private", ctx=_mock_ctx(fake_client))
fake_client.webdav.find_by_name.assert_not_called()
async def test_find_by_type_raises_when_scope_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Private"})
fn = webdav_tools["nc_webdav_find_by_type"].fn
with pytest.raises(ToolError, match="excluded tag"):
await fn(mime_type="text/plain", scope="/Private", ctx=_mock_ctx(fake_client))
fake_client.webdav.find_by_type.assert_not_called()
async def test_list_favorites_raises_when_scope_excluded(
webdav_tools, fake_client, patch_get_client, patch_excluded
):
patch_get_client(fake_client)
patch_excluded({"Private"})
fn = webdav_tools["nc_webdav_list_favorites"].fn
with pytest.raises(ToolError, match="excluded tag"):
await fn(ctx=_mock_ctx(fake_client), scope="/Private")
fake_client.webdav.list_favorites.assert_not_called()