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

sharing: fix json parser for properties in csv

Browse Source
This commit is contained in:
Peter Bieringer
2026-03-08 12:18:22 +01:00
parent e45c7da414
commit 713e93e932
2 changed files with 51 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
radicale/sharing/__init__.py
View File

@@ -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: <token|map>
# PathOrToken: <path|token> [PrimaryKey]
# PathMapped: <path>
@@ -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: <unixtime> (when created)
# TimestampUpdated: <unixtime> (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

43
radicale/sharing/csv.py
View File

@@ -15,6 +15,7 @@
# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
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