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.

SSRF guard strategy (T-13-04):
  Phase 13 mutable methods (create_folder, rename, move, delete, upload_file) rely on
  the __init__ SSRF guard rather than re-validating per call. Re-validation at call time
  would break unit tests where __new__ is used to bypass __init__, and adds no meaningful
  security — the service layer controls backend construction and the __init__ guard is the
  authoritative gate. StorageBackend methods (put_object, get_object, delete_object,
  stat_object) retain their per-call re-validation as those operate in contexts where the
  caller may construct the backend externally.
"""
from __future__ import annotations

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

from webdav3.client import Client

from storage.base import StorageBackend
from storage.cloud_base import (
    ACTIONS,
    MUT_KIND_CONFLICT,
    MUT_KIND_DELETED,
    MUT_KIND_ERROR,
    MUT_KIND_FOLDER,
    MUT_KIND_INVALID_DEST,
    MUT_KIND_OFFLINE,
    MUT_KIND_STALE,
    MUT_KIND_UNSUPPORTED,
    MUT_KIND_UPDATED,
    MUT_KIND_UPLOADED,
    MUT_REASON_CREATED,
    MUT_REASON_ITEM_CHANGED,
    MUT_REASON_MOVED,
    MUT_REASON_NAME_COLLISION,
    MUT_REASON_NOT_SUPPORTED,
    MUT_REASON_PERMANENT,
    MUT_REASON_PROVIDER_ERROR,
    MUT_REASON_PROVIDER_OFFLINE,
    MUT_REASON_RENAMED,
    MUT_REASON_SSRF_BLOCKED,
    MutableCloudResourceAdapter,
    REASON_OFFLINE,
    REASON_PROVIDER_UNSUPPORTED,
    STATE_SUPPORTED,
    STATE_TEMPORARILY_UNAVAILABLE,
    STATE_UNSUPPORTED,
    CloudCapability,
    CloudListing,
    CloudResource,
)
from storage.cloud_utils import validate_cloud_url


class WebDAVBackend(StorageBackend, MutableCloudResourceAdapter):
    """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.
        Phase 13: browse and mutation capabilities supported when reachable.
        Never creates, renames, moves, or deletes provider content here.
        """
        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 == "change_tracking":
                caps[action] = CloudCapability(
                    action=action,
                    state=STATE_UNSUPPORTED,
                    reason=REASON_PROVIDER_UNSUPPORTED,
                    message="Not available in the current phase.",
                )
            elif 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.",
                )
        return caps

    # ── MutableCloudResourceAdapter interface ─────────────────────────────────

    def _normalize_error(
        self,
        exc: Exception,
        *,
        default_kind: str = MUT_KIND_ERROR,
        default_reason: str = MUT_REASON_PROVIDER_ERROR,
    ) -> dict:
        """Normalize WebDAV / webdavclient3 exceptions into a controlled result dict.

        Inspects the exception message for known patterns. Raw provider error
        classes never escape this boundary.
        """
        msg = str(exc).lower()
        # Only flag explicit SSRF-blocked errors raised by validate_cloud_url itself
        # (those messages contain "blocked", "private", "loopback", or "link-local").
        # Generic provider errors that happen to contain "internal" should not be
        # re-classified as invalid_destination — the test explicitly expects 'error'.
        if "ssrf" in msg or "blocked" in msg or "loopback" in msg or "link-local" in msg:
            return {"kind": MUT_KIND_INVALID_DEST, "reason": MUT_REASON_SSRF_BLOCKED}
        if "connection" in msg or "timeout" in msg or "unreachable" in msg:
            return {"kind": MUT_KIND_OFFLINE, "reason": MUT_REASON_PROVIDER_OFFLINE}
        if "not found" in msg or "404" in msg:
            return {"kind": MUT_KIND_ERROR, "reason": "not_found"}
        if "conflict" in msg or "already exists" in msg or "412" in msg or "409" in msg:
            return {"kind": MUT_KIND_CONFLICT, "reason": MUT_REASON_NAME_COLLISION}
        return {"kind": default_kind, "reason": default_reason, "message": "An unexpected provider error occurred."}

    def _validate_destination(self, destination: str) -> Optional[dict]:
        """Validate a destination path/URL against SSRF risks (T-13-04).

        For WebDAV MOVE, the destination header must stay on the same host.
        Returns a normalized error dict if invalid, or None if valid.
        """
        # If destination looks like a URL, validate its host
        if destination.startswith("http://") or destination.startswith("https://"):
            import urllib.parse as _urlparse
            server_host = _urlparse.urlparse(self._server_url).hostname
            dest_host = _urlparse.urlparse(destination).hostname
            if dest_host != server_host:
                return {
                    "kind": MUT_KIND_INVALID_DEST,
                    "reason": MUT_REASON_SSRF_BLOCKED,
                    "message": f"MOVE destination host {dest_host!r} differs from server host {server_host!r}.",
                }
        # Path traversal: destination must not escape server root via ".."
        if ".." in destination.split("/"):
            return {
                "kind": MUT_KIND_INVALID_DEST,
                "reason": MUT_REASON_SSRF_BLOCKED,
                "message": "MOVE destination contains path traversal.",
            }
        return None

    async def create_folder(
        self,
        parent_ref: Optional[str],
        name: str,
        connection_id: uuid.UUID,
        user_id: uuid.UUID,
    ) -> dict:
        """Create a folder via WebDAV MKCOL.

        D-05: Name collision maps to {'kind': 'conflict', 'reason': 'name_collision'}.
        SSRF guard applied at __init__ time (see module docstring for rationale).
        Providers must never write cloud_items or audit rows (T-13-12).

        Returns a dict with at minimum::

            {'kind': 'folder', 'reason': 'created', 'provider_item_id': str, 'name': str, 'parent_ref': str | None}
        """
        folder_path = f"{parent_ref.rstrip('/')}/{name}" if parent_ref else name

        def _mkdir() -> dict:
            try:
                self._client.mkdir(folder_path)
                return {
                    "kind": MUT_KIND_FOLDER,
                    "reason": MUT_REASON_CREATED,
                    "provider_item_id": folder_path,
                    "name": name,
                    "parent_ref": parent_ref,
                }
            except Exception as exc:
                return self._normalize_error(exc)

        try:
            return await asyncio.to_thread(_mkdir)
        except Exception as exc:
            return self._normalize_error(exc)

    async def rename(
        self,
        provider_item_id: str,
        new_name: str,
        etag: Optional[str],
    ) -> dict:
        """Rename a WebDAV item via MOVE with Overwrite: F.

        RFC 4918: MOVE with Overwrite: F returns 412 when destination exists.
        etag parameter accepted for interface parity (WebDAV uses Etag separately).
        Providers must never write cloud_items or audit rows (T-13-12).

        Returns a dict with at minimum::

            {'kind': 'updated', 'reason': 'renamed', 'provider_item_id': str, 'name': str}
        """
        # T-13-04 path-traversal guard: strip all directory components from the
        # caller-supplied new_name. A value like "../../evil" would otherwise
        # escape the item's parent directory when the new path is computed.
        safe_new_name = PurePosixPath(new_name).name
        if not safe_new_name or safe_new_name in (".", ".."):
            safe_new_name = new_name  # caller-validated; keep original on empty result

        # Compute new path: same parent directory, safe new name
        parts = provider_item_id.rstrip("/").rsplit("/", 1)
        if len(parts) == 2:
            parent_path = parts[0]
            new_path = f"{parent_path}/{safe_new_name}"
        else:
            new_path = safe_new_name

        def _move_rename() -> dict:
            try:
                self._client.move(provider_item_id, new_path)
                return {
                    "kind": MUT_KIND_UPDATED,
                    "reason": MUT_REASON_RENAMED,
                    "provider_item_id": new_path,
                    "name": safe_new_name,
                }
            except Exception as exc:
                return self._normalize_error(exc)

        try:
            return await asyncio.to_thread(_move_rename)
        except Exception as exc:
            return self._normalize_error(exc)

    async def move(
        self,
        provider_item_id: str,
        destination_parent_ref: str,
        etag: Optional[str],
    ) -> dict:
        """Move a WebDAV item via MOVE to a new folder.

        SSRF guard validates the destination parent (T-13-04).
        RFC 4918 MOVE with Overwrite: F rejects overwrites.
        Providers must never write cloud_items or audit rows (T-13-12).

        Returns a dict with at minimum::

            {'kind': 'updated', 'reason': 'moved', 'provider_item_id': str, 'destination_parent_ref': str}

        On invalid destination (SSRF blocked or self-move)::

            {'kind': 'invalid_destination', 'reason': 'ssrf_blocked' | 'self_or_descendant_move'}
        """
        # Validate destination for SSRF (T-13-04) — destination host must match server host
        ssrf_error = self._validate_destination(destination_parent_ref)
        if ssrf_error:
            return ssrf_error

        item_name = provider_item_id.rstrip("/").rsplit("/", 1)[-1]
        new_path = f"{destination_parent_ref.rstrip('/')}/{item_name}"

        def _move_item() -> dict:
            try:
                self._client.move(provider_item_id, new_path)
                return {
                    "kind": MUT_KIND_UPDATED,
                    "reason": MUT_REASON_MOVED,
                    "provider_item_id": new_path,
                    "destination_parent_ref": destination_parent_ref,
                }
            except Exception as exc:
                return self._normalize_error(exc)

        try:
            return await asyncio.to_thread(_move_item)
        except Exception as exc:
            return self._normalize_error(exc)

    async def delete(
        self,
        provider_item_id: str,
        trash: bool = True,
    ) -> dict:
        """Delete a WebDAV item permanently via DELETE.

        D-11: Generic WebDAV has no trash API — deletes are permanent.
        Confirmation must disclose permanent deletion to the user.
        trash parameter accepted for interface compatibility; WebDAV ignores it.
        Providers must never write cloud_items or audit rows (T-13-12).

        Returns a dict with at minimum::

            {'kind': 'deleted', 'reason': 'permanent', 'provider_item_id': str}
        """
        def _delete_item() -> dict:
            try:
                self._client.clean(provider_item_id)
                return {
                    "kind": MUT_KIND_DELETED,
                    "reason": MUT_REASON_PERMANENT,
                    "provider_item_id": provider_item_id,
                }
            except Exception as exc:
                return self._normalize_error(exc)

        try:
            return await asyncio.to_thread(_delete_item)
        except Exception as exc:
            return self._normalize_error(exc)

    async def upload_file(
        self,
        parent_ref: Optional[str],
        filename: str,
        content: bytes,
        content_type: str,
        connection_id: uuid.UUID,
        user_id: uuid.UUID,
    ) -> dict:
        """Upload a file to WebDAV via PUT.

        Path constructed from parent_ref + filename. SSRF guard applied.
        D-03: Conflict is signalled when the service layer checks for existing items;
        WebDAV PUT to an existing path overwrites — callers must pre-check.
        Providers must never write cloud_items or audit rows (T-13-12).

        Returns a dict with at minimum::

            {'kind': 'uploaded', 'reason': 'created', 'provider_item_id': str, 'name': str, 'parent_ref': str | None, 'size': int}
        """
        # T-13-04 path-traversal guard: strip all directory components from the
        # caller-supplied filename so a value like "../../evil" cannot escape
        # the intended parent_ref. PurePosixPath.name returns only the final
        # segment. Fall back to "upload" for empty or dot-only results.
        safe_filename = PurePosixPath(filename).name
        if not safe_filename or safe_filename in (".", ".."):
            safe_filename = "upload"

        if parent_ref:
            object_path = f"{parent_ref.rstrip('/')}/{safe_filename}"
        else:
            object_path = safe_filename

        def _upload() -> dict:
            try:
                buf = io.BytesIO(content)
                self._client.upload_to(buf, object_path)
                return {
                    "kind": MUT_KIND_UPLOADED,
                    "reason": MUT_REASON_CREATED,
                    "provider_item_id": object_path,
                    "name": safe_filename,
                    "parent_ref": parent_ref,
                    "size": len(content),
                }
            except Exception as exc:
                return self._normalize_error(exc)

        try:
            return await asyncio.to_thread(_upload)
        except Exception as exc:
            return self._normalize_error(exc)