590 lines
28 KiB
Python
590 lines
28 KiB
Python
# 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-2021 Unrud <unrud@outlook.com>
|
|
# Copyright © 2025-2026 Peter Bieringer <pb@bieringer.de>
|
|
#
|
|
# 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 <http://www.gnu.org/licenses/>.
|
|
|
|
import collections
|
|
import itertools
|
|
import logging
|
|
import posixpath
|
|
import socket
|
|
import xml.etree.ElementTree as ET
|
|
from http import client
|
|
from typing import (Dict, Iterable, Iterator, List, Optional, Sequence, Tuple,
|
|
Union)
|
|
|
|
from radicale import (httputils, pathutils, rights, storage, types, utils,
|
|
xmlutils)
|
|
from radicale.app.base import Access, ApplicationBase
|
|
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, 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.
|
|
|
|
The collections parameter is a list of collections that are to be included
|
|
in the output.
|
|
|
|
"""
|
|
# A client may choose not to submit a request body. An empty PROPFIND
|
|
# request body MUST be treated as if it were an 'allprop' request.
|
|
top_element = (xml_request[0] if xml_request is not None else
|
|
ET.Element(xmlutils.make_clark("D:allprop")))
|
|
|
|
props: List[str] = []
|
|
allprop = False
|
|
propname = False
|
|
if top_element.tag == xmlutils.make_clark("D:allprop"):
|
|
allprop = True
|
|
elif top_element.tag == xmlutils.make_clark("D:propname"):
|
|
propname = True
|
|
elif top_element.tag == xmlutils.make_clark("D:prop"):
|
|
props.extend(prop.tag for prop in top_element)
|
|
|
|
if xmlutils.make_clark("D:current-user-principal") in props and not user:
|
|
# Ask for authentication
|
|
# Returning the DAV:unauthenticated pseudo-principal as specified in
|
|
# RFC 5397 doesn't seem to work with DAVx5.
|
|
return None
|
|
|
|
# Writing answer
|
|
multistatus = ET.Element(xmlutils.make_clark("D:multistatus"))
|
|
|
|
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, shares=shares, sharetype=sharetype))
|
|
|
|
return multistatus
|
|
|
|
|
|
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, 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")
|
|
|
|
if isinstance(item, storage.BaseCollection):
|
|
is_collection = True
|
|
is_leaf = item.tag in ("VADDRESSBOOK", "VCALENDAR", "VSUBSCRIBED")
|
|
collection = item
|
|
# Some clients expect collections to end with `/`
|
|
uri = pathutils.unstrip_path(item.path, True)
|
|
else:
|
|
is_collection = is_leaf = False
|
|
assert item.collection is not None
|
|
assert item.href
|
|
collection = item.collection
|
|
uri = pathutils.unstrip_path(posixpath.join(
|
|
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: check entry=%r", entry)
|
|
if entry is not None:
|
|
if logger.isEnabledFor(logging.DEBUG):
|
|
logger.debug("TRACE/PROPFIND/xml_propfind: PathMapped=%r uri=%r", shares[entry]['PathMapped'], uri)
|
|
if uri.startswith(shares[entry]['PathMapped']):
|
|
if sharetype is not None and shares[entry]['ShareType'] == sharetype:
|
|
share = shares[entry]
|
|
if logger.isEnabledFor(logging.DEBUG):
|
|
logger.debug("TRACE/PROPFIND/xml_propfind: found share=%r", share)
|
|
break
|
|
|
|
share_bday_automap = False
|
|
if share and share['ShareType'] == "bday":
|
|
share_bday_automap = True
|
|
|
|
if share:
|
|
# backmap
|
|
uri = uri.replace(share['PathMapped'], share['PathOrToken'])
|
|
if share_bday_automap and not uri.endswith("/"):
|
|
uri = uri.rstrip(".vcf") + ".ics"
|
|
|
|
href.text = xmlutils.make_href(base_prefix, uri)
|
|
response.append(href)
|
|
|
|
if propname or allprop:
|
|
props = []
|
|
# Should list all properties that can be retrieved by the code below
|
|
props.append(xmlutils.make_clark("D:principal-collection-set"))
|
|
props.append(xmlutils.make_clark("D:current-user-principal"))
|
|
props.append(xmlutils.make_clark("D:current-user-privilege-set"))
|
|
props.append(xmlutils.make_clark("D:supported-report-set"))
|
|
props.append(xmlutils.make_clark("D:resourcetype"))
|
|
props.append(xmlutils.make_clark("D:owner"))
|
|
if not allprop:
|
|
# RFC4791#5.2.5: SHOULD NOT be returned by a PROPFIND DAV:allprop request
|
|
props.append(xmlutils.make_clark("C:max-resource-size"))
|
|
|
|
if is_collection and collection.is_principal:
|
|
props.append(xmlutils.make_clark("C:calendar-user-address-set"))
|
|
props.append(xmlutils.make_clark("D:principal-URL"))
|
|
props.append(xmlutils.make_clark("CR:addressbook-home-set"))
|
|
props.append(xmlutils.make_clark("C:calendar-home-set"))
|
|
|
|
if not is_collection or is_leaf:
|
|
props.append(xmlutils.make_clark("D:getetag"))
|
|
props.append(xmlutils.make_clark("D:getlastmodified"))
|
|
props.append(xmlutils.make_clark("D:getcontenttype"))
|
|
props.append(xmlutils.make_clark("D:getcontentlength"))
|
|
|
|
if is_collection:
|
|
if is_leaf:
|
|
props.append(xmlutils.make_clark("D:displayname"))
|
|
if share_bday_automap:
|
|
# bday share has no sync-token support
|
|
pass
|
|
else:
|
|
props.append(xmlutils.make_clark("D:sync-token"))
|
|
if collection.tag == "VCALENDAR" or share_bday_automap:
|
|
props.append(xmlutils.make_clark("CS:getctag"))
|
|
props.append(
|
|
xmlutils.make_clark("C:supported-calendar-component-set"))
|
|
if collection.tag == "VADDRESSBOOK" and not share_bday_automap:
|
|
props.append(xmlutils.make_clark("CS:getctag"))
|
|
props.append(
|
|
xmlutils.make_clark("CR:supported-address-data"))
|
|
|
|
meta = collection.get_meta()
|
|
for tag in meta:
|
|
if tag == "tag":
|
|
continue
|
|
clark_tag = xmlutils.make_clark(tag)
|
|
if clark_tag not in props:
|
|
props.append(clark_tag)
|
|
|
|
responses: Dict[int, List[ET.Element]] = collections.defaultdict(list)
|
|
if propname:
|
|
for tag in props:
|
|
responses[200].append(ET.Element(tag))
|
|
props = []
|
|
for tag in props:
|
|
element = ET.Element(tag)
|
|
is404 = False
|
|
if tag == xmlutils.make_clark("D:getetag"):
|
|
if not is_collection or is_leaf:
|
|
element.text = item.etag
|
|
else:
|
|
is404 = True
|
|
elif tag == xmlutils.make_clark("D:getlastmodified"):
|
|
if not is_collection or is_leaf:
|
|
element.text = item.last_modified
|
|
else:
|
|
is404 = True
|
|
elif tag == xmlutils.make_clark("D:principal-collection-set"):
|
|
child_element = ET.Element(xmlutils.make_clark("D:href"))
|
|
child_element.text = xmlutils.make_href(base_prefix, "/")
|
|
element.append(child_element)
|
|
elif (tag in (xmlutils.make_clark("C:calendar-user-address-set"),
|
|
xmlutils.make_clark("D:principal-URL"),
|
|
xmlutils.make_clark("CR:addressbook-home-set"),
|
|
xmlutils.make_clark("C:calendar-home-set")) and
|
|
is_collection and collection.is_principal):
|
|
child_element = ET.Element(xmlutils.make_clark("D:href"))
|
|
child_element.text = xmlutils.make_href(base_prefix, path)
|
|
if share:
|
|
# backmap
|
|
child_element.text = child_element.text.replace(share['PathMapped'], share['PathOrToken'])
|
|
if share_bday_automap:
|
|
child_element.text = child_element.text.rstrip(".vcf") + ".ics"
|
|
element.append(child_element)
|
|
elif tag == xmlutils.make_clark("C:supported-calendar-component-set"):
|
|
human_tag = xmlutils.make_human_tag(tag)
|
|
if is_collection and is_leaf:
|
|
components_text = collection.get_meta(human_tag)
|
|
if components_text:
|
|
components = components_text.split(",")
|
|
else:
|
|
components = ["VTODO", "VEVENT", "VJOURNAL"]
|
|
if share_bday_automap:
|
|
# enforce VEVENT-only
|
|
components = ["VEVENT"]
|
|
for component in components:
|
|
comp = ET.Element(xmlutils.make_clark("C:comp"))
|
|
comp.set("name", component)
|
|
element.append(comp)
|
|
else:
|
|
is404 = True
|
|
elif tag == xmlutils.make_clark("CR:supported-address-data"):
|
|
if is_collection and is_leaf and collection.tag == "VADDRESSBOOK" and not share_bday_automap:
|
|
# Advertise supported vCard versions per RFC 6352 section 6.2.2
|
|
# vCard 4.0 requires vobject >= 1.0.0
|
|
versions: Sequence[str] = (("4.0", "3.0")
|
|
if utils.vobject_supports_vcard4()
|
|
else ("3.0",))
|
|
for version in versions:
|
|
address_data_type = ET.Element(
|
|
xmlutils.make_clark("CR:address-data-type"))
|
|
address_data_type.set("content-type", "text/vcard")
|
|
address_data_type.set("version", version)
|
|
element.append(address_data_type)
|
|
else:
|
|
is404 = True
|
|
elif tag == xmlutils.make_clark("D:current-user-principal"):
|
|
if user:
|
|
child_element = ET.Element(xmlutils.make_clark("D:href"))
|
|
child_element.text = xmlutils.make_href(
|
|
base_prefix, "/%s/" % user)
|
|
if share:
|
|
# backmap
|
|
child_element.text = child_element.text.replace(share['Owner'], share['User'])
|
|
if share_bday_automap:
|
|
child_element.text = child_element.text.rstrip(".vcf") + ".ics"
|
|
element.append(child_element)
|
|
else:
|
|
element.append(ET.Element(
|
|
xmlutils.make_clark("D:unauthenticated")))
|
|
elif tag == xmlutils.make_clark("D:current-user-privilege-set"):
|
|
privileges = ["D:read"]
|
|
if share:
|
|
if write:
|
|
if "P" in share['Permissions']:
|
|
privileges.append("D:write-properties")
|
|
if "w" in share['Permissions']:
|
|
if not share_bday_automap:
|
|
privileges.append("D:write-content")
|
|
elif write:
|
|
privileges.append("D:all")
|
|
privileges.append("D:write")
|
|
privileges.append("D:write-properties")
|
|
privileges.append("D:write-content")
|
|
for human_tag in privileges:
|
|
privilege = ET.Element(xmlutils.make_clark("D:privilege"))
|
|
privilege.append(ET.Element(
|
|
xmlutils.make_clark(human_tag)))
|
|
element.append(privilege)
|
|
elif tag == xmlutils.make_clark("D:supported-report-set"):
|
|
# These 3 reports are not implemented
|
|
reports = ["D:expand-property",
|
|
"D:principal-search-property-set",
|
|
"D:principal-property-search"]
|
|
if is_collection and is_leaf:
|
|
if not share_bday_automap:
|
|
reports.append("D:sync-collection")
|
|
if collection.tag == "VADDRESSBOOK" and not share_bday_automap:
|
|
reports.append("CR:addressbook-multiget")
|
|
reports.append("CR:addressbook-query")
|
|
elif collection.tag == "VCALENDAR" or share_bday_automap:
|
|
reports.append("C:calendar-multiget")
|
|
reports.append("C:calendar-query")
|
|
for human_tag in reports:
|
|
supported_report = ET.Element(
|
|
xmlutils.make_clark("D:supported-report"))
|
|
report_element = ET.Element(xmlutils.make_clark("D:report"))
|
|
report_element.append(
|
|
ET.Element(xmlutils.make_clark(human_tag)))
|
|
supported_report.append(report_element)
|
|
element.append(supported_report)
|
|
elif tag == xmlutils.make_clark("D:getcontentlength"):
|
|
if not is_collection or is_leaf:
|
|
if collection.tag == "VADDRESSBOOK" and share_bday_automap and isinstance(item, storage.BaseCollection):
|
|
if logger.isEnabledFor(logging.DEBUG):
|
|
logger.debug("TRACE/PROPFIND/xml_propfind_response/getcontentlength: start bday automap handling")
|
|
length = 0
|
|
for entry in item.get_all():
|
|
item_ics = entry.convert_vcf_to_ics()
|
|
if item_ics is None:
|
|
continue
|
|
length += len(item_ics.vobject_item.serialize().encode(encoding))
|
|
element.text = str(length)
|
|
else:
|
|
element.text = str(len(item.serialize().encode(encoding)))
|
|
else:
|
|
is404 = True
|
|
elif tag == xmlutils.make_clark("D:owner"):
|
|
# return empty elment, if no owner available (rfc3744-5.1)
|
|
if collection.owner:
|
|
child_element = ET.Element(xmlutils.make_clark("D:href"))
|
|
child_element.text = xmlutils.make_href(
|
|
base_prefix, "/%s/" % collection.owner)
|
|
element.append(child_element)
|
|
elif tag == xmlutils.make_clark("C:max-resource-size"):
|
|
# RFC4791#5.2.5
|
|
element.text = str(max_resource_size)
|
|
elif is_collection:
|
|
if tag == xmlutils.make_clark("D:getcontenttype"):
|
|
if is_leaf:
|
|
element.text = xmlutils.MIMETYPES[
|
|
collection.tag]
|
|
if share_bday_automap:
|
|
# overwrite
|
|
element.text = xmlutils.MIMETYPES["VCALENDAR"]
|
|
else:
|
|
is404 = True
|
|
elif tag == xmlutils.make_clark("D:resourcetype"):
|
|
if collection.is_principal:
|
|
child_element = ET.Element(
|
|
xmlutils.make_clark("D:principal"))
|
|
element.append(child_element)
|
|
if is_leaf:
|
|
if collection.tag == "VADDRESSBOOK" and not share_bday_automap:
|
|
child_element = ET.Element(
|
|
xmlutils.make_clark("CR:addressbook"))
|
|
element.append(child_element)
|
|
elif collection.tag == "VCALENDAR" or share_bday_automap:
|
|
child_element = ET.Element(
|
|
xmlutils.make_clark("C:calendar"))
|
|
element.append(child_element)
|
|
elif collection.tag == "VSUBSCRIBED":
|
|
child_element = ET.Element(
|
|
xmlutils.make_clark("CS:subscribed"))
|
|
element.append(child_element)
|
|
child_element = ET.Element(xmlutils.make_clark("D:collection"))
|
|
element.append(child_element)
|
|
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
|
|
if share_bday_automap:
|
|
element.text += " (BDAY)"
|
|
else:
|
|
is404 = True
|
|
elif tag == xmlutils.make_clark("RADICALE:getcontentcount"):
|
|
# Only for internal use by the web interface
|
|
if isinstance(item, storage.BaseCollection) and not collection.is_principal:
|
|
if collection.tag == "VADDRESSBOOK" and share_bday_automap:
|
|
if logger.isEnabledFor(logging.DEBUG):
|
|
logger.debug("TRACE/PROPFIND/xml_propfind_response/getcontentcount: start bday automap handling")
|
|
items = []
|
|
for entry in item.get_all():
|
|
item_ics = entry.convert_vcf_to_ics()
|
|
if item_ics is None:
|
|
continue
|
|
items.append(item_ics.vobject_item)
|
|
element.text = str(sum(1 for x in items))
|
|
else:
|
|
element.text = str(sum(1 for x in item.get_all()))
|
|
else:
|
|
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:
|
|
element.text = displayname
|
|
if share_bday_automap:
|
|
element.text += " (BDAY)"
|
|
else:
|
|
is404 = True
|
|
elif tag == xmlutils.make_clark("CS:getctag"):
|
|
if is_leaf:
|
|
element.text = collection.etag
|
|
else:
|
|
is404 = True
|
|
elif tag == xmlutils.make_clark("D:sync-token"):
|
|
if is_leaf:
|
|
element.text, _ = collection.sync()
|
|
else:
|
|
is404 = True
|
|
elif tag == xmlutils.make_clark("CS:source"):
|
|
if is_leaf:
|
|
child_element = ET.Element(xmlutils.make_clark("D:href"))
|
|
child_element.text = collection.get_meta('CS:source')
|
|
element.append(child_element)
|
|
else:
|
|
is404 = True
|
|
else:
|
|
human_tag = xmlutils.make_human_tag(tag)
|
|
tag_text = collection.get_meta(human_tag)
|
|
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'][human_tag] is not None:
|
|
tag_text = share['Properties'][human_tag]
|
|
if tag_text is not None:
|
|
element.text = tag_text
|
|
else:
|
|
is404 = True
|
|
# Not for collections
|
|
elif tag == xmlutils.make_clark("D:getcontenttype"):
|
|
assert not isinstance(item, storage.BaseCollection)
|
|
element.text = xmlutils.get_content_type(item, encoding)
|
|
if share_bday_automap:
|
|
# overwrite
|
|
element.text = xmlutils.MIMETYPES["VCALENDAR"]
|
|
elif tag == xmlutils.make_clark("D:resourcetype"):
|
|
# resourcetype must be returned empty for non-collection elements
|
|
pass
|
|
else:
|
|
is404 = True
|
|
|
|
responses[404 if is404 else 200].append(element)
|
|
|
|
for status_code, children in responses.items():
|
|
if not children:
|
|
continue
|
|
propstat = ET.Element(xmlutils.make_clark("D:propstat"))
|
|
response.append(propstat)
|
|
prop = ET.Element(xmlutils.make_clark("D:prop"))
|
|
prop.extend(children)
|
|
propstat.append(prop)
|
|
status = ET.Element(xmlutils.make_clark("D:status"))
|
|
status.text = xmlutils.make_response(status_code)
|
|
propstat.append(status)
|
|
|
|
return response
|
|
|
|
|
|
class ApplicationPartPropfind(ApplicationBase):
|
|
|
|
def _collect_allowed_items(
|
|
self, items: Iterable[types.CollectionOrItem], user: str
|
|
) -> Iterator[Tuple[types.CollectionOrItem, str]]:
|
|
"""Get items from request that user is allowed to access."""
|
|
for item in items:
|
|
if isinstance(item, storage.BaseCollection):
|
|
path = pathutils.unstrip_path(item.path, True)
|
|
if logger.isEnabledFor(logging.DEBUG):
|
|
logger.debug("TRACE/PROPFIND/_collect_allowed_items/BaseCollection: path=%r user=%r", path, user)
|
|
if item.tag:
|
|
permissions = rights.intersect(
|
|
self._rights.authorization(user, path), "rw")
|
|
target = "collection with tag %r" % item.path
|
|
else:
|
|
permissions = rights.intersect(
|
|
self._rights.authorization(user, path), "RW")
|
|
target = "collection %r" % item.path
|
|
else:
|
|
assert item.collection is not None
|
|
path = pathutils.unstrip_path(item.collection.path, True)
|
|
permissions = rights.intersect(
|
|
self._rights.authorization(user, path), "rw")
|
|
target = "item %r from %r" % (item.href, item.collection.path)
|
|
if rights.intersect(permissions, "Ww"):
|
|
permission = "w"
|
|
status = "write"
|
|
elif rights.intersect(permissions, "Rr"):
|
|
permission = "r"
|
|
status = "read"
|
|
else:
|
|
permission = ""
|
|
status = "NO"
|
|
logger.debug(
|
|
"%s has %s access to %s",
|
|
repr(user) if user else "anonymous user", status, target)
|
|
if permission:
|
|
yield item, permission
|
|
|
|
def do_PROPFIND(self, environ: types.WSGIEnviron, base_prefix: str,
|
|
path: str, user: str, remote_host: str, remote_useragent: str) -> types.WSGIResponse:
|
|
"""Manage PROPFIND request."""
|
|
http_depth = environ.get("HTTP_DEPTH", "0")
|
|
permissions_filter = 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)
|
|
if share:
|
|
# overwrite and run through extended permission check
|
|
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
|
|
try:
|
|
xml_content = self._read_xml_request_body(environ)
|
|
except RuntimeError as e:
|
|
logger.warning(
|
|
"Bad PROPFIND 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
|
|
with self._storage.acquire_lock("r", user):
|
|
if logger.isEnabledFor(logging.DEBUG):
|
|
logger.debug("TRACE/PROPFIND: discover path=%r depth=%s", path, http_depth)
|
|
items_iter = iter(self._storage.discover(
|
|
path, http_depth,
|
|
None, self._rights._user_groups))
|
|
# take root item for rights checking
|
|
item = next(items_iter, None)
|
|
if not item:
|
|
return httputils.NOT_FOUND
|
|
if not access.check("r", item):
|
|
return httputils.NOT_ALLOWED
|
|
# put item back
|
|
items_iter = itertools.chain([item], items_iter)
|
|
for item, permission in list(self._collect_allowed_items(items_iter, user)):
|
|
if self._sharing._enabled and share:
|
|
if share['ShareType'] == "bday" and not isinstance(item, storage.BaseCollection):
|
|
if not item.convert_vcf_to_ics():
|
|
continue
|
|
allowed_items.append((item, permission, share['ShareType']))
|
|
else:
|
|
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_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']
|
|
c_permissions_filter = share['Permissions']
|
|
if logger.isEnabledFor(logging.DEBUG):
|
|
logger.debug("TRACE/PROPFIND: test shared collection: PathOrToken=%r PathMapped=%r Owner=%r Permissions=%s", c_share, c_path, c_user, c_permissions_filter)
|
|
c_access = Access(self._rights, c_user, c_path, c_permissions_filter)
|
|
if not c_access.check("r"):
|
|
if logger.isEnabledFor(logging.DEBUG):
|
|
logger.debug("TRACE/PROPFIND: skip shared collection: PathOrToken=%r PathMapped=%r Owner=%r Permissions=%s (permissions not matching)", c_share, c_path, c_user, c_permissions_filter)
|
|
continue
|
|
if logger.isEnabledFor(logging.DEBUG):
|
|
logger.debug("TRACE/PROPFIND: append shared collection: PathOrToken=%r PathMapped=%r Owner=%r", c_share, c_path, c_user)
|
|
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))
|
|
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, 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)
|