diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2c20bd9c..9de58012 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,7 @@
* Performance: improve `path_to_filesystem()`
* Performance: preload access rights from file
* Add: [server] delay_on_error option
+* Add: [logging] limit_content option
## 3.6.1
diff --git a/DOCUMENTATION.md b/DOCUMENTATION.md
index 9e910026..d1864280 100644
--- a/DOCUMENTATION.md
+++ b/DOCUMENTATION.md
@@ -1679,6 +1679,14 @@ Available levels are:
Default: `warning` _(< 3.2.0)_ / `info` _(>= 3.2.0)_
+##### limit_content
+
+_(> 3.7.0)_
+
+Limit content of wrapped text (chars)
+
+Default: `3000`
+
##### trace_on_debug
_(> 3.5.4)_
diff --git a/SHARING.md b/SHARING.md
index c359e123..df9853f2 100644
--- a/SHARING.md
+++ b/SHARING.md
@@ -322,6 +322,7 @@ Create a share by mapping a collection of an `Owner` to an `User`.
* Authorization
* `PathMapped` is existing and a collection
+ * `PathMapped` is not existing already as a share target for same `User`
* Authenticated user as `Owner` has at least read access to `PathMapped`
* Provided `User` has at least read access to `PathOrToken`
* Global permitted by `permit_create_map = True` or `rights` permission `m`
diff --git a/config b/config
index 86363049..399d6d9e 100644
--- a/config
+++ b/config
@@ -362,6 +362,9 @@
# Value: debug | info | warning | error | critical
#level = info
+# Limit content of wrapped text (chars)
+#limit_content = 3000
+
# do not filter debug messages starting with 'TRACE'
#trace_on_debug = False
diff --git a/radicale/app/__init__.py b/radicale/app/__init__.py
index e52934dd..b64e2c4b 100644
--- a/radicale/app/__init__.py
+++ b/radicale/app/__init__.py
@@ -83,6 +83,7 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
_extra_headers: Mapping[str, str]
_profiling_per_request: bool = False
_profiling_per_request_method: bool = False
+ _limit_content: int
profiler_per_request_method: dict[str, cProfile.Profile] = {}
profiler_per_request_method_counter: dict[str, int] = {}
profiler_per_request_method_starttime: datetime.datetime
@@ -119,6 +120,8 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
logger.debug("log request content on debug: %s", self._request_content_on_debug)
logger.debug("log response header on debug: %s", self._response_header_on_debug)
logger.debug("log response content on debug: %s", self._response_content_on_debug)
+ self._limit_content = configuration.get("logging", "limit_content")
+ logger.debug("log limit for content: %d", self._limit_content)
self._auth_delay = configuration.get("auth", "delay")
self._auth_type = configuration.get("auth", "type")
self._web_type = configuration.get("web", "type")
@@ -258,7 +261,7 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
if isinstance(answer, str):
if self._response_content_on_debug:
if logger.isEnabledFor(logging.DEBUG):
- logger.debug("Response content (nonXML):\n%s", utils.textwrap_str(answer))
+ logger.debug("Response content (nonXML):\n%s", utils.textwrap_str(answer, self._limit_content))
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response content: suppressed by config/option [logging] response_content_on_debug")
@@ -283,7 +286,7 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
if self._response_header_on_debug:
if logger.isEnabledFor(logging.DEBUG):
- logger.debug("Response header:\n%s", utils.textwrap_str(pprint.pformat(headers)))
+ logger.debug("Response header:\n%s", utils.textwrap_str(pprint.pformat(headers), self._limit_content))
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("Response header: suppressed by config/option [logging] response_header_on_debug")
@@ -397,7 +400,7 @@ class Application(ApplicationPartDelete, ApplicationPartHead,
remote_host, remote_useragent, https_info)
if self._request_header_on_debug:
logger.debug("Request header:\n%s",
- utils.textwrap_str(pprint.pformat(self._scrub_headers(environ))))
+ utils.textwrap_str(pprint.pformat(self._scrub_headers(environ)), self._limit_content))
else:
logger.debug("Request header: suppressed by config/option [logging] request_header_on_debug")
diff --git a/radicale/app/base.py b/radicale/app/base.py
index 44f064e7..cff58ec1 100644
--- a/radicale/app/base.py
+++ b/radicale/app/base.py
@@ -57,6 +57,7 @@ class ApplicationBase:
self._log_bad_put_request_content = configuration.get("logging", "bad_put_request_content")
self._response_content_on_debug = configuration.get("logging", "response_content_on_debug")
self._request_content_on_debug = configuration.get("logging", "request_content_on_debug")
+ self._limit_content = configuration.get("logging", "limit_content")
self._hook = hook.load(configuration)
def _read_xml_request_body(self, environ: types.WSGIEnviron
@@ -83,7 +84,7 @@ class ApplicationBase:
if logger.isEnabledFor(logging.DEBUG):
if self._response_content_on_debug:
logger.debug("Response content (XML):\n%s",
- utils.textwrap_str(xmlutils.pretty_xml(xml_content)))
+ utils.textwrap_str(xmlutils.pretty_xml(xml_content), self._limit_content))
else:
logger.debug("Response content (XML): suppressed by config/option [logging] response_content_on_debug")
f = io.BytesIO()
diff --git a/radicale/app/mkcalendar.py b/radicale/app/mkcalendar.py
index 02a821e1..bbcd2636 100644
--- a/radicale/app/mkcalendar.py
+++ b/radicale/app/mkcalendar.py
@@ -56,9 +56,9 @@ class ApplicationPartMkcalendar(ApplicationBase):
return httputils.BAD_REQUEST
if self._sharing._enabled:
# check for shared collections (all)
- collections_share_map = self._sharing.sharing_collection_map_list()
- if collections_share_map:
- for share in collections_share_map:
+ collections_share_list = self._sharing.sharing_collection_list()
+ if collections_share_list:
+ for share in collections_share_list:
if share['PathOrToken'] == path:
return httputils.CONFLICT
# TODO: use this?
diff --git a/radicale/app/mkcol.py b/radicale/app/mkcol.py
index 66b0fd15..f0a5a131 100644
--- a/radicale/app/mkcol.py
+++ b/radicale/app/mkcol.py
@@ -63,9 +63,9 @@ class ApplicationPartMkcol(ApplicationBase):
return httputils.NOT_ALLOWED
if self._sharing._enabled:
# check for shared collections (all)
- collections_share_map = self._sharing.sharing_collection_map_list()
- if collections_share_map:
- for share in collections_share_map:
+ collections_share_list = self._sharing.sharing_collection_list()
+ if collections_share_list:
+ for share in collections_share_list:
if share['PathOrToken'] == path:
return httputils.CONFLICT
with self._storage.acquire_lock("w", user, path=path, request="MKCOL"):
diff --git a/radicale/app/propfind.py b/radicale/app/propfind.py
index 98ee950b..4af86d2b 100644
--- a/radicale/app/propfind.py
+++ b/radicale/app/propfind.py
@@ -36,8 +36,8 @@ from radicale.log import logger
def xml_propfind(base_prefix: str, path: str,
xml_request: Optional[ET.Element],
- allowed_items: Iterable[Tuple[types.CollectionOrItem, str]],
- user: str, encoding: str, max_resource_size: int, share: Union[dict, None] = None) -> Optional[ET.Element]:
+ allowed_items: Iterable[Tuple[types.CollectionOrItem, str, str]],
+ user: str, encoding: str, max_resource_size: int, shares: dict = {}) -> Optional[ET.Element]:
"""Read and answer PROPFIND requests.
Read rfc4918-9.1 for info.
@@ -70,11 +70,14 @@ def xml_propfind(base_prefix: str, path: str,
# Writing answer
multistatus = ET.Element(xmlutils.make_clark("D:multistatus"))
- for item, permission in allowed_items:
+ if logger.isEnabledFor(logging.DEBUG):
+ logger.debug("TRACE/PROPFIND/xml_propfind: shares=%r", shares)
+
+ for item, permission, sharetype in allowed_items:
write = permission == "w"
multistatus.append(xml_propfind_response(
base_prefix, path, item, props, user, encoding, write=write,
- allprop=allprop, propname=propname, max_resource_size=max_resource_size, share=share))
+ allprop=allprop, propname=propname, max_resource_size=max_resource_size, shares=shares, sharetype=sharetype))
return multistatus
@@ -82,7 +85,7 @@ def xml_propfind(base_prefix: str, path: str,
def xml_propfind_response(
base_prefix: str, path: str, item: types.CollectionOrItem,
props: Sequence[str], user: str, encoding: str, max_resource_size: int, write: bool = False,
- propname: bool = False, allprop: bool = False, share: Union[dict, None] = None) -> ET.Element:
+ propname: bool = False, allprop: bool = False, shares: dict = {}, sharetype: Union[str, None] = None) -> ET.Element:
"""Build and return a PROPFIND response."""
if propname and allprop or (props and (propname or allprop)):
raise ValueError("Only use one of props, propname and allprops")
@@ -102,6 +105,22 @@ def xml_propfind_response(
collection.path, item.href))
response = ET.Element(xmlutils.make_clark("D:response"))
href = ET.Element(xmlutils.make_clark("D:href"))
+
+ # lookup share
+ share = None
+ if logger.isEnabledFor(logging.DEBUG):
+ logger.debug("TRACE/PROPFIND/xml_propfind: sharetype=%r item.path=%r", sharetype, uri)
+ for entry in shares:
+ if logger.isEnabledFor(logging.DEBUG):
+ logger.debug("TRACE/PROPFIND/xml_propfind: entry=%r", entry)
+ if entry is not None:
+ if shares[entry]['PathMapped'] == uri:
+ if sharetype is None or shares[entry]['ShareType'] == sharetype:
+ share = shares[entry]
+ if logger.isEnabledFor(logging.DEBUG):
+ logger.debug("TRACE/PROPFIND/xml_propfind: share=%r", share)
+ break
+
if share:
# backmap
uri = uri.replace(share['PathMapped'], share['PathOrToken'])
@@ -230,7 +249,13 @@ def xml_propfind_response(
xmlutils.make_clark("D:unauthenticated")))
elif tag == xmlutils.make_clark("D:current-user-privilege-set"):
privileges = ["D:read"]
- if write:
+ if share:
+ if write:
+ if "P" in share['Permissions']:
+ privileges.append("D:write-properties")
+ if "w" in share['Permissions']:
+ privileges.append("D:write-content")
+ elif write:
privileges.append("D:all")
privileges.append("D:write")
privileges.append("D:write-properties")
@@ -306,6 +331,8 @@ def xml_propfind_response(
elif tag == xmlutils.make_clark("RADICALE:displayname"):
# Only for internal use by the web interface
displayname = collection.get_meta("D:displayname")
+ if share and 'Properties' in share and share['Properties'] is not None and "D:displayname" in share['Properties']:
+ displayname = share['Properties']["D:displayname"]
if displayname is not None:
element.text = displayname
else:
@@ -318,6 +345,8 @@ def xml_propfind_response(
is404 = True
elif tag == xmlutils.make_clark("D:displayname"):
displayname = collection.get_meta("D:displayname")
+ if share and 'Properties' in share and share['Properties'] is not None and "D:displayname" in share['Properties']:
+ displayname = share['Properties']["D:displayname"]
if not displayname and is_leaf:
displayname = collection.path
if displayname is not None:
@@ -344,12 +373,10 @@ def xml_propfind_response(
else:
human_tag = xmlutils.make_human_tag(tag)
tag_text = collection.get_meta(human_tag)
- if share:
+ if share and 'Properties' in share and share['Properties'] is not None and human_tag in share['Properties']:
# map/add from overlay
- if share['Properties']:
- if human_tag in share['Properties']:
- if share['Properties'][human_tag] is not None:
- tag_text = share['Properties'][human_tag]
+ if share['Properties'][human_tag] is not None:
+ tag_text = share['Properties'][human_tag]
if tag_text is not None:
element.text = tag_text
else:
@@ -426,7 +453,8 @@ class ApplicationPartPropfind(ApplicationBase):
"""Manage PROPFIND request."""
http_depth = environ.get("HTTP_DEPTH", "0")
permissions_filter = None
- share = None
+ shares: dict = {}
+ allowed_items: list = []
if self._sharing._enabled:
# Sharing by token or map (if enabled)
share = self._sharing.sharing_collection_resolver(path, user)
@@ -435,6 +463,7 @@ class ApplicationPartPropfind(ApplicationBase):
path = share['PathMapped']
user = share['Owner']
permissions_filter = share['Permissions']
+ shares[share['PathOrToken']] = share
access = Access(self._rights, user, path, permissions_filter)
if not access.check("r"):
return httputils.NOT_ALLOWED
@@ -461,15 +490,16 @@ class ApplicationPartPropfind(ApplicationBase):
return httputils.NOT_ALLOWED
# put item back
items_iter = itertools.chain([item], items_iter)
- allowed_items = list(self._collect_allowed_items(items_iter, user))
+ for item, permission in list(self._collect_allowed_items(items_iter, user)):
+ allowed_items.append((item, permission, None))
if self._sharing._enabled:
if http_depth == "1":
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/PROPFIND: get shared collections")
# check for shared collections related to user, Enabled and not Hidden
- collections_share_map = self._sharing.sharing_collection_map_list(User=user, Enabled=True, Hidden=False)
- if collections_share_map:
- for share in collections_share_map:
+ collections_share_list = self._sharing.sharing_collection_list(User=user, Enabled=True, Hidden=False)
+ if collections_share_list:
+ for share in collections_share_list:
c_share = share['PathOrToken']
c_path = share['PathMapped']
c_user = share['Owner']
@@ -486,11 +516,14 @@ class ApplicationPartPropfind(ApplicationBase):
with self._storage.acquire_lock("r", c_user):
c_items_iter = iter(self._storage.discover(c_path, "0"))
c_allowed_items = list(self._collect_allowed_items(c_items_iter, c_user))
- allowed_items = allowed_items + c_allowed_items
+ for item, permission in c_allowed_items:
+ allowed_items.append((item, permission, share['ShareType']))
+ shares[c_share] = share
+
headers = {"DAV": httputils.DAV_HEADERS,
"Content-Type": "text/xml; charset=%s" % self._encoding}
xml_answer = xml_propfind(base_prefix, path, xml_content,
- allowed_items, user, self._encoding, max_resource_size=self._max_resource_size, share=share)
+ allowed_items, user, self._encoding, max_resource_size=self._max_resource_size, shares=shares)
if xml_answer is None:
return httputils.NOT_ALLOWED
return client.MULTI_STATUS, headers, self._xml_response(xml_answer), xmlutils.pretty_xml(xml_content)
diff --git a/radicale/app/report.py b/radicale/app/report.py
index b940fc92..ef18cfd8 100644
--- a/radicale/app/report.py
+++ b/radicale/app/report.py
@@ -255,7 +255,7 @@ def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
# Retrieve everything required for finishing the request.
retrieved_items = list(retrieve_items(
- base_prefix, path, collection, hreferences, main_filters, multistatus))
+ base_prefix, path, collection, hreferences, main_filters, multistatus, share))
collection_tag = collection.tag
# !!! Don't access storage after this !!!
unlock_storage_fn()
@@ -743,26 +743,29 @@ def xml_item_response(base_prefix: str, href: str,
def retrieve_items(
base_prefix: str, path: str, collection: storage.BaseCollection,
hreferences: Iterable[str], filters: Sequence[ET.Element],
- multistatus: ET.Element) -> Iterator[Tuple[radicale_item.Item, bool]]:
+ multistatus: ET.Element, share: Union[dict, None]) -> Iterator[Tuple[radicale_item.Item, bool]]:
"""Retrieves all items that are referenced in ``hreferences`` from
``collection`` and adds 404 responses for missing and invalid items
to ``multistatus``."""
collection_requested = False
- def get_names() -> Iterator[str]:
+ def get_names(share: Union[dict, None]) -> Iterator[str]:
"""Extracts all names from references in ``hreferences`` and adds
404 responses for invalid references to ``multistatus``.
If the whole collections is referenced ``collection_requested``
gets set to ``True``."""
nonlocal collection_requested
for hreference in hreferences:
+ if share:
+ # map back to owner
+ hreference = hreference.replace(share['PathOrToken'], share['PathMapped'])
try:
name = pathutils.name_from_path(hreference, collection)
except ValueError as e:
logger.warning("Skipping invalid path %r in REPORT request on "
"%r: %s", hreference, path, e)
response = xml_item_response(base_prefix, hreference,
- found_item=False)
+ found_item=False, share=share)
multistatus.append(response)
continue
if name:
@@ -772,10 +775,10 @@ def retrieve_items(
# Reference is a collection
collection_requested = True
- for name, item in collection.get_multi(get_names()):
+ for name, item in collection.get_multi(get_names(share)):
if not item:
uri = pathutils.unstrip_path(posixpath.join(collection.path, name))
- response = xml_item_response(base_prefix, uri, found_item=False)
+ response = xml_item_response(base_prefix, uri, found_item=False, share=share)
multistatus.append(response)
else:
yield item, False
diff --git a/radicale/config.py b/radicale/config.py
index 76e4fdbd..0d50dc7b 100644
--- a/radicale/config.py
+++ b/radicale/config.py
@@ -37,7 +37,7 @@ from configparser import RawConfigParser
from typing import (Any, Callable, ClassVar, Iterable, List, Optional,
Sequence, Tuple, TypeVar, Union)
-from radicale import auth, hook, rights, sharing, storage, types, web
+from radicale import auth, hook, rights, sharing, storage, types, utils, web
from radicale.hook import email
from radicale.item import check_and_sanitize_props
@@ -614,6 +614,10 @@ This is an automated message. Please do not reply.""",
"value": "info",
"help": "threshold for the logger",
"type": logging_level}),
+ ("limit_content", {
+ "value": str(utils.DEFAULT_LIMIT_CONTENT),
+ "help": "limit content of wrapped text (chars)",
+ "type": positive_int}),
("trace_on_debug", {
"value": "False",
"help": "do not filter debug messages starting with 'TRACE'",
diff --git a/radicale/httputils.py b/radicale/httputils.py
index d2829ea8..1d9d08b6 100644
--- a/radicale/httputils.py
+++ b/radicale/httputils.py
@@ -154,12 +154,12 @@ def read_request_body(configuration: "config.Configuration",
environ: types.WSGIEnviron) -> str:
content = decode_request(configuration, environ,
read_raw_request_body(configuration, environ))
- if configuration.get("logging", "request_content_on_debug"):
- if logger.isEnabledFor(logging.DEBUG):
+ if logger.isEnabledFor(logging.DEBUG):
+ if configuration.get("logging", "request_content_on_debug"):
+ _limit_content = configuration.get("logging", "limit_content")
logger.debug("Request content (sha256sum): %s", utils.sha256_str(content))
- logger.debug("Request content:\n%s", utils.textwrap_str(content))
- else:
- if logger.isEnabledFor(logging.DEBUG):
+ logger.debug("Request content:\n%s", utils.textwrap_str(content, _limit_content))
+ else:
logger.debug("Request content: suppressed by config/option [logging] request_content_on_debug")
return content
diff --git a/radicale/sharing/__init__.py b/radicale/sharing/__init__.py
index 28de14a5..6a6ee954 100644
--- a/radicale/sharing/__init__.py
+++ b/radicale/sharing/__init__.py
@@ -111,7 +111,7 @@ PATH_PATTERN: str = "([a-zA-Z0-9/.\\-]+)" # TODO: extend or find better source
USER_PATTERN: str = "([a-zA-Z0-9@]+)" # TODO: extend or find better source
-OVERLAY_PROPERTIES_WHITELIST: Sequence[str] = ("C:calendar-description", "ICAL:calendar-color", "CR:addressbook-description", "INF:addressbook-color")
+OVERLAY_PROPERTIES_WHITELIST: Sequence[str] = ("C:calendar-description", "ICAL:calendar-color", "CR:addressbook-description", "INF:addressbook-color", "D:displayname")
def load(configuration: "config.Configuration") -> "BaseSharing":
@@ -322,16 +322,17 @@ class BaseSharing:
return True
# *** sharing functions called by request methods ***
- # list sharings of type "map"
- def sharing_collection_map_list(self, User: Union[str, None] = None, Enabled: Union[bool, None] = None, Hidden: Union[bool, None] = None) -> list[dict]:
+ # list sharings
+ def sharing_collection_list(self, User: Union[str, None] = None, Enabled: Union[bool, None] = None, Hidden: Union[bool, None] = None) -> list[dict]:
""" returning dict with shared collections by filter(User/Enabled/Hidden) or None if not found"""
+ sharing_collection_list = []
+
if not self.sharing_collection_by_map:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map: not active")
- return [{}]
-
- # retrieve collections depending on filter
- shared_collection_list = self.database_list_sharing(
+ else:
+ # retrieve collections depending on filter
+ sharing_collection_list += self.database_list_sharing(
ShareType="map",
OwnerOrUser=User,
User=User,
@@ -340,43 +341,60 @@ class BaseSharing:
HiddenByOwner=Hidden,
HiddenByUser=Hidden)
- # final
- return shared_collection_list
+ return sharing_collection_list
# resolves a path to a share
def sharing_collection_resolver(self, path: str, user: str) -> Union[dict, None]:
""" returning dict with PathMapped, Owner, Permissions or None if not found"""
+ share = None
+
+ if path == "/":
+ # not supported
+ return None
+
if self.sharing_collection_by_token:
- result = self.sharing_collection_by_token_resolver(path)
- if result is not None:
- return result
- else:
- # check for map
- pass
+ if share is None:
+ share = self.sharing_collection_by_token_resolver(path)
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/token: not active")
- return None
if self.sharing_collection_by_map:
- result = self.sharing_collection_by_map_resolver(path, user)
- if result is not None:
- return result
+ if share is None:
+ share = self.sharing_collection_by_map_resolver(path, user)
else:
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/map: not active")
- return None
- return None
+ if share is not None:
+ if self.permit_properties_overlay:
+ if share['Permissions'] and "p" not in share['Permissions']:
+ # add permit permission
+ share['Permissions'] += "P"
+ else:
+ if share['Permissions'] and "P" not in share['Permissions']:
+ # add deny permission
+ share['Permissions'] += "p"
+
+ return share
# adjust a share
def sharing_collection_update(self, ShareType: str, PathOrToken: str, OwnerOrUser: str, Properties: dict) -> None:
""" returning dict with PathMapped, Owner, Permissions or None if not found"""
logger.info("Sharing/collection/update: ShareType=%r PathOrToken=%r OwnerOrUser=%r", ShareType, PathOrToken, OwnerOrUser)
+ # Filter properies for permitted ones
+ properties_filtered: dict = {}
+ for prop in Properties:
+ if prop in OVERLAY_PROPERTIES_WHITELIST:
+ properties_filtered[prop] = Properties[prop]
+ else:
+ if logger.isEnabledFor(logging.DEBUG):
+ logger.debug("TRACE/sharing/collection_update: silent discard unsupported property: %r", prop)
+
self.database_update_sharing(ShareType=ShareType,
PathOrToken=PathOrToken,
OwnerOrUser=OwnerOrUser,
- Properties=Properties)
+ Properties=properties_filtered)
# *** internal sharing functions ***
# resolves a token "path" to a share
@@ -494,8 +512,12 @@ class BaseSharing:
Status in JSON/TEXT (TEXT can be parsed by shell)
"""
+ # initial log prefix
+ api_info = "Sharing/API/POST"
+
if not self._enabled:
# API is not enabled
+ logger.warning(api_info + ": API is not enabled")
return httputils.NOT_FOUND
if user == "":
@@ -504,6 +526,7 @@ class BaseSharing:
# supported API version check
if not path.startswith("/.sharing/v1/"):
+ logger.warning(api_info + ": leading part of path not matching supported API version")
return httputils.NOT_FOUND
# split into ShareType and action
@@ -517,6 +540,9 @@ class BaseSharing:
ShareType = match.group(1)
action = match.group(2)
+ # append ShareType
+ api_info = api_info + "/" + ShareType
+
# check for valid ShareTypes
if ShareType:
if ShareType not in SHARE_TYPES:
@@ -525,12 +551,14 @@ class BaseSharing:
return httputils.NOT_FOUND
# check for enabled ShareTypes
- if not self.sharing_collection_by_map and ShareType == "map":
- # API "map" is not enabled
- return httputils.NOT_FOUND
-
if not self.sharing_collection_by_token and ShareType == "token":
# API "token" is not enabled
+ logger.warning(api_info + ": not enabled by config (collection_by_token)")
+ return httputils.NOT_FOUND
+
+ if not self.sharing_collection_by_map and ShareType == "map":
+ # API "map" is not enabled
+ logger.warning(api_info + ": not enabled by config (collection_by_map)")
return httputils.NOT_FOUND
# check for valid API hooks
@@ -539,6 +567,9 @@ class BaseSharing:
logger.debug("TRACE/sharing/API: action not whitelisted: %r", action)
return httputils.NOT_FOUND
+ # append action
+ api_info = api_info + "/" + action
+
if logger.isEnabledFor(logging.DEBUG):
logger.debug("TRACE/sharing/API: called by authenticated user: %r", user)
# read POST data
@@ -551,9 +582,6 @@ class BaseSharing:
logger.debug("Client timed out", exc_info=True)
return httputils.REQUEST_TIMEOUT
- # initial log prefix
- api_info = "Sharing/API/POST/" + ShareType + "/" + action
-
# parse body according to content-type
content_type = environ.get("CONTENT_TYPE", "")
if 'application/json' in content_type:
@@ -658,12 +686,12 @@ class BaseSharing:
if not re.search('^' + TOKEN_PATTERN_V1 + '$', request_data[key]):
logger.warning(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for PathOrToken")
- elif ShareType == "map":
+ else:
if not re.search('^' + PATH_PATTERN + '$', request_data[key]):
logger.warning(api_info + ": unsupported " + key)
return httputils.bad_request("Invalid value for PathOrToken")
- elif not request_data[key].endswith("/"):
- return httputils.bad_request("PathOrToken not ending with /")
+ if not request_data[key].endswith("/"):
+ return httputils.bad_request("PathOrToken not ending with /")
elif key == "PathMapped":
if not re.search('^' + PATH_PATTERN + '$', request_data[key]):
logger.warning(api_info + ": unsupported " + key)
@@ -731,6 +759,7 @@ class BaseSharing:
if not self.sharing_collection_by_map and not self.sharing_collection_by_token:
if not action == 'info':
# API is not enabled
+ logger.warning(api_info + ": API is not enabled")
return httputils.NOT_FOUND
# action: list
@@ -776,6 +805,7 @@ class BaseSharing:
with self._storage.acquire_lock("r", user, path=PathMapped):
item = next(iter(self._storage.discover(PathMapped)), None)
if not item:
+ logger.warning(api_info + ": cannot find PathMapped=%r", PathMapped)
return httputils.NOT_FOUND
if not isinstance(item, storage.BaseCollection):
return httputils.METHOD_NOT_ALLOWED
@@ -869,6 +899,12 @@ class BaseSharing:
else:
User = str(User)
+ # lookup existing shares with requested PathMapped for same User
+ shares = self.database_list_sharing(ShareType=ShareType, PathMapped=PathMapped, User=User)
+ if len(shares) > 0:
+ logger.warning(api_info + ": share already exists with PathMapped=%r User=%r", PathMapped, User)
+ return httputils.CONFLICT
+
# check access Permissions
access = Access(self._rights, user, PathMapped, None) # PathMapped is mandatory
if not access.check("r") and "i" not in access.permissions:
diff --git a/radicale/tests/test_sharing.py b/radicale/tests/test_sharing.py
index 6bac36e4..9de2734d 100644
--- a/radicale/tests/test_sharing.py
+++ b/radicale/tests/test_sharing.py
@@ -71,7 +71,24 @@ class TestSharingApiSanity(BaseTest):
_, headers, answer = self._sharing_api(sharing_type, action, check, login, data, content_type, accept)
return _, headers, answer
- def _propfind_calendar_color(self, path, login):
+ def _propfind_allprop(self, path: str, login) -> dict:
+ propfind_allprop = get_file_content("allprop.xml")
+ _, responses = self.propfind(path=path, data=propfind_allprop, login=login)
+ logging.info("response: %r", responses)
+ response = responses[path]
+ assert not isinstance(response, int)
+ return response
+
+ def _propfind_priviledges(self, path: str, login) -> list[str]:
+ response = self._propfind_allprop(path, login)
+ status, prop = response["D:current-user-privilege-set"]
+ logging.debug("prop: %r", prop)
+ priviledges = prop.findall(xmlutils.make_clark("D:privilege"))
+ assert len(priviledges) >= 1
+ priviledges_list = [xmlutils.make_human_tag(priviledge.findall("*")[0].tag) for priviledge in priviledges]
+ return priviledges_list
+
+ def _propfind_calendar_color(self, path, login) -> Union[str, None]:
propfind_calendar_color = get_file_content("propfind_calendar_color.xml")
_, responses = self.propfind(path=path, data=propfind_calendar_color, login=login)
logging.info("response: %r", responses)
@@ -477,15 +494,15 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}})
- logging.info("\n*** create token without PathMapped (form) -> should fail")
+ logging.info("\n*** create token without PathMapped (form) -> 400")
form_array = []
_, headers, answer = self._sharing_api_form("token", "create", 400, login="owner:ownerpw", form_array=form_array)
- logging.info("\n*** create token without PathMapped (json) -> should fail")
+ logging.info("\n*** create token without PathMapped (json) -> 400")
json_dict = {}
_, headers, answer = self._sharing_api_json("token", "create", 400, login="owner:ownerpw", json_dict=json_dict)
- logging.info("\n*** create token#1 without existing collection (form->text)")
+ logging.info("\n*** create token#1 without existing collection (form->text) -> 404")
form_array = ["PathMapped=" + path_base1]
_, headers, answer = self._sharing_api_form("token", "create", check=404, login="owner:ownerpw", form_array=form_array)
@@ -493,7 +510,11 @@ class TestSharingApiSanity(BaseTest):
self.mkcalendar(path_base1, login="owner:ownerpw")
self.mkcalendar(path_base2, login="owner:ownerpw")
- logging.info("\n*** create token#1 with existing collection (form->text)")
+ logging.info("\n*** create token#1 with existing collection (form->text) but no trailing / -> 400")
+ form_array = ["PathMapped=" + path_base1.rstrip('/')]
+ _, headers, answer = self._sharing_api_form("token", "create", check=400, login="owner:ownerpw", form_array=form_array)
+
+ logging.info("\n*** create token#1 with existing collection (form->text) -> 200")
form_array = ["PathMapped=" + path_base1]
_, headers, answer = self._sharing_api_form("token", "create", check=200, login="owner:ownerpw", form_array=form_array)
assert "Status='success'" in answer
@@ -789,24 +810,75 @@ class TestSharingApiSanity(BaseTest):
json_dict: dict
+ path_owner = "/owner/calendar.ics/"
+ path_user = "/user/calendar-owner.ics/"
+ path_user2 = "/user/calendar-owner2.ics/"
+ self.mkcalendar(path_owner, login="owner:ownerpw")
+
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
+ self.configure({"sharing": {"permit_create_map": "False"}})
+
logging.info("\n*** test: %s", db_type)
self.configure({"sharing": {"type": db_type}})
- logging.info("\n*** create map without PathMapped (json) -> should fail")
+ logging.info("\n*** create map without PathMapped (json) -> 400")
json_dict = {}
_, headers, answer = self._sharing_api_json("map", "create", 400, login="owner:ownerpw", json_dict=json_dict)
- logging.info("\n*** create map without PathMapped but User (json) -> should fail")
+ logging.info("\n*** create map without PathMapped but User (json) -> 400")
json_dict = {'User': "user"}
_, headers, answer = self._sharing_api_json("map", "create", 400, login="owner:ownerpw", json_dict=json_dict)
- logging.info("\n*** create map without PathMapped but User and PathOrToken (json) -> should fail")
+ logging.info("\n*** create map without PathMapped but User and PathOrToken (json) -> 400")
json_dict = {}
json_dict['User'] = "user"
- json_dict['PathOrToken'] = "/owner/calendar.ics"
+ json_dict['PathOrToken'] = path_user
_, headers, answer = self._sharing_api_json("map", "create", 400, login="owner:ownerpw", json_dict=json_dict)
+ logging.info("\n*** create map with PathMapped, User, PathOrToken without trailing / (json) -> 400")
+ json_dict = {}
+ json_dict['User'] = "user"
+ json_dict['PathOrToken'] = path_user
+ json_dict['PathMapped'] = path_owner.rstrip('/')
+ _, headers, answer = self._sharing_api_json("map", "create", 400, login="owner:ownerpw", json_dict=json_dict)
+
+ logging.info("\n*** create map with PathMapped without trailing /, User, PathOrToken (json) -> 400")
+ json_dict = {}
+ json_dict['User'] = "user"
+ json_dict['PathOrToken'] = path_user.rstrip('/')
+ json_dict['PathMapped'] = path_owner
+ _, headers, answer = self._sharing_api_json("map", "create", 400, login="owner:ownerpw", json_dict=json_dict)
+
+ logging.info("\n*** create map with PathMapped, User, PathOrToken - not permitted (json) -> 403")
+ json_dict = {}
+ json_dict['User'] = "user"
+ json_dict['PathOrToken'] = path_user
+ json_dict['PathMapped'] = path_owner
+ _, headers, answer = self._sharing_api_json("map", "create", 403, login="owner:ownerpw", json_dict=json_dict)
+
+ self.configure({"sharing": {"permit_create_map": "True"}})
+
+ logging.info("\n*** create map with PathMapped, User, PathOrToken (json) -> 200")
+ json_dict = {}
+ json_dict['User'] = "user"
+ json_dict['PathOrToken'] = path_user
+ json_dict['PathMapped'] = path_owner
+ _, headers, answer = self._sharing_api_json("map", "create", 200, login="owner:ownerpw", json_dict=json_dict)
+
+ logging.info("\n*** create map with PathMapped, User, PathOrToken2 (json) -> 409")
+ json_dict = {}
+ json_dict['User'] = "user"
+ json_dict['PathOrToken'] = path_user2
+ json_dict['PathMapped'] = path_owner
+ _, headers, answer = self._sharing_api_json("map", "create", 409, login="owner:ownerpw", json_dict=json_dict)
+
+ logging.info("\n*** create map with PathMapped, User, PathOrToken=PathOwner (json) -> 409")
+ json_dict = {}
+ json_dict['User'] = "owner"
+ json_dict['PathOrToken'] = path_owner
+ json_dict['PathMapped'] = path_owner
+ _, headers, answer = self._sharing_api_json("map", "create", 409, login="owner:ownerpw", json_dict=json_dict)
+
def test_sharing_api_map_usage(self) -> None:
"""share-by-map API usage tests."""
self.configure({"auth": {"type": "htpasswd",
@@ -1348,8 +1420,8 @@ class TestSharingApiSanity(BaseTest):
json_dict: dict
path_shared_r = "/user/calendar-shared-by-owner-r.ics/"
- path_shared_w = "/user/calendar-shared-by-owner-w.ics/"
- path_shared_rw = "/user/calendar-shared-by-owner-rw.ics/"
+ path_shared_w = "/user1/calendar-shared-by-owner-w.ics/"
+ path_shared_rw = "/user2/calendar-shared-by-owner-rw.ics/"
path_mapped = "/owner/calendar.ics/"
logging.info("\n*** prepare and test access")
@@ -1387,9 +1459,9 @@ class TestSharingApiSanity(BaseTest):
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
- logging.info("\n*** create map user/owner:w -> ok")
+ logging.info("\n*** create map user1/owner:w -> ok")
json_dict = {}
- json_dict['User'] = "user"
+ json_dict['User'] = "user1"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_w
json_dict['Permissions'] = "w"
@@ -1398,9 +1470,9 @@ class TestSharingApiSanity(BaseTest):
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
- logging.info("\n*** create map user/owner:rw -> ok")
+ logging.info("\n*** create map user2/owner:rw -> ok")
json_dict = {}
- json_dict['User'] = "user"
+ json_dict['User'] = "user2"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_rw
json_dict['Permissions'] = "rw"
@@ -1419,10 +1491,10 @@ class TestSharingApiSanity(BaseTest):
_, headers, answer = self.request("GET", path_shared_r, check=404, login="user:userpw")
logging.info("\n*** fetch collection via map:w -> n/a")
- _, headers, answer = self.request("GET", path_shared_r, check=404, login="user:userpw")
+ _, headers, answer = self.request("GET", path_shared_w, check=404, login="user1:user1pw")
logging.info("\n*** fetch collection via map:rw -> n/a")
- _, headers, answer = self.request("GET", path_shared_r, check=404, login="user:userpw")
+ _, headers, answer = self.request("GET", path_shared_rw, check=404, login="user2:user2pw")
# enable maps by user
logging.info("\n*** enable map by user:r")
@@ -1430,15 +1502,15 @@ class TestSharingApiSanity(BaseTest):
json_dict['PathOrToken'] = path_shared_r
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
- logging.info("\n*** enable map by user:w")
+ logging.info("\n*** enable map by user1:w")
json_dict = {}
json_dict['PathOrToken'] = path_shared_w
- _, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
+ _, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user1:user1pw", json_dict=json_dict)
- logging.info("\n*** enable map by user:rw")
+ logging.info("\n*** enable map by user2:rw")
json_dict = {}
json_dict['PathOrToken'] = path_shared_rw
- _, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
+ _, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user2:user2pw", json_dict=json_dict)
# list adjusted maps
logging.info("\n*** list (json->text)")
@@ -1450,10 +1522,10 @@ class TestSharingApiSanity(BaseTest):
_, headers, answer = self.request("GET", path_shared_r, check=200, login="user:userpw")
logging.info("\n*** fetch collection via map:w -> fail")
- _, headers, answer = self.request("GET", path_shared_w, check=403, login="user:userpw")
+ _, headers, answer = self.request("GET", path_shared_w, check=403, login="user1:user1pw")
logging.info("\n*** fetch collection via map:rw -> ok")
- _, headers, answer = self.request("GET", path_shared_rw, check=200, login="user:userpw")
+ _, headers, answer = self.request("GET", path_shared_rw, check=200, login="user2:user2pw")
# list adjusted maps
logging.info("\n*** list (json->text)")
@@ -1466,10 +1538,10 @@ class TestSharingApiSanity(BaseTest):
path = path_shared_r + "/event2.ics"
self.put(path, event, check=403, login="user:userpw")
- logging.info("\n*** put to collection by user via map:w -> ok")
+ logging.info("\n*** put to collection by user1 via map:w -> ok")
event = get_file_content("event2.ics")
path = path_shared_w + "event2.ics"
- self.put(path, event, check=201, login="user:userpw")
+ self.put(path, event, check=201, login="user1:user1pw")
# check result
logging.info("\n*** fetch event via map:r -> ok")
@@ -1478,10 +1550,10 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** fetch event as owner -> ok")
_, headers, answer = self.request("GET", path_mapped + "event2.ics", check=200, login="owner:ownerpw")
- logging.info("\n*** put to collection by user via map:rw -> ok")
+ logging.info("\n*** put to collection by user2 via map:rw -> ok")
event = get_file_content("event3.ics")
path = path_shared_rw + "event3.ics"
- self.put(path, event, check=201, login="user:userpw")
+ self.put(path, event, check=201, login="user2:user2pw")
# check result
logging.info("\n*** fetch event via map:r -> ok")
@@ -1491,10 +1563,10 @@ class TestSharingApiSanity(BaseTest):
_, headers, answer = self.request("GET", path_shared_r + "event3.ics", check=200, login="user:userpw")
logging.info("\n*** fetch event via map:rw -> ok")
- _, headers, answer = self.request("GET", path_shared_rw + "event2.ics", check=200, login="user:userpw")
+ _, headers, answer = self.request("GET", path_shared_rw + "event2.ics", check=200, login="user2:user2pw")
logging.info("\n*** fetch event via map:rw -> ok")
- _, headers, answer = self.request("GET", path_shared_rw + "event3.ics", check=200, login="user:userpw")
+ _, headers, answer = self.request("GET", path_shared_rw + "event3.ics", check=200, login="user2:user2pw")
logging.info("\n*** fetch event as owner -> ok")
_, headers, answer = self.request("GET", path_mapped + "event1.ics", check=200, login="owner:ownerpw")
@@ -1510,10 +1582,10 @@ class TestSharingApiSanity(BaseTest):
_, headers, answer = self.request("DELETE", path_shared_r + "event1.ics", check=403, login="user:userpw")
logging.info("\n*** DELETE from collection by user via map:rw -> ok")
- _, headers, answer = self.request("DELETE", path_shared_rw + "event2.ics", check=200, login="user:userpw")
+ _, headers, answer = self.request("DELETE", path_shared_rw + "event2.ics", check=200, login="user2:user2pw")
logging.info("\n*** DELETE from collection by user via map:w -> ok")
- _, headers, answer = self.request("DELETE", path_shared_w + "event3.ics", check=200, login="user:userpw")
+ _, headers, answer = self.request("DELETE", path_shared_w + "event3.ics", check=200, login="user1:user1pw")
# check results
logging.info("\n*** fetch event as owner -> ok")
@@ -1769,7 +1841,7 @@ class TestSharingApiSanity(BaseTest):
response = responses[path_shared]
assert isinstance(response, dict)
- def test_sharing_api_map_propfind(self) -> None:
+ def test_sharing_api_map_propfind_base(self) -> None:
"""share-by-map API usage tests related to propfind."""
self.configure({"auth": {"type": "htpasswd",
"htpasswd_filename": self.htpasswd_file_path,
@@ -1889,8 +1961,8 @@ class TestSharingApiSanity(BaseTest):
path_mapped = "/owner/calendarPP.ics/"
path_shared_r = "/user/calendarPP-shared-by-owner-r.ics/"
- path_shared_w = "/user/calendarPP-shared-by-owner-w.ics/"
- path_shared_rw = "/user/calendarPP-shared-by-owner-rw.ics/"
+ path_shared_w = "/user1/calendarPP-shared-by-owner-w.ics/"
+ path_shared_rw = "/user2/calendarPP-shared-by-owner-rw.ics/"
logging.info("\n*** prepare and test access")
self.mkcalendar(path_mapped, login="owner:ownerpw")
@@ -1937,8 +2009,8 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** PROPPATCH collection as user -> 404")
proppatch = get_file_content("proppatch_remove_calendar_color.xml")
_, responses = self.proppatch(path_shared_r, proppatch, login="user:userpw", check=404)
- _, responses = self.proppatch(path_shared_w, proppatch, login="user:userpw", check=404)
- _, responses = self.proppatch(path_shared_rw, proppatch, login="user:userpw", check=404)
+ _, responses = self.proppatch(path_shared_w, proppatch, login="user1:user1pw", check=404)
+ _, responses = self.proppatch(path_shared_rw, proppatch, login="user2:user2pw", check=404)
# create map
logging.info("\n*** create map user/owner:r -> ok")
@@ -1953,9 +2025,9 @@ class TestSharingApiSanity(BaseTest):
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
- logging.info("\n*** create map user/owner:w -> ok")
+ logging.info("\n*** create map user1/owner:w -> ok")
json_dict = {}
- json_dict['User'] = "user"
+ json_dict['User'] = "user1"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_w
json_dict['Permissions'] = "w"
@@ -1965,9 +2037,9 @@ class TestSharingApiSanity(BaseTest):
answer_dict = json.loads(answer)
assert answer_dict['Status'] == "success"
- logging.info("\n*** create map user/owner:rw -> ok")
+ logging.info("\n*** create map user2/owner:rw -> ok")
json_dict = {}
- json_dict['User'] = "user"
+ json_dict['User'] = "user2"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_rw
json_dict['Permissions'] = "rw"
@@ -1981,30 +2053,27 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** PROPPATCH collection as user -> 403")
proppatch = get_file_content("proppatch_set_calendar_color.xml")
_, responses = self.proppatch(path_shared_r, proppatch, login="user:userpw", check=404)
- _, responses = self.proppatch(path_shared_w, proppatch, login="user:userpw", check=404)
- _, responses = self.proppatch(path_shared_rw, proppatch, login="user:userpw", check=404)
+ _, responses = self.proppatch(path_shared_w, proppatch, login="user1:user1pw", check=404)
+ _, responses = self.proppatch(path_shared_rw, proppatch, login="user2:user2pw", check=404)
# enable map by user
logging.info("\n*** enable map by user")
json_dict = {}
- json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_r
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
- logging.info("\n*** enable map by user")
+ logging.info("\n*** enable map by user1")
json_dict = {}
- json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_w
- _, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
+ _, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user1:user1pw", json_dict=json_dict)
- logging.info("\n*** enable map by user")
+ logging.info("\n*** enable map by user2")
json_dict = {}
- json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped
json_dict['PathOrToken'] = path_shared_rw
- _, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
+ _, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user2:user2pw", json_dict=json_dict)
# check PROPPATCH as user
proppatch = get_file_content("proppatch_remove_calendar_color.xml")
@@ -2012,11 +2081,11 @@ class TestSharingApiSanity(BaseTest):
_, responses = self.proppatch(path_shared_r, proppatch, login="user:userpw", check=403)
logging.info("\n*** PROPPATCH collection as user:w -> ok")
- _, responses = self.proppatch(path_shared_w, proppatch, login="user:userpw")
+ _, responses = self.proppatch(path_shared_w, proppatch, login="user1:user1pw")
logging.info("response: %r", responses)
logging.info("\n*** PROPPATCH collection as user:rw -> ok")
- _, responses = self.proppatch(path_shared_rw, proppatch, login="user:userpw")
+ _, responses = self.proppatch(path_shared_rw, proppatch, login="user2:user2pw")
logging.info("response: %r", responses)
# check PROPFIND as owner
@@ -2057,6 +2126,7 @@ class TestSharingApiSanity(BaseTest):
path_user = "/user/calendarM.ics/"
path_mapped1 = "/owner/calendar1M.ics/"
+ path_mapped1r = "/owner/calendar1MR.ics/"
path_mapped2 = "/owner/calendar2M.ics/"
path_shared1_r = "/user/calendar1M-shared-by-owner-r.ics/"
path_shared1_rw = "/user/calendar1M-shared-by-owner-rw.ics/"
@@ -2067,6 +2137,8 @@ class TestSharingApiSanity(BaseTest):
event = get_file_content("event1.ics")
self.put(os.path.join(path_mapped1, "event1.ics"), event, login="owner:ownerpw")
+ self.mkcalendar(path_mapped1r, login="owner:ownerpw")
+
self.mkcalendar(path_mapped2, login="owner:ownerpw")
event = get_file_content("event2.ics")
self.put(os.path.join(path_mapped2, "event2.ics"), event, login="owner:ownerpw")
@@ -2106,7 +2178,7 @@ class TestSharingApiSanity(BaseTest):
logging.info("\n*** create map user/owner:r -> ok")
json_dict = {}
json_dict['User'] = "user"
- json_dict['PathMapped'] = path_mapped1
+ json_dict['PathMapped'] = path_mapped1r
json_dict['PathOrToken'] = path_shared1_r
json_dict['Permissions'] = "r"
json_dict['Enabled'] = True
@@ -2153,21 +2225,18 @@ class TestSharingApiSanity(BaseTest):
# enable map by user
logging.info("\n*** enable map shared1_r by user")
json_dict = {}
- json_dict['User'] = "user"
- json_dict['PathMapped'] = path_mapped1
+ json_dict['PathMapped'] = path_mapped1r
json_dict['PathOrToken'] = path_shared1_r
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
logging.info("\n*** enable map shared1_rw by user")
json_dict = {}
- json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped1
json_dict['PathOrToken'] = path_shared1_rw
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
logging.info("\n*** enable map shared2_rw by user")
json_dict = {}
- json_dict['User'] = "user"
json_dict['PathMapped'] = path_mapped2
json_dict['PathOrToken'] = path_shared2_rw
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
@@ -2204,13 +2273,13 @@ class TestSharingApiSanity(BaseTest):
HTTP_DESTINATION="http://127.0.0.1/"+os.path.join(path_shared1_rw, "event1.ics"))
# check GET as user
- logging.info("\n*** GET event1 as user -> ok")
- _, headers, answer = self.request("GET", os.path.join(path_shared1_r, "event1.ics"), check=200, login="user:userpw")
+ logging.info("\n*** GET event1 from r as user -> 404")
+ _, headers, answer = self.request("GET", os.path.join(path_shared1_r, "event1.ics"), check=404, login="user:userpw")
- logging.info("\n*** GET event1 as user -> ok")
+ logging.info("\n*** GET event1 from 1/rw as user -> ok")
_, headers, answer = self.request("GET", os.path.join(path_shared1_rw, "event1.ics"), check=200, login="user:userpw")
- logging.info("\n*** GET event1 as user -> 404")
+ logging.info("\n*** GET event1 from 2/rw as user -> 404")
_, headers, answer = self.request("GET", os.path.join(path_shared2_rw, "event1.ics"), check=404, login="user:userpw")
# check MOVE as user between shares and own calendar
@@ -2731,45 +2800,50 @@ permissions: RrWw""")
logging.info("\n*** create map user1/owner1, globally disabled / not granted M -> 403")
json_dict['PathMapped'] = path_owner1_M
- json_dict['PathOrToken'] = path_user1 + "dM-uc" + db_type
+ json_dict['PathOrToken'] = path_user1.replace(".ics", "dM-uc" + db_type + ".ics")
_, headers, answer = self._sharing_api_json("map", "create", check=403, login="owner1:owner1pw", json_dict=json_dict)
logging.info("\n*** create map user1/owner1, globally disabled / not granted T -> 403")
json_dict['PathMapped'] = path_owner1_T
- json_dict['PathOrToken'] = path_user1 + "dT-uc" + db_type
+ json_dict['PathOrToken'] = path_user1.replace(".ics", "dT-uc" + db_type + ".ics")
_, headers, answer = self._sharing_api_json("map", "create", check=403, login="owner1:owner1pw", json_dict=json_dict)
logging.info("\n*** create map user1/owner1, globally disabled / not granted t -> 403")
json_dict['PathMapped'] = path_owner1_t
- json_dict['PathOrToken'] = path_user1 + "dt-lc" + db_type
+ json_dict['PathOrToken'] = path_user1.replace(".ics", "dt-lc" + db_type + ".ics")
_, headers, answer = self._sharing_api_json("map", "create", check=403, login="owner1:owner1pw", json_dict=json_dict)
logging.info("\n*** create map user1/owner1, globally disabled / granted m -> 200")
json_dict['PathMapped'] = path_owner1_m
- json_dict['PathOrToken'] = path_user1 + "dm-lc" + db_type
+ json_dict['PathOrToken'] = path_user1.replace(".ics", "dm-lc" + db_type + ".ics")
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner1:owner1pw", json_dict=json_dict)
+ logging.info("\n*** deletee map user1/owner1, globally disabled / granted m -> 200")
+ json_dict['PathMapped'] = path_owner1_m
+ json_dict['PathOrToken'] = path_user1.replace(".ics", "dm-lc" + db_type + ".ics")
+ _, headers, answer = self._sharing_api_json("map", "delete", check=200, login="owner1:owner1pw", json_dict=json_dict)
+
logging.info("\n*** create map user1/owner1, globally enabled")
self.configure({"sharing": {"permit_create_map": "True"}})
logging.info("\n*** create map user1/owner1, globally enabled / not granted M -> 403")
json_dict['PathMapped'] = path_owner1_M
- json_dict['PathOrToken'] = path_user1 + "eM-uc" + db_type
+ json_dict['PathOrToken'] = path_user1.replace(".ics", "eM-uc" + db_type + ".ics")
_, headers, answer = self._sharing_api_json("map", "create", check=403, login="owner1:owner1pw", json_dict=json_dict)
logging.info("\n*** create map user1/owner1, globally enabled / ignore T -> 200")
json_dict['PathMapped'] = path_owner1_T
- json_dict['PathOrToken'] = path_user1 + "eT-uc" + db_type
+ json_dict['PathOrToken'] = path_user1.replace(".ics", "eT-uc" + db_type + ".ics")
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner1:owner1pw", json_dict=json_dict)
logging.info("\n*** create map user1/owner1, globally enabled / ignore t -> 200")
json_dict['PathMapped'] = path_owner1_t
- json_dict['PathOrToken'] = path_user1 + "et-lc" + db_type
+ json_dict['PathOrToken'] = path_user1.replace(".ics", "et-lc" + db_type + ".ics")
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner1:owner1pw", json_dict=json_dict)
logging.info("\n*** create map user1/owner1, globally enabled / ignore m -> 200")
json_dict['PathMapped'] = path_owner1_m
- json_dict['PathOrToken'] = path_user1 + "em-lc" + db_type
+ json_dict['PathOrToken'] = path_user1.replace(".ics", "em-lc" + db_type + ".ics")
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner1:owner1pw", json_dict=json_dict)
# create token
@@ -2825,17 +2899,23 @@ permissions: RrWw""")
"collection_by_map": "True",
"collection_by_token": "True"},
"logging": {"request_header_on_debug": "False",
- "response_content_on_debug": "False",
+ "response_content_on_debug": "True",
"request_content_on_debug": "True"},
"rights": {"type": "owner_only"}})
json_dict: dict
- path_user1 = "/user1/calendarPGu1.ics/"
path_owner1 = "/owner1/calendarPGo1.ics/"
+ path_owner1_rw = "/owner1/calendarPGo1rw.ics/"
+ path_owner1_RrWw = "/owner1/calendarPGo1RrWw.ics/"
+ path_user1_r = "/user1/calendarPGu1-r.ics/"
+ path_user1_rw = "/user1/calendarPGu1-rw.ics/"
+ path_user1_RrWw = "/user1/calendarPGu1-RrWw.ics/"
logging.info("\n*** prepare")
self.mkcalendar(path_owner1, login="owner1:owner1pw")
+ self.mkcalendar(path_owner1_rw, login="owner1:owner1pw")
+ self.mkcalendar(path_owner1_RrWw, login="owner1:owner1pw")
for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
logging.info("\n*** test: %s", db_type)
@@ -2844,14 +2924,21 @@ permissions: RrWw""")
# create map
self.configure({"sharing": {"default_permissions_create_map": "r"}})
+ logging.info("\n*** create map user1/owner1 r -> 200")
json_dict = {}
json_dict['User'] = "user1"
json_dict['PathMapped'] = path_owner1
-
- logging.info("\n*** create map user1/owner1 r -> 200")
- json_dict['PathOrToken'] = path_user1 + "r"
+ json_dict['PathOrToken'] = path_user1_r
+ json_dict['Enabled'] = True
+ json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner1:owner1pw", json_dict=json_dict)
+ # enable map by user
+ logging.info("\n*** enable map by user1")
+ json_dict = {}
+ json_dict['PathOrToken'] = path_user1_r
+ _, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user1:user1pw", json_dict=json_dict)
+
logging.info("\n*** list (json->json)")
_, headers, answer = self._sharing_api_json("map", "list", check=200, login="owner1:owner1pw", json_dict=json_dict)
answer_dict = json.loads(answer)
@@ -2859,9 +2946,23 @@ permissions: RrWw""")
assert answer_dict['Lines'] == 1
assert answer_dict['Content'][0]['Permissions'] == "r"
+ # check PROPFIND/priviledges item as user
+ logging.info("\n*** PROPFIND/priviledges item as user")
+ priviledges_list = self._propfind_priviledges(path_user1_r, login="user1:user1pw")
+ assert "D:read" in priviledges_list
+ assert "D:write-content" not in priviledges_list
+ assert "D:write-properties" not in priviledges_list
+ assert "D:write" not in priviledges_list
+ assert "D:all" not in priviledges_list
+
logging.info("\n*** create map user1/owner1 rw -> 200")
- json_dict['PathOrToken'] = path_user1 + "rw"
+ json_dict = {}
+ json_dict['User'] = "user1"
+ json_dict['PathMapped'] = path_owner1_rw
+ json_dict['PathOrToken'] = path_user1_rw
json_dict['Permissions'] = "rw"
+ json_dict['Enabled'] = True
+ json_dict['Hidden'] = False
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner1:owner1pw", json_dict=json_dict)
logging.info("\n*** list (json->json)")
@@ -2871,10 +2972,27 @@ permissions: RrWw""")
assert answer_dict['Lines'] == 1
assert answer_dict['Content'][0]['Permissions'] == "rw"
+ # enable map by user
+ logging.info("\n*** enable map by user1")
+ json_dict = {}
+ json_dict['PathOrToken'] = path_user1_rw
+ _, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user1:user1pw", json_dict=json_dict)
+
+ # check PROPFIND/priviledges item as user
+ logging.info("\n*** PROPFIND/priviledges item as user")
+ priviledges_list = self._propfind_priviledges(path_user1_rw, login="user1:user1pw")
+ assert "D:read" in priviledges_list
+ assert "D:write-content" in priviledges_list
+ assert "D:write-properties" not in priviledges_list
+ assert "D:write" not in priviledges_list
+ assert "D:all" not in priviledges_list
+
logging.info("\n*** create map user1/owner1 with adjusted default permissions -> 200")
self.configure({"sharing": {"default_permissions_create_map": "RrWw"}})
- json_dict['PathOrToken'] = path_user1 + "RrRw"
- del json_dict['Permissions']
+ json_dict = {}
+ json_dict['User'] = "user1"
+ json_dict['PathMapped'] = path_owner1_RrWw
+ json_dict['PathOrToken'] = path_user1_RrWw
_, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner1:owner1pw", json_dict=json_dict)
logging.info("\n*** list (json->json)")
@@ -2934,6 +3052,96 @@ permissions: RrWw""")
assert answer_dict['Lines'] == 1
assert answer_dict['Content'][0]['Permissions'] == "RrWw"
+ def test_sharing_api_map_report_base(self) -> None:
+ """share-by-map API usage tests related to report."""
+ self.configure({"auth": {"type": "htpasswd",
+ "htpasswd_filename": self.htpasswd_file_path,
+ "htpasswd_encryption": "plain"},
+ "sharing": {
+ "type": "csv",
+ "permit_create_map": True,
+ "permit_create_token": True,
+ "permit_properties_overlay": True,
+ "collection_by_map": "True",
+ "collection_by_token": "True"},
+ "logging": {"request_header_on_debug": "False",
+ "response_content_on_debug": "True",
+ "request_content_on_debug": "True"},
+ "rights": {"type": "owner_only"}})
+
+ json_dict: dict
+
+ path_mapped = "/owner/abook1.vcf/"
+ path_shared_r = "/user/abook-shared-by-owner.vcf/"
+
+ logging.info("\n*** prepare and test access")
+ self.create_addressbook(path_mapped, login="owner:ownerpw")
+ contact = get_file_content("contact1.vcf")
+ path_mapped_item = path_mapped + "contact.vcf"
+ path_shared_item = path_shared_r + "contact.vcf"
+ self.put(path_mapped_item, contact, login="owner:ownerpw")
+
+ for db_type in list(filter(lambda item: item != "none", sharing.INTERNAL_TYPES)):
+ logging.info("\n*** test: %s", db_type)
+ self.configure({"sharing": {"type": db_type}})
+
+ # create map
+ logging.info("\n*** create map user/owner:r -> ok")
+ json_dict = {}
+ json_dict['User'] = "user"
+ json_dict['PathMapped'] = path_mapped
+ json_dict['PathOrToken'] = path_shared_r
+ json_dict['Permissions'] = "r"
+ json_dict['Enabled'] = True
+ json_dict['Hidden'] = False
+ _, headers, answer = self._sharing_api_json("map", "create", check=200, login="owner:ownerpw", json_dict=json_dict)
+ answer_dict = json.loads(answer)
+ assert answer_dict['Status'] == "success"
+
+ # enable map by user
+ logging.info("\n*** enable map by user")
+ json_dict = {}
+ json_dict['User'] = "user"
+ json_dict['PathMapped'] = path_mapped
+ json_dict['PathOrToken'] = path_shared_r
+ _, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
+
+ # check REPORT as owner
+ logging.info("\n*** REPORT collection owner -> ok")
+ _, responses = self.report(path_mapped, """\
+
+
+
+
+
+
+ """ + path_mapped_item + """
+""", login="owner:ownerpw")
+ assert len(responses) == 1
+ logging.info("response: %r", responses)
+ response = responses[path_mapped_item]
+ assert isinstance(response, dict)
+ status, prop = response["D:getetag"]
+ assert status == 200 and prop.text
+
+ # check REPORT as user
+ logging.info("\n*** REPORT collection user -> ok")
+ _, responses = self.report(path_shared_r, """\
+
+
+
+
+
+
+ """ + path_shared_r + """
+""", login="user:userpw")
+ assert len(responses) == 1
+ logging.info("response: %r", responses)
+ response = responses[path_shared_item]
+ assert isinstance(response, dict)
+ status, prop = response["D:getetag"]
+ assert status == 200 and prop.text
+
def test_sharing_api_map_propfind_overlay_api_base(self) -> None:
"""share-by-map API usage tests related to proppatch."""
self.configure({"auth": {"type": "htpasswd",
@@ -3037,7 +3245,7 @@ permissions: RrWw""")
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
# verify PROPFIND as user
- logging.info("\n*** PROPFIND collection user -> ok")
+ logging.info("\n*** PROPFIND collection owner -> ok")
propfind_calendar_color = get_file_content("propfind_multiple.xml")
_, responses = self.propfind(path_mapped, propfind_calendar_color, login="owner:ownerpw")
logging.info("response: %r", responses)
@@ -3485,8 +3693,8 @@ permissions: RrWw""")
"htpasswd_encryption": "plain"},
"sharing": {
"type": "csv",
- "permit_create_map": True,
- "permit_create_token": True,
+ "permit_create_map": "True",
+ "permit_create_token": "True",
"collection_by_map": "True",
"collection_by_token": "True"},
"logging": {"request_header_on_debug": "False",
@@ -3553,8 +3761,17 @@ permissions: RrWw""")
json_dict['PathOrToken'] = path_shared_r
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
+ # check PROPFIND/priviledges item as user
+ logging.info("\n*** PROPFIND/priviledges item as user -> calendar")
+ priviledges_list = self._propfind_priviledges(path_shared_r, login="user:userpw")
+ assert "D:read" in priviledges_list
+ assert "D:write-content" not in priviledges_list
+ assert "D:write-properties" in priviledges_list
+ assert "D:write" not in priviledges_list
+ assert "D:all" not in priviledges_list
+
# verify PROPPATCH as user
- logging.info("\n*** PROPFIND collection user -> ok")
+ logging.info("\n*** PROPFIND collection user -> color #AAAAAA")
color = self._propfind_calendar_color(path_shared_r, login="user:userpw")
assert color == "#AAAAAA"
@@ -3770,7 +3987,7 @@ permissions: RrWw""")
_, headers, answer = self._sharing_api_json("map", "enable", check=200, login="user:userpw", json_dict=json_dict)
# verify PROPPATCH as user
- logging.info("\n*** PROPFIND color collection collection user -> ok")
+ logging.info("\n*** PROPFIND color collection user -> #AAAAAA")
color = self._propfind_calendar_color(path_shared_r, login="user:userpw")
assert color == "#AAAAAA"
@@ -3809,7 +4026,7 @@ permissions: RrWw""")
assert answer_dict['Content'][0]['Properties']['C:calendar-description'] == "USER"
# verify PROPPATCH as user
- logging.info("\n*** PROPFIND color collection collection user -> ok")
+ logging.info("\n*** PROPFIND color collection user -> #BBBBBB")
color = self._propfind_calendar_color(path_shared_r, login="user:userpw")
assert color == "#BBBBBB"
diff --git a/radicale/utils.py b/radicale/utils.py
index 9e59dc7b..d25e08e0 100644
--- a/radicale/utils.py
+++ b/radicale/utils.py
@@ -66,6 +66,10 @@ UNIT_M: int = (1024 * 1024)
UNIT_K: int = (1024)
+# Limits
+DEFAULT_LIMIT_CONTENT: int = 3000
+
+
def load_plugin(internal_types: Sequence[str], module_name: str,
class_name: str, base_class: Type[_T_co],
configuration: "config.Configuration") -> _T_co:
@@ -385,7 +389,7 @@ def limit_str(content: str, limit: int) -> str:
return content
-def textwrap_str(content: str, limit: int = 3000) -> str:
+def textwrap_str(content: str, limit: int = DEFAULT_LIMIT_CONTENT) -> str:
# TODO: add support for config option and prefix
return textwrap.indent(limit_str(content, limit), " ", lambda line: True)
diff --git a/radicale/web/internal_data/js/utils/misc.js b/radicale/web/internal_data/js/utils/misc.js
index 6433a17a..cb5e37e0 100644
--- a/radicale/web/internal_data/js/utils/misc.js
+++ b/radicale/web/internal_data/js/utils/misc.js
@@ -106,5 +106,5 @@ export function bytesToHumanReadable(bytes) {
const units = ['b', 'kb', 'mb', 'gb', 'tb'];
let i = bytes == 0 ? 0 : Math.floor(Math.log(bytes) / Math.log(1024));
i = Math.min(i, units.length - 1);
- return (bytes / Math.pow(1024, i)) + ' ' + units[i];
-}
\ No newline at end of file
+ return Math.round((bytes / Math.pow(1024, i)) * 100) / 100 + ' ' + units[i];
+}