diff --git a/backend/storage/cloud_backend_factory.py b/backend/storage/cloud_backend_factory.py index f7c710d..87dedc5 100644 --- a/backend/storage/cloud_backend_factory.py +++ b/backend/storage/cloud_backend_factory.py @@ -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"], ) diff --git a/backend/storage/cloud_utils.py b/backend/storage/cloud_utils.py index fa1e5d7..9997f03 100644 --- a/backend/storage/cloud_utils.py +++ b/backend/storage/cloud_utils.py @@ -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://[/]/remote.php/dav/files// + + 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. diff --git a/backend/storage/nextcloud_backend.py b/backend/storage/nextcloud_backend.py index adea487..4296f86 100644 --- a/backend/storage/nextcloud_backend.py +++ b/backend/storage/nextcloud_backend.py @@ -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. diff --git a/backend/tests/test_webdav_backend.py b/backend/tests/test_webdav_backend.py index 8ba23e4..e4ecfd6 100644 --- a/backend/tests/test_webdav_backend.py +++ b/backend/tests/test_webdav_backend.py @@ -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")