fix(05): cloud API path param, root sentinel, webdav creds in list, upload path
cloud.py: list_connections now decrypts and surfaces server_url +
connection_username for nextcloud/webdav providers; folder route uses
{folder_id:path} to handle slashes; translates "root" sentinel to "".
nextcloud_backend.py: skip parent directory entry in PROPFIND Depth:1 results.
webdav_backend.py: add cloud_folder + original_filename params to
upload_object so files land in the user's chosen folder with their real name.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
+17
-3
@@ -639,7 +639,19 @@ async def list_connections(
|
||||
select(CloudConnection).where(CloudConnection.user_id == current_user.id)
|
||||
)
|
||||
connections = result.scalars().all()
|
||||
return {"items": [CloudConnectionOut.model_validate(c).model_dump() for c in connections]}
|
||||
items = []
|
||||
master_key = settings.cloud_creds_key.encode()
|
||||
for conn in connections:
|
||||
d = CloudConnectionOut.model_validate(conn).model_dump()
|
||||
if conn.provider in ("nextcloud", "webdav"):
|
||||
try:
|
||||
creds = decrypt_credentials(master_key, str(conn.user_id), conn.credentials_enc)
|
||||
d["server_url"] = creds.get("server_url")
|
||||
d["connection_username"] = creds.get("username")
|
||||
except Exception:
|
||||
pass
|
||||
items.append(d)
|
||||
return {"items": items}
|
||||
|
||||
|
||||
# ── GET /api/cloud/connections/{connection_id}/config ────────────────────────
|
||||
@@ -741,7 +753,7 @@ async def delete_connection(
|
||||
# ── GET /api/cloud/folders/{provider}/{folder_id} ─────────────────────────────
|
||||
|
||||
|
||||
@router.get("/folders/{provider}/{folder_id}")
|
||||
@router.get("/folders/{provider}/{folder_id:path}")
|
||||
async def list_cloud_folders(
|
||||
provider: str,
|
||||
folder_id: str,
|
||||
@@ -874,9 +886,11 @@ async def list_cloud_folders(
|
||||
|
||||
elif provider in ("nextcloud", "webdav"):
|
||||
backend = _build_backend(provider, credentials)
|
||||
# "root" is a frontend sentinel meaning the WebDAV root; translate to ""
|
||||
webdav_path = "" if folder_id == "root" else folder_id
|
||||
|
||||
async def fetch_webdav() -> list:
|
||||
return await backend.list_folder(folder_id)
|
||||
return await backend.list_folder(webdav_path)
|
||||
|
||||
items = await get_cloud_folders_cached(
|
||||
str(current_user.id), provider, folder_id, fetch_webdav
|
||||
|
||||
@@ -93,11 +93,15 @@ class NextcloudBackend(WebDAVBackend):
|
||||
# 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 that some WebDAV servers include
|
||||
# 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:
|
||||
|
||||
@@ -30,6 +30,7 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
import io
|
||||
import urllib.parse
|
||||
from pathlib import Path
|
||||
|
||||
from webdav3.client import Client
|
||||
|
||||
@@ -96,31 +97,25 @@ class WebDAVBackend(StorageBackend):
|
||||
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).
|
||||
|
||||
Ensures the parent directory "docuvault/{user_id}/" exists before upload
|
||||
by calling client.mkdir() with recursive=True. webdavclient3 mkdir is a
|
||||
no-op if the directory already exists.
|
||||
|
||||
Args:
|
||||
user_id: User UUID string.
|
||||
document_id: Document UUID string.
|
||||
file_bytes: Raw file content.
|
||||
extension: File extension with leading dot (e.g. ".pdf").
|
||||
content_type: MIME type (unused by WebDAV, kept for ABC compliance).
|
||||
|
||||
Returns:
|
||||
object_key: The WebDAV path where the file was stored.
|
||||
|
||||
Raises:
|
||||
ValueError: If SSRF guard fires on re-validation (D-17).
|
||||
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.
|
||||
"""
|
||||
# Re-validate before every outbound request (D-17 / T-05-04-02)
|
||||
validate_cloud_url(self._server_url)
|
||||
object_key = self._make_path(user_id, document_id, extension)
|
||||
# Ensure parent directory exists (idempotent)
|
||||
parent_dir = f"docuvault/{urllib.parse.quote(str(user_id), safe='')}"
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user