From 713e93e932d357b6841567908a8108ea23bc54db Mon Sep 17 00:00:00 2001 From: Peter Bieringer Date: Sun, 8 Mar 2026 12:18:22 +0100 Subject: [PATCH] sharing: fix json parser for properties in csv --- radicale/sharing/__init__.py | 24 ++++++++++++++++---- radicale/sharing/csv.py | 43 ++++++++++++++++++++++++++---------- 2 files changed, 51 insertions(+), 16 deletions(-) diff --git a/radicale/sharing/__init__.py b/radicale/sharing/__init__.py index d8b9b93f..49514353 100644 --- a/radicale/sharing/__init__.py +++ b/radicale/sharing/__init__.py @@ -34,9 +34,6 @@ from radicale.log import logger INTERNAL_TYPES: Sequence[str] = ("csv", "files", "none") DB_FIELDS_V1: Sequence[str] = ('ShareType', 'PathOrToken', 'PathMapped', 'Owner', 'User', 'Permissions', 'EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser', 'TimestampCreated', 'TimestampUpdated', 'Properties') -DB_FIELDS_V1_BOOL: Sequence[str] = ('EnabledByOwner', 'EnabledByUser', 'HiddenByOwner', 'HiddenByUser') -DB_FIELDS_V1_INT: Sequence[str] = ('TimestampCreated', 'TimestampUpdated') -DB_FIELDS_V1_USER_PERMITTED: Sequence[str] = ('EnabledByUser', 'HiddenByUser', 'Properties') # ShareType: # PathOrToken: [PrimaryKey] # PathMapped: @@ -49,9 +46,28 @@ DB_FIELDS_V1_USER_PERMITTED: Sequence[str] = ('EnabledByUser', 'HiddenByUser', ' # HiddenByUser: True|False (share exposure controlled by user) - check skipped if Owner==User # TimestampCreated: (when created) # TimestampUpdated: (last update) -# Properties: Overlay of collection properties +# Properties: Overlay of collection properties in JSON + +DB_TYPES_V1: dict[str, type] = { + "ShareType": str, + "PathOrToken": str, + "PathMapped": str, + "Owner": str, + "User": str, + "Permissions": str, + "EnabledByOwner": bool, + "HiddenByOwner": bool, + "EnabledByUser": bool, + "HiddenByUser": bool, + "TimestampCreated": int, + "TimestampUpdated": int, + "Properties": dict +} + +DB_FIELDS_V1_USER_PERMITTED: Sequence[str] = ('EnabledByUser', 'HiddenByUser', 'Properties') SHARE_TYPES: Sequence[str] = ('token', 'map', 'all') + SHARE_TYPES_V1: Sequence[str] = ('token', 'map') # token: share by secret token (does not require authentication) # map : share by mapping collection of one user to another as virtual diff --git a/radicale/sharing/csv.py b/radicale/sharing/csv.py index 62c8a9dc..4d65de5e 100644 --- a/radicale/sharing/csv.py +++ b/radicale/sharing/csv.py @@ -15,6 +15,7 @@ # along with Radicale. If not, see . import csv +import json import logging import os from typing import Union @@ -406,21 +407,39 @@ class Sharing(sharing.BaseSharing): return False # convert txt to bool or int if self._lines > 0: - for fieldname in sharing.DB_FIELDS_V1_BOOL: - try: - row[fieldname] = config._convert_to_bool(row[fieldname]) - except Exception as e: - logger.error("sharing database row error fieldname=%r row=%r error: %r", fieldname, row, e) - for fieldname in sharing.DB_FIELDS_V1_INT: - row[fieldname] = int(row[fieldname]) + for fieldname in row: + if logger.isEnabledFor(logging.DEBUG): + logger.debug("TRACE/sharing/_load: test fieldname=%r", fieldname) + if fieldname not in sharing.DB_TYPES_V1: + logger.error("sharing database row error, unsupported fieldname found: %r", fieldname) + return False + if sharing.DB_TYPES_V1[fieldname] is bool: + try: + row[fieldname] = config._convert_to_bool(row[fieldname]) + except Exception as e: + logger.error("sharing database row error in type conversion fieldname=%r row=%r error: %r", fieldname, row, e) + return False + elif sharing.DB_TYPES_V1[fieldname] is int: + try: + row[fieldname] = int(row[fieldname]) + except Exception as e: + logger.error("sharing database row error in type conversion fieldname=%r row=%r error: %r", fieldname, row, e) + return False + elif sharing.DB_TYPES_V1[fieldname] is dict: + if row[fieldname] is None or row[fieldname] == '': + row[fieldname] = {} + else: + field = row[fieldname].lstrip('"').rstrip('"').replace("'", '"') + try: + row[fieldname] = json.loads(field) + except Exception as e: + logger.error("sharing database row error in type conversion fieldname=%r field=%r row=%r error: %r", fieldname, field, row, e) + return False # check for duplicates - dup = False for row_cached in self._sharing_cache: if row == row_cached: - dup = True - break - if dup: - continue + logger.error("sharing database row duplicate row=%r", row) + return False # logger.debug("sharing database load add: %r", row) self._sharing_cache.append(row) self._lines += 1