assert sanitized and stripped paths

radicale/app/__init__.py

@@ -324,8 +324,8 @@ class Application(
         if permissions and self.Rights.authorized(user, path, permissions):
             return True
         if parent_permissions:
-            parent_path = pathutils.sanitize_path(
+            parent_path = pathutils.unstrip_path(
-                "/%s/" % posixpath.dirname(path.strip("/")))
+                posixpath.dirname(pathutils.strip_path(path)), True)
             if self.Rights.authorized(user, parent_path, parent_permissions):
                 return True
         return False

radicale/app/get.py

@@ -28,7 +28,7 @@ import posixpath
 from http import client
 from urllib.parse import quote
 
-from radicale import httputils, storage, xmlutils
+from radicale import httputils, pathutils, storage, xmlutils
 from radicale.log import logger
 
 
@@ -66,7 +66,7 @@ class ApplicationGetMixin:
     def do_GET(self, environ, base_prefix, path, user):
         """Manage GET request."""
         # Redirect to .web if the root URL is requested
-        if not path.strip("/"):
+        if not pathutils.strip_path(path):
             web_path = ".web"
             if not environ.get("PATH_INFO"):
                 web_path = posixpath.join(posixpath.basename(base_prefix),

radicale/app/mkcalendar.py

@@ -63,8 +63,8 @@ class ApplicationMkcalendarMixin:
             if item:
                 return self.webdav_error_response(
                     "D", "resource-must-be-null")
-            parent_path = pathutils.sanitize_path(
+            parent_path = pathutils.unstrip_path(
-                "/%s/" % posixpath.dirname(path.strip("/")))
+                posixpath.dirname(pathutils.strip_path(path)), True)
             parent_item = next(self.Collection.discover(parent_path), None)
             if not parent_item:
                 return httputils.CONFLICT

radicale/app/mkcol.py

@@ -64,8 +64,8 @@ class ApplicationMkcolMixin:
             item = next(self.Collection.discover(path), None)
             if item:
                 return httputils.METHOD_NOT_ALLOWED
-            parent_path = pathutils.sanitize_path(
+            parent_path = pathutils.unstrip_path(
-                "/%s/" % posixpath.dirname(path.strip("/")))
+                posixpath.dirname(pathutils.strip_path(path)), True)
             parent_item = next(self.Collection.discover(parent_path), None)
             if not parent_item:
                 return httputils.CONFLICT

radicale/app/move.py

@@ -66,8 +66,8 @@ class ApplicationMoveMixin:
             to_item = next(self.Collection.discover(to_path), None)
             if isinstance(to_item, storage.BaseCollection):
                 return httputils.FORBIDDEN
-            to_parent_path = pathutils.sanitize_path(
+            to_parent_path = pathutils.unstrip_path(
-                "/%s/" % posixpath.dirname(to_path.strip("/")))
+                posixpath.dirname(pathutils.strip_path(to_path)), True)
             to_collection = next(
                 self.Collection.discover(to_parent_path), None)
             if not to_collection:
@@ -83,7 +83,7 @@ class ApplicationMoveMixin:
                     to_collection.has_uid(item.uid)):
                 return self.webdav_error_response(
                     "C" if tag == "VCALENDAR" else "CR", "no-uid-conflict")
-            to_href = posixpath.basename(to_path.strip("/"))
+            to_href = posixpath.basename(pathutils.strip_path(to_path))
             try:
                 self.Collection.move(item, to_collection, to_href)
             except ValueError as e:

radicale/app/propfind.py

@@ -95,9 +95,10 @@ def xml_propfind_response(base_prefix, path, item, props, user, write=False,
     href = ET.Element(xmlutils.make_tag("D", "href"))
     if is_collection:
         # Some clients expect collections to end with /
-        uri = "/%s/" % item.path if item.path else "/"
+        uri = pathutils.unstrip_path(item.path, True)
     else:
-        uri = "/" + posixpath.join(collection.path, item.href)
+        uri = pathutils.unstrip_path(
+            posixpath.join(collection.path, item.href))
 
     href.text = xmlutils.make_href(base_prefix, uri)
     response.append(href)
@@ -335,7 +336,7 @@ class ApplicationPropfindMixin:
         """Get items from request that user is allowed to access."""
         for item in items:
             if isinstance(item, storage.BaseCollection):
-                path = pathutils.sanitize_path("/%s/" % item.path)
+                path = pathutils.unstrip_path(item.path, True)
                 if item.get_meta("tag"):
                     permissions = self.Rights.authorized(user, path, "rw")
                     target = "collection with tag %r" % item.path
@@ -343,7 +344,7 @@ class ApplicationPropfindMixin:
                     permissions = self.Rights.authorized(user, path, "RW")
                     target = "collection %r" % item.path
             else:
-                path = pathutils.sanitize_path("/%s/" % item.collection.path)
+                path = pathutils.unstrip_path(item.collection.path, True)
                 permissions = self.Rights.authorized(user, path, "rw")
                 target = "item %r from %r" % (item.href, item.collection.path)
             if rights.intersect_permissions(permissions, "Ww"):

radicale/app/put.py

@@ -52,8 +52,8 @@ class ApplicationPutMixin:
             logger.debug("client timed out", exc_info=True)
             return httputils.REQUEST_TIMEOUT
         # Prepare before locking
-        parent_path = pathutils.sanitize_path(
+        parent_path = pathutils.unstrip_path(
-            "/%s/" % posixpath.dirname(path.strip("/")))
+            posixpath.dirname(pathutils.strip_path(path)), True)
         permissions = self.Rights.authorized(user, path, "Ww")
         parent_permissions = self.Rights.authorized(user, parent_path, "w")
 
@@ -69,7 +69,7 @@ class ApplicationPutMixin:
                     vobject_items, tags.get(content_type))
                 if not tag:
                     raise ValueError("Can't determine collection tag")
-                collection_path = pathutils.sanitize_path(path).strip("/")
+                collection_path = pathutils.strip_path(path)
             elif (write_whole_collection is not None and
                     not write_whole_collection or
                     not permissions and parent_permissions):
@@ -78,7 +78,7 @@ class ApplicationPutMixin:
                     tag = storage.predict_tag_of_parent_collection(
                         vobject_items)
                 collection_path = posixpath.dirname(
-                    pathutils.sanitize_path(path).strip("/"))
+                    pathutils.strip_path(path))
             props = None
             stored_exc_info = None
             items = []
