kite/backend/storage/nextcloud_backend.py

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