# This file is part of Radicale - CalDAV and CardDAV server # Copyright © 2008 Nicolas Kandel # Copyright © 2008 Pascal Halter # Copyright © 2008-2017 Guillaume Ayoub # Copyright © 2017-2020 Unrud # Copyright © 2020-2020 Tuna Celik # Copyright © 2025-2026 Peter Bieringer # # This library is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This library is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with Radicale. If not, see . import errno import logging import re import socket import xml.etree.ElementTree as ET from http import client from typing import Dict, Optional, Union, cast import defusedxml.ElementTree as DefusedET import radicale.item as radicale_item from radicale import httputils, sharing, storage, types, xmlutils from radicale.app.base import Access, ApplicationBase from radicale.hook import HookNotificationItem, HookNotificationItemTypes from radicale.log import logger def xml_proppatch(base_prefix: str, path: str, xml_request: Optional[ET.Element], collection: Union[storage.BaseCollection, None], share: Union[dict, None] = None, share_overlay: bool = False, _sharing: Union[sharing.BaseSharing, None] = None) -> ET.Element: """Read and answer PROPPATCH requests. Read rfc4918-9.2 for info. """ multistatus = ET.Element(xmlutils.make_clark("D:multistatus")) response = ET.Element(xmlutils.make_clark("D:response")) multistatus.append(response) href = ET.Element(xmlutils.make_clark("D:href")) href.text = xmlutils.make_href(base_prefix, path) if share: # backmap href.text = href.text.replace(share['PathMapped'], share['PathOrToken']) response.append(href) # Create D:propstat element for props with status 200 OK propstat = ET.Element(xmlutils.make_clark("D:propstat")) status = ET.Element(xmlutils.make_clark("D:status")) status.text = xmlutils.make_response(200) props_ok = ET.Element(xmlutils.make_clark("D:prop")) propstat.append(props_ok) propstat.append(status) response.append(propstat) props_with_remove = xmlutils.props_from_request(xml_request) if share and share_overlay: # PROPPATCH overlay adjustment logger.debug("TRACE/PROPPATCH/xml_proppatch: share+share_overlay is active: %r", share) if share['Properties'] is not None: all_props_with_remove = cast(Dict[str, Optional[str]], radicale_item.check_and_sanitize_props(share['Properties'])) else: all_props_with_remove = {} all_props_with_remove.update(props_with_remove) all_props = radicale_item.check_and_sanitize_props(all_props_with_remove) logger.debug("TRACE/PROPPATCH/xml_proppatch: share+share_overlay result: %r", all_props) else: if collection is not None: # always the case, but makes mypy happy all_props_with_remove = cast(Dict[str, Optional[str]], dict(collection.get_meta())) all_props_with_remove.update(props_with_remove) all_props = radicale_item.check_and_sanitize_props(all_props_with_remove) if share and share_overlay and _sharing is not None: # _sharing is not None: always the case, but makes mypy happy _sharing.sharing_collection_update(ShareType=share['ShareType'], PathOrToken=share['PathOrToken'], OwnerOrUser=share['User'], Properties=cast(Dict[str, str], all_props)) else: if collection is not None: # always the case, but makes mypy happy collection.set_meta(all_props) for short_name in props_with_remove: props_ok.append(ET.Element(xmlutils.make_clark(short_name))) return multistatus class ApplicationPartProppatch(ApplicationBase): def do_PROPPATCH(self, environ: types.WSGIEnviron, base_prefix: str, path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse: """Manage PROPPATCH request.""" permissions_filter = None share = None share_overlay = False path_orig = path if self._sharing._enabled: # Sharing by token or map (if enabled) share = self._sharing.sharing_collection_resolver(path, user) if share: # overwrite and run through extended permission check path = share['PathMapped'] user = share['Owner'] permissions_filter = share['Permissions'] access = Access(self._rights, user, path, permissions_filter) raw_permissions = self._rights.authorization(user, path) if not access.check("w"): logger.debug("TRACE/PROPPATCH/xml_proppatch: no native write-access: %r", path) if share: # priority share->rights->global if logger.isEnabledFor(logging.DEBUG): logger.debug("TRACE/PROPPATCH/share: raw_permissions=%r share[Permissions]=%r permit_properties_overlay=%s enforce_properties_overlay=%s", raw_permissions, share['Permissions'], self._sharing.permit_properties_overlay, self._sharing.enforce_properties_overlay) if ("P" in share['Permissions'] or ("P" in raw_permissions and "p" not in share['Permissions']) or (self._sharing.permit_properties_overlay and "p" not in raw_permissions and "p" not in share['Permissions']) ) and not ( "p" in share['Permissions'] or ("p" in raw_permissions and "P" not in share['Permissions']) or (self._sharing.permit_properties_overlay and "P" not in raw_permissions and "P" not in share['Permissions'])): logger.info("PROPPATCH request on shared %r: write-access", path_orig) if permissions_filter is not None and "e" in permissions_filter: logger.info("PROPPATCH request on shared %r: write-access, overlay enforced, but disabled by share permission 'e'", path_orig) elif "e" in raw_permissions: logger.info("PROPPATCH request on shared %r: write-access, overlay enforced, but disabled by rights permission 'e'", path_orig) else: share_overlay = True else: logger.info("PROPPATCH request on shared %r: no write-access", path_orig) return httputils.NOT_ALLOWED else: return httputils.NOT_ALLOWED else: logger.debug("TRACE/PROPPATCH/xml_proppatch: write-access: %r", path) if share: # write access -> check for enforced properties overlay logger.debug("TRACE/PROPPATCH/xml_proppatch: write-access/sharing: %r", path_orig) if self._sharing.enforce_properties_overlay: if permissions_filter is not None and "e" in permissions_filter: logger.info("PROPPATCH request on shared %r: write-permissions, overlay enforced, but disabled by share permission 'e'", path_orig) elif "e" in raw_permissions: logger.info("PROPPATCH request on shared %r: write-permissions, overlay enforced, but disabled by rights permission 'e'", path_orig) else: share_overlay = True else: if permissions_filter is not None and "E" in permissions_filter: logger.info("PROPPATCH request on shared %r: write-permissions, overlay not enforced, but enforced by permission 'E'", path_orig) share_overlay = True try: xml_content = self._read_xml_request_body(environ) except RuntimeError as e: logger.warning( "Bad PROPPATCH request on %r: %s", path, e, exc_info=True) return httputils.BAD_REQUEST except socket.timeout: logger.debug("Client timed out", exc_info=True) return httputils.REQUEST_TIMEOUT if share_overlay: # call API function internally and no not trigger any hook headers = {"DAV": httputils.DAV_HEADERS, "Content-Type": "text/xml; charset=%s" % self._encoding} try: xml_answer = xml_proppatch(base_prefix, path, xml_content, None, share, share_overlay, self._sharing) if xml_content is not None: content = DefusedET.tostring( xml_content, encoding=self._encoding ).decode(encoding=self._encoding) except ValueError as e: # return better matching HTTP result in case errno is provided and catched errno_match = re.search("\\[Errno ([0-9]+)\\]", str(e)) if errno_match: logger.error( "Failed PROPPATCH request on %r: %s", path, e, exc_info=True) errno_e = int(errno_match.group(1)) if errno_e == errno.ENOSPC: return httputils.INSUFFICIENT_STORAGE elif errno_e in [errno.EPERM, errno.EACCES]: return httputils.FORBIDDEN else: return httputils.INTERNAL_SERVER_ERROR else: logger.warning( "Bad PROPPATCH request on %r: %s", path, e, exc_info=True) return httputils.BAD_REQUEST return client.MULTI_STATUS, headers, self._xml_response(xml_answer), xmlutils.pretty_xml(xml_content) with self._storage.acquire_lock("w", user, path=path, request="PROPPATCH"): item = next(iter(self._storage.discover(path)), None) if not item: return httputils.NOT_FOUND if not access.check("w", item): return httputils.NOT_ALLOWED if not isinstance(item, storage.BaseCollection): return httputils.FORBIDDEN headers = {"DAV": httputils.DAV_HEADERS, "Content-Type": "text/xml; charset=%s" % self._encoding} try: xml_answer = xml_proppatch(base_prefix, path, xml_content, item, share) if xml_content is not None: content = DefusedET.tostring( xml_content, encoding=self._encoding ).decode(encoding=self._encoding) if self._hook.enabled: hook_notification_item = HookNotificationItem( notification_item_type=HookNotificationItemTypes.CPATCH, path=access.path, content=content, content_type=None, # Can't easily determine content type, won't trigger email hook uid=None, old_content=None, new_content=content ) self._hook.notify(hook_notification_item) except ValueError as e: # return better matching HTTP result in case errno is provided and catched errno_match = re.search("\\[Errno ([0-9]+)\\]", str(e)) if errno_match: logger.error( "Failed PROPPATCH request on %r: %s", path, e, exc_info=True) errno_e = int(errno_match.group(1)) if errno_e == errno.ENOSPC: return httputils.INSUFFICIENT_STORAGE elif errno_e in [errno.EPERM, errno.EACCES]: return httputils.FORBIDDEN else: return httputils.INTERNAL_SERVER_ERROR else: logger.warning( "Bad PROPPATCH request on %r: %s", path, e, exc_info=True) return httputils.BAD_REQUEST return client.MULTI_STATUS, headers, self._xml_response(xml_answer), xmlutils.pretty_xml(xml_content)