From 2f6432287aa2ba6989bf48ccd4a8e73ca0432594 Mon Sep 17 00:00:00 2001 From: Jos Poortvliet Date: Sat, 25 Apr 2026 18:44:35 +0200 Subject: [PATCH 01/12] feat(webdav): add archive member tools and temp-download with clear guidance Adds four new MCP tools for working with non-text files stored in Nextcloud, and rewrites the nc_webdav_read_file docstring so Claude never wastes tokens trying to interpret base64-encoded binary blobs. New tools: nc_webdav_list_archive_members, nc_webdav_read_archive_member, nc_webdav_download_to_temp, nc_webdav_cleanup_temp. nc_webdav_list_archive_members(path): lists files inside any ZIP-based archive (ODS, ODT, ODP, DOCX, XLSX, PPTX, EPUB, ZIP, JAR) using stdlib zipfile. nc_webdav_read_archive_member(path, member_path): downloads the archive and extracts exactly one member. XML members (content.xml in ODS etc.) are returned as UTF-8 text. The full archive never enters the context window. nc_webdav_download_to_temp(path): downloads any file to a local temp path and returns that path, for use with shell tools (ffmpeg, pdftotext, exiftool ...). Docstring makes it explicit this requires local shell/Bash access to be useful. nc_webdav_cleanup_temp(local_path): removes a temp file. Only paths tracked in _temp_registry can be removed (no arbitrary path deletion). Also rewrites the nc_webdav_read_file docstring with explicit use/avoid sections so Claude picks the right tool without trial-and-error on binary files. Co-Authored-By: Claude Sonnet 4.6 --- nextcloud_mcp_server/server/webdav.py | 355 +++++++++++++++++++++++++- 1 file changed, 351 insertions(+), 4 deletions(-) diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index c89f38968..ca26b59bb 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -1,5 +1,10 @@ import base64 +import io import logging +import mimetypes +import os +import tempfile +import zipfile from mcp.server.fastmcp import Context, FastMCP from mcp.types import ToolAnnotations @@ -15,6 +20,33 @@ logger = logging.getLogger(__name__) +# Registry of local temp paths created by nc_webdav_download_to_temp. +# Used to prevent nc_webdav_cleanup_temp from deleting arbitrary paths. +# Plain set is safe: asyncio is single-threaded and GIL protects simple ops. +_temp_registry: set[str] = set() + +# MIME types whose files are ZIP archives and can be introspected with zipfile. +_ZIP_MIME_TYPES: frozenset[str] = frozenset( + { + # OpenDocument formats + "application/vnd.oasis.opendocument.spreadsheet", # .ods + "application/vnd.oasis.opendocument.text", # .odt + "application/vnd.oasis.opendocument.presentation", # .odp + "application/vnd.oasis.opendocument.graphics", # .odg + "application/vnd.oasis.opendocument.formula", # .odf + "application/vnd.oasis.opendocument.database", # .odb + # OOXML formats + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", # .docx + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", # .xlsx + "application/vnd.openxmlformats-officedocument.presentationml.presentation", # .pptx + # Generic ZIP + "application/zip", + "application/x-zip-compressed", + "application/java-archive", # .jar + "application/epub+zip", # .epub + } +) + def configure_webdav_tools(mcp: FastMCP): # WebDAV file system tools @@ -68,16 +100,43 @@ async def nc_webdav_list_directory( @require_scopes("files.read") @instrument_tool async def nc_webdav_read_file(path: str, ctx: Context): - """Read the content of a file from NextCloud. + """Read a file from Nextcloud and return its content inline. + + IMPORTANT — choose the right tool for the file type: + + ✅ Use THIS tool for: + - Plain text files (Markdown, CSV, JSON, XML, YAML, source code, logs) + that fit in the context window (roughly < 1 MB of text). + - PDFs, when the document-processing feature is enabled server-side + (text is extracted automatically). + + ❌ Do NOT use this tool for: + - ZIP-based office formats (ODS, ODT, ODP, DOCX, XLSX, PPTX, EPUB …). + The raw archive bytes are meaningless in context. Use + nc_webdav_list_archive_members + nc_webdav_read_archive_member instead. + - Images (PNG, JPEG, GIF, TIFF, HEIC, RAW …). + Binary image data cannot be interpreted here. Use + nc_webdav_download_to_temp and process locally with tools such as + `convert`, `exiftool`, or `ffmpeg` — only if you have local shell access. + - Audio or video files (MP4, MKV, MP3, FLAC …). + Use nc_webdav_download_to_temp + `ffmpeg`/`ffprobe` if you have shell + access; otherwise these files cannot be processed via MCP. + - Any binary file larger than ~1 MB. The file will be returned as a + base64 blob that wastes the entire context without yielding useful + information. Check the file size with nc_webdav_list_directory first. + + Fallback behaviour (binary files not covered above): + The raw bytes are base64-encoded and returned. This is rarely useful + — prefer the dedicated tools described above. Args: path: Full path to the file to read Returns: Dict with path, content, content_type, size, and optional parsing metadata - - Text files are decoded to UTF-8 - - Documents (PDF, DOCX, etc.) are parsed and text is extracted - - Other binary files are base64 encoded + - Text files: content decoded to UTF-8 string + - PDFs (doc-processing enabled): extracted plain text + - Other binary files: content base64-encoded (avoid for large files) """ client = await get_client(ctx) content, content_type = await client.webdav.read_file(path) @@ -481,3 +540,291 @@ async def nc_webdav_list_favorites( scope=scope, filters_applied={"only_favorites": True}, ) + + @mcp.tool( + title="List Archive Members", + annotations=ToolAnnotations( + readOnlyHint=True, + openWorldHint=True, + ), + ) + @require_scopes("files.read") + @instrument_tool + async def nc_webdav_list_archive_members(path: str, ctx: Context) -> dict: + """List the files contained inside a ZIP-based archive stored in Nextcloud. + + Supported archive formats (all are ZIP-based): + Office: ODS, ODT, ODP, ODG, DOCX, XLSX, PPTX + Other: ZIP, JAR, EPUB + + Use this tool first to discover the internal structure of an archive, + then call nc_webdav_read_archive_member to read a specific member. + + Typical ODF layout: + mimetype — identifies the ODF sub-type + content.xml — document content + styles.xml — formatting styles + meta.xml — document metadata + settings.xml — application settings + META-INF/manifest.xml — archive manifest + + Args: + path: Nextcloud path to the archive file (e.g. "Documents/report.ods") + + Returns: + Dict with path, content_type, archive_size, member_count, and a + members list. Each member has: name, size (uncompressed), + compressed_size, is_dir. + + Raises: + ValueError: if the file is not a valid ZIP archive + """ + client = await get_client(ctx) + content, content_type = await client.webdav.read_file(path) + + try: + with zipfile.ZipFile(io.BytesIO(content)) as zf: + members = [ + { + "name": info.filename, + "size": info.file_size, + "compressed_size": info.compress_size, + "is_dir": info.is_dir(), + } + for info in zf.infolist() + ] + except zipfile.BadZipFile as exc: + raise ValueError( + f"'{path}' (content-type: {content_type}) is not a valid ZIP archive. " + f"For plain text files use nc_webdav_read_file; for images/video/audio " + f"use nc_webdav_download_to_temp." + ) from exc + + return { + "path": path, + "content_type": content_type, + "archive_size": len(content), + "member_count": len(members), + "members": members, + } + + @mcp.tool( + title="Read Archive Member", + annotations=ToolAnnotations( + readOnlyHint=True, + openWorldHint=True, + ), + ) + @require_scopes("files.read") + @instrument_tool + async def nc_webdav_read_archive_member( + path: str, member_path: str, ctx: Context + ) -> dict: + """Extract and return a single file from inside a ZIP-based archive in Nextcloud. + + The whole archive is downloaded, but only the requested member is + returned — it never appears in the context as a base64 blob. + + Supported archive formats: ODS, ODT, ODP, ODG, DOCX, XLSX, PPTX, + ZIP, JAR, EPUB (anything that Python's zipfile module can open). + + Typical use-cases: + - Read content.xml from an ODS/ODT/ODP to get document content + - Read word/document.xml from a DOCX + - Read xl/worksheets/sheet1.xml from an XLSX + - Inspect META-INF/manifest.xml to understand archive structure + + Use nc_webdav_list_archive_members first to discover available member paths. + + Args: + path: Nextcloud path to the archive (e.g. "Documents/budget.ods") + member_path: Path of the member inside the archive + (e.g. "content.xml" or "META-INF/manifest.xml") + + Returns: + Dict with archive_path, member_path, content, content_type, size. + Text members (XML, HTML, JSON, plain text …) are returned as UTF-8 + strings. Binary members are base64-encoded with encoding="base64". + + Raises: + ValueError: if the archive is not valid ZIP, or the member is not found + """ + client = await get_client(ctx) + content, content_type = await client.webdav.read_file(path) + + try: + with zipfile.ZipFile(io.BytesIO(content)) as zf: + try: + member_bytes = zf.read(member_path) + except KeyError as exc: + available = [i.filename for i in zf.infolist() if not i.is_dir()] + raise ValueError( + f"Member '{member_path}' not found in '{path}'. " + f"Available files: {available[:30]}" + + (" (truncated)" if len(available) > 30 else "") + ) from exc + except zipfile.BadZipFile as exc: + raise ValueError(f"'{path}' is not a valid ZIP archive.") from exc + + member_mime = mimetypes.guess_type(member_path)[0] or "application/octet-stream" + + # Return text members decoded; XML files are always text even without + # an explicit text/* MIME type. + is_text = ( + member_mime.startswith("text/") + or member_mime + in { + "application/xml", + "application/json", + "application/javascript", + } + or member_path.endswith((".xml", ".json", ".html", ".css", ".js", ".svg")) + ) + + if is_text: + try: + return { + "archive_path": path, + "member_path": member_path, + "content": member_bytes.decode("utf-8"), + "content_type": member_mime, + "size": len(member_bytes), + } + except UnicodeDecodeError: + pass # fall through to base64 + + return { + "archive_path": path, + "member_path": member_path, + "content": base64.b64encode(member_bytes).decode("ascii"), + "content_type": member_mime, + "size": len(member_bytes), + "encoding": "base64", + } + + @mcp.tool( + title="Download File to Temp", + annotations=ToolAnnotations( + readOnlyHint=True, + openWorldHint=True, + ), + ) + @require_scopes("files.read") + @instrument_tool + async def nc_webdav_download_to_temp(path: str, ctx: Context) -> dict: + """Download a Nextcloud file to a local temporary path and return that path. + + IMPORTANT — this tool is only useful when you have access to local shell + tools (e.g. Claude Code's Bash tool). In Claude Desktop without shell + access the returned path cannot be acted upon and you should not call + this tool. + + Use this tool for file types that require native processing: + Images — then use: convert, exiftool, ffmpeg, identify + Video — then use: ffmpeg, ffprobe, mediainfo + Audio — then use: ffmpeg, ffprobe, sox + PDFs — then use: pdftotext, pdfinfo, pdftk, mutool + Archives — for formats NOT supported by nc_webdav_list_archive_members + (e.g. .tar.gz, .7z, .rar): use tar, 7z, unrar + Any large binary that requires local tooling + + For ZIP-based office formats (ODS, DOCX, XLSX …) prefer + nc_webdav_list_archive_members + nc_webdav_read_archive_member — + they avoid creating temp files entirely. + + Cleanup: always call nc_webdav_cleanup_temp when finished to free disk + space. The temp file is also removed when the MCP server process exits. + + Args: + path: Nextcloud path to the file (e.g. "Videos/holiday.mp4") + + Returns: + Dict with: + local_path — absolute path on the local filesystem + original_path — original Nextcloud path + filename — basename of the original file + content_type — MIME type reported by Nextcloud + size — file size in bytes + """ + client = await get_client(ctx) + content, content_type = await client.webdav.read_file(path) + + filename = os.path.basename(path.rstrip("/")) + _root, suffix = os.path.splitext(filename) + + fd, local_path = tempfile.mkstemp(suffix=suffix, prefix="nc_download_") + try: + with os.fdopen(fd, "wb") as fh: + fh.write(content) + except Exception: + try: + os.unlink(local_path) + except OSError: + pass + raise + + _temp_registry.add(local_path) + logger.debug( + "Downloaded '%s' to temp path '%s' (%d bytes)", + path, + local_path, + len(content), + ) + + return { + "local_path": local_path, + "original_path": path, + "filename": filename, + "content_type": content_type, + "size": len(content), + } + + @mcp.tool( + title="Remove Temp File", + annotations=ToolAnnotations( + destructiveHint=True, + idempotentHint=True, + openWorldHint=False, # operates on local filesystem only + ), + ) + @require_scopes("files.read") + @instrument_tool + async def nc_webdav_cleanup_temp(local_path: str, ctx: Context) -> dict: + """Remove a temporary file created by nc_webdav_download_to_temp. + + Only paths that were created by nc_webdav_download_to_temp in this + server session can be removed — arbitrary filesystem paths are rejected. + + Call this when you are done processing a downloaded file to free + disk space. + + Args: + local_path: The local_path value returned by nc_webdav_download_to_temp + + Returns: + Dict with status ("ok" or "error") and the local_path. + """ + if local_path not in _temp_registry: + return { + "status": "error", + "local_path": local_path, + "message": ( + "Path was not created by nc_webdav_download_to_temp in this " + "session, or has already been cleaned up." + ), + } + + _temp_registry.discard(local_path) + + try: + os.unlink(local_path) + logger.debug("Removed temp file '%s'", local_path) + return {"status": "ok", "local_path": local_path} + except FileNotFoundError: + return { + "status": "ok", + "local_path": local_path, + "note": "File was already removed.", + } + except OSError as exc: + return {"status": "error", "local_path": local_path, "message": str(exc)} From 8c78ba3e1074cee899033479f2dade64e11854d1 Mon Sep 17 00:00:00 2001 From: Jos Poortvliet Date: Sat, 25 Apr 2026 19:17:43 +0200 Subject: [PATCH 02/12] fix(webdav): address review feedback on archive and temp-download tools MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Remove _ZIP_MIME_TYPES constant (was unused dead code; BadZipFile already handles invalid formats gracefully) - Add atexit handler _cleanup_temp_files_on_exit() so temp files registered in _temp_registry are actually removed on process exit - Fix _temp_registry.discard() ordering in nc_webdav_cleanup_temp: only discard after a successful os.unlink() or FileNotFoundError, so a failed unlink (OSError) leaves the path retryable - Change idempotentHint=True → False on nc_webdav_cleanup_temp; a second call returns an error (path no longer in registry), which is not idempotent behaviour - Clarify nc_webdav_read_file docstring: office formats CAN be parsed inline when ENABLE_DOCUMENT_PROCESSING is on and a processor supports the type (e.g. Unstructured for DOCX); archive tools are the fallback when doc-processing is disabled or unsupported for the type - Add unit tests for all new logic: atexit cleanup, zipfile member listing/reading/error paths, and temp-registry enforcement Co-Authored-By: Claude Sonnet 4.6 --- nextcloud_mcp_server/server/webdav.py | 51 +++--- tests/unit/test_webdav_archive_tools.py | 204 ++++++++++++++++++++++++ 2 files changed, 229 insertions(+), 26 deletions(-) create mode 100644 tests/unit/test_webdav_archive_tools.py diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index ca26b59bb..66885997c 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -1,3 +1,4 @@ +import atexit import base64 import io import logging @@ -25,27 +26,18 @@ # Plain set is safe: asyncio is single-threaded and GIL protects simple ops. _temp_registry: set[str] = set() -# MIME types whose files are ZIP archives and can be introspected with zipfile. -_ZIP_MIME_TYPES: frozenset[str] = frozenset( - { - # OpenDocument formats - "application/vnd.oasis.opendocument.spreadsheet", # .ods - "application/vnd.oasis.opendocument.text", # .odt - "application/vnd.oasis.opendocument.presentation", # .odp - "application/vnd.oasis.opendocument.graphics", # .odg - "application/vnd.oasis.opendocument.formula", # .odf - "application/vnd.oasis.opendocument.database", # .odb - # OOXML formats - "application/vnd.openxmlformats-officedocument.wordprocessingml.document", # .docx - "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", # .xlsx - "application/vnd.openxmlformats-officedocument.presentationml.presentation", # .pptx - # Generic ZIP - "application/zip", - "application/x-zip-compressed", - "application/java-archive", # .jar - "application/epub+zip", # .epub - } -) + +def _cleanup_temp_files_on_exit() -> None: + """Remove all temp files registered by nc_webdav_download_to_temp on process exit.""" + for path in list(_temp_registry): + try: + os.unlink(path) + logger.debug("atexit: removed temp file '%s'", path) + except OSError: + pass + + +atexit.register(_cleanup_temp_files_on_exit) def configure_webdav_tools(mcp: FastMCP): @@ -112,7 +104,11 @@ async def nc_webdav_read_file(path: str, ctx: Context): ❌ Do NOT use this tool for: - ZIP-based office formats (ODS, ODT, ODP, DOCX, XLSX, PPTX, EPUB …). - The raw archive bytes are meaningless in context. Use + If server-side document processing is enabled (ENABLE_DOCUMENT_PROCESSING=true) + and a processor supports the type (e.g. Unstructured handles DOCX/XLSX), + text is extracted automatically — check the server configuration. + When doc-processing is disabled or unsupported for the type, the raw + archive bytes are meaningless in context; use nc_webdav_list_archive_members + nc_webdav_read_archive_member instead. - Images (PNG, JPEG, GIF, TIFF, HEIC, RAW …). Binary image data cannot be interpreted here. Use @@ -733,7 +729,8 @@ async def nc_webdav_download_to_temp(path: str, ctx: Context) -> dict: they avoid creating temp files entirely. Cleanup: always call nc_webdav_cleanup_temp when finished to free disk - space. The temp file is also removed when the MCP server process exits. + space. All remaining temp files are also removed automatically when the + MCP server process exits (via an atexit handler). Args: path: Nextcloud path to the file (e.g. "Videos/holiday.mp4") @@ -783,7 +780,7 @@ async def nc_webdav_download_to_temp(path: str, ctx: Context) -> dict: title="Remove Temp File", annotations=ToolAnnotations( destructiveHint=True, - idempotentHint=True, + idempotentHint=False, # errors on second call (path no longer in registry) openWorldHint=False, # operates on local filesystem only ), ) @@ -814,17 +811,19 @@ async def nc_webdav_cleanup_temp(local_path: str, ctx: Context) -> dict: ), } - _temp_registry.discard(local_path) - try: os.unlink(local_path) + _temp_registry.discard(local_path) logger.debug("Removed temp file '%s'", local_path) return {"status": "ok", "local_path": local_path} except FileNotFoundError: + # File already gone — treat as success and clean up registry. + _temp_registry.discard(local_path) return { "status": "ok", "local_path": local_path, "note": "File was already removed.", } except OSError as exc: + # Do NOT discard — leave in registry so the caller can retry. return {"status": "error", "local_path": local_path, "message": str(exc)} diff --git a/tests/unit/test_webdav_archive_tools.py b/tests/unit/test_webdav_archive_tools.py new file mode 100644 index 000000000..31b1fc018 --- /dev/null +++ b/tests/unit/test_webdav_archive_tools.py @@ -0,0 +1,204 @@ +"""Unit tests for WebDAV archive-member and temp-download tools. + +These tests exercise the pure Python logic (zipfile handling, temp registry +management) without a live Nextcloud or full MCP server stack. +""" + +import io +import os +import zipfile + +import pytest + +import nextcloud_mcp_server.server.webdav as webdav_module + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def make_zip(members: dict[str, bytes]) -> bytes: + """Build an in-memory ZIP archive from a {name: content} mapping.""" + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", compression=zipfile.ZIP_STORED) as zf: + for name, data in members.items(): + zf.writestr(name, data) + return buf.getvalue() + + +# --------------------------------------------------------------------------- +# _cleanup_temp_files_on_exit +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_atexit_handler_removes_registered_files(tmp_path): + """atexit handler deletes every path currently in the registry.""" + # Create real files and register them + paths = [] + for i in range(3): + p = tmp_path / f"nc_download_test_{i}.bin" + p.write_bytes(b"data") + paths.append(str(p)) + webdav_module._temp_registry.add(str(p)) + + try: + webdav_module._cleanup_temp_files_on_exit() + for path in paths: + assert not os.path.exists(path) + finally: + for path in paths: + webdav_module._temp_registry.discard(path) + + +@pytest.mark.unit +def test_atexit_handler_tolerates_already_deleted_files(tmp_path): + """atexit handler does not raise if a registered file was already removed.""" + p = tmp_path / "nc_download_gone.bin" + # Do NOT create the file — it's already missing + webdav_module._temp_registry.add(str(p)) + try: + webdav_module._cleanup_temp_files_on_exit() # must not raise + finally: + webdav_module._temp_registry.discard(str(p)) + + +# --------------------------------------------------------------------------- +# ZIP member listing (core logic) +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_zip_member_listing_correct_names(): + """zipfile.ZipFile correctly lists members — validates our iteration logic.""" + content = make_zip( + { + "mimetype": b"application/vnd.oasis.opendocument.spreadsheet", + "content.xml": b"", + "META-INF/manifest.xml": b"", + } + ) + with zipfile.ZipFile(io.BytesIO(content)) as zf: + names = [i.filename for i in zf.infolist()] + assert "content.xml" in names + assert "META-INF/manifest.xml" in names + assert "mimetype" in names + + +@pytest.mark.unit +def test_zip_bad_zip_raises(): + """BadZipFile is raised for non-ZIP bytes — our tools catch this correctly.""" + with pytest.raises(zipfile.BadZipFile): + with zipfile.ZipFile(io.BytesIO(b"this is not a zip")): + pass + + +@pytest.mark.unit +def test_zip_member_read_returns_correct_content(): + """zf.read() returns exact bytes written for a member.""" + xml = b"hello" + content = make_zip({"content.xml": xml}) + with zipfile.ZipFile(io.BytesIO(content)) as zf: + assert zf.read("content.xml") == xml + + +@pytest.mark.unit +def test_zip_missing_member_raises_key_error(): + """Missing member raises KeyError — our tool wraps this into ValueError.""" + content = make_zip({"content.xml": b""}) + with zipfile.ZipFile(io.BytesIO(content)) as zf: + with pytest.raises(KeyError): + zf.read("nonexistent.xml") + + +# --------------------------------------------------------------------------- +# _temp_registry enforcement (cleanup_temp logic) +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_cleanup_temp_rejects_unregistered_path(tmp_path): + """cleanup_temp refuses paths not in _temp_registry.""" + p = tmp_path / "arbitrary.bin" + p.write_bytes(b"secret") + + path = str(p) + assert path not in webdav_module._temp_registry + + # Simulate what cleanup_temp does for unregistered paths + if path not in webdav_module._temp_registry: + result = { + "status": "error", + "local_path": path, + "message": "Path was not created by nc_webdav_download_to_temp in this session, or has already been cleaned up.", + } + assert result["status"] == "error" + assert os.path.exists(path) # file must NOT have been removed + + +@pytest.mark.unit +def test_cleanup_temp_discard_happens_after_unlink(tmp_path): + """Registry entry is only discarded after a successful unlink.""" + p = tmp_path / "nc_download_test.bin" + p.write_bytes(b"payload") + path = str(p) + webdav_module._temp_registry.add(path) + + try: + # Successful unlink path + os.unlink(path) + webdav_module._temp_registry.discard(path) + + assert not os.path.exists(path) + assert path not in webdav_module._temp_registry + finally: + webdav_module._temp_registry.discard(path) + + +@pytest.mark.unit +def test_cleanup_temp_registry_preserved_on_oserror(tmp_path, monkeypatch): + """Registry entry is NOT discarded when os.unlink raises OSError.""" + p = tmp_path / "nc_download_locked.bin" + p.write_bytes(b"payload") + path = str(p) + webdav_module._temp_registry.add(path) + + def _raise(*_a, **_kw): + raise OSError("permission denied") + + monkeypatch.setattr(os, "unlink", _raise) + + try: + try: + os.unlink(path) + webdav_module._temp_registry.discard(path) + except OSError: + pass # do NOT discard + + # Path should still be in the registry so caller can retry + assert path in webdav_module._temp_registry + finally: + webdav_module._temp_registry.discard(path) + # Restore real unlink to remove the file + monkeypatch.undo() + if p.exists(): + p.unlink() + + +@pytest.mark.unit +def test_cleanup_temp_file_not_found_still_discards(tmp_path): + """FileNotFoundError (already deleted) still removes entry from registry.""" + path = str(tmp_path / "nc_download_gone.bin") + # Register without creating the file + webdav_module._temp_registry.add(path) + + try: + try: + os.unlink(path) + webdav_module._temp_registry.discard(path) + except FileNotFoundError: + webdav_module._temp_registry.discard(path) + + assert path not in webdav_module._temp_registry + finally: + webdav_module._temp_registry.discard(path) From df55f48eeb50501d42b06f93e46f2e8601066aa3 Mon Sep 17 00:00:00 2001 From: Jos Poortvliet Date: Sat, 25 Apr 2026 19:27:22 +0200 Subject: [PATCH 03/12] fix(webdav): drop unused content_type, add zip-bomb size guard - nc_webdav_read_archive_member: rename unused 'content_type' return value to '_' (Ruff F841 / unused-variable) - Add _MAX_MEMBER_BYTES = 50 MB module constant; check ZipInfo.file_size via zf.getinfo() *before* zf.read() so an oversized member (including zip-bomb expansions) raises ValueError with a clear message pointing to nc_webdav_download_to_temp for local extraction - Add unit tests: ZipInfo.file_size is readable pre-extraction, _MAX_MEMBER_BYTES is positive, monkeypatched low limit triggers the expected ValueError Co-Authored-By: Claude Sonnet 4.6 --- nextcloud_mcp_server/server/webdav.py | 20 ++++++++++++-- tests/unit/test_webdav_archive_tools.py | 36 +++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 2 deletions(-) diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index 66885997c..b004e9ece 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -21,6 +21,12 @@ logger = logging.getLogger(__name__) +# Maximum uncompressed size (bytes) allowed when extracting a single archive +# member. Guards against zip-bomb attacks where a tiny compressed archive +# expands to an enormous member in memory. 50 MB is generous for XML/text +# content while still bounding worst-case memory use. +_MAX_MEMBER_BYTES: int = 50 * 1024 * 1024 # 50 MB + # Registry of local temp paths created by nc_webdav_download_to_temp. # Used to prevent nc_webdav_cleanup_temp from deleting arbitrary paths. # Plain set is safe: asyncio is single-threaded and GIL protects simple ops. @@ -646,12 +652,12 @@ async def nc_webdav_read_archive_member( ValueError: if the archive is not valid ZIP, or the member is not found """ client = await get_client(ctx) - content, content_type = await client.webdav.read_file(path) + content, _ = await client.webdav.read_file(path) try: with zipfile.ZipFile(io.BytesIO(content)) as zf: try: - member_bytes = zf.read(member_path) + info = zf.getinfo(member_path) except KeyError as exc: available = [i.filename for i in zf.infolist() if not i.is_dir()] raise ValueError( @@ -659,6 +665,16 @@ async def nc_webdav_read_archive_member( f"Available files: {available[:30]}" + (" (truncated)" if len(available) > 30 else "") ) from exc + + if info.file_size > _MAX_MEMBER_BYTES: + raise ValueError( + f"Member '{member_path}' uncompressed size " + f"({info.file_size:,} bytes) exceeds the " + f"{_MAX_MEMBER_BYTES // (1024 * 1024)} MB limit. " + f"Use nc_webdav_download_to_temp and extract locally." + ) + + member_bytes = zf.read(member_path) except zipfile.BadZipFile as exc: raise ValueError(f"'{path}' is not a valid ZIP archive.") from exc diff --git a/tests/unit/test_webdav_archive_tools.py b/tests/unit/test_webdav_archive_tools.py index 31b1fc018..d79b7a991 100644 --- a/tests/unit/test_webdav_archive_tools.py +++ b/tests/unit/test_webdav_archive_tools.py @@ -111,6 +111,42 @@ def test_zip_missing_member_raises_key_error(): zf.read("nonexistent.xml") +@pytest.mark.unit +def test_zip_member_size_check_via_getinfo(): + """ZipInfo.file_size is available before extraction — used for the size guard.""" + xml = b"" * 100 + content = make_zip({"content.xml": xml}) + with zipfile.ZipFile(io.BytesIO(content)) as zf: + info = zf.getinfo("content.xml") + assert info.file_size == len(xml) + + +@pytest.mark.unit +def test_max_member_bytes_constant_is_positive(): + """_MAX_MEMBER_BYTES is defined and positive.""" + assert webdav_module._MAX_MEMBER_BYTES > 0 + + +@pytest.mark.unit +def test_member_size_exceeds_limit_raises(monkeypatch): + """A member whose file_size exceeds _MAX_MEMBER_BYTES raises ValueError before extraction.""" + # Lower the limit to 10 bytes for this test + monkeypatch.setattr(webdav_module, "_MAX_MEMBER_BYTES", 10) + + large_content = b"x" * 100 + archive = make_zip({"big.xml": large_content}) + + with zipfile.ZipFile(io.BytesIO(archive)) as zf: + info = zf.getinfo("big.xml") + if info.file_size > webdav_module._MAX_MEMBER_BYTES: + with pytest.raises(ValueError, match="exceeds the"): + raise ValueError( + f"Member 'big.xml' uncompressed size " + f"({info.file_size:,} bytes) exceeds the " + f"{webdav_module._MAX_MEMBER_BYTES // (1024 * 1024)} MB limit." + ) + + # --------------------------------------------------------------------------- # _temp_registry enforcement (cleanup_temp logic) # --------------------------------------------------------------------------- From 0f806570949f99ba78c9e06bd3e06504ab16197b Mon Sep 17 00:00:00 2001 From: Jos Poortvliet Date: Sat, 25 Apr 2026 23:49:04 +0200 Subject: [PATCH 04/12] refactor: extract zip helpers as pure functions and fix text detection - Extract _list_zip_members() and _read_zip_member() as module-level pure functions so unit tests can call real production code directly, not just stdlib zipfile behavior in isolation. - Fix dotfile extension detection: os.path.splitext("_rels/.rels") returns "" because Python treats the basename as a hidden file; now the whole basename is used as the extension when splitext yields nothing for a dotfile. - Expand _TEXT_EXTENSIONS to include .rels, .opf, .ncx, .xhtml, .rdf, .plist so OOXML relationship files and EPUB packaging files are returned as UTF-8 text rather than base64. - Rewrite unit tests to import and call the helpers directly; covers member listing structure, bad-zip errors, text/binary detection per extension, size-limit enforcement, atexit cleanup, and registry guard semantics. Co-Authored-By: Claude Sonnet 4.6 --- nextcloud_mcp_server/server/webdav.py | 243 ++++++++++++------- tests/unit/test_webdav_archive_tools.py | 297 ++++++++++++++---------- 2 files changed, 334 insertions(+), 206 deletions(-) diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index b004e9ece..9eff687ab 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -45,6 +45,161 @@ def _cleanup_temp_files_on_exit() -> None: atexit.register(_cleanup_temp_files_on_exit) +# --------------------------------------------------------------------------- +# Pure helpers — no MCP context required, fully unit-testable +# --------------------------------------------------------------------------- + +# Extensions always treated as UTF-8 text regardless of MIME type. +# Covers XML-based OOXML internals (.rels, .opf, .xhtml, .ncx) that +# mimetypes.guess_type() returns None or application/octet-stream for. +_TEXT_EXTENSIONS: frozenset[str] = frozenset( + { + ".xml", + ".json", + ".html", + ".xhtml", + ".css", + ".js", + ".svg", + ".txt", + ".md", + ".rels", # OOXML relationship files + ".opf", # EPUB Open Packaging Format + ".ncx", # EPUB Navigation Control + ".rdf", # RDF/XML metadata + ".plist", # Apple property list (XML form) + } +) + +# MIME types treated as text even when the extension doesn't match. +_TEXT_MIME_TYPES: frozenset[str] = frozenset( + { + "application/xml", + "application/json", + "application/javascript", + "application/xhtml+xml", + } +) + + +def _list_zip_members(content: bytes, path: str, content_type: str) -> dict: + """Return the member listing of a ZIP archive as a plain dict. + + Args: + content: Raw bytes of the archive. + path: Nextcloud path (used only in error messages). + content_type: MIME type reported by Nextcloud (included in result). + + Returns: + Dict with path, content_type, archive_size, member_count, members. + + Raises: + ValueError: if *content* is not a valid ZIP archive. + """ + try: + with zipfile.ZipFile(io.BytesIO(content)) as zf: + members = [ + { + "name": info.filename, + "size": info.file_size, + "compressed_size": info.compress_size, + "is_dir": info.is_dir(), + } + for info in zf.infolist() + ] + except zipfile.BadZipFile as exc: + raise ValueError( + f"'{path}' (content-type: {content_type}) is not a valid ZIP archive. " + f"For plain text files use nc_webdav_read_file; for images/video/audio " + f"use nc_webdav_download_to_temp." + ) from exc + + return { + "path": path, + "content_type": content_type, + "archive_size": len(content), + "member_count": len(members), + "members": members, + } + + +def _read_zip_member(content: bytes, path: str, member_path: str) -> dict: + """Extract and return a single member from a ZIP archive. + + Text members (detected by MIME type or file extension) are returned as + UTF-8 strings. Binary members are base64-encoded. + + Args: + content: Raw bytes of the archive. + path: Nextcloud path (used only in error messages). + member_path: Path of the member inside the archive. + + Returns: + Dict with archive_path, member_path, content, content_type, size, + and optionally encoding="base64" for binary members. + + Raises: + ValueError: if the archive is invalid, the member is missing, or + the uncompressed member size exceeds _MAX_MEMBER_BYTES. + """ + try: + with zipfile.ZipFile(io.BytesIO(content)) as zf: + try: + info = zf.getinfo(member_path) + except KeyError as exc: + available = [i.filename for i in zf.infolist() if not i.is_dir()] + raise ValueError( + f"Member '{member_path}' not found in '{path}'. " + f"Available files: {available[:30]}" + + (" (truncated)" if len(available) > 30 else "") + ) from exc + + if info.file_size > _MAX_MEMBER_BYTES: + raise ValueError( + f"Member '{member_path}' uncompressed size " + f"({info.file_size:,} bytes) exceeds the " + f"{_MAX_MEMBER_BYTES // (1024 * 1024)} MB limit. " + f"Use nc_webdav_download_to_temp and extract locally." + ) + + member_bytes = zf.read(member_path) + except zipfile.BadZipFile as exc: + raise ValueError(f"'{path}' is not a valid ZIP archive.") from exc + + member_mime = mimetypes.guess_type(member_path)[0] or "application/octet-stream" + basename = os.path.basename(member_path) + ext = os.path.splitext(basename)[1].lower() + # Dotfiles like ".rels" have no extension per splitext; treat the whole name as the extension. + if not ext and basename.startswith("."): + ext = basename.lower() + + is_text = ( + member_mime.startswith("text/") + or member_mime in _TEXT_MIME_TYPES + or ext in _TEXT_EXTENSIONS + ) + + if is_text: + try: + return { + "archive_path": path, + "member_path": member_path, + "content": member_bytes.decode("utf-8"), + "content_type": member_mime, + "size": len(member_bytes), + } + except UnicodeDecodeError: + pass # fall through to base64 + + return { + "archive_path": path, + "member_path": member_path, + "content": base64.b64encode(member_bytes).decode("ascii"), + "content_type": member_mime, + "size": len(member_bytes), + "encoding": "base64", + } + def configure_webdav_tools(mcp: FastMCP): # WebDAV file system tools @@ -583,32 +738,7 @@ async def nc_webdav_list_archive_members(path: str, ctx: Context) -> dict: """ client = await get_client(ctx) content, content_type = await client.webdav.read_file(path) - - try: - with zipfile.ZipFile(io.BytesIO(content)) as zf: - members = [ - { - "name": info.filename, - "size": info.file_size, - "compressed_size": info.compress_size, - "is_dir": info.is_dir(), - } - for info in zf.infolist() - ] - except zipfile.BadZipFile as exc: - raise ValueError( - f"'{path}' (content-type: {content_type}) is not a valid ZIP archive. " - f"For plain text files use nc_webdav_read_file; for images/video/audio " - f"use nc_webdav_download_to_temp." - ) from exc - - return { - "path": path, - "content_type": content_type, - "archive_size": len(content), - "member_count": len(members), - "members": members, - } + return _list_zip_members(content, path, content_type) @mcp.tool( title="Read Archive Member", @@ -653,66 +783,7 @@ async def nc_webdav_read_archive_member( """ client = await get_client(ctx) content, _ = await client.webdav.read_file(path) - - try: - with zipfile.ZipFile(io.BytesIO(content)) as zf: - try: - info = zf.getinfo(member_path) - except KeyError as exc: - available = [i.filename for i in zf.infolist() if not i.is_dir()] - raise ValueError( - f"Member '{member_path}' not found in '{path}'. " - f"Available files: {available[:30]}" - + (" (truncated)" if len(available) > 30 else "") - ) from exc - - if info.file_size > _MAX_MEMBER_BYTES: - raise ValueError( - f"Member '{member_path}' uncompressed size " - f"({info.file_size:,} bytes) exceeds the " - f"{_MAX_MEMBER_BYTES // (1024 * 1024)} MB limit. " - f"Use nc_webdav_download_to_temp and extract locally." - ) - - member_bytes = zf.read(member_path) - except zipfile.BadZipFile as exc: - raise ValueError(f"'{path}' is not a valid ZIP archive.") from exc - - member_mime = mimetypes.guess_type(member_path)[0] or "application/octet-stream" - - # Return text members decoded; XML files are always text even without - # an explicit text/* MIME type. - is_text = ( - member_mime.startswith("text/") - or member_mime - in { - "application/xml", - "application/json", - "application/javascript", - } - or member_path.endswith((".xml", ".json", ".html", ".css", ".js", ".svg")) - ) - - if is_text: - try: - return { - "archive_path": path, - "member_path": member_path, - "content": member_bytes.decode("utf-8"), - "content_type": member_mime, - "size": len(member_bytes), - } - except UnicodeDecodeError: - pass # fall through to base64 - - return { - "archive_path": path, - "member_path": member_path, - "content": base64.b64encode(member_bytes).decode("ascii"), - "content_type": member_mime, - "size": len(member_bytes), - "encoding": "base64", - } + return _read_zip_member(content, path, member_path) @mcp.tool( title="Download File to Temp", diff --git a/tests/unit/test_webdav_archive_tools.py b/tests/unit/test_webdav_archive_tools.py index d79b7a991..43f2df961 100644 --- a/tests/unit/test_webdav_archive_tools.py +++ b/tests/unit/test_webdav_archive_tools.py @@ -1,7 +1,9 @@ """Unit tests for WebDAV archive-member and temp-download tools. -These tests exercise the pure Python logic (zipfile handling, temp registry -management) without a live Nextcloud or full MCP server stack. +All tests call the real production functions (_list_zip_members, +_read_zip_member, _cleanup_temp_files_on_exit, _temp_registry) so that +regressions in the implementation are caught rather than just verifying +stdlib zipfile behaviour. """ import io @@ -11,6 +13,11 @@ import pytest import nextcloud_mcp_server.server.webdav as webdav_module +from nextcloud_mcp_server.server.webdav import ( + _list_zip_members, + _read_zip_member, + _temp_registry, +) # --------------------------------------------------------------------------- # Helpers @@ -27,177 +34,230 @@ def make_zip(members: dict[str, bytes]) -> bytes: # --------------------------------------------------------------------------- -# _cleanup_temp_files_on_exit +# _list_zip_members # --------------------------------------------------------------------------- @pytest.mark.unit -def test_atexit_handler_removes_registered_files(tmp_path): - """atexit handler deletes every path currently in the registry.""" - # Create real files and register them - paths = [] - for i in range(3): - p = tmp_path / f"nc_download_test_{i}.bin" - p.write_bytes(b"data") - paths.append(str(p)) - webdav_module._temp_registry.add(str(p)) +def test_list_members_returns_expected_structure(): + """_list_zip_members returns correct member names, sizes, and metadata.""" + content = make_zip( + { + "mimetype": b"application/vnd.oasis.opendocument.spreadsheet", + "content.xml": b"", + "META-INF/manifest.xml": b"", + } + ) + result = _list_zip_members( + content, "test.ods", "application/vnd.oasis.opendocument.spreadsheet" + ) - try: - webdav_module._cleanup_temp_files_on_exit() - for path in paths: - assert not os.path.exists(path) - finally: - for path in paths: - webdav_module._temp_registry.discard(path) + assert result["path"] == "test.ods" + assert result["member_count"] == 3 + assert result["archive_size"] == len(content) + + names = {m["name"] for m in result["members"]} + assert names == {"mimetype", "content.xml", "META-INF/manifest.xml"} + + content_xml = next(m for m in result["members"] if m["name"] == "content.xml") + assert content_xml["size"] == len(b"") + assert content_xml["is_dir"] is False @pytest.mark.unit -def test_atexit_handler_tolerates_already_deleted_files(tmp_path): - """atexit handler does not raise if a registered file was already removed.""" - p = tmp_path / "nc_download_gone.bin" - # Do NOT create the file — it's already missing - webdav_module._temp_registry.add(str(p)) - try: - webdav_module._cleanup_temp_files_on_exit() # must not raise - finally: - webdav_module._temp_registry.discard(str(p)) +def test_list_members_bad_zip_raises_value_error(): + """_list_zip_members raises ValueError (not BadZipFile) for non-ZIP bytes.""" + with pytest.raises(ValueError, match="not a valid ZIP archive"): + _list_zip_members(b"this is not a zip", "bad.ods", "application/octet-stream") + + +@pytest.mark.unit +def test_list_members_includes_content_type_in_result(): + """content_type from Nextcloud is passed through to the result dict.""" + content = make_zip({"x.xml": b""}) + mime = "application/vnd.oasis.opendocument.spreadsheet" + result = _list_zip_members(content, "sheet.ods", mime) + assert result["content_type"] == mime # --------------------------------------------------------------------------- -# ZIP member listing (core logic) +# _read_zip_member — text detection # --------------------------------------------------------------------------- @pytest.mark.unit -def test_zip_member_listing_correct_names(): - """zipfile.ZipFile correctly lists members — validates our iteration logic.""" - content = make_zip( - { - "mimetype": b"application/vnd.oasis.opendocument.spreadsheet", - "content.xml": b"", - "META-INF/manifest.xml": b"", - } - ) - with zipfile.ZipFile(io.BytesIO(content)) as zf: - names = [i.filename for i in zf.infolist()] - assert "content.xml" in names - assert "META-INF/manifest.xml" in names - assert "mimetype" in names +def test_read_member_xml_returned_as_utf8(): + """XML member (content.xml) is returned as a UTF-8 string, not base64.""" + xml = b"hello" + content = make_zip({"content.xml": xml}) + result = _read_zip_member(content, "test.ods", "content.xml") + + assert result["content"] == xml.decode("utf-8") + assert "encoding" not in result + assert result["size"] == len(xml) @pytest.mark.unit -def test_zip_bad_zip_raises(): - """BadZipFile is raised for non-ZIP bytes — our tools catch this correctly.""" - with pytest.raises(zipfile.BadZipFile): - with zipfile.ZipFile(io.BytesIO(b"this is not a zip")): - pass +def test_read_member_rels_returned_as_utf8(): + """.rels files (OOXML relationship files) are returned as text, not base64.""" + rels = b'' + content = make_zip({"_rels/.rels": rels}) + result = _read_zip_member(content, "test.docx", "_rels/.rels") + assert result["content"] == rels.decode("utf-8") + assert "encoding" not in result @pytest.mark.unit -def test_zip_member_read_returns_correct_content(): - """zf.read() returns exact bytes written for a member.""" - xml = b"hello" - content = make_zip({"content.xml": xml}) - with zipfile.ZipFile(io.BytesIO(content)) as zf: - assert zf.read("content.xml") == xml +def test_read_member_xhtml_returned_as_utf8(): + """.xhtml files (common in EPUB) are returned as text.""" + xhtml = b"hello" + content = make_zip({"OEBPS/chapter1.xhtml": xhtml}) + result = _read_zip_member(content, "book.epub", "OEBPS/chapter1.xhtml") + assert result["content"] == xhtml.decode("utf-8") + assert "encoding" not in result @pytest.mark.unit -def test_zip_missing_member_raises_key_error(): - """Missing member raises KeyError — our tool wraps this into ValueError.""" - content = make_zip({"content.xml": b""}) - with zipfile.ZipFile(io.BytesIO(content)) as zf: - with pytest.raises(KeyError): - zf.read("nonexistent.xml") +def test_read_member_opf_returned_as_utf8(): + """.opf files (EPUB Open Packaging Format) are returned as text.""" + opf = b"" + content = make_zip({"OEBPS/content.opf": opf}) + result = _read_zip_member(content, "book.epub", "OEBPS/content.opf") + assert result["content"] == opf.decode("utf-8") + assert "encoding" not in result @pytest.mark.unit -def test_zip_member_size_check_via_getinfo(): - """ZipInfo.file_size is available before extraction — used for the size guard.""" - xml = b"" * 100 - content = make_zip({"content.xml": xml}) - with zipfile.ZipFile(io.BytesIO(content)) as zf: - info = zf.getinfo("content.xml") - assert info.file_size == len(xml) +def test_read_member_binary_returned_as_base64(): + """Binary members (e.g. embedded images) are base64-encoded.""" + import base64 + + png_bytes = b"\x89PNG\r\n\x1a\n" + b"\x00" * 20 + content = make_zip({"image.png": png_bytes}) + result = _read_zip_member(content, "test.ods", "image.png") + + assert result["encoding"] == "base64" + assert base64.b64decode(result["content"]) == png_bytes + + +# --------------------------------------------------------------------------- +# _read_zip_member — error paths +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_read_member_missing_member_raises_value_error(): + """Missing member raises ValueError with the available file list.""" + content = make_zip({"content.xml": b""}) + with pytest.raises(ValueError, match="not found"): + _read_zip_member(content, "test.ods", "nonexistent.xml") @pytest.mark.unit -def test_max_member_bytes_constant_is_positive(): - """_MAX_MEMBER_BYTES is defined and positive.""" - assert webdav_module._MAX_MEMBER_BYTES > 0 +def test_read_member_bad_zip_raises_value_error(): + """Non-ZIP bytes raise ValueError (not BadZipFile).""" + with pytest.raises(ValueError, match="not a valid ZIP archive"): + _read_zip_member(b"garbage", "test.ods", "content.xml") @pytest.mark.unit -def test_member_size_exceeds_limit_raises(monkeypatch): - """A member whose file_size exceeds _MAX_MEMBER_BYTES raises ValueError before extraction.""" - # Lower the limit to 10 bytes for this test +def test_read_member_size_limit_enforced(monkeypatch): + """Members exceeding _MAX_MEMBER_BYTES raise ValueError before extraction.""" monkeypatch.setattr(webdav_module, "_MAX_MEMBER_BYTES", 10) - large_content = b"x" * 100 - archive = make_zip({"big.xml": large_content}) + large_data = b"x" * 100 # well above the patched 10-byte limit + content = make_zip({"big.xml": large_data}) - with zipfile.ZipFile(io.BytesIO(archive)) as zf: - info = zf.getinfo("big.xml") - if info.file_size > webdav_module._MAX_MEMBER_BYTES: - with pytest.raises(ValueError, match="exceeds the"): - raise ValueError( - f"Member 'big.xml' uncompressed size " - f"({info.file_size:,} bytes) exceeds the " - f"{webdav_module._MAX_MEMBER_BYTES // (1024 * 1024)} MB limit." - ) + with pytest.raises(ValueError, match="exceeds the"): + _read_zip_member(content, "test.ods", "big.xml") + + +@pytest.mark.unit +def test_read_member_size_limit_not_triggered_for_small_member(monkeypatch): + """Members within _MAX_MEMBER_BYTES are extracted without error.""" + monkeypatch.setattr(webdav_module, "_MAX_MEMBER_BYTES", 200) + + small_data = b"" * 10 # 40 bytes, well within 200 + content = make_zip({"small.xml": small_data}) + + result = _read_zip_member(content, "test.ods", "small.xml") + assert result["content"] == small_data.decode("utf-8") # --------------------------------------------------------------------------- -# _temp_registry enforcement (cleanup_temp logic) +# _cleanup_temp_files_on_exit and _temp_registry # --------------------------------------------------------------------------- +@pytest.mark.unit +def test_atexit_handler_removes_registered_files(tmp_path): + """atexit handler deletes every path currently in the registry.""" + paths = [] + for i in range(3): + p = tmp_path / f"nc_download_test_{i}.bin" + p.write_bytes(b"data") + paths.append(str(p)) + _temp_registry.add(str(p)) + + try: + webdav_module._cleanup_temp_files_on_exit() + for path in paths: + assert not os.path.exists(path) + finally: + for path in paths: + _temp_registry.discard(path) + + +@pytest.mark.unit +def test_atexit_handler_tolerates_already_deleted_files(tmp_path): + """atexit handler does not raise if a registered file was already removed.""" + p = tmp_path / "nc_download_gone.bin" + _temp_registry.add(str(p)) + try: + webdav_module._cleanup_temp_files_on_exit() # must not raise + finally: + _temp_registry.discard(str(p)) + + @pytest.mark.unit def test_cleanup_temp_rejects_unregistered_path(tmp_path): - """cleanup_temp refuses paths not in _temp_registry.""" + """Paths not in _temp_registry must not be removed.""" p = tmp_path / "arbitrary.bin" p.write_bytes(b"secret") - path = str(p) - assert path not in webdav_module._temp_registry - - # Simulate what cleanup_temp does for unregistered paths - if path not in webdav_module._temp_registry: - result = { - "status": "error", - "local_path": path, - "message": "Path was not created by nc_webdav_download_to_temp in this session, or has already been cleaned up.", - } - assert result["status"] == "error" - assert os.path.exists(path) # file must NOT have been removed + + assert path not in _temp_registry + # Verify the guard condition the tool uses + assert path not in _temp_registry + # File is untouched + assert os.path.exists(path) @pytest.mark.unit -def test_cleanup_temp_discard_happens_after_unlink(tmp_path): - """Registry entry is only discarded after a successful unlink.""" +def test_cleanup_temp_discard_only_after_successful_unlink(tmp_path): + """Registry entry is removed only after os.unlink() succeeds.""" p = tmp_path / "nc_download_test.bin" p.write_bytes(b"payload") path = str(p) - webdav_module._temp_registry.add(path) + _temp_registry.add(path) try: - # Successful unlink path os.unlink(path) - webdav_module._temp_registry.discard(path) + _temp_registry.discard(path) assert not os.path.exists(path) - assert path not in webdav_module._temp_registry + assert path not in _temp_registry finally: - webdav_module._temp_registry.discard(path) + _temp_registry.discard(path) @pytest.mark.unit def test_cleanup_temp_registry_preserved_on_oserror(tmp_path, monkeypatch): - """Registry entry is NOT discarded when os.unlink raises OSError.""" + """Registry entry is NOT discarded when os.unlink raises OSError (allows retry).""" p = tmp_path / "nc_download_locked.bin" p.write_bytes(b"payload") path = str(p) - webdav_module._temp_registry.add(path) + _temp_registry.add(path) def _raise(*_a, **_kw): raise OSError("permission denied") @@ -207,34 +267,31 @@ def _raise(*_a, **_kw): try: try: os.unlink(path) - webdav_module._temp_registry.discard(path) + _temp_registry.discard(path) except OSError: pass # do NOT discard - # Path should still be in the registry so caller can retry - assert path in webdav_module._temp_registry + assert path in _temp_registry finally: - webdav_module._temp_registry.discard(path) - # Restore real unlink to remove the file + _temp_registry.discard(path) monkeypatch.undo() if p.exists(): p.unlink() @pytest.mark.unit -def test_cleanup_temp_file_not_found_still_discards(tmp_path): - """FileNotFoundError (already deleted) still removes entry from registry.""" +def test_cleanup_temp_file_not_found_discards_registry(tmp_path): + """FileNotFoundError (file already gone) still removes the registry entry.""" path = str(tmp_path / "nc_download_gone.bin") - # Register without creating the file - webdav_module._temp_registry.add(path) + _temp_registry.add(path) try: try: os.unlink(path) - webdav_module._temp_registry.discard(path) + _temp_registry.discard(path) except FileNotFoundError: - webdav_module._temp_registry.discard(path) + _temp_registry.discard(path) - assert path not in webdav_module._temp_registry + assert path not in _temp_registry finally: - webdav_module._temp_registry.discard(path) + _temp_registry.discard(path) From 4ce657acccb24067bd7110dde5029c22072ff941 Mon Sep 17 00:00:00 2001 From: Jos Poortvliet Date: Wed, 29 Apr 2026 23:33:39 +0200 Subject: [PATCH 05/12] docs(webdav): clarify download_to_temp is local-process-only nc_webdav_download_to_temp writes to the MCP server's local filesystem. Over a remote streamable-HTTP connection the temp path is inaccessible to the client's shell tools. Add an explicit note in the docstring so callers understand this is only meaningful in stdio/localhost mode. Co-Authored-By: Claude Sonnet 4.6 --- nextcloud_mcp_server/server/webdav.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index 9eff687ab..55ab2bb30 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -797,7 +797,14 @@ async def nc_webdav_read_archive_member( async def nc_webdav_download_to_temp(path: str, ctx: Context) -> dict: """Download a Nextcloud file to a local temporary path and return that path. - IMPORTANT — this tool is only useful when you have access to local shell + IMPORTANT — this tool only makes sense when the MCP server is running as + a local process on the same machine as the client (stdio transport or + localhost SSE). Over a remote streamable-HTTP connection the temp file is + written to the *server's* filesystem, where local shell tools cannot + reach it. In that case use nc_webdav_read_file or the archive member + tools instead. + + Even in local mode this tool is only useful when you have access to shell tools (e.g. Claude Code's Bash tool). In Claude Desktop without shell access the returned path cannot be acted upon and you should not call this tool. From fb60ef738bc705b6842a896cf0c073dfd4cfb99a Mon Sep 17 00:00:00 2001 From: Jos Poortvliet Date: Tue, 5 May 2026 14:36:13 +0200 Subject: [PATCH 06/12] Update tests/unit/test_webdav_archive_tools.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> --- tests/unit/test_webdav_archive_tools.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/unit/test_webdav_archive_tools.py b/tests/unit/test_webdav_archive_tools.py index 43f2df961..954518afc 100644 --- a/tests/unit/test_webdav_archive_tools.py +++ b/tests/unit/test_webdav_archive_tools.py @@ -226,7 +226,6 @@ def test_cleanup_temp_rejects_unregistered_path(tmp_path): p.write_bytes(b"secret") path = str(p) - assert path not in _temp_registry # Verify the guard condition the tool uses assert path not in _temp_registry # File is untouched From 4d74f46d9f5eda61f1261dfb9fecf26caf9551ea Mon Sep 17 00:00:00 2001 From: Jos Poortvliet Date: Tue, 5 May 2026 14:41:51 +0200 Subject: [PATCH 07/12] refactor: extract _cleanup_temp_path helper and fix test coverage - Extract _cleanup_temp_path() as a module-level pure function (same pattern as _list_zip_members/_read_zip_member) so unit tests can call real production logic instead of reimplementing it. - nc_webdav_cleanup_temp tool closure becomes a one-line wrapper. - Rewrite cleanup-temp tests to call _cleanup_temp_path() directly, asserting actual return values and registry state for the success, FileNotFoundError, OSError, and unregistered-path cases. - Remove duplicate assertion in test_cleanup_temp_rejects_unregistered_path. Co-Authored-By: Claude Sonnet 4.6 --- nextcloud_mcp_server/server/webdav.py | 70 ++++++++++++++++--------- tests/unit/test_webdav_archive_tools.py | 49 +++++++++-------- 2 files changed, 72 insertions(+), 47 deletions(-) diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index 55ab2bb30..2b1bc41c2 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -201,6 +201,49 @@ def _read_zip_member(content: bytes, path: str, member_path: str) -> dict: } +def _cleanup_temp_path(local_path: str) -> dict: + """Remove a temp file that was registered by nc_webdav_download_to_temp. + + Only paths present in *_temp_registry* may be removed; any other path is + rejected. The registry entry is discarded only after a successful unlink + (or when the file is already gone); it is retained on OSError so the caller + can retry. + + Args: + local_path: The path previously returned by nc_webdav_download_to_temp. + + Returns: + Dict with ``status`` ("ok" or "error"), ``local_path``, and an optional + ``message`` / ``note`` field. + """ + if local_path not in _temp_registry: + return { + "status": "error", + "local_path": local_path, + "message": ( + "Path was not created by nc_webdav_download_to_temp in this " + "session, or has already been cleaned up." + ), + } + + try: + os.unlink(local_path) + _temp_registry.discard(local_path) + logger.debug("Removed temp file '%s'", local_path) + return {"status": "ok", "local_path": local_path} + except FileNotFoundError: + # File already gone — treat as success and clean up registry. + _temp_registry.discard(local_path) + return { + "status": "ok", + "local_path": local_path, + "note": "File was already removed.", + } + except OSError as exc: + # Do NOT discard — leave in registry so the caller can retry. + return {"status": "error", "local_path": local_path, "message": str(exc)} + + def configure_webdav_tools(mcp: FastMCP): # WebDAV file system tools @mcp.tool( @@ -895,29 +938,4 @@ async def nc_webdav_cleanup_temp(local_path: str, ctx: Context) -> dict: Returns: Dict with status ("ok" or "error") and the local_path. """ - if local_path not in _temp_registry: - return { - "status": "error", - "local_path": local_path, - "message": ( - "Path was not created by nc_webdav_download_to_temp in this " - "session, or has already been cleaned up." - ), - } - - try: - os.unlink(local_path) - _temp_registry.discard(local_path) - logger.debug("Removed temp file '%s'", local_path) - return {"status": "ok", "local_path": local_path} - except FileNotFoundError: - # File already gone — treat as success and clean up registry. - _temp_registry.discard(local_path) - return { - "status": "ok", - "local_path": local_path, - "note": "File was already removed.", - } - except OSError as exc: - # Do NOT discard — leave in registry so the caller can retry. - return {"status": "error", "local_path": local_path, "message": str(exc)} + return _cleanup_temp_path(local_path) diff --git a/tests/unit/test_webdav_archive_tools.py b/tests/unit/test_webdav_archive_tools.py index 43f2df961..f17689d24 100644 --- a/tests/unit/test_webdav_archive_tools.py +++ b/tests/unit/test_webdav_archive_tools.py @@ -1,9 +1,9 @@ """Unit tests for WebDAV archive-member and temp-download tools. All tests call the real production functions (_list_zip_members, -_read_zip_member, _cleanup_temp_files_on_exit, _temp_registry) so that -regressions in the implementation are caught rather than just verifying -stdlib zipfile behaviour. +_read_zip_member, _cleanup_temp_path, _cleanup_temp_files_on_exit, +_temp_registry) so that regressions in the implementation are caught rather +than just verifying stdlib zipfile behaviour. """ import io @@ -14,6 +14,7 @@ import nextcloud_mcp_server.server.webdav as webdav_module from nextcloud_mcp_server.server.webdav import ( + _cleanup_temp_path, _list_zip_members, _read_zip_member, _temp_registry, @@ -219,32 +220,41 @@ def test_atexit_handler_tolerates_already_deleted_files(tmp_path): _temp_registry.discard(str(p)) +# --------------------------------------------------------------------------- +# _cleanup_temp_path — calls real production helper +# --------------------------------------------------------------------------- + + @pytest.mark.unit def test_cleanup_temp_rejects_unregistered_path(tmp_path): - """Paths not in _temp_registry must not be removed.""" + """_cleanup_temp_path returns an error dict for paths not in _temp_registry.""" p = tmp_path / "arbitrary.bin" p.write_bytes(b"secret") path = str(p) assert path not in _temp_registry - # Verify the guard condition the tool uses - assert path not in _temp_registry - # File is untouched + + result = _cleanup_temp_path(path) + + assert result["status"] == "error" + assert "session" in result["message"].lower() + # File must be untouched assert os.path.exists(path) @pytest.mark.unit -def test_cleanup_temp_discard_only_after_successful_unlink(tmp_path): - """Registry entry is removed only after os.unlink() succeeds.""" +def test_cleanup_temp_success(tmp_path): + """_cleanup_temp_path deletes the file and removes it from the registry.""" p = tmp_path / "nc_download_test.bin" p.write_bytes(b"payload") path = str(p) _temp_registry.add(path) try: - os.unlink(path) - _temp_registry.discard(path) + result = _cleanup_temp_path(path) + assert result["status"] == "ok" + assert result["local_path"] == path assert not os.path.exists(path) assert path not in _temp_registry finally: @@ -265,12 +275,11 @@ def _raise(*_a, **_kw): monkeypatch.setattr(os, "unlink", _raise) try: - try: - os.unlink(path) - _temp_registry.discard(path) - except OSError: - pass # do NOT discard + result = _cleanup_temp_path(path) + assert result["status"] == "error" + assert "permission denied" in result["message"] + # Entry must remain so the caller can retry. assert path in _temp_registry finally: _temp_registry.discard(path) @@ -283,15 +292,13 @@ def _raise(*_a, **_kw): def test_cleanup_temp_file_not_found_discards_registry(tmp_path): """FileNotFoundError (file already gone) still removes the registry entry.""" path = str(tmp_path / "nc_download_gone.bin") + # Register a path for a file that does NOT exist on disk. _temp_registry.add(path) try: - try: - os.unlink(path) - _temp_registry.discard(path) - except FileNotFoundError: - _temp_registry.discard(path) + result = _cleanup_temp_path(path) + assert result["status"] == "ok" assert path not in _temp_registry finally: _temp_registry.discard(path) From ddc54dbbe093b8fa7833fda0424dd78865eb7053 Mon Sep 17 00:00:00 2001 From: Jos Poortvliet Date: Tue, 5 May 2026 14:56:42 +0200 Subject: [PATCH 08/12] fix(webdav): content-sniff extensionless archive members as text Extension/MIME heuristics miss members that have no extension, such as the ODF mandatory 'mimetype' entry. Add a UTF-8 decode + null-byte probe as a fallback (the classic binary-vs-text check used by file(1)): if the bytes decode as valid UTF-8 and contain no null bytes they are returned as a UTF-8 string rather than base64. Add a regression test for 'mimetype'. Co-Authored-By: Claude Sonnet 4.6 --- nextcloud_mcp_server/server/webdav.py | 10 ++++++++++ tests/unit/test_webdav_archive_tools.py | 10 ++++++++++ 2 files changed, 20 insertions(+) diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index 2b1bc41c2..29466cf44 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -179,6 +179,16 @@ def _read_zip_member(content: bytes, path: str, member_path: str) -> dict: or ext in _TEXT_EXTENSIONS ) + # Content-sniff fallback: if the extension/MIME heuristics didn't fire + # (e.g. extensionless members like ODF's "mimetype"), try UTF-8 decoding + # and reject if null bytes are present (the classic binary-vs-text probe). + if not is_text and b"\x00" not in member_bytes: + try: + member_bytes.decode("utf-8") + is_text = True + except UnicodeDecodeError: + pass + if is_text: try: return { diff --git a/tests/unit/test_webdav_archive_tools.py b/tests/unit/test_webdav_archive_tools.py index f17689d24..5b4f3023f 100644 --- a/tests/unit/test_webdav_archive_tools.py +++ b/tests/unit/test_webdav_archive_tools.py @@ -128,6 +128,16 @@ def test_read_member_opf_returned_as_utf8(): assert "encoding" not in result +@pytest.mark.unit +def test_read_member_extensionless_text_returned_as_utf8(): + """Extensionless text members (e.g. ODF 'mimetype') are detected via content sniff.""" + mime_content = b"application/vnd.oasis.opendocument.spreadsheet" + content = make_zip({"mimetype": mime_content}) + result = _read_zip_member(content, "test.ods", "mimetype") + assert result["content"] == mime_content.decode("utf-8") + assert "encoding" not in result + + @pytest.mark.unit def test_read_member_binary_returned_as_base64(): """Binary members (e.g. embedded images) are base64-encoded.""" From 1f53baf956416445cf4d65bd60de8b7b51287eea Mon Sep 17 00:00:00 2001 From: Jos Poortvliet Date: Tue, 5 May 2026 15:00:32 +0200 Subject: [PATCH 09/12] fix(webdav): scope temp-file registry to owning username _temp_registry was a process-global set[str], allowing any session in multi-user deployments to delete another session's temp files by supplying a known path. Change _temp_registry to dict[str, str] (local_path -> username). nc_webdav_download_to_temp stores client.username alongside the path; _cleanup_temp_path gains an owner parameter and rejects callers whose username doesn't match the registered owner. The atexit handler is unaffected (it runs without a user context and iterates the dict keys). Add test_cleanup_temp_rejects_wrong_owner to cover the cross-session denial path. Co-Authored-By: Claude Sonnet 4.6 --- nextcloud_mcp_server/server/webdav.py | 44 +++++++++++++++------- tests/unit/test_webdav_archive_tools.py | 50 ++++++++++++++++++------- 2 files changed, 66 insertions(+), 28 deletions(-) diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index 29466cf44..624474c07 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -28,9 +28,11 @@ _MAX_MEMBER_BYTES: int = 50 * 1024 * 1024 # 50 MB # Registry of local temp paths created by nc_webdav_download_to_temp. -# Used to prevent nc_webdav_cleanup_temp from deleting arbitrary paths. -# Plain set is safe: asyncio is single-threaded and GIL protects simple ops. -_temp_registry: set[str] = set() +# Maps local_path -> owning_username so nc_webdav_cleanup_temp can verify +# that the caller is the same user who created the file, preventing one +# multi-user session from deleting another session's temp files. +# Dict mutation is safe in asyncio: single-threaded, GIL protects simple ops. +_temp_registry: dict[str, str] = {} def _cleanup_temp_files_on_exit() -> None: @@ -211,22 +213,28 @@ def _read_zip_member(content: bytes, path: str, member_path: str) -> dict: } -def _cleanup_temp_path(local_path: str) -> dict: +def _cleanup_temp_path(local_path: str, owner: str | None = None) -> dict: """Remove a temp file that was registered by nc_webdav_download_to_temp. - Only paths present in *_temp_registry* may be removed; any other path is - rejected. The registry entry is discarded only after a successful unlink - (or when the file is already gone); it is retained on OSError so the caller - can retry. + Only paths present in *_temp_registry* may be removed. When *owner* is + supplied (the Nextcloud username of the calling session) the registry entry + must also match that username, preventing one multi-user session from + deleting another session's temp files. + + The registry entry is discarded only after a successful unlink (or when the + file is already gone); it is retained on OSError so the caller can retry. Args: local_path: The path previously returned by nc_webdav_download_to_temp. + owner: Username of the requesting session. Pass ``None`` only in + contexts where ownership cannot be determined (e.g. atexit). Returns: Dict with ``status`` ("ok" or "error"), ``local_path``, and an optional ``message`` / ``note`` field. """ - if local_path not in _temp_registry: + registered_owner = _temp_registry.get(local_path) + if registered_owner is None: return { "status": "error", "local_path": local_path, @@ -236,21 +244,28 @@ def _cleanup_temp_path(local_path: str) -> dict: ), } + if owner is not None and registered_owner != owner: + return { + "status": "error", + "local_path": local_path, + "message": "Permission denied: this temp file belongs to a different session.", + } + try: os.unlink(local_path) - _temp_registry.discard(local_path) + del _temp_registry[local_path] logger.debug("Removed temp file '%s'", local_path) return {"status": "ok", "local_path": local_path} except FileNotFoundError: # File already gone — treat as success and clean up registry. - _temp_registry.discard(local_path) + _temp_registry.pop(local_path, None) return { "status": "ok", "local_path": local_path, "note": "File was already removed.", } except OSError as exc: - # Do NOT discard — leave in registry so the caller can retry. + # Do NOT remove from registry — leave so the caller can retry. return {"status": "error", "local_path": local_path, "message": str(exc)} @@ -907,7 +922,7 @@ async def nc_webdav_download_to_temp(path: str, ctx: Context) -> dict: pass raise - _temp_registry.add(local_path) + _temp_registry[local_path] = client.username logger.debug( "Downloaded '%s' to temp path '%s' (%d bytes)", path, @@ -948,4 +963,5 @@ async def nc_webdav_cleanup_temp(local_path: str, ctx: Context) -> dict: Returns: Dict with status ("ok" or "error") and the local_path. """ - return _cleanup_temp_path(local_path) + client = await get_client(ctx) + return _cleanup_temp_path(local_path, owner=client.username) diff --git a/tests/unit/test_webdav_archive_tools.py b/tests/unit/test_webdav_archive_tools.py index 5b4f3023f..08f8a0734 100644 --- a/tests/unit/test_webdav_archive_tools.py +++ b/tests/unit/test_webdav_archive_tools.py @@ -208,7 +208,7 @@ def test_atexit_handler_removes_registered_files(tmp_path): p = tmp_path / f"nc_download_test_{i}.bin" p.write_bytes(b"data") paths.append(str(p)) - _temp_registry.add(str(p)) + _temp_registry[str(p)] = "testuser" try: webdav_module._cleanup_temp_files_on_exit() @@ -216,18 +216,18 @@ def test_atexit_handler_removes_registered_files(tmp_path): assert not os.path.exists(path) finally: for path in paths: - _temp_registry.discard(path) + _temp_registry.pop(path, None) @pytest.mark.unit def test_atexit_handler_tolerates_already_deleted_files(tmp_path): """atexit handler does not raise if a registered file was already removed.""" p = tmp_path / "nc_download_gone.bin" - _temp_registry.add(str(p)) + _temp_registry[str(p)] = "testuser" try: webdav_module._cleanup_temp_files_on_exit() # must not raise finally: - _temp_registry.discard(str(p)) + _temp_registry.pop(str(p), None) # --------------------------------------------------------------------------- @@ -244,7 +244,7 @@ def test_cleanup_temp_rejects_unregistered_path(tmp_path): assert path not in _temp_registry - result = _cleanup_temp_path(path) + result = _cleanup_temp_path(path, owner="alice") assert result["status"] == "error" assert "session" in result["message"].lower() @@ -252,23 +252,45 @@ def test_cleanup_temp_rejects_unregistered_path(tmp_path): assert os.path.exists(path) +@pytest.mark.unit +def test_cleanup_temp_rejects_wrong_owner(tmp_path): + """_cleanup_temp_path rejects callers who don't own the file.""" + p = tmp_path / "nc_download_alice.bin" + p.write_bytes(b"payload") + path = str(p) + _temp_registry[path] = "alice" + + try: + result = _cleanup_temp_path(path, owner="bob") + + assert result["status"] == "error" + assert "permission" in result["message"].lower() + # File must be untouched + assert os.path.exists(path) + assert path in _temp_registry + finally: + _temp_registry.pop(path, None) + if p.exists(): + p.unlink() + + @pytest.mark.unit def test_cleanup_temp_success(tmp_path): """_cleanup_temp_path deletes the file and removes it from the registry.""" p = tmp_path / "nc_download_test.bin" p.write_bytes(b"payload") path = str(p) - _temp_registry.add(path) + _temp_registry[path] = "alice" try: - result = _cleanup_temp_path(path) + result = _cleanup_temp_path(path, owner="alice") assert result["status"] == "ok" assert result["local_path"] == path assert not os.path.exists(path) assert path not in _temp_registry finally: - _temp_registry.discard(path) + _temp_registry.pop(path, None) @pytest.mark.unit @@ -277,7 +299,7 @@ def test_cleanup_temp_registry_preserved_on_oserror(tmp_path, monkeypatch): p = tmp_path / "nc_download_locked.bin" p.write_bytes(b"payload") path = str(p) - _temp_registry.add(path) + _temp_registry[path] = "alice" def _raise(*_a, **_kw): raise OSError("permission denied") @@ -285,14 +307,14 @@ def _raise(*_a, **_kw): monkeypatch.setattr(os, "unlink", _raise) try: - result = _cleanup_temp_path(path) + result = _cleanup_temp_path(path, owner="alice") assert result["status"] == "error" assert "permission denied" in result["message"] # Entry must remain so the caller can retry. assert path in _temp_registry finally: - _temp_registry.discard(path) + _temp_registry.pop(path, None) monkeypatch.undo() if p.exists(): p.unlink() @@ -303,12 +325,12 @@ def test_cleanup_temp_file_not_found_discards_registry(tmp_path): """FileNotFoundError (file already gone) still removes the registry entry.""" path = str(tmp_path / "nc_download_gone.bin") # Register a path for a file that does NOT exist on disk. - _temp_registry.add(path) + _temp_registry[path] = "alice" try: - result = _cleanup_temp_path(path) + result = _cleanup_temp_path(path, owner="alice") assert result["status"] == "ok" assert path not in _temp_registry finally: - _temp_registry.discard(path) + _temp_registry.pop(path, None) From decc16334e03d04506d1f980a7b0d9927b36f029 Mon Sep 17 00:00:00 2001 From: Jos Poortvliet Date: Tue, 5 May 2026 17:53:59 +0200 Subject: [PATCH 10/12] test(webdav): add MCP tool wiring tests via FastMCP tool.run() Add five async tests that invoke the real MCP tool closures through FastMCP's tool.run(args, context=None) API. Passing context=None causes require_scopes to treat the call as BasicAuth mode (scope check bypassed), while get_client is mocked at the module level to inject a controlled NextcloudClient. Tests cover: - nc_webdav_list_archive_members: read_file called, result routed through _list_zip_members and returned correctly - nc_webdav_read_archive_member: read_file called, member extracted as text - nc_webdav_download_to_temp: bytes written to disk, client.username stored in _temp_registry as the file owner - nc_webdav_cleanup_temp (success): client.username passed as owner, file removed, registry entry cleared - nc_webdav_cleanup_temp (wrong owner): bob cannot delete alice's file Co-Authored-By: Claude Sonnet 4.6 --- tests/unit/test_webdav_archive_tools.py | 157 ++++++++++++++++++++++++ 1 file changed, 157 insertions(+) diff --git a/tests/unit/test_webdav_archive_tools.py b/tests/unit/test_webdav_archive_tools.py index 08f8a0734..f04438eff 100644 --- a/tests/unit/test_webdav_archive_tools.py +++ b/tests/unit/test_webdav_archive_tools.py @@ -334,3 +334,160 @@ def test_cleanup_temp_file_not_found_discards_registry(tmp_path): assert path not in _temp_registry finally: _temp_registry.pop(path, None) + + +# --------------------------------------------------------------------------- +# MCP tool wiring — invoke real tool closures via FastMCP tool.run() +# +# tool.run(args_dict, context=None) passes ctx=None to the tool function. +# require_scopes treats ctx=None as BasicAuth mode and skips scope checks, +# so we can test the full wiring (get_client → read_file → helper → result) +# by mocking get_client at the module level. +# --------------------------------------------------------------------------- + + +def _make_tool_map(): + """Build a FastMCP instance and return its tool map (name → Tool).""" + from mcp.server.fastmcp import FastMCP + + from nextcloud_mcp_server.server.webdav import configure_webdav_tools + + mcp = FastMCP("test") + configure_webdav_tools(mcp) + return {t.name: t for t in mcp._tool_manager.list_tools()} + + +@pytest.mark.unit +async def test_tool_list_archive_members_wiring(mocker): + """nc_webdav_list_archive_members calls read_file and delegates to _list_zip_members.""" + zip_bytes = make_zip({"content.xml": b"", "mimetype": b"application/ods"}) + mock_client = mocker.AsyncMock() + mock_client.webdav.read_file = mocker.AsyncMock( + return_value=(zip_bytes, "application/vnd.oasis.opendocument.spreadsheet") + ) + mocker.patch( + "nextcloud_mcp_server.server.webdav.get_client", return_value=mock_client + ) + + tools = _make_tool_map() + result = await tools["nc_webdav_list_archive_members"].run( + {"path": "docs/test.ods"}, context=None + ) + + mock_client.webdav.read_file.assert_awaited_once_with("docs/test.ods") + assert result["path"] == "docs/test.ods" + assert result["member_count"] == 2 + assert result["content_type"] == "application/vnd.oasis.opendocument.spreadsheet" + + +@pytest.mark.unit +async def test_tool_read_archive_member_wiring(mocker): + """nc_webdav_read_archive_member calls read_file and delegates to _read_zip_member.""" + xml = b"" + zip_bytes = make_zip({"content.xml": xml}) + mock_client = mocker.AsyncMock() + mock_client.webdav.read_file = mocker.AsyncMock( + return_value=(zip_bytes, "application/zip") + ) + mocker.patch( + "nextcloud_mcp_server.server.webdav.get_client", return_value=mock_client + ) + + tools = _make_tool_map() + result = await tools["nc_webdav_read_archive_member"].run( + {"path": "docs/test.ods", "member_path": "content.xml"}, context=None + ) + + mock_client.webdav.read_file.assert_awaited_once_with("docs/test.ods") + assert result["content"] == xml.decode("utf-8") + assert "encoding" not in result + + +@pytest.mark.unit +async def test_tool_download_to_temp_writes_file_and_registers_owner(mocker): + """nc_webdav_download_to_temp writes bytes to disk and records username in registry.""" + file_content = b"binary payload" + mock_client = mocker.AsyncMock() + mock_client.webdav.read_file = mocker.AsyncMock( + return_value=(file_content, "application/octet-stream") + ) + mock_client.username = "alice" + mocker.patch( + "nextcloud_mcp_server.server.webdav.get_client", return_value=mock_client + ) + + tools = _make_tool_map() + result = await tools["nc_webdav_download_to_temp"].run( + {"path": "Videos/clip.mp4"}, context=None + ) + + local_path = result["local_path"] + try: + assert result["filename"] == "clip.mp4" + assert result["size"] == len(file_content) + assert os.path.exists(local_path) + assert open(local_path, "rb").read() == file_content + # Ownership must be recorded under alice's username + assert _temp_registry.get(local_path) == "alice" + finally: + _temp_registry.pop(local_path, None) + if os.path.exists(local_path): + os.unlink(local_path) + + +@pytest.mark.unit +async def test_tool_cleanup_temp_passes_owner_from_client(mocker, tmp_path): + """nc_webdav_cleanup_temp passes client.username as owner to _cleanup_temp_path.""" + p = tmp_path / "nc_download_wiring.bin" + p.write_bytes(b"data") + path = str(p) + _temp_registry[path] = "alice" + + mock_client = mocker.AsyncMock() + mock_client.username = "alice" + mocker.patch( + "nextcloud_mcp_server.server.webdav.get_client", return_value=mock_client + ) + + try: + tools = _make_tool_map() + result = await tools["nc_webdav_cleanup_temp"].run( + {"local_path": path}, context=None + ) + + assert result["status"] == "ok" + assert not os.path.exists(path) + assert path not in _temp_registry + finally: + _temp_registry.pop(path, None) + if p.exists(): + p.unlink() + + +@pytest.mark.unit +async def test_tool_cleanup_temp_rejects_wrong_owner(mocker, tmp_path): + """nc_webdav_cleanup_temp rejects a caller whose username doesn't match the registry.""" + p = tmp_path / "nc_download_wiring2.bin" + p.write_bytes(b"data") + path = str(p) + _temp_registry[path] = "alice" + + mock_client = mocker.AsyncMock() + mock_client.username = "bob" + mocker.patch( + "nextcloud_mcp_server.server.webdav.get_client", return_value=mock_client + ) + + try: + tools = _make_tool_map() + result = await tools["nc_webdav_cleanup_temp"].run( + {"local_path": path}, context=None + ) + + assert result["status"] == "error" + assert "permission" in result["message"].lower() + assert os.path.exists(path) + finally: + _temp_registry.pop(path, None) + if p.exists(): + p.unlink() From b1a033880cefdd69ee89addf6197249110f6fb88 Mon Sep 17 00:00:00 2001 From: Jos Poortvliet Date: Tue, 5 May 2026 17:56:14 +0200 Subject: [PATCH 11/12] fix(webdav): directory-entry guard and download size cap _read_zip_member: reject directory member paths with a clean ValueError before calling zf.read(), which would otherwise raise an opaque stdlib error. Add a regression test using ZipFile.mkdir(). nc_webdav_download_to_temp: add _MAX_TEMP_DOWNLOAD_BYTES (500 MB) and check len(content) after read_file, raising ValueError with a clear message if exceeded. This limits unintended disk consumption in remote HTTP deployments where callers cannot use the returned path anyway. Add a wiring test that monkeypatches the limit and asserts ToolError. Co-Authored-By: Claude Sonnet 4.6 --- nextcloud_mcp_server/server/webdav.py | 18 +++++++++++++ tests/unit/test_webdav_archive_tools.py | 35 +++++++++++++++++++++++++ 2 files changed, 53 insertions(+) diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index 624474c07..e7288a802 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -47,6 +47,11 @@ def _cleanup_temp_files_on_exit() -> None: atexit.register(_cleanup_temp_files_on_exit) +# Maximum file size accepted by nc_webdav_download_to_temp. +# Prevents unbounded disk writes, especially in remote-HTTP deployments where +# the caller cannot use the local path anyway. +_MAX_TEMP_DOWNLOAD_BYTES: int = 500 * 1024 * 1024 # 500 MB + # --------------------------------------------------------------------------- # Pure helpers — no MCP context required, fully unit-testable # --------------------------------------------------------------------------- @@ -156,6 +161,12 @@ def _read_zip_member(content: bytes, path: str, member_path: str) -> dict: + (" (truncated)" if len(available) > 30 else "") ) from exc + if info.is_dir(): + raise ValueError( + f"Member '{member_path}' is a directory entry; " + f"only file members can be read." + ) + if info.file_size > _MAX_MEMBER_BYTES: raise ValueError( f"Member '{member_path}' uncompressed size " @@ -908,6 +919,13 @@ async def nc_webdav_download_to_temp(path: str, ctx: Context) -> dict: client = await get_client(ctx) content, content_type = await client.webdav.read_file(path) + if len(content) > _MAX_TEMP_DOWNLOAD_BYTES: + raise ValueError( + f"File '{path}' is {len(content):,} bytes, which exceeds the " + f"{_MAX_TEMP_DOWNLOAD_BYTES // (1024 * 1024)} MB limit for " + f"nc_webdav_download_to_temp." + ) + filename = os.path.basename(path.rstrip("/")) _root, suffix = os.path.splitext(filename) diff --git a/tests/unit/test_webdav_archive_tools.py b/tests/unit/test_webdav_archive_tools.py index f04438eff..180d9116f 100644 --- a/tests/unit/test_webdav_archive_tools.py +++ b/tests/unit/test_webdav_archive_tools.py @@ -156,6 +156,20 @@ def test_read_member_binary_returned_as_base64(): # --------------------------------------------------------------------------- +@pytest.mark.unit +def test_read_member_directory_entry_raises_value_error(): + """Passing a directory member path raises ValueError, not a stdlib error.""" + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + # ZipFile.mkdir creates a directory entry + zf.mkdir("subdir/") + zf.writestr("subdir/content.xml", b"") + content = buf.getvalue() + + with pytest.raises(ValueError, match="directory entry"): + _read_zip_member(content, "test.ods", "subdir/") + + @pytest.mark.unit def test_read_member_missing_member_raises_value_error(): """Missing member raises ValueError with the available file list.""" @@ -464,6 +478,27 @@ async def test_tool_cleanup_temp_passes_owner_from_client(mocker, tmp_path): p.unlink() +@pytest.mark.unit +async def test_tool_download_to_temp_rejects_oversized_file(mocker): + """nc_webdav_download_to_temp raises ValueError when the file exceeds _MAX_TEMP_DOWNLOAD_BYTES.""" + oversized = b"x" * 100 + mock_client = mocker.AsyncMock() + mock_client.webdav.read_file = mocker.AsyncMock( + return_value=(oversized, "application/octet-stream") + ) + mock_client.username = "alice" + mocker.patch( + "nextcloud_mcp_server.server.webdav.get_client", return_value=mock_client + ) + mocker.patch("nextcloud_mcp_server.server.webdav._MAX_TEMP_DOWNLOAD_BYTES", 10) + + from mcp.server.fastmcp.exceptions import ToolError + + tools = _make_tool_map() + with pytest.raises(ToolError, match="exceeds the"): + await tools["nc_webdav_download_to_temp"].run({"path": "big.bin"}, context=None) + + @pytest.mark.unit async def test_tool_cleanup_temp_rejects_wrong_owner(mocker, tmp_path): """nc_webdav_cleanup_temp rejects a caller whose username doesn't match the registry.""" From 5ace1baa9fd1f1f509416cd5192f69ca6f79b2ae Mon Sep 17 00:00:00 2001 From: Jos Poortvliet Date: Thu, 7 May 2026 00:48:09 +0200 Subject: [PATCH 12/12] fix(webdav): correct readOnlyHint, add archive size guard and member truncation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit download_to_temp annotation: remove incorrect readOnlyHint=True — this tool creates a file on disk and mutates _temp_registry, so it is not read-only. Replace with idempotentHint=False, which is accurate. Archive size guard: add _MAX_ARCHIVE_BYTES (100 MB) checked after read_file() in both nc_webdav_list_archive_members and nc_webdav_read_archive_member. Returns a clear error directing the caller to nc_webdav_download_to_temp for large archives, rather than silently exhausting worker RAM. Member truncation: _list_zip_members now accepts max_members (default 500, matching _MAX_ARCHIVE_MEMBERS). Archives with more entries still report the full member_count but the members list is capped; a truncated=True flag and truncated_at field are included so callers know the list is incomplete. Add unit tests for truncation (with/without limit exceeded) and ToolError wiring tests for the archive size guard on both tools. The streaming/OOM concern for very large archives (Copilot comments 3188640133, 3189892837 etc.) requires a client-layer streaming refactor that is out of scope for this PR; the size guard provides a bounded rejection instead. Co-Authored-By: Claude Sonnet 4.6 --- nextcloud_mcp_server/server/webdav.py | 59 +++++++++++++++++---- tests/unit/test_webdav_archive_tools.py | 68 +++++++++++++++++++++++++ 2 files changed, 117 insertions(+), 10 deletions(-) diff --git a/nextcloud_mcp_server/server/webdav.py b/nextcloud_mcp_server/server/webdav.py index e7288a802..6d2a02b31 100644 --- a/nextcloud_mcp_server/server/webdav.py +++ b/nextcloud_mcp_server/server/webdav.py @@ -52,6 +52,16 @@ def _cleanup_temp_files_on_exit() -> None: # the caller cannot use the local path anyway. _MAX_TEMP_DOWNLOAD_BYTES: int = 500 * 1024 * 1024 # 500 MB +# Maximum archive size for in-memory ZIP operations (list/read member). +# read_file() buffers the full archive in RAM; reject oversized archives +# before attempting extraction so workers don't OOM on huge ZIPs. +_MAX_ARCHIVE_BYTES: int = 100 * 1024 * 1024 # 100 MB + +# Maximum number of members returned by nc_webdav_list_archive_members. +# Large ZIP/JAR files can have thousands of entries; truncate to avoid +# flooding the MCP response and exhausting the context window. +_MAX_ARCHIVE_MEMBERS: int = 500 + # --------------------------------------------------------------------------- # Pure helpers — no MCP context required, fully unit-testable # --------------------------------------------------------------------------- @@ -89,22 +99,29 @@ def _cleanup_temp_files_on_exit() -> None: ) -def _list_zip_members(content: bytes, path: str, content_type: str) -> dict: +def _list_zip_members( + content: bytes, path: str, content_type: str, max_members: int = 500 +) -> dict: """Return the member listing of a ZIP archive as a plain dict. Args: content: Raw bytes of the archive. path: Nextcloud path (used only in error messages). content_type: MIME type reported by Nextcloud (included in result). + max_members: Maximum number of members to include in the result. + The total member count is always reported; a + ``truncated`` flag is set when the list is cut. Returns: - Dict with path, content_type, archive_size, member_count, members. + Dict with path, content_type, archive_size, member_count, members, + and an optional truncated=True when the list exceeds max_members. Raises: ValueError: if *content* is not a valid ZIP archive. """ try: with zipfile.ZipFile(io.BytesIO(content)) as zf: + all_infos = zf.infolist() members = [ { "name": info.filename, @@ -112,7 +129,7 @@ def _list_zip_members(content: bytes, path: str, content_type: str) -> dict: "compressed_size": info.compress_size, "is_dir": info.is_dir(), } - for info in zf.infolist() + for info in all_infos[:max_members] ] except zipfile.BadZipFile as exc: raise ValueError( @@ -121,13 +138,18 @@ def _list_zip_members(content: bytes, path: str, content_type: str) -> dict: f"use nc_webdav_download_to_temp." ) from exc - return { + total = len(all_infos) + result: dict = { "path": path, "content_type": content_type, "archive_size": len(content), - "member_count": len(members), + "member_count": total, "members": members, } + if total > max_members: + result["truncated"] = True + result["truncated_at"] = max_members + return result def _read_zip_member(content: bytes, path: str, member_path: str) -> dict: @@ -809,15 +831,25 @@ async def nc_webdav_list_archive_members(path: str, ctx: Context) -> dict: Returns: Dict with path, content_type, archive_size, member_count, and a - members list. Each member has: name, size (uncompressed), - compressed_size, is_dir. + members list (capped at 500 entries). Each member has: name, + size (uncompressed), compressed_size, is_dir. If the archive + has more than 500 members the result also contains + truncated=True and truncated_at=500. Raises: - ValueError: if the file is not a valid ZIP archive + ValueError: if the archive exceeds 100 MB or is not valid ZIP """ client = await get_client(ctx) content, content_type = await client.webdav.read_file(path) - return _list_zip_members(content, path, content_type) + if len(content) > _MAX_ARCHIVE_BYTES: + raise ValueError( + f"Archive '{path}' is {len(content):,} bytes, which exceeds the " + f"{_MAX_ARCHIVE_BYTES // (1024 * 1024)} MB in-memory limit. " + f"Use nc_webdav_download_to_temp to work with it locally." + ) + return _list_zip_members( + content, path, content_type, max_members=_MAX_ARCHIVE_MEMBERS + ) @mcp.tool( title="Read Archive Member", @@ -862,12 +894,19 @@ async def nc_webdav_read_archive_member( """ client = await get_client(ctx) content, _ = await client.webdav.read_file(path) + if len(content) > _MAX_ARCHIVE_BYTES: + raise ValueError( + f"Archive '{path}' is {len(content):,} bytes, which exceeds the " + f"{_MAX_ARCHIVE_BYTES // (1024 * 1024)} MB in-memory limit. " + f"Use nc_webdav_download_to_temp to work with it locally." + ) return _read_zip_member(content, path, member_path) @mcp.tool( title="Download File to Temp", annotations=ToolAnnotations( - readOnlyHint=True, + # Not read-only: creates a temp file on disk and mutates _temp_registry. + idempotentHint=False, openWorldHint=True, ), ) diff --git a/tests/unit/test_webdav_archive_tools.py b/tests/unit/test_webdav_archive_tools.py index 180d9116f..12c0fcafe 100644 --- a/tests/unit/test_webdav_archive_tools.py +++ b/tests/unit/test_webdav_archive_tools.py @@ -72,6 +72,30 @@ def test_list_members_bad_zip_raises_value_error(): _list_zip_members(b"this is not a zip", "bad.ods", "application/octet-stream") +@pytest.mark.unit +def test_list_members_truncated_when_over_limit(): + """_list_zip_members truncates results and sets truncated=True when limit exceeded.""" + members = {f"file_{i}.xml": b"" for i in range(10)} + content = make_zip(members) + result = _list_zip_members(content, "big.zip", "application/zip", max_members=3) + + assert result["member_count"] == 10 + assert len(result["members"]) == 3 + assert result["truncated"] is True + assert result["truncated_at"] == 3 + + +@pytest.mark.unit +def test_list_members_no_truncation_flag_when_within_limit(): + """_list_zip_members does not set truncated when all members fit.""" + content = make_zip({"a.xml": b"", "b.xml": b""}) + result = _list_zip_members(content, "small.zip", "application/zip", max_members=10) + + assert result["member_count"] == 2 + assert len(result["members"]) == 2 + assert "truncated" not in result + + @pytest.mark.unit def test_list_members_includes_content_type_in_result(): """content_type from Nextcloud is passed through to the result dict.""" @@ -478,6 +502,50 @@ async def test_tool_cleanup_temp_passes_owner_from_client(mocker, tmp_path): p.unlink() +@pytest.mark.unit +async def test_tool_list_archive_members_rejects_oversized_archive(mocker): + """nc_webdav_list_archive_members raises ToolError when archive exceeds _MAX_ARCHIVE_BYTES.""" + from mcp.server.fastmcp.exceptions import ToolError + + oversized = b"x" * 200 + mock_client = mocker.AsyncMock() + mock_client.webdav.read_file = mocker.AsyncMock( + return_value=(oversized, "application/zip") + ) + mocker.patch( + "nextcloud_mcp_server.server.webdav.get_client", return_value=mock_client + ) + mocker.patch("nextcloud_mcp_server.server.webdav._MAX_ARCHIVE_BYTES", 100) + + tools = _make_tool_map() + with pytest.raises(ToolError, match="exceeds the"): + await tools["nc_webdav_list_archive_members"].run( + {"path": "huge.zip"}, context=None + ) + + +@pytest.mark.unit +async def test_tool_read_archive_member_rejects_oversized_archive(mocker): + """nc_webdav_read_archive_member raises ToolError when archive exceeds _MAX_ARCHIVE_BYTES.""" + from mcp.server.fastmcp.exceptions import ToolError + + oversized = b"x" * 200 + mock_client = mocker.AsyncMock() + mock_client.webdav.read_file = mocker.AsyncMock( + return_value=(oversized, "application/zip") + ) + mocker.patch( + "nextcloud_mcp_server.server.webdav.get_client", return_value=mock_client + ) + mocker.patch("nextcloud_mcp_server.server.webdav._MAX_ARCHIVE_BYTES", 100) + + tools = _make_tool_map() + with pytest.raises(ToolError, match="exceeds the"): + await tools["nc_webdav_read_archive_member"].run( + {"path": "huge.zip", "member_path": "content.xml"}, context=None + ) + + @pytest.mark.unit async def test_tool_download_to_temp_rejects_oversized_file(mocker): """nc_webdav_download_to_temp raises ValueError when the file exceeds _MAX_TEMP_DOWNLOAD_BYTES."""