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

Merge pull request #2024 from pbiering/sharing-review-4

Browse Source 
Sharing review 4
This commit is contained in:
Peter Bieringer
2026-03-15 07:56:17 +01:00
committed by GitHub
parent 33bfe29a69 75571b85ce
commit c6c9562a4a
16 changed files with 471 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@@ -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

8
DOCUMENTATION.md
View File

@@ -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)_

1
SHARING.md
View File

@@ -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`

3
config
View File

@@ -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

9
radicale/app/__init__.py
View File

@@ -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")

3
radicale/app/base.py
View File

@@ -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()

6
radicale/app/mkcalendar.py
View File

@@ -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?

6
radicale/app/mkcol.py
View File

@@ -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"):

69
radicale/app/propfind.py
View File

@@ -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)

15
radicale/app/report.py
View File

@@ -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

6
radicale/config.py
View File

@@ -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'",

10
radicale/httputils.py
View File

@@ -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

100
radicale/sharing/__init__.py
View File

@@ -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:

381
radicale/tests/test_sharing.py
View File

@@ -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, """\
<?xml version="1.0"?>
<CR:addressbook-multiget xmlns="DAV:" xmlns:CR="urn:ietf:params:xml:ns:carddav">
<prop>
<getetag />
<CR:address-data />
</prop>
<href>""" + path_mapped_item + """</href>
</CR:addressbook-multiget>""", 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, """\
<?xml version="1.0"?>
<CR:addressbook-multiget xmlns="DAV:" xmlns:CR="urn:ietf:params:xml:ns:carddav">
<prop>
<getetag />
<CR:address-data />
</prop>
<href>""" + path_shared_r + """</href>
</CR:addressbook-multiget>""", 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"

6
radicale/utils.py
View File

@@ -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)

4
radicale/web/internal_data/js/utils/misc.js
View File

@@ -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];
}
return Math.round((bytes / Math.pow(1024, i)) * 100) / 100 + ' ' + units[i];
}