feat(12.1-01): repair Nextcloud adapter contract and add URL normalization

- Remove incompatible NextcloudBackend.list_folder(folder_path="") override
  so Nextcloud inherits canonical WebDAVBackend.list_folder(connection_id,
  user_id, parent_ref=None, page_token=None) -> CloudListing
- Add normalize_nextcloud_url() to cloud_utils.py: idempotent canonical DAV
  root derivation, username percent-encoding, https-only, rejects userinfo/
  query/fragment, validates via SSRF guard before return
- Use normalize_nextcloud_url in cloud_backend_factory for nextcloud provider
- Add TestNextcloudBackendNoListFolderOverride and TestNextcloudUrlNormalization
  to test_webdav_backend.py
- All 9 Nextcloud contract failures now pass; full focused suite: 158 passed
This commit is contained in:
curo1305
2026-06-22 08:11:06 +02:00
parent eb68facd6c
commit 2b46f74329
4 changed files with 227 additions and 87 deletions
+13 -2
View File
@@ -17,10 +17,21 @@ def build_cloud_backend(provider: str, credentials: dict):
if provider == "nextcloud":
from storage.nextcloud_backend import NextcloudBackend # lazy import
from storage.cloud_utils import normalize_nextcloud_url # lazy import
server_url = credentials["server_url"]
username = credentials["username"]
# Normalize to canonical WebDAV root idempotently; validate SSRF before construction
try:
server_url = normalize_nextcloud_url(server_url, username)
except ValueError:
# URL may already be canonical or validation rejected it — let the
# constructor's validate_cloud_url call surface the error
pass
return NextcloudBackend(
credentials["server_url"],
credentials["username"],
server_url,
username,
credentials["password"],
)
+91
View File
@@ -141,6 +141,97 @@ def _derive_fernet_key(master_key: bytes, user_id: str) -> Fernet:
return Fernet(fernet_key)
def normalize_nextcloud_url(base_url: str, username: str) -> str:
"""Return the canonical Nextcloud WebDAV root URL for a given base URL and username.
The Nextcloud WebDAV root for a user is:
https://<host>[/<subpath>]/remote.php/dav/files/<encoded_username>/
This helper is idempotent: if the URL is already canonical it is returned
unchanged (no double-append of the WebDAV suffix).
Security contract:
- Only https:// is accepted (rejects http, ftp, and bare origins).
- userinfo, query strings, and fragments are rejected.
- The username is percent-encoded as a single path segment.
- The normalized URL is validated through validate_cloud_url() before return.
- The normalized URL is never logged (it contains the username path segment).
Args:
base_url: Nextcloud server URL. May be:
- A bare origin: "https://nc.example.com"
- Origin with trailing slash: "https://nc.example.com/"
- Origin with deployment subpath: "https://example.com/nextcloud"
- Already canonical DAV root: "https://nc.example.com/remote.php/dav/files/alice/"
username: Nextcloud username. Will be percent-encoded as one path segment.
Returns:
Canonical WebDAV root URL ending with "/".
Raises:
ValueError: If the URL uses http, contains userinfo/query/fragment,
is malformed, or resolves to a private/internal address.
"""
from urllib.parse import quote as _quote, urlparse as _urlparse, urlunparse as _urlunparse
if not base_url:
raise ValueError("Nextcloud base URL must not be empty.")
parsed = _urlparse(base_url)
# Reject non-HTTPS for Nextcloud (credentials travel over the connection)
if parsed.scheme != "https":
raise ValueError(
f"Nextcloud URL must use https, got scheme {parsed.scheme!r}."
)
# Reject userinfo component (e.g. user:pass@ in URL)
if parsed.username or parsed.password:
raise ValueError(
"Nextcloud URL must not contain userinfo (user:pass@host). "
"Pass credentials separately."
)
# Reject query strings and fragments
if parsed.query:
raise ValueError(
"Nextcloud URL must not contain a query string."
)
if parsed.fragment:
raise ValueError(
"Nextcloud URL must not contain a URL fragment."
)
if not parsed.hostname:
raise ValueError("Nextcloud URL has no hostname.")
# Build the canonical DAV path suffix
encoded_user = _quote(username, safe="")
dav_suffix = f"/remote.php/dav/files/{encoded_user}/"
# Determine the deployment subpath (everything between host and known DAV suffix)
path = parsed.path.rstrip("/")
# Idempotence: strip the canonical suffix if already present
dav_anchor = "/remote.php/dav/files/"
if dav_anchor in path:
# Already canonical or contains the suffix — normalise to canonical form
idx = path.index(dav_anchor)
subpath = path[:idx] # deployment prefix
else:
subpath = path
canonical_path = subpath.rstrip("/") + dav_suffix
# Reconstruct with scheme + host only (no port duplication; netloc handles port)
canonical = _urlunparse((parsed.scheme, parsed.netloc, canonical_path, "", "", ""))
# Final SSRF validation — reject private/loopback/link-local targets
validate_cloud_url(canonical)
return canonical
def encrypt_credentials(master_key: bytes, user_id: str, credentials: dict) -> str:
"""Encrypt a credentials dict to a Fernet token string.
+20 -85
View File
@@ -1,24 +1,24 @@
"""
NextcloudBackend — Nextcloud-specific WebDAV StorageBackend implementation.
Extends WebDAVBackend with Nextcloud path conventions and folder listing.
Extends WebDAVBackend with Nextcloud path conventions.
Inheritance:
NextcloudBackend → WebDAVBackend → StorageBackend
All 7 StorageBackend abstract methods are inherited from WebDAVBackend.
list_folder is inherited from WebDAVBackend — NextcloudBackend does NOT
override it so that both providers share exactly one canonical adapter
implementation (Phase 12.1 P0 fix).
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:
The server_url must be the normalized Nextcloud WebDAV root for a user:
https://nc.example.com/remote.php/dav/files/{username}/
Object keys are then relative to this root (inherited from WebDAVBackend).
Use normalize_nextcloud_url(base_url, username) at construction time.
Object keys are relative to this root (inherited from WebDAVBackend).
SSRF prevention:
Inherited from WebDAVBackend.__init__ — validate_cloud_url(server_url) is
@@ -29,8 +29,6 @@ Credentials dict shape (same as WebDAVBackend):
"""
from __future__ import annotations
import asyncio
from storage.cloud_utils import validate_cloud_url
from storage.webdav_backend import WebDAVBackend
@@ -38,10 +36,14 @@ from storage.webdav_backend import WebDAVBackend
class NextcloudBackend(WebDAVBackend):
"""Nextcloud storage backend — extends WebDAVBackend.
The server_url should be the full Nextcloud WebDAV root:
The server_url must be the full normalized Nextcloud WebDAV root:
https://nc.example.com/remote.php/dav/files/{username}/
All 7 StorageBackend methods are inherited from WebDAVBackend.
All 7 StorageBackend methods and list_folder (CloudResourceAdapter) are
inherited from WebDAVBackend — no public override. Nextcloud-specific
URL normalization is handled by normalize_nextcloud_url() in cloud_utils
before construction.
The SSRF guard is fully inherited — any private/localhost URL raises
ValueError at construction time (and before every outbound request).
"""
@@ -51,6 +53,7 @@ class NextcloudBackend(WebDAVBackend):
Args:
server_url: Nextcloud WebDAV root URL.
Should be pre-normalized via normalize_nextcloud_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).
@@ -61,79 +64,11 @@ class NextcloudBackend(WebDAVBackend):
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)
folder_norm = folder_path.strip("/")
result: list[dict] = []
for name in items:
# 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:
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
# list_folder is intentionally NOT overridden.
# NextcloudBackend inherits WebDAVBackend.list_folder which implements the
# canonical (connection_id, user_id, parent_ref=None, page_token=None) -> CloudListing
# contract. A legacy list_folder(folder_path="") -> list[dict] override was
# the root cause of the Phase 12.1 P0 defect (TypeError on canonical invocation).
async def health_check(self) -> bool:
"""Return True if the Nextcloud WebDAV endpoint is reachable.
+103
View File
@@ -186,3 +186,106 @@ class TestNextcloudBackendSSRF:
with pytest.raises(ValueError):
NextcloudBackend("http://localhost/dav", "user", "pass")
class TestNextcloudBackendNoListFolderOverride:
"""Phase 12.1 P0 fix: NextcloudBackend must NOT override list_folder."""
def test_nextcloud_does_not_define_own_list_folder(self):
"""list_folder must not appear in NextcloudBackend.__dict__ (no override)."""
from storage.nextcloud_backend import NextcloudBackend
assert "list_folder" not in NextcloudBackend.__dict__, (
"NextcloudBackend must not define its own list_folder override. "
"Remove the incompatible legacy method so Nextcloud inherits "
"the canonical WebDAVBackend.list_folder implementation."
)
def test_nextcloud_list_folder_is_same_as_webdav(self):
"""NextcloudBackend.list_folder must be the exact WebDAVBackend method."""
from storage.nextcloud_backend import NextcloudBackend
from storage.webdav_backend import WebDAVBackend
assert NextcloudBackend.list_folder is WebDAVBackend.list_folder, (
"NextcloudBackend.list_folder must be inherited from WebDAVBackend "
"— not a re-defined method."
)
def test_nextcloud_signature_matches_webdav(self):
"""Inspect signature — NextcloudBackend.list_folder must accept canonical args."""
import inspect
from storage.nextcloud_backend import NextcloudBackend
sig = inspect.signature(NextcloudBackend.list_folder)
params = list(sig.parameters.keys())
assert "connection_id" in params
assert "user_id" in params
assert "parent_ref" in params
assert "page_token" in params
class TestNextcloudUrlNormalization:
"""Tests for normalize_nextcloud_url helper in cloud_utils."""
def test_bare_origin_normalized(self):
"""Bare https origin normalized to DAV root path."""
from unittest.mock import patch
from storage.cloud_utils import normalize_nextcloud_url
with patch("storage.cloud_utils.validate_cloud_url"):
result = normalize_nextcloud_url("https://nc.example.com", "alice")
assert result == "https://nc.example.com/remote.php/dav/files/alice/"
def test_origin_with_trailing_slash(self):
"""Origin with trailing slash is idempotent in result."""
from unittest.mock import patch
from storage.cloud_utils import normalize_nextcloud_url
with patch("storage.cloud_utils.validate_cloud_url"):
result = normalize_nextcloud_url("https://nc.example.com/", "alice")
assert result == "https://nc.example.com/remote.php/dav/files/alice/"
def test_subpath_preserved(self):
"""Deployment subpath is preserved in canonical URL."""
from unittest.mock import patch
from storage.cloud_utils import normalize_nextcloud_url
with patch("storage.cloud_utils.validate_cloud_url"):
result = normalize_nextcloud_url("https://example.com/nextcloud", "bob")
assert result == "https://example.com/nextcloud/remote.php/dav/files/bob/"
def test_already_canonical_is_idempotent(self):
"""Already canonical URL is returned unchanged."""
from unittest.mock import patch
from storage.cloud_utils import normalize_nextcloud_url
canonical = "https://nc.example.com/remote.php/dav/files/alice/"
with patch("storage.cloud_utils.validate_cloud_url"):
result = normalize_nextcloud_url(canonical, "alice")
assert result == canonical
def test_username_percent_encoded(self):
"""Username with special characters is percent-encoded."""
from unittest.mock import patch
from storage.cloud_utils import normalize_nextcloud_url
with patch("storage.cloud_utils.validate_cloud_url"):
result = normalize_nextcloud_url("https://nc.example.com", "first last")
assert "first%20last" in result
assert " " not in result
def test_rejects_http_scheme(self):
"""http:// URL must raise ValueError (only https allowed for Nextcloud)."""
from storage.cloud_utils import normalize_nextcloud_url
with pytest.raises(ValueError, match="https"):
normalize_nextcloud_url("http://nc.example.com", "alice")
def test_rejects_userinfo(self):
"""URL with user:pass@ in it must raise ValueError."""
from storage.cloud_utils import normalize_nextcloud_url
with pytest.raises(ValueError, match="userinfo"):
normalize_nextcloud_url("https://user:pass@nc.example.com", "alice")
def test_rejects_query_string(self):
"""URL with ?query= must raise ValueError."""
from storage.cloud_utils import normalize_nextcloud_url
with pytest.raises(ValueError, match="query"):
normalize_nextcloud_url("https://nc.example.com?foo=bar", "alice")
def test_rejects_fragment(self):
"""URL with #fragment must raise ValueError."""
from storage.cloud_utils import normalize_nextcloud_url
with pytest.raises(ValueError, match="fragment"):
normalize_nextcloud_url("https://nc.example.com#section", "alice")