feat(05-04): implement NextcloudBackend extending WebDAVBackend
- 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
This commit is contained in:
@@ -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
|
||||
Reference in New Issue
Block a user