@@ -218,7 +218,7 @@ class ApplicationPutMixin:
                         "C" if tag == "VCALENDAR" else "CR",
                         "no-uid-conflict")
 
-                href = posixpath.basename(path.strip("/"))
+                href = posixpath.basename(pathutils.strip_path(path))
                 try:
                     etag = parent_item.upload(href, prepared_item).etag
                 except ValueError as e:

radicale/app/report.py

@@ -100,7 +100,8 @@ def xml_report(base_prefix, path, xml_request, collection, unlock_storage_fn):
                            old_sync_token, e, exc_info=True)
             return (client.CONFLICT,
                     xmlutils.webdav_error("D", "valid-sync-token"))
-        hreferences = ("/" + posixpath.join(collection.path, n) for n in names)
+        hreferences = (pathutils.unstrip_path(
+            posixpath.join(collection.path, n)) for n in names)
         # Append current sync token to response
         sync_token_element = ET.Element(xmlutils.make_tag("D", "sync-token"))
         sync_token_element.text = sync_token
@@ -142,7 +143,8 @@ def xml_report(base_prefix, path, xml_request, collection, unlock_storage_fn):
 
         for name, item in collection.get_multi(get_names()):
             if not item:
-                uri = "/" + posixpath.join(collection.path, name)
+                uri = pathutils.unstrip_path(
+                    posixpath.join(collection.path, name))
                 response = xml_item_response(base_prefix, uri,
                                              found_item=False)
                 multistatus.append(response)
@@ -223,7 +225,8 @@ def xml_report(base_prefix, path, xml_request, collection, unlock_storage_fn):
             else:
                 not_found_props.append(element)
 
-        uri = "/" + posixpath.join(collection.path, item.href)
+        uri = pathutils.unstrip_path(
+            posixpath.join(collection.path, item.href))
         multistatus.append(xml_item_response(
             base_prefix, uri, found_props=found_props,
             not_found_props=not_found_props, found_item=True))

radicale/item/__init__.py

@@ -25,6 +25,7 @@ from random import getrandbits
 
 import vobject
 
+from radicale import pathutils
 from radicale.item import filter as radicale_filter
 
 
@@ -297,6 +298,8 @@ class Item:
                 raise ValueError("at least one of 'collection_path' or "
                                  "'collection' must be set")
             collection_path = collection.path
