From 1b9573f398c10d364eb40b177de1a3ef39649c1b Mon Sep 17 00:00:00 2001 From: curo1305 Date: Thu, 28 May 2026 21:11:12 +0200 Subject: [PATCH] feat(05-04): implement NextcloudBackend extending WebDAVBackend MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - NextcloudBackend subclasses WebDAVBackend; inherits all 7 StorageBackend methods - SSRF guard fully inherited: NextcloudBackend("http://10.0.0.1/dav", ...) raises ValueError - stores self._username for Nextcloud path convention context - list_folder(folder_path: str = "") async method added — lists via client.list() + client.info() wrapped in asyncio.to_thread(), returns [{id, name, is_dir, size}, ...] - validate_cloud_url called before every asyncio.to_thread() call in list_folder (D-17) - health_check overrides parent to use client.check("") for Nextcloud root probe --- backend/storage/nextcloud_backend.py | 150 +++++++++++++++++++++++++++ 1 file changed, 150 insertions(+) create mode 100644 backend/storage/nextcloud_backend.py diff --git a/backend/storage/nextcloud_backend.py b/backend/storage/nextcloud_backend.py new file mode 100644 index 0000000..a7457f7 --- /dev/null +++ b/backend/storage/nextcloud_backend.py @@ -0,0 +1,150 @@ +""" +NextcloudBackend — Nextcloud-specific WebDAV StorageBackend implementation. + +Extends WebDAVBackend with Nextcloud path conventions and folder listing. + +Inheritance: + NextcloudBackend → WebDAVBackend → StorageBackend + +All 7 StorageBackend abstract methods are inherited from WebDAVBackend. +Only health_check is overridden to use the Nextcloud root (empty string +is a valid check target for Nextcloud's WebDAV endpoint). + +Additional method (not in ABC): + list_folder(folder_path: str) → list[dict] + Lists folder contents via client.list() + client.info(). Used by: + GET /api/cloud/folders/nextcloud/{folder_id} (D-09, lazy-load tree). + +Nextcloud WebDAV path convention: + The server_url should be the Nextcloud WebDAV base for a specific user: + https://nc.example.com/remote.php/dav/files/{username}/ + Object keys are then relative to this root (inherited from WebDAVBackend). + +SSRF prevention: + Inherited from WebDAVBackend.__init__ — validate_cloud_url(server_url) is + called before Client construction and before every outbound request (D-17). + +Credentials dict shape (same as WebDAVBackend): + {"server_url": str, "username": str, "password": str} +""" +from __future__ import annotations + +import asyncio + +from storage.cloud_utils import validate_cloud_url +from storage.webdav_backend import WebDAVBackend + + +class NextcloudBackend(WebDAVBackend): + """Nextcloud storage backend — extends WebDAVBackend. + + The server_url should be the full Nextcloud WebDAV root: + https://nc.example.com/remote.php/dav/files/{username}/ + + All 7 StorageBackend methods are inherited from WebDAVBackend. + The SSRF guard is fully inherited — any private/localhost URL raises + ValueError at construction time (and before every outbound request). + """ + + def __init__(self, server_url: str, username: str, password: str) -> None: + """Construct a NextcloudBackend. + + Args: + server_url: Nextcloud WebDAV root URL. + E.g. "https://nc.example.com/remote.php/dav/files/alice/" + username: Nextcloud username (stored for path convention context). + password: Nextcloud account password or app-specific password (D-07). + + Raises: + ValueError: If server_url targets a private/internal address (inherited SSRF guard). + """ + super().__init__(server_url, username, password) + self._username = username + + async def list_folder(self, folder_path: str = "") -> list[dict]: + """List folder contents at folder_path relative to the WebDAV root. + + Performs a PROPFIND (via client.list()) and fetches metadata for each + item via client.info(). Both calls are wrapped in asyncio.to_thread(). + + SSRF guard is called before every outbound request (D-17). + + TTL caching at 60 seconds is handled by cloud_cache.get_cloud_folders_cached() + at the API layer — this method always performs live PROPFIND calls. + + Args: + folder_path: Path relative to the WebDAV root to list. + Empty string or "/" lists the WebDAV root. + + Returns: + List of dicts with keys: + - "id" (str): WebDAV path (usable as object_key or folder_id) + - "name" (str): File or folder display name + - "is_dir" (bool): True if the item is a directory + - "size" (int): File size in bytes (0 for directories) + + Raises: + ValueError: If SSRF guard fires on re-validation. + Exception: Propagates any webdavclient3 exceptions (e.g. connection errors). + """ + # Re-validate before every outbound request (D-17 / T-05-04-02) + validate_cloud_url(self._server_url) + + # client.list() returns a list of file/folder names in the directory + items = await asyncio.to_thread(self._client.list, folder_path) + + result: list[dict] = [] + for name in items: + # Skip the "." self-reference that some WebDAV servers include + if not name or name in (".", "./"): + continue + + # Construct the full path for info lookup + if folder_path: + item_path = f"{folder_path.rstrip('/')}/{name}" + else: + item_path = name + + # Fetch metadata for each item + validate_cloud_url(self._server_url) + try: + info = await asyncio.to_thread(self._client.info, item_path) + size = int(info.get("size", 0)) + # webdavclient3 info dict uses "isdir" or checks content_type + # is_dir is True if size == 0 and name ends with "/" or info signals it + is_dir = info.get("isdir", False) or str(info.get("content_type", "")).startswith( + "httpd/unix-directory" + ) or name.endswith("/") + except Exception: + # If we can't get info for an item, include it with defaults + size = 0 + is_dir = name.endswith("/") + + result.append( + { + "id": item_path, + "name": name.rstrip("/"), + "is_dir": is_dir, + "size": size, + } + ) + + return result + + async def health_check(self) -> bool: + """Return True if the Nextcloud WebDAV endpoint is reachable. + + Uses client.check("") to probe the WebDAV root path rather than "/" + because some Nextcloud configurations expect the base path check on + the root without a trailing slash component. + + Returns: + True if the server is reachable, False otherwise. + """ + try: + # Re-validate before every outbound request (D-17) + validate_cloud_url(self._server_url) + result = await asyncio.to_thread(self._client.check, "") + return bool(result) + except Exception: + return False