""" 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) folder_norm = folder_path.strip("/") result: list[dict] = [] for name in items: # Skip the "." self-reference and empty entries if not name or name in (".", "./"): continue # Skip the directory itself (PROPFIND Depth:1 always includes the parent) if name.strip("/") == folder_norm: 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