From b857cb9eeabd4dbc83736b753dcd6f67c9778e39 Mon Sep 17 00:00:00 2001 From: Peter Bieringer Date: Tue, 24 Feb 2026 06:28:55 +0100 Subject: [PATCH] sharing: file-based config store --- radicale/sharing/files.py | 445 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 445 insertions(+) create mode 100644 radicale/sharing/files.py diff --git a/radicale/sharing/files.py b/radicale/sharing/files.py new file mode 100644 index 00000000..e268099b --- /dev/null +++ b/radicale/sharing/files.py @@ -0,0 +1,445 @@ +# This file is part of Radicale Server - Calendar Server +# Copyright © 2026-2026 Peter Bieringer +# +# This library is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with Radicale. If not, see . + +import logging +import os +import pickle +import urllib +from typing import Union + +from radicale import sharing +from radicale.log import logger + +""" File 'database' based sharing by token or map """ + +DB_VERSION: str = "1" + + +class Sharing(sharing.BaseSharing): + _sharing_db_path_ShareType: dict = {} + + # Overloaded functions + def init_database(self) -> bool: + logger.debug("sharing database initialization for type 'files'") + sharing_db_path = self.configuration.get("sharing", "database_path") + if sharing_db_path == "": + folder = self.configuration.get("storage", "filesystem_folder") + folder_db = os.path.join(folder, "collection-db") + sharing_db_path = os.path.join(folder_db, "files") + logger.info("sharing database path not provided, use default: %r", sharing_db_path) + else: + logger.info("sharing database path: %r", sharing_db_path) + + if not os.path.exists(folder_db): + logger.warning("sharing database folder is not existing: %r (create now)", folder_db) + try: + os.mkdir(folder_db) + except Exception as e: + logger.error("sharing database folder cannot be created (check permissions): %r (%r)", folder_db, e) + return False + logger.info("sharing database folder successfully created: %r", folder_db) + + if not os.path.exists(sharing_db_path): + logger.warning("sharing database path is not existing: %r", sharing_db_path) + try: + os.mkdir(sharing_db_path) + except Exception as e: + logger.error("sharing database path cannot be created (check permissions): %r (%r)", sharing_db_path, e) + return False + logger.info("sharing database path successfully created: %r", sharing_db_path) + + for ShareType in sharing.SHARE_TYPES_V1: + path = os.path.join(sharing_db_path, ShareType) + self._sharing_db_path_ShareType[ShareType] = path + if not os.path.exists(path): + logger.warning("sharing database path for %r is not existing: %r", ShareType, path) + try: + os.mkdir(path) + except Exception as e: + logger.error("sharing database path for %r cannot be created (check permissions): %r (%r)", ShareType, path, e) + return False + logger.info("sharing database path for %r successfully created: %r", ShareType, path) + return True + + def get_database_info(self) -> Union[dict, None]: + database_info = {'type': "files"} + return database_info + + def verify_database(self) -> bool: + logger.info("sharing database (files) verification begin") + for ShareType in sharing.SHARE_TYPES_V1: + logger.info("sharing database (files) path for %r: %r", ShareType, self._sharing_db_path_ShareType[ShareType]) + # TODO: count amount of files + logger.info("sharing database (files) verification end") + return True + + def get_sharing(self, + ShareType: str, + PathOrToken: str, + User: Union[str, None] = None) -> Union[dict, None]: + """ retrieve sharing target and attributes by map """ + # Lookup + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/get: PathOrToken=%r User=%r)", ShareType, PathOrToken, User) + + sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken)) + + if not os.path.isfile(sharing_config_file): + return None + + # read content + with self._storage.acquire_lock("r", User): + # read file + with open(sharing_config_file, "rb") as fb: + (version, row) = pickle.load(fb) + + if version != DB_VERSION: + return {"status": "error"} + + if User is not None and row['User'] != User: + return None + elif row['EnabledByOwner'] is not True: + return None + elif row['ShareType'] == "map": + if row['EnabledByUser'] is not True: + return None + + PathMapped = row['PathMapped'] + Owner = row['Owner'] + UserShare = row['User'] + Permissions = row['Permissions'] + Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser']) + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing: map %r to %r (Owner=%r User=%r Permissions=%r Hidden=%s)", PathOrToken, PathMapped, Owner, UserShare, Permissions, Hidden) + return { + "mapped": True, + "PathOrToken": PathOrToken, + "PathMapped": PathMapped, + "Owner": Owner, + "User": UserShare, + "Hidden": Hidden, + "Permissions": Permissions} + + return None + + def list_sharing(self, + OwnerOrUser: Union[str, None] = None, + ShareType: Union[str, None] = None, + PathOrToken: Union[str, None] = None, + PathMapped: Union[str, None] = None, + User: Union[str, None] = None, + EnabledByOwner: Union[bool, None] = None, + EnabledByUser: Union[bool, None] = None, + HiddenByOwner: Union[bool, None] = None, + HiddenByUser: Union[bool, None] = None) -> list[dict]: + """ retrieve sharing """ + result = [] + + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/list/called: ShareType=%r OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r HiddenByOwner=%s HiddenByUser=%s", ShareType, OwnerOrUser, User, PathOrToken, PathMapped, HiddenByOwner, HiddenByUser) + + for _ShareType in sharing.SHARE_TYPES_V1: + if ShareType is not None and _ShareType != ShareType: + # skip + continue + + path = self._sharing_db_path_ShareType[_ShareType] + with self._storage.acquire_lock("r", OwnerOrUser, path=path): + for entry in os.scandir(path): + if not entry.is_file(): + continue + + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/list: check file: %r", entry.name) + # read file + with open(entry, "rb") as fb: + (version, row) = pickle.load(fb) + + if version != DB_VERSION: + # skip + continue + + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/list/row: test: %r", row) + if ShareType is not None and row['ShareType'] != ShareType: + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/list/row: skip by ShareType") + pass + elif OwnerOrUser is not None and (row['Owner'] != OwnerOrUser and row['User'] != OwnerOrUser): + pass + elif User is not None and row['User'] != User: + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/list/row: skip by User") + pass + elif PathOrToken is not None and row['PathOrToken'] != PathOrToken: + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/list/row: skip by PathOrToken") + pass + elif PathMapped is not None and row['PathMapped'] != PathMapped: + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/list/row: skip by PathMapped") + pass + elif EnabledByOwner is not None and row['EnabledByOwner'] != EnabledByOwner: + pass + elif EnabledByUser is not None and row['EnabledByUser'] != EnabledByUser: + pass + elif HiddenByOwner is not None and row['HiddenByOwner'] != HiddenByOwner: + pass + elif HiddenByUser is not None and row['HiddenByUser'] != HiddenByUser: + pass + else: + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/list/row: add: %r", row) + result.append(row) + + return result + + def create_sharing(self, + ShareType: str, + PathOrToken: str, PathMapped: str, + Owner: str, User: str, + Permissions: str = "r", + EnabledByOwner: bool = False, EnabledByUser: bool = False, + HiddenByOwner: bool = True, HiddenByUser: bool = True, + Timestamp: int = 0) -> dict: + """ create sharing """ + row: dict + + sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken)) + + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/create: sharing_config_file=%r", ShareType, sharing_config_file) + logger.debug("TRACE/sharing/%s/create: PathOrToken=%r Owner=%r PathMapped=%r User=%r Permissions=%r", ShareType, PathOrToken, Owner, PathMapped, User, Permissions) + if os.path.isfile(sharing_config_file): + return {"status": "conflict"} + + row = {"ShareType": ShareType, + "PathOrToken": PathOrToken, + "PathMapped": PathMapped, + "Owner": Owner, + "User": User, + "Permissions": Permissions, + "EnabledByOwner": EnabledByOwner, + "EnabledByUser": EnabledByUser, + "HiddenByOwner": HiddenByOwner, + "HiddenByUser": HiddenByUser, + "TimestampCreated": Timestamp, + "TimestampUpdated": Timestamp} + + version = DB_VERSION + + try: + with self._storage.acquire_lock("w", Owner, path=sharing_config_file): + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/create: store share-config: %r into file %r", ShareType, row, sharing_config_file) + # write file + with open(sharing_config_file, "wb") as fb: + pickle.dump((version, row), fb) + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/*/create: share-config file stored: %r", sharing_config_file) + return {"status": "success"} + except Exception as e: + logger.error("sharing/%s/create: cannot store share-config: %r (%r)", ShareType, sharing_config_file, e) + return {"status": "error"} + + def update_sharing(self, + ShareType: str, + PathOrToken: str, + Owner: Union[str, None] = None, + User: Union[str, None] = None, + PathMapped: Union[str, None] = None, + Permissions: Union[str, None] = None, + EnabledByOwner: Union[bool, None] = None, + HiddenByOwner: Union[bool, None] = None, + Timestamp: int = 0) -> dict: + """ update sharing """ + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/update: PathOrToken=%r Owner=%r User=%r", ShareType, PathOrToken, Owner, User) + + sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken)) + + if not os.path.isfile(sharing_config_file): + return {"status": "not-found"} + + # read content + with self._storage.acquire_lock("w", Owner, path=sharing_config_file): + # read file + with open(sharing_config_file, "rb") as fb: + (version, row) = pickle.load(fb) + + if version != DB_VERSION: + return {"status": "error"} + + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/update: check: %r", ShareType, row) + + if Owner is not None and row['Owner'] != Owner: + return {"status": "permission-denied"} + if User is not None and row['User'] != User: + return {"status": "permission-denied"} + + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/update: orig row=%r", ShareType, row) + + if PathMapped is not None: + row["PathMapped"] = PathMapped + if Permissions is not None: + row["Permissions"] = Permissions + if User is not None: + row["User"] = User + if EnabledByOwner is not None: + row["EnabledByOwner"] = EnabledByOwner + if HiddenByOwner is not None: + row["HiddenByOwner"] = HiddenByOwner + # update timestamp + row["TimestampUpdated"] = Timestamp + + logger.debug("TRACE/sharing/%s/update: adj row=%r", ShareType, row) + + try: + # write file + with open(sharing_config_file, "wb") as fb: + pickle.dump((version, row), fb) + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/create: share-config file stored: %r", ShareType, sharing_config_file) + return {"status": "success"} + except Exception as e: + logger.error("sharing/%s/create: cannot store share-config: %r (%r)", ShareType, sharing_config_file, e) + return {"status": "error"} + + def delete_sharing(self, + ShareType: str, + PathOrToken: str, Owner: str, + PathMapped: Union[str, None] = None) -> dict: + """ delete sharing """ + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r Owner=%r", ShareType, PathOrToken, Owner) + + sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken)) + + if not os.path.isfile(sharing_config_file): + return {"status": "not-found"} + + # read content + with self._storage.acquire_lock("r", Owner, path=sharing_config_file): + # read file + with open(sharing_config_file, "rb") as fb: + (version, row) = pickle.load(fb) + + if version != DB_VERSION: + return {"status": "error"} + + # verify owner + if row['Owner'] != Owner: + return {"status": "permission-denied"} + + try: + os.remove(sharing_config_file) + except Exception as e: + logger.error("sharing/%s/delete: cannot remove share-config: %r (%r)", ShareType, sharing_config_file, e) + return {"status": "error"} + + logger.debug("sharing/%s/delete: successful removed share-config: %r", ShareType, sharing_config_file) + return {"status": "success"} + + def toggle_sharing(self, + ShareType: str, + PathOrToken: str, + OwnerOrUser: str, + Action: str, + PathMapped: Union[str, None] = None, + User: Union[str, None] = None, + Timestamp: int = 0) -> dict: + """ toggle sharing """ + row: dict + + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/%s: OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r", ShareType, Action, OwnerOrUser, User, PathOrToken, PathMapped) + + if Action not in sharing.API_SHARE_TOGGLES_V1: + # should not happen + raise + + sharing_config_file = os.path.join(self._sharing_db_path_ShareType[ShareType], self._encode_path(PathOrToken)) + + if not os.path.isfile(sharing_config_file): + return {"status": "not-found"} + + # read content + with self._storage.acquire_lock("w", OwnerOrUser, path=sharing_config_file): + # read file + with open(sharing_config_file, "rb") as fb: + (version, row) = pickle.load(fb) + + if version != DB_VERSION: + return {"status": "error"} + + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/%s: check: %r", ShareType, Action, row) + + # verify ownership or user + if User is not None and row['User'] != User: + return {"status": "permission-denied"} + elif row['Owner'] == OwnerOrUser: + pass + elif row['User'] == OwnerOrUser: + pass + else: + return {"status": "permission-denied"} + + if row['Owner'] == OwnerOrUser: + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/%s: Owner=%r User=%r PathOrToken=%r", ShareType, Action, OwnerOrUser, User, PathOrToken) + if Action == "disable": + row['EnabledByOwner'] = False + elif Action == "enable": + row['EnabledByOwner'] = True + elif Action == "hide": + row['HiddenByOwner'] = True + elif Action == "unhide": + row['HiddenByOwner'] = False + row['TimestampUpdated'] = Timestamp + if row['User'] == OwnerOrUser: + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/%s: User=%r PathOrToken=%r", ShareType, Action, OwnerOrUser, PathOrToken) + if Action == "disable": + row['EnabledByUser'] = False + elif Action == "enable": + row['EnabledByUser'] = True + elif Action == "hide": + row['HiddenByUser'] = True + elif Action == "unhide": + row['HiddenByUser'] = False + + row['TimestampUpdated'] = Timestamp + + try: + # write file + with open(sharing_config_file, "wb") as fb: + pickle.dump((version, row), fb) + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/%s/create: share-config file stored: %r", ShareType, sharing_config_file) + return {"status": "success"} + except Exception as e: + logger.error("sharing/%s/create: cannot store share-config: %r (%r)", ShareType, sharing_config_file, e) + return {"status": "error"} + + # local functions + def _encode_path(self, path: str) -> str: + return urllib.parse.quote(path, safe="") + + def _decode_path(self, path: str) -> str: + return urllib.parse.unquote(path)