+        assert collection_path == pathutils.strip_path(
+            pathutils.sanitize_path(collection_path))
         self._collection_path = collection_path
         self.collection = collection
         self.href = href

radicale/pathutils.py

@@ -125,6 +125,20 @@ def fsync(fd):
         os.fsync(fd)
 
 
+def strip_path(path):
+    assert sanitize_path(path) == path
+    return path.strip("/")
+
+
+def unstrip_path(stripped_path, trailing_slash=False):
+    assert strip_path(sanitize_path(stripped_path)) == stripped_path
+    assert stripped_path or trailing_slash
+    path = "/%s" % stripped_path
+    if trailing_slash and not path.endswith("/"):
+        path += "/"
+    return path
+
+
 def sanitize_path(path):
     """Make path absolute with leading slash to prevent access to other data.
 
@@ -165,30 +179,31 @@ def is_safe_filesystem_path_component(path):
         is_safe_path_component(path))
 
 
-def path_to_filesystem(root, *paths):
+def path_to_filesystem(root, sane_path):
-    """Convert path to a local filesystem path relative to base_folder.
+    """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.
 
-    Conversion of `paths` is done in a secure manner, or raises ``ValueError``.
+    `sane_path` must be a sanitized path without leading or trailing ``/``.
+
+    Conversion of `sane_path` is done in a secure manner,
+    or raises ``ValueError``.
 
     """
-    paths = [sanitize_path(path).strip("/") for path in paths]
+    assert sane_path == strip_path(sanitize_path(sane_path))
     safe_path = root
-    for path in paths:
+    parts = sane_path.split("/") if sane_path else []
-        if not path:
+    for part in parts:
-            continue
+        if not is_safe_filesystem_path_component(part):
-        for part in path.split("/"):
+            raise UnsafePathError(part)
-            if not is_safe_filesystem_path_component(part):
+        safe_path_parent = safe_path
-                raise UnsafePathError(part)
+        safe_path = os.path.join(safe_path, part)
-            safe_path_parent = safe_path
+        # Check for conflicting files (e.g. case-insensitive file systems
-            safe_path = os.path.join(safe_path, part)
+        # or short names on Windows file systems)
-            # Check for conflicting files (e.g. case-insensitive file systems
+        if (os.path.lexists(safe_path) and
-            # or short names on Windows file systems)
+                part not in (e.name for e in
-            if (os.path.lexists(safe_path) and
+                             os.scandir(safe_path_parent))):
-                    part not in (e.name for e in
+            raise CollidingPathError(part)
-                                 os.scandir(safe_path_parent))):
-                raise CollidingPathError(part)
     return safe_path
 
 
@@ -206,11 +221,11 @@ class CollidingPathError(ValueError):
 
 def name_from_path(path, collection):
     """Return Radicale item name from ``path``."""
-    path = path.strip("/") + "/"
+    assert sanitize_path(path) == path
-    start = collection.path + "/"
+    start = unstrip_path(collection.path, True)
-    if not path.startswith(start):
+    if not (path + "/").startswith(start):
         raise ValueError("%r doesn't start with %r" % (path, start))
-    name = path[len(start):][:-1]
+    name = path[len(start):]
     if name and not is_safe_path_component(name):
         raise ValueError("%r is not a component in collection %r" %
                          (name, collection.path))

radicale/rights/authenticated.py

@@ -27,7 +27,7 @@ class Rights(rights.BaseRights):
     def authorized(self, user, path, permissions):
         if self._verify_user and not user:
             return ""
-        sane_path = pathutils.sanitize_path(path).strip("/")
+        sane_path = pathutils.strip_path(path)
         if "/" not in sane_path:
             return rights.intersect_permissions(permissions, "RW")
         if sane_path.count("/") == 1:

radicale/rights/from_file.py

@@ -30,7 +30,7 @@ class Rights(rights.BaseRights):
 
     def authorized(self, user, path, permissions):
         user = user or ""
-        sane_path = pathutils.sanitize_path(path).strip("/")
+        sane_path = pathutils.strip_path(path)
         # Prevent "regex injection"
         user_escaped = re.escape(user)
         sane_path_escaped = re.escape(sane_path)

radicale/rights/owner_only.py

@@ -23,7 +23,7 @@ class Rights(authenticated.Rights):
     def authorized(self, user, path, permissions):
         if self._verify_user and not user:
             return ""
-        sane_path = pathutils.sanitize_path(path).strip("/")
+        sane_path = pathutils.strip_path(path)
         if not sane_path:
             return rights.intersect_permissions(permissions, "R")
         if self._verify_user and user != sane_path.split("/", maxsplit=1)[0]:

radicale/rights/owner_write.py

@@ -23,7 +23,7 @@ class Rights(authenticated.Rights):
     def authorized(self, user, path, permissions):
         if self._verify_user and not user:
             return ""
-        sane_path = pathutils.sanitize_path(path).strip("/")
+        sane_path = pathutils.strip_path(path)
         if not sane_path:
             return rights.intersect_permissions(permissions, "R")
         if self._verify_user:

radicale/storage/multifilesystem.py

@@ -54,7 +54,7 @@ class Collection(storage.BaseCollection):
     def __init__(self, path, filesystem_path=None):
         folder = self._get_collection_root_folder()
         # Path should already be sanitized
-        self.path = pathutils.sanitize_path(path).strip("/")
+        self.path = pathutils.strip_path(path)
         self._encoding = self.configuration.get("encoding", "stock")
         if filesystem_path is None:
             filesystem_path = pathutils.path_to_filesystem(folder, self.path)
@@ -142,7 +142,7 @@ class Collection(storage.BaseCollection):
     def discover(cls, path, depth="0", child_context_manager=(
                  lambda path, href=None: contextlib.ExitStack())):
         # Path should already be sanitized
-        sane_path = pathutils.sanitize_path(path).strip("/")
+        sane_path = pathutils.strip_path(path)
         attributes = sane_path.split("/") if sane_path else []
 
         folder = cls._get_collection_root_folder()
@@ -166,7 +166,7 @@ class Collection(storage.BaseCollection):
             href = None
 
         sane_path = "/".join(attributes)
-        collection = cls(sane_path)
+        collection = cls(pathutils.unstrip_path(sane_path, True))
 
         if href:
             yield collection.get(href)
@@ -178,7 +178,8 @@ class Collection(storage.BaseCollection):
             return
 
         for href in collection.list():
-            with child_context_manager(sane_path, href):
+            with child_context_manager(
+                    pathutils.unstrip_path(sane_path, True), href):
                 yield collection.get(href)
 
         for entry in os.scandir(filesystem_path):
@@ -190,7 +191,8 @@ class Collection(storage.BaseCollection):
                     logger.debug("Skipping collection %r in %r",
                                  href, sane_path)
                 continue
-            child_path = posixpath.join(sane_path, href)
+            child_path = pathutils.unstrip_path(
+                posixpath.join(sane_path, href), True)
             with child_context_manager(child_path):
                 yield cls(child_path)
 
@@ -201,21 +203,23 @@ class Collection(storage.BaseCollection):
         @contextlib.contextmanager
         def exception_cm(path, href=None):
             nonlocal item_errors, collection_errors
+            sane_path = pathutils.strip_path(path)
             try:
                 yield
             except Exception as e:
                 if href:
                     item_errors += 1
-                    name = "item %r in %r" % (href, path.strip("/"))
+                    name = "item %r in %r" % (href, sane_path)
                 else:
                     collection_errors += 1
-                    name = "collection %r" % path.strip("/")
+                    name = "collection %r" % sane_path
                 logger.error("Invalid %s: %s", name, e, exc_info=True)
 
-        remaining_paths = [""]
+        remaining_sane_paths = [""]
-        while remaining_paths:
+        while remaining_sane_paths:
-            path = remaining_paths.pop(0)
+            sane_path = remaining_sane_paths.pop(0)
-            logger.debug("Verifying collection %r", path)
+            path = pathutils.unstrip_path(sane_path, True)
+            logger.debug("Verifying collection %r", sane_path)
             with exception_cm(path):
                 saved_item_errors = item_errors
                 collection = None
@@ -228,19 +232,20 @@ class Collection(storage.BaseCollection):
                         continue
                     if isinstance(item, storage.BaseCollection):
                         has_child_collections = True
-                        remaining_paths.append(item.path)
+                        remaining_sane_paths.append(item.path)
                     elif item.uid in uids:
                         cls.logger.error(
                             "Invalid item %r in %r: UID conflict %r",
-                            item.href, path.strip("/"), item.uid)
+                            item.href, sane_path, item.uid)
                     else:
                         uids.add(item.uid)
-                        logger.debug("Verified item %r in %r", item.href, path)
+                        logger.debug("Verified item %r in %r",
+                                     item.href, sane_path)
                 if item_errors == saved_item_errors:
                     collection.sync()
                 if has_child_collections and collection.get_meta("tag"):
                     cls.logger.error("Invalid collection %r: %r must not have "
-                                     "child collections", path.strip("/"),
+                                     "child collections", sane_path,
                                      collection.get_meta("tag"))
         return item_errors == 0 and collection_errors == 0
 
@@ -249,12 +254,12 @@ class Collection(storage.BaseCollection):
         folder = cls._get_collection_root_folder()
 
         # Path should already be sanitized
-        sane_path = pathutils.sanitize_path(href).strip("/")
+        sane_path = pathutils.strip_path(href)
         filesystem_path = pathutils.path_to_filesystem(folder, sane_path)
 
         if not props:
             cls._makedirs_synced(filesystem_path)
-            return cls(sane_path)
+            return cls(pathutils.unstrip_path(sane_path, True))
 
         parent_dir = os.path.dirname(filesystem_path)
         cls._makedirs_synced(parent_dir)
@@ -265,7 +270,8 @@ class Collection(storage.BaseCollection):
             # The temporary directory itself can't be renamed
             tmp_filesystem_path = os.path.join(tmp_dir, "collection")
             os.makedirs(tmp_filesystem_path)
-            self = cls(sane_path, filesystem_path=tmp_filesystem_path)
+            self = cls(pathutils.unstrip_path(sane_path, True),
+                       filesystem_path=tmp_filesystem_path)
             self.set_meta(props)
             if items is not None:
                 if props.get("tag") == "VCALENDAR":
@@ -281,7 +287,7 @@ class Collection(storage.BaseCollection):
             os.rename(tmp_filesystem_path, filesystem_path)
             cls._sync_directory(parent_dir)
 
-        return cls(sane_path)
+        return cls(pathutils.unstrip_path(sane_path, True))
 
     def _upload_all_nonatomic(self, items, suffix=""):
         """Upload a new set of items.

radicale/tests/custom/rights.py

@@ -19,11 +19,12 @@ Custom rights management.
 
 """
 
-from radicale import rights
+from radicale import pathutils, rights
 
 
 class Rights(rights.BaseRights):
     def authorized(self, user, path, permissions):
-        if path.strip("/") not in ("tmp", "other"):
+        sane_path = pathutils.strip_path(path)
+        if sane_path not in ("tmp", "other"):
             return ""
         return rights.intersect_permissions(permissions)

radicale/web/internal.py

@@ -48,9 +48,11 @@ class Web(web.BaseWeb):
                                                       "internal_data")
 
     def get(self, environ, base_prefix, path, user):
+        assert path == "/.web" or path.startswith("/.web/")
+        assert pathutils.sanitize_path(path) == path
         try:
             filesystem_path = pathutils.path_to_filesystem(
-                self.folder, path[len("/.web"):])
+                self.folder, path[len("/.web"):].strip("/"))
         except ValueError as e:
             logger.debug("Web content with unsafe path %r requested: %s",
                          path, e, exc_info=True)

radicale/web/none.py

@@ -16,11 +16,13 @@
 
 from http import client
 
-from radicale import httputils, web
+from radicale import httputils, pathutils, web
 
 
 class Web(web.BaseWeb):
     def get(self, environ, base_prefix, path, user):
+        assert path == "/.web" or path.startswith("/.web/")
+        assert pathutils.sanitize_path(path) == path
         if path != "/.web":
             return httputils.NOT_FOUND
         return client.OK, {"Content-Type": "text/plain"}, "Radicale works!"

radicale/xmlutils.py

@@ -33,6 +33,8 @@ from collections import OrderedDict
 from http import client
 from urllib.parse import quote
 
+from radicale import pathutils
+
 MIMETYPES = {
     "VADDRESSBOOK": "text/vcard",
     "VCALENDAR": "text/calendar"}
@@ -118,6 +120,7 @@ def make_response(code):
 
 def make_href(base_prefix, href):
     """Return prefixed href."""
+    assert href == pathutils.sanitize_path(href)
     return quote("%s%s" % (base_prefix, href))