Avii/Radicale
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

storage features: code review

Browse Source
This commit is contained in:
Peter Bieringer
2026-04-22 20:55:46 +02:00
parent 86fda2b6c3
commit f965d46bdf
10 changed files with 32 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
radicale/app/__init__.py
View File

@@ -163,12 +163,12 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
# Format checks # Format checks
self._validate_user_value = configuration.get("server", "validate_user_value") self._validate_user_value = configuration.get("server", "validate_user_value")
self._validate_path_value = configuration.get("server", "validate_path_value") self._validate_path_value = configuration.get("server", "validate_path_value")
if not self._storage._filesystem_root_folder_supports_unicode: if not self._storage._supports_unicode:
if self._validate_user_value not in ["strict", "no-unicode"]: if self._validate_user_value not in ["strict", "no-unicode"]:
self._validate_user_value = "no-unicode" self._validate_user_value = "no-unicode"
if self._validate_path_value not in ["strict", "no-unicode"]: if self._validate_path_value not in ["strict", "no-unicode"]:
self._validate_path_value = "no-unicode" self._validate_path_value = "no-unicode"
if not self._storage._filesystem_root_folder_supports_problematic_chars or not self._storage._filesystem_root_folder_supports_trailing_whitespace: if not self._storage._supports_problematic_chars or not self._storage._supports_trailing_whitespace:
if self._validate_user_value not in ["strict"]: if self._validate_user_value not in ["strict"]:
self._validate_user_value = "strict" self._validate_user_value = "strict"
if self._validate_path_value not in ["strict"]: if self._validate_path_value not in ["strict"]:

5
radicale/storage/__init__.py
View File

@@ -302,6 +302,11 @@ class BaseCollection:
class BaseStorage: class BaseStorage:
_is_collision_free: bool = False
_supports_unicode: bool = False
_supports_trailing_whitespace: bool = False
_supports_problematic_chars: bool = False
def __init__(self, configuration: "config.Configuration") -> None: def __init__(self, configuration: "config.Configuration") -> None:
"""Initialize BaseStorage. """Initialize BaseStorage.

24
radicale/storage/multifilesystem/__init__.py
View File

@@ -173,19 +173,19 @@ class Storage(
self._makedirs_synced(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 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())) 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()) 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()) 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 self._is_collision_free = is_collision_free_case_sensitive and is_collision_free_no_short_filename
self._filesystem_root_folder_supports_unicode = pathutils.path_supports_unicode(self._get_collection_root_folder()) self._supports_unicode = pathutils.path_supports_unicode(self._get_collection_root_folder())
self._filesystem_root_folder_supports_trailing_whitespace = pathutils.path_supports_trailing_whitespace(self._get_collection_root_folder()) self._supports_trailing_whitespace = pathutils.path_supports_trailing_whitespace(self._get_collection_root_folder())
self._filesystem_root_folder_supports_problematic_chars = pathutils.path_supports_problematic_chars(self._get_collection_root_folder()) self._supports_problematic_chars = pathutils.path_supports_problematic_chars(self._get_collection_root_folder())
logger.info("Storage location subfolder is collision free: %s (case-sensitive=%s no-short-filename=%s)", logger.info("Storage location subfolder is collision free: %s (case-sensitive=%s no-short-filename=%s)",
self._filesystem_root_folder_is_collision_free, self._is_collision_free,
filesystem_root_folder_is_collision_free_case_sensitive, is_collision_free_case_sensitive,
filesystem_root_folder_is_collision_free_no_short_filename) is_collision_free_no_short_filename)
logger.info("Storage location subfolder supports unicode: %s", self._filesystem_root_folder_supports_unicode) logger.info("Storage location subfolder supports unicode: %s", self._supports_unicode)
logger.info("Storage location subfolder supports trailing whitespace: %s", self._filesystem_root_folder_supports_trailing_whitespace) logger.info("Storage location subfolder supports trailing whitespace: %s", self._supports_trailing_whitespace)
logger.info("Storage location subfolder supports problematic chars: %s", self._filesystem_root_folder_supports_problematic_chars) logger.info("Storage location subfolder supports problematic chars: %s", self._supports_problematic_chars)
logger.info("Storage cache subfolder usage for 'item': %s", self._use_cache_subfolder_for_item) 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 'history': %s", self._use_cache_subfolder_for_history)
logger.info("Storage cache subfolder usage for 'sync-token': %s", self._use_cache_subfolder_for_synctoken) logger.info("Storage cache subfolder usage for 'sync-token': %s", self._use_cache_subfolder_for_synctoken)

16
radicale/storage/multifilesystem/base.py
View File

@@ -32,10 +32,6 @@ class CollectionBase(storage.BaseCollection):
_path: str _path: str
_encoding: str _encoding: str
_filesystem_path: str _filesystem_path: str
_filesystem_root_folder_is_collision_free: bool
_filesystem_root_folder_supports_unicode: bool
_filesystem_root_folder_supports_trailing_whitespace: bool
_filesystem_root_folder_supports_problematic_chars: bool
def __init__(self, storage_: "multifilesystem.Storage", path: str, def __init__(self, storage_: "multifilesystem.Storage", path: str,
filesystem_path: Optional[str] = None) -> None: filesystem_path: Optional[str] = None) -> None:
@@ -46,12 +42,12 @@ class CollectionBase(storage.BaseCollection):
self._path = pathutils.strip_path(path) self._path = pathutils.strip_path(path)
self._encoding = storage_.configuration.get("encoding", "stock") self._encoding = storage_.configuration.get("encoding", "stock")
self._skip_broken_item = storage_.configuration.get("storage", "skip_broken_item") 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 self._is_collision_free = storage_._is_collision_free
self._filesystem_root_folder_supports_unicode = storage_._filesystem_root_folder_supports_unicode self._supports_unicode = storage_._supports_unicode
self._filesystem_root_folder_supports_trailing_whitespace = storage_._filesystem_root_folder_supports_trailing_whitespace self._supports_trailing_whitespace = storage_._supports_trailing_whitespace
self._filesystem_root_folder_supports_problematic_chars = storage_._filesystem_root_folder_supports_problematic_chars self._supports_problematic_chars = storage_._supports_problematic_chars
if filesystem_path is None: if filesystem_path is None:
filesystem_path = pathutils.path_to_filesystem(folder, self.path, self._filesystem_root_folder_is_collision_free) filesystem_path = pathutils.path_to_filesystem(folder, self.path, self._is_collision_free)
self._filesystem_path = filesystem_path self._filesystem_path = filesystem_path
# TODO: better fix for "mypy" # TODO: better fix for "mypy"
@@ -87,8 +83,6 @@ class StorageBase(storage.BaseStorage):
_folder_umask: str _folder_umask: str
_config_umask: int _config_umask: int
_max_resource_size: int _max_resource_size: int
_filesystem_root_folder_is_collision_free: bool = False
_filesystem_root_folder_supports_unicode: bool = False
def __init__(self, configuration: config.Configuration) -> None: def __init__(self, configuration: config.Configuration) -> None:
super().__init__(configuration) super().__init__(configuration)

2
radicale/storage/multifilesystem/create_collection.py
View File

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

2
radicale/storage/multifilesystem/delete.py
View File

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

4
radicale/storage/multifilesystem/discover.py
View File

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

2
radicale/storage/multifilesystem/get.py
View File

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

4
radicale/storage/multifilesystem/move.py
View File

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

2
radicale/storage/multifilesystem/upload.py
View File

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