kite/backend/storage/webdav_backend.py

"""
WebDAVBackend — Generic WebDAV StorageBackend implementation.

Security design (D-17 / T-05-04-01 / T-05-04-02):
  validate_cloud_url() is called at two mandatory points:
    1. In __init__, BEFORE constructing the webdavclient3 Client — blocks SSRF at
       construction time so a private-IP URL never reaches the SDK.
    2. Before EVERY asyncio.to_thread() call — defends against DNS-rebinding attacks
       where a hostname passes the initial check but later resolves to an internal IP.

Path encoding (Pitfall 2 from RESEARCH.md):
  _make_path() percent-encodes each path segment using urllib.parse.quote()
  so non-ASCII characters and spaces in UUIDs (which don't occur in practice,
  but are handled defensively) are properly escaped for Nextcloud/WebDAV.

Object key scheme:
  object_key = WebDAV path: "docuvault/{user_id}/{document_id}{extension}"
  This is the path relative to the WebDAV root, stored in the database as
  document.object_key for later get/delete operations.

WebDAV credentials dict shape:
  {"server_url": str, "username": str, "password": str}

Not implemented (D-14):
  presigned_get_url and generate_presigned_put_url raise NotImplementedError.
  Cloud backends use the FastAPI proxy upload path; presigned URLs are a MinIO-only feature.
"""
from __future__ import annotations

import asyncio
import io
import uuid
import urllib.parse
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

from webdav3.client import Client

from storage.base import StorageBackend
from storage.cloud_base import (
    ACTIONS,
    REASON_OFFLINE,
    REASON_PROVIDER_UNSUPPORTED,
    STATE_SUPPORTED,
    STATE_TEMPORARILY_UNAVAILABLE,
    STATE_UNSUPPORTED,
    CloudCapability,
    CloudListing,
    CloudResource,
    CloudResourceAdapter,
)
from storage.cloud_utils import validate_cloud_url


class WebDAVBackend(StorageBackend, CloudResourceAdapter):
    """Generic WebDAV storage backend implementing all 7 StorageBackend abstract methods.

    All synchronous webdavclient3 calls are wrapped in asyncio.to_thread() to avoid
    blocking the FastAPI event loop (mirrors MinIOBackend pattern — RESEARCH.md Pattern 5).

    The SSRF guard (validate_cloud_url) is called in __init__ and before every
    asyncio.to_thread() call for defense-in-depth against DNS rebinding (D-17).
    """

    def __init__(self, server_url: str, username: str, password: str) -> None:
        """Construct a WebDAVBackend.

        Args:
            server_url: The WebDAV root URL (e.g. "https://dav.example.com/remote.php/dav/").
            username:   HTTP Basic Auth username.
            password:   HTTP Basic Auth password or app-specific password (D-07).

        Raises:
            ValueError: If server_url targets a private/internal address (SSRF guard, D-17).
        """
        # SSRF guard: validate before any SDK construction — T-05-04-01
        validate_cloud_url(server_url)
        self._server_url = server_url

        options = {
            "webdav_hostname": server_url,
            "webdav_login": username,
            "webdav_password": password,
        }
        self._client = Client(options)

    def _make_path(self, user_id: str, document_id: str, extension: str) -> str:
        """Construct a WebDAV path for a document.

        Path schema: "docuvault/{user_id}/{document_id}{extension}"

        Each segment is percent-encoded (Pitfall 2 — Nextcloud/WebDAV compatibility
        with non-ASCII characters and spaces, even though UUIDs are alphanumeric).

        Args:
            user_id:     User UUID string.
            document_id: Document UUID string.
            extension:   File extension including leading dot (e.g. ".pdf").

        Returns:
            WebDAV path string suitable for use as object_key.
        """
        encoded_uid = urllib.parse.quote(str(user_id), safe="")
        encoded_did = urllib.parse.quote(str(document_id), safe="")
        return f"docuvault/{encoded_uid}/{encoded_did}{extension}"

    async def put_object(
        self,
        user_id: str,
        document_id: str,
        file_bytes: bytes,
        extension: str,
        content_type: str,
        cloud_folder: str | None = None,
        original_filename: str | None = None,
    ) -> str:
        """Upload bytes to WebDAV and return the object_key (WebDAV path).

        When cloud_folder is provided the file is stored inside that folder
        (e.g. "Documents/") using the original filename so it appears naturally
        in the user's cloud folder browser. When omitted the default
        DocuVault-managed UUID path is used.
        """
        validate_cloud_url(self._server_url)
        if cloud_folder:
            parent_dir = cloud_folder.rstrip("/")
            # Use original filename (basename only — path traversal guard)
            safe_name = Path(original_filename).name if original_filename else f"{document_id}{extension}"
            object_key = f"{parent_dir}/{safe_name}" if parent_dir else safe_name
        else:
            object_key = self._make_path(user_id, document_id, extension)
            parent_dir = f"docuvault/{urllib.parse.quote(str(user_id), safe='')}"
        await asyncio.to_thread(self._client.mkdir, parent_dir, True)
        buf = io.BytesIO(file_bytes)
        await asyncio.to_thread(self._client.upload_to, buf, object_key)
        return object_key

    async def get_object(self, object_key: str) -> bytes:
        """Download bytes from WebDAV by object_key (WebDAV path).

        Args:
            object_key: WebDAV path returned by put_object.

        Returns:
            File content as bytes.

        Raises:
            ValueError: If SSRF guard fires on re-validation (D-17).
        """
        # Re-validate before every outbound request (D-17)
        validate_cloud_url(self._server_url)
        buf = io.BytesIO()
        await asyncio.to_thread(self._client.download_from, buf, object_key)
        return buf.getvalue()

    async def delete_object(self, object_key: str) -> None:
        """Delete an object from WebDAV by object_key.

        Silently ignores missing files (no-op if object_key does not exist).
        Any WebDAV exception during delete is swallowed — consistent with the
        StorageBackend ABC contract ("No-op if the key does not exist").

        Args:
            object_key: WebDAV path returned by put_object.
        """
        # Re-validate before every outbound request (D-17)
        validate_cloud_url(self._server_url)
        try:
            await asyncio.to_thread(self._client.clean, object_key)
        except Exception:
            # Covers FileNotFoundError, WebDavException, and any other exception
            # for missing or already-deleted files
            pass

    async def presigned_get_url(self, object_key: str, expires_minutes: int = 60) -> str:
        """Not supported for WebDAV backends.

        WebDAV does not have a concept of presigned URLs. Cloud document downloads
        use the FastAPI proxy endpoint (D-15 — same as all cloud backends).

        Raises:
            NotImplementedError: Always.
        """
        raise NotImplementedError("WebDAV backend does not support presigned GET URLs")

    async def generate_presigned_put_url(
        self, object_key: str, expires_minutes: int = 15
    ) -> str:
        """Not supported for WebDAV backends.

        Cloud backends use the FastAPI intermediary upload path (D-14).
        Presigned PUT URLs are a MinIO/S3-only feature.

        Raises:
            NotImplementedError: Always.
        """
        raise NotImplementedError(
            "WebDAV backend does not support presigned PUT URLs — use the upload proxy endpoint"
        )

    async def stat_object(self, object_key: str) -> int:
        """Return the file size in bytes from the WebDAV server.

        Uses client.info() which performs a PROPFIND request and returns a dict
        including a "size" key with the content length.

        Args:
            object_key: WebDAV path returned by put_object.

        Returns:
            File size in bytes. Returns 0 if the server does not report a size.

        Raises:
            ValueError: If SSRF guard fires on re-validation (D-17).
        """
        # Re-validate before every outbound request (D-17)
        validate_cloud_url(self._server_url)
        info = await asyncio.to_thread(self._client.info, object_key)
        return int(info.get("size", 0))

    async def health_check(self) -> bool:
        """Return True if the WebDAV server root ("/") is reachable.

        Uses a lightweight client.check("/") PROPFIND call — no file read/write.
        Per D-08: connection health is validated before credentials are stored.

        Returns:
            True if the server is reachable and the WebDAV root exists, 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

    # ── CloudResourceAdapter interface ────────────────────────────────────────

    async def list_folder(
        self,
        connection_id: uuid.UUID,
        user_id: uuid.UUID,
        parent_ref: Optional[str] = None,
        page_token: Optional[str] = None,
    ) -> CloudListing:
        """List direct children of a WebDAV folder via non-mutating PROPFIND.

        parent_ref=None lists the WebDAV root.
        Uses client.list() + client.info() — no file byte download, no mutation.
        SSRF guard is re-validated before every outbound request (D-17).
        Absent properties treated as unknown (not unsupported).
        """
        folder_path = parent_ref if parent_ref else ""

        def _propfind() -> list[dict]:
            validate_cloud_url(self._server_url)
            items = self._client.list(folder_path)
            folder_norm = folder_path.strip("/")
            result: list[dict] = []
            for name in items:
                if not name or name in (".", "./"):
                    continue
                if name.strip("/") == folder_norm:
                    continue
                if folder_path:
                    item_path = f"{folder_path.rstrip('/')}/{name}"
                else:
                    item_path = name
                validate_cloud_url(self._server_url)
                try:
                    info = self._client.info(item_path)
                    size_raw = info.get("size")
                    size = int(size_raw) if size_raw is not None else 0
                    is_dir = (
                        info.get("isdir", False)
                        or str(info.get("content_type", "")).startswith("httpd/unix-directory")
                        or name.endswith("/")
                    )
                    etag = info.get("etag") or info.get("getetag")
                    mod_str = info.get("modified") or info.get("getlastmodified")
                    content_type = info.get("content_type") or info.get("getcontenttype")
                except Exception:
                    size = 0
                    is_dir = name.endswith("/")
                    etag = None
                    mod_str = None
                    content_type = None
                result.append({
                    "path": item_path,
                    "name": name.rstrip("/"),
                    "is_dir": is_dir,
                    "size": size,
                    "etag": etag,
                    "modified": mod_str,
                    "content_type": content_type,
                })
            return result

        try:
            raw_items = await asyncio.to_thread(_propfind)
            complete = True
        except Exception:
            return CloudListing(items=(), complete=False)

        resources: list[CloudResource] = []
        for item in raw_items:
            kind = "folder" if item["is_dir"] else "file"
            size: Optional[int] = None if item["is_dir"] else item["size"]
            modified_at: Optional[datetime] = None
            if item.get("modified"):
                try:
                    modified_at = datetime.fromisoformat(item["modified"])
                except (ValueError, TypeError):
                    pass
            resources.append(
                CloudResource(
                    id=uuid.uuid4(),
                    provider_item_id=item["path"],
                    connection_id=connection_id,
                    user_id=user_id,
                    name=item["name"],
                    kind=kind,
                    parent_ref=parent_ref,
                    content_type=item.get("content_type") if not item["is_dir"] else None,
                    size=size,
                    modified_at=modified_at,
                    etag=item.get("etag"),
                )
            )
        return CloudListing(items=tuple(resources), complete=complete)

    async def get_capabilities(
        self,
        connection_id: uuid.UUID,
        user_id: uuid.UUID,
    ) -> dict[str, CloudCapability]:
        """Return connection-level capabilities for WebDAV.

        Uses non-mutating OPTIONS probe to verify server reachability.
        WebDAV supports browse; mutations pending Phase 13.
        Never creates, renames, moves, or deletes provider content.
        """
        try:
            validate_cloud_url(self._server_url)
            reachable = await asyncio.to_thread(self._client.check, "/")
        except Exception:
            reachable = False

        caps: dict[str, CloudCapability] = {}
        for action in ACTIONS:
            if action == "browse":
                if reachable:
                    caps[action] = CloudCapability(action=action, state=STATE_SUPPORTED)
                else:
                    caps[action] = CloudCapability(
                        action=action,
                        state=STATE_TEMPORARILY_UNAVAILABLE,
                        reason=REASON_OFFLINE,
                        message="WebDAV server is unreachable.",
                    )
            else:
                caps[action] = CloudCapability(
                    action=action,
                    state=STATE_UNSUPPORTED,
                    reason=REASON_PROVIDER_UNSUPPORTED,
                    message="Not available in the current phase.",
                )
        return caps