Merge pull request #2087 from pbiering/issue-2081

* Fix: broken storage/mtime granularity detection on vfat
* Improve: `path_to_filesystem()` by pre-detection of collision-free file system
* Adjustment: MKCOL/MKCALENDAR return now CONFLICT instead of BADREQUEST of file name collision

CHANGELOG.md

@@ -1,6 +1,9 @@
 # Changelog
 
 ## 3.7.2.dev
+* Fix: broken storage/mtime granularity detection on vfat
+* Improve: `path_to_filesystem()` by pre-detection of collision-free file system
+* Adjustment: MKCOL/MKCALENDAR return now CONFLICT instead of BADREQUEST of file name collision
 
 ## 3.7.1
 * Fix: share address book collection as birthday calendar not working on non-DEBUG level

radicale/app/mkcalendar.py

@@ -90,6 +90,8 @@ class ApplicationPartMkcalendar(ApplicationBase):
                         return httputils.FORBIDDEN
                     else:
                         return httputils.INTERNAL_SERVER_ERROR
+                elif type(e) is pathutils.CollidingPathError:
+                    return httputils.CONFLICT
                 else:
                     logger.warning(
                         "Bad MKCALENDAR request on %r: %s", path, e, exc_info=True)

radicale/app/mkcol.py

@@ -94,6 +94,8 @@ class ApplicationPartMkcol(ApplicationBase):
                         return httputils.FORBIDDEN
                     else:
                         return httputils.INTERNAL_SERVER_ERROR
+                elif type(e) is pathutils.CollidingPathError:
+                    return httputils.CONFLICT
                 else:
                     logger.warning(
                         "Bad MKCOL request on %r (type:%s): %s", path, collection_type, e, exc_info=True)

radicale/pathutils.py

@@ -27,6 +27,7 @@ import os
 import pathlib
 import posixpath
 import sys
+import tempfile
 import threading
 from tempfile import TemporaryDirectory
 from typing import Iterator, Type, Union
@@ -266,28 +267,45 @@ def is_safe_filesystem_path_component(path: str) -> bool:
         is_safe_path_component(path))
 
 
-def path_to_filesystem(root: str, sane_path: str) -> str:
+def path_to_filesystem(root: str, sane_path: str, path_is_collision_free: bool = False) -> str:
     """Convert `sane_path` to a local filesystem path relative to `root`.
 
     `root` must be a secure filesystem path, it will be prepend to the path.
 
     `sane_path` must be a sanitized path without leading or trailing ``/``.
 
+    `path_is_collision_free` is a toggle whether it was earlier detected as collision-free
+
     Conversion of `sane_path` is done in a secure manner,
     or raises ``ValueError``.
 
     """
+    # logger.trace("path_to_filesystem root=%r sane_path=%r path_is_collision_free=%s", root, sane_path, path_is_collision_free)
     assert sane_path == strip_path(sanitize_path(sane_path))
     safe_path = root
     parts = sane_path.split("/") if sane_path else []
     for part in parts:
         if not is_safe_filesystem_path_component(part):
             raise UnsafePathError(part)
+        safe_path_parent = safe_path
         safe_path = os.path.join(safe_path, part)
         # Check for conflicting files (e.g. case-insensitive file systems
         # or short names on Windows file systems)
-        if (os.path.lexists(safe_path) and not os.path.realpath(safe_path).endswith(part)) and not os.path.islink(safe_path):
-            raise CollidingPathError(part)
+        if not path_is_collision_free:
+            if sys.platform == "win32" and False:  # temporary for testing
+                # logger.trace("path_to_filesystem check (win32): %r", part)
+                # if (os.path.lexists(safe_path) and not os.path.realpath(safe_path).endswith(part)) and not os.path.islink(safe_path):
+                if (os.path.lexists(safe_path) and not os.path.realpath(safe_path).endswith(part)):
+                    raise CollidingPathError(part)
+            else:
+                # logger.trace("path_to_filesystem check (!win32): %r", part)
+                if os.path.lexists(safe_path):
+                    with os.scandir(safe_path_parent) as entries:
+                        if part not in (e.name for e in entries):
+                            raise CollidingPathError(part)
+        else:
+            # logger.trace("path_to_filesystem check (skipped): %r", part)
+            pass
     return safe_path
 
 
@@ -365,3 +383,72 @@ def file_check_size(path: str, limit: int):
             logger.warning("file skipped because size exceeds limit %s > %s: %r", utils.format_unit(size, binary=True), utils.format_unit(limit, binary=True), path)
             return False
     return True
+
+
+def path_supports_symlink(path):
+    """Check whether path supports symlink."""
+    if not os.path.isdir(path):
+        raise ValueError("%r is not a path" % (path))
+    result = True
+    test_dir1 = tempfile.mkdtemp(dir=path)
+    test_dir2 = tempfile.mkdtemp(dir=path)
+    os.rmdir(test_dir2)
+    try:
+        os.symlink(test_dir1, test_dir2)
+    except PermissionError:
+        result = False
+    else:
+        # cleanup
+        os.remove(test_dir2)
+    finally:
+        # cleanup
+        os.rmdir(test_dir1)
+    return result
+
+
+def path_is_collision_free_case_sensitive(path):
+    # Test: case sensitive
+    if not os.path.isdir(path):
+        raise ValueError("%r is not a path" % (path))
+    base_dir = tempfile.mkdtemp(dir=path)
+    test_dir = "TESTDIR"
+    test_dir_uc = os.path.join(base_dir, test_dir.upper())
+    test_dir_lc = os.path.join(base_dir, test_dir.lower())
+    os.mkdir(test_dir_uc)
+    result = True
+    try:
+        os.mkdir(test_dir_lc)
+    except FileExistsError:
+        result = False
+    else:
+        # cleanup
+        os.rmdir(test_dir_lc)
+    # cleanup
+    os.rmdir(test_dir_uc)
+    os.rmdir(base_dir)
+    logger.debug("path_is_collision_free (case-sensitive): path=%r result=%s", path, result)
+    return result
+
+
+def path_is_collision_free_no_short_filename(path):
+    """Check whether path supports short-filename collision-free entries."""
+    if not os.path.isdir(path):
+        raise ValueError("%r is not a path" % (path))
+    base_dir = tempfile.mkdtemp(dir=path)
+    test_dir = "TESTDIRLONG"
+    test_dir_long = os.path.join(base_dir, test_dir)
+    test_dir_short = os.path.join(base_dir, test_dir[:6] + "~1")
+    os.mkdir(test_dir_long)
+    result = True
+    try:
+        os.mkdir(test_dir_short)
+    except FileExistsError:
+        result = False
+    else:
+        # cleanup
+        os.rmdir(test_dir_short)
+    # cleanup
+    os.rmdir(test_dir_long)
+    os.rmdir(base_dir)
+    logger.debug("path_is_collision_free (no short-filename): path=%r result=%s", path, result)
+    return result

radicale/storage/multifilesystem/__init__.py

@@ -104,15 +104,16 @@ class Storage(
             logger.warning("Storage item mtime resolution test not possible, cannot write file: %r (%s)", path, e)
             raise
         # set mtime_ns for tests
+        mtime_ns = os.stat(path).st_mtime_ns
         try:
-            os.utime(path, times=None, ns=(MTIME_NS_TEST, MTIME_NS_TEST))
+            os.utime(path, times=None, ns=(mtime_ns + MTIME_NS_TEST, mtime_ns + MTIME_NS_TEST))
         except Exception as e:
             logger.warning("Storage item mtime resolution test not possible, cannot set utime on file: %r (%s)", path, e)
             os.remove(path)
             raise
-        logger.debug("Storage item mtime resoultion test set: %d" % MTIME_NS_TEST)
-        mtime_ns = os.stat(path).st_mtime_ns
-        logger.debug("Storage item mtime resoultion test get: %d" % mtime_ns)
+        logger.debug("Storage item mtime resoultion test set: %d ns" % MTIME_NS_TEST)
+        mtime_ns = os.stat(path).st_mtime_ns - mtime_ns
+        logger.debug("Storage item mtime resoultion test get: %d ns" % mtime_ns)
         # start analysis
         precision = 1
         mtime_ns_test = MTIME_NS_TEST
@@ -171,6 +172,14 @@ class Storage(
             logger.warning("Storage location subfolder: %r does not exist, creating now", self._get_collection_root_folder())
             self._makedirs_synced(self._get_collection_root_folder())
         logger.info("Storage location subfolder permissions: %s", pathutils.path_permissions_as_string(self._get_collection_root_folder()))
+        logger.info("Storage location subfolder softlink support: %s", pathutils.path_supports_symlink(self._get_collection_root_folder()))
+        filesystem_root_folder_is_collision_free_case_sensitive = pathutils.path_is_collision_free_case_sensitive(self._get_collection_root_folder())
+        filesystem_root_folder_is_collision_free_no_short_filename = pathutils.path_is_collision_free_no_short_filename(self._get_collection_root_folder())
+        self._filesystem_root_folder_is_collision_free = filesystem_root_folder_is_collision_free_case_sensitive and filesystem_root_folder_is_collision_free_no_short_filename
+        logger.info("Storage location subfolder is collision free: %s (case-sensitive=%s no-short-filename=%s)",
+                    self._filesystem_root_folder_is_collision_free,
+                    filesystem_root_folder_is_collision_free_case_sensitive,
+                    filesystem_root_folder_is_collision_free_no_short_filename)
         logger.info("Storage cache subfolder usage for 'item': %s", self._use_cache_subfolder_for_item)
         logger.info("Storage cache subfolder usage for 'history': %s", self._use_cache_subfolder_for_history)
         logger.info("Storage cache subfolder usage for 'sync-token': %s", self._use_cache_subfolder_for_synctoken)

radicale/storage/multifilesystem/base.py

@@ -32,6 +32,7 @@ class CollectionBase(storage.BaseCollection):
     _path: str
     _encoding: str
     _filesystem_path: str
+    _filesystem_root_folder_is_collision_free: bool
 
     def __init__(self, storage_: "multifilesystem.Storage", path: str,
                  filesystem_path: Optional[str] = None) -> None:
@@ -42,8 +43,9 @@ class CollectionBase(storage.BaseCollection):
         self._path = pathutils.strip_path(path)
         self._encoding = storage_.configuration.get("encoding", "stock")
         self._skip_broken_item = storage_.configuration.get("storage", "skip_broken_item")
+        self._filesystem_root_folder_is_collision_free = storage_._filesystem_root_folder_is_collision_free
         if filesystem_path is None:
-            filesystem_path = pathutils.path_to_filesystem(folder, self.path)
+            filesystem_path = pathutils.path_to_filesystem(folder, self.path, self._filesystem_root_folder_is_collision_free)
         self._filesystem_path = filesystem_path
 
     # TODO: better fix for "mypy"
@@ -79,6 +81,7 @@ class StorageBase(storage.BaseStorage):
     _folder_umask: str
     _config_umask: int
     _max_resource_size: int
+    _filesystem_root_folder_is_collision_free: bool = False
 
     def __init__(self, configuration: config.Configuration) -> None:
         super().__init__(configuration)

radicale/storage/multifilesystem/create_collection.py

@@ -65,7 +65,7 @@ class StoragePartCreateCollection(StorageBase):
 
         # Path should already be sanitized
         sane_path = pathutils.strip_path(href)
-        filesystem_path = pathutils.path_to_filesystem(folder, sane_path)
+        filesystem_path = pathutils.path_to_filesystem(folder, sane_path, self._filesystem_root_folder_is_collision_free)
         logger.debug("Create collection: %r" % filesystem_path)
 
         if not props:

radicale/storage/multifilesystem/delete.py

@@ -46,7 +46,7 @@ class CollectionPartDelete(CollectionPartHistory, CollectionBase):
             # Delete an item
             if not pathutils.is_safe_filesystem_path_component(href):
                 raise pathutils.UnsafePathError(href)
-            path = pathutils.path_to_filesystem(self._filesystem_path, href)
+            path = pathutils.path_to_filesystem(self._filesystem_path, href, self._filesystem_root_folder_is_collision_free)
             if not os.path.isfile(path):
                 raise storage.ComponentNotFoundError(href)
             os.remove(path)

radicale/storage/multifilesystem/discover.py

@@ -52,11 +52,11 @@ class StoragePartDiscover(StorageBase):
         # Create the root collection
         self._makedirs_synced(folder)
         try:
-            filesystem_path = pathutils.path_to_filesystem(folder, sane_path)
+            filesystem_path = pathutils.path_to_filesystem(folder, sane_path, self._filesystem_root_folder_is_collision_free)
         except ValueError as e:
             # Path is unsafe
-            logger.debug("Unsafe path %r requested from storage: %s",
-                         sane_path, e, exc_info=True)
+            logger.warning("Unsafe path %r requested from storage: %s",
+                           sane_path, e, exc_info=False)
             return
 
         # Check if the path exists and if it leads to a collection or an item
@@ -110,7 +110,7 @@ class StoragePartDiscover(StorageBase):
             href = base64.b64encode(group.encode('utf-8')).decode('ascii')
             logger.debug(f"searching for group calendar {group} {href}")
             sane_child_path = f"GROUPS/{href}"
-            if not os.path.isdir(pathutils.path_to_filesystem(folder, sane_child_path)):
+            if not os.path.isdir(pathutils.path_to_filesystem(folder, sane_child_path, self._filesystem_root_folder_is_collision_free)):
                 continue
             child_path = f"/GROUPS/{href}/"
             with child_context_manager(sane_child_path, None):

radicale/storage/multifilesystem/get.py

@@ -59,7 +59,8 @@ class CollectionPartGet(CollectionPartCache, CollectionPartLock,
                 if not pathutils.is_safe_filesystem_path_component(href):
                     raise pathutils.UnsafePathError(href)
                 path = pathutils.path_to_filesystem(self._filesystem_path,
-                                                    href)
+                                                    href,
+                                                    self._filesystem_root_folder_is_collision_free)
             except ValueError as e:
                 logger.debug(
                     "Can't translate name %r safely to filesystem in %r: %s",

radicale/storage/multifilesystem/move.py

@@ -35,8 +35,8 @@ class StoragePartMove(StorageBase):
         assert isinstance(to_collection, multifilesystem.Collection)
         assert isinstance(item.collection, multifilesystem.Collection)
         assert item.href
-        move_from = pathutils.path_to_filesystem(item.collection._filesystem_path, item.href)
-        move_to = pathutils.path_to_filesystem(to_collection._filesystem_path, to_href)
+        move_from = pathutils.path_to_filesystem(item.collection._filesystem_path, item.href, self._filesystem_root_folder_is_collision_free)
+        move_to = pathutils.path_to_filesystem(to_collection._filesystem_path, to_href, self._filesystem_root_folder_is_collision_free)
         try:
             os.replace(move_from, move_to)
         except OSError as e:

radicale/storage/multifilesystem/upload.py

@@ -39,7 +39,7 @@ class CollectionPartUpload(CollectionPartGet, CollectionPartCache,
                ) -> Tuple[radicale_item.Item, Optional[radicale_item.Item]]:
         if not pathutils.is_safe_filesystem_path_component(href):
             raise pathutils.UnsafePathError(href)
-        path = pathutils.path_to_filesystem(self._filesystem_path, href)
+        path = pathutils.path_to_filesystem(self._filesystem_path, href, self._filesystem_root_folder_is_collision_free)
         old_item = self._get(href, verify_href=False)
         try:
             with self._atomic_write(path, newline="") as fo:  # type: ignore

radicale/tests/test_storage.py

@@ -26,13 +26,13 @@ import logging
 import os
 import re
 import shutil
-import sys
+import tempfile
 from typing import ClassVar, cast
 
 import pytest
 
 import radicale.tests.custom.storage_simple_sync
-from radicale import logger
+from radicale import logger, pathutils
 from radicale.tests import BaseTest
 from radicale.tests.helpers import get_file_content
 from radicale.tests.test_base import TestBaseRequests as _TestBaseRequests
@@ -190,7 +190,7 @@ class TestMultiFileSystem(BaseTest):
             assert answer is not None
             assert "\r\nUID:%s\r\n" % uid in answer
 
-    @pytest.mark.skipif(sys.platform == 'win32', reason="Not supported on Windows")
+    @pytest.mark.skipif(not pathutils.path_supports_symlink(tempfile.mkdtemp()), reason="TEMP is not supporting symlink")
     def test_collection_sharing_by_softlink(self) -> None:
         """Test collection sharing by softlink."""
         self.configure({"auth": {"type": "none"}})
@@ -198,11 +198,11 @@ class TestMultiFileSystem(BaseTest):
         file_item = "event1.ics"
         self.mkcalendar(path_group_col, login="group:grouppw")
         event = get_file_content(file_item)
-        self.put(os.path.join(path_group_col, file_item), event)
-        fs_path_group_col = self.colpath + "/collection-root" + path_group_col
-        fs_path_group_col_rel = ".." + path_group_col
-        fs_path_user = self.colpath + "/collection-root/user"
-        fs_path_user_col = self.colpath + "/collection-root/user/calendar-group.ics"
+        self.put(path_group_col + file_item, event)
+        fs_path_group_col = os.path.join(self.colpath, "collection-root", "group", "calendar-shared.ics")
+        fs_path_group_col_rel = os.path.join("..", "group", "calendar-shared.ics")
+        fs_path_user = os.path.join(self.colpath, "collection-root", "user")
+        fs_path_user_col = os.path.join(self.colpath, "collection-root", "user", "calendar-group.ics")
         logger.debug("colpath=%r fs_path_group_col=%r", self.colpath, fs_path_group_col)
         assert os.path.isdir(fs_path_group_col)
         # create user directory and check
@@ -218,6 +218,73 @@ class TestMultiFileSystem(BaseTest):
         assert os.path.islink(fs_path_user_col)
         self.propfind("/user/", login="user:userpw", HTTP_DEPTH="1")
 
+    def test_colliding_items_by_file_case_insensitive(self, caplog) -> None:
+        """Test for colliding files on file systems."""
+        caplog.set_level(logging.WARNING)
+        self.configure({"logging": {"request_content_on_debug": "False"}})
+        fs_collision_free = pathutils.path_is_collision_free_case_sensitive(self.colpath)
+        file_item = "EvEnT1.iCs"
+        path_coll = "/calendar.ics/"
+        self.mkcalendar(path_coll)
+        event = get_file_content(file_item.lower())
+        path_item = os.path.join(path_coll, file_item)
+        path_uc = os.path.join(path_coll, file_item.upper())
+        path_lc = os.path.join(path_coll, file_item.lower())
+        self.put(path_item, event)
+        self.put(path_uc, event, check=409)
+        if not fs_collision_free:
+            logs = caplog.messages
+            assert len([log for log in logs if "File name collision" in log]) == 1
+        self.put(path_lc, event, check=409)
+        if not fs_collision_free:
+            logs = caplog.messages
+            assert len([log for log in logs if "File name collision" in log]) == 2
+
+    def test_colliding_items_by_dir_case_insensitive(self, caplog) -> None:
+        """Test for colliding dirs on file systems."""
+        caplog.set_level(logging.WARNING)
+        self.configure({"logging": {"request_content_on_debug": "False"}})
+        fs_collision_free = pathutils.path_is_collision_free_case_sensitive(self.colpath)
+        path_coll = "/CaLeNdAr.ics/"
+        self.mkcalendar(path_coll)
+        if fs_collision_free:
+            self.mkcalendar(path_coll.lower(), check=201)
+        else:
+            self.mkcalendar(path_coll.lower(), check=409)
+            logs = caplog.messages
+            assert len([log for log in logs if "File name collision" in log]) == 1
+        if fs_collision_free:
+            self.mkcalendar(path_coll.upper(), check=201)
+        else:
+            self.mkcalendar(path_coll.upper(), check=409)
+            logs = caplog.messages
+            assert len([log for log in logs if "File name collision" in log]) == 2
+
+    def test_colliding_items_by_dir_shortname(self, caplog) -> None:
+        """Test for colliding dirs (shortname) on file systems."""
+        caplog.set_level(logging.WARNING)
+        self.configure({"logging": {"request_content_on_debug": "False"}})
+        fs_collision_free = pathutils.path_is_collision_free_no_short_filename(self.colpath)
+        fs_collision_free_case_sensitive = pathutils.path_is_collision_free_case_sensitive(self.colpath)
+        path_coll = "/calendarlongname.ics/"
+        path_coll_short = "/calend~1.ics/"
+        self.mkcalendar(path_coll)
+        if fs_collision_free:
+            self.mkcalendar(path_coll_short.lower(), check=201)
+            if not fs_collision_free_case_sensitive:
+                # cleanup to avoid collision below
+                self.delete(path_coll_short.lower())
+        else:
+            self.mkcalendar(path_coll_short.lower(), check=409)
+            logs = caplog.messages
+            assert len([log for log in logs if "File name collision" in log]) == 1
+        if fs_collision_free:
+            self.mkcalendar(path_coll_short.upper(), check=201)
+        else:
+            self.mkcalendar(path_coll_short.upper(), check=409)
+            logs = caplog.messages
+            assert len([log for log in logs if "File name collision" in log]) == 2
+
     @pytest.mark.skipif(not shutil.which("flock"), reason="flock command not found")
     @pytest.mark.skipif(radicale.log.logger.getEffectiveLevel() == logging.INFO, reason="requires loglevel DEBUG")
     def test_hook_placeholders_PUT(self, caplog) -> None: