8580040934

Committed 06 Apr 2024 08:54AM UTC coverage: 82.934% (+0.2%) from 82.779%

Build # 8580040934

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1465 from metallerok/processing-expand-property

Validation fixes for processing expand property

Run Details

1594 of 2051 branches covered (77.72%)

Branch coverage included in aggregate %.

10 of 10 new or added lines in 1 file covered. (100.0%)

39 existing lines in 1 file now uncovered.

3601 of 4213 relevant lines covered (85.47%)

19.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

78.9

/radicale/app/report.py

# 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-2018 Unrud <unrud@outlook.com>
#
# 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 contextlib
import copy
import datetime
import posixpath
import socket
import xml.etree.ElementTree as ET
from http import client
from typing import (Callable, Iterable, Iterator, List, Optional, Sequence,
                    Tuple, Union)
from urllib.parse import unquote, urlparse

import vobject.base
from vobject.base import ContentLine

import radicale.item as radicale_item
from radicale import httputils, pathutils, storage, types, xmlutils
from radicale.app.base import Access, ApplicationBase
from radicale.item import filter as radicale_filter
from radicale.log import logger


def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],
               collection: storage.BaseCollection, encoding: str,
               unlock_storage_fn: Callable[[], None]
               ) -> Tuple[int, ET.Element]:
    """Read and answer REPORT requests.

    Read rfc3253-3.6 for info.

    """
    multistatus = ET.Element(xmlutils.make_clark("D:multistatus"))
    if xml_request is None:
        return client.MULTI_STATUS, multistatus
    root = xml_request
    if root.tag in (xmlutils.make_clark("D:principal-search-property-set"),
                    xmlutils.make_clark("D:principal-property-search"),
                    xmlutils.make_clark("D:expand-property")):
        # We don't support searching for principals or indirect retrieving of
        # properties, just return an empty result.
        # InfCloud asks for expand-property reports (even if we don't announce
        # support for them) and stops working if an error code is returned.
        logger.warning("Unsupported REPORT method %r on %r requested",
                       xmlutils.make_human_tag(root.tag), path)
        return client.MULTI_STATUS, multistatus
    if (root.tag == xmlutils.make_clark("C:calendar-multiget") and
            collection.tag != "VCALENDAR" or
            root.tag == xmlutils.make_clark("CR:addressbook-multiget") and
            collection.tag != "VADDRESSBOOK" or
            root.tag == xmlutils.make_clark("D:sync-collection") and
            collection.tag not in ("VADDRESSBOOK", "VCALENDAR")):
        logger.warning("Invalid REPORT method %r on %r requested",
                       xmlutils.make_human_tag(root.tag), path)
        return client.FORBIDDEN, xmlutils.webdav_error("D:supported-report")

    props: Union[ET.Element, List] = root.find(xmlutils.make_clark("D:prop")) or []

    hreferences: Iterable[str]
    if root.tag in (
            xmlutils.make_clark("C:calendar-multiget"),
            xmlutils.make_clark("CR:addressbook-multiget")):
        # Read rfc4791-7.9 for info
        hreferences = set()
        for href_element in root.findall(xmlutils.make_clark("D:href")):
            temp_url_path = urlparse(href_element.text).path
            assert isinstance(temp_url_path, str)
            href_path = pathutils.sanitize_path(unquote(temp_url_path))
            if (href_path + "/").startswith(base_prefix + "/"):
                hreferences.add(href_path[len(base_prefix):])
            else:
                logger.warning("Skipping invalid path %r in REPORT request on "
                               "%r", href_path, path)
    elif root.tag == xmlutils.make_clark("D:sync-collection"):
        old_sync_token_element = root.find(
            xmlutils.make_clark("D:sync-token"))
        old_sync_token = ""
        if old_sync_token_element is not None and old_sync_token_element.text:
            old_sync_token = old_sync_token_element.text.strip()
        logger.debug("Client provided sync token: %r", old_sync_token)
        try:
            sync_token, names = collection.sync(old_sync_token)
        except ValueError as e:
            # Invalid sync token
            logger.warning("Client provided invalid sync token %r: %s",
                           old_sync_token, e, exc_info=True)
            # client.CONFLICT doesn't work with some clients (e.g. InfCloud)
            return (client.FORBIDDEN,
                    xmlutils.webdav_error("D:valid-sync-token"))
        hreferences = (pathutils.unstrip_path(
            posixpath.join(collection.path, n)) for n in names)
        # Append current sync token to response
        sync_token_element = ET.Element(xmlutils.make_clark("D:sync-token"))
        sync_token_element.text = sync_token
        multistatus.append(sync_token_element)
    else:
        hreferences = (path,)
    filters = (
        root.findall(xmlutils.make_clark("C:filter")) +
        root.findall(xmlutils.make_clark("CR:filter")))

    # Retrieve everything required for finishing the request.
    retrieved_items = list(retrieve_items(
        base_prefix, path, collection, hreferences, filters, multistatus))
    collection_tag = collection.tag
    # !!! Don't access storage after this !!!
    unlock_storage_fn()

    while retrieved_items:
        # ``item.vobject_item`` might be accessed during filtering.
        # Don't keep reference to ``item``, because VObject requires a lot of
        # memory.
        item, filters_matched = retrieved_items.pop(0)
        if filters and not filters_matched:
            try:
                if not all(test_filter(collection_tag, item, filter_)
                           for filter_ in filters):
                    continue
            except ValueError as e:
                raise ValueError("Failed to filter item %r from %r: %s" %
                                 (item.href, collection.path, e)) from e
            except Exception as e:
                raise RuntimeError("Failed to filter item %r from %r: %s" %
                                   (item.href, collection.path, e)) from e

        found_props = []
        not_found_props = []

        for prop in props:
            element = ET.Element(prop.tag)
            if prop.tag == xmlutils.make_clark("D:getetag"):
                element.text = item.etag
                found_props.append(element)
            elif prop.tag == xmlutils.make_clark("D:getcontenttype"):
                element.text = xmlutils.get_content_type(item, encoding)
                found_props.append(element)
            elif prop.tag in (
                    xmlutils.make_clark("C:calendar-data"),
                    xmlutils.make_clark("CR:address-data")):
                element.text = item.serialize()

                expand = prop.find(xmlutils.make_clark("C:expand"))
                if expand is not None:
                    start = expand.get('start')
                    end = expand.get('end')

                    if (start is None) or (end is None):
                        return client.FORBIDDEN, \
                            xmlutils.webdav_error("C:expand")

                    start = datetime.datetime.strptime(
                        start, '%Y%m%dT%H%M%SZ'
                    ).replace(tzinfo=datetime.timezone.utc)
                    end = datetime.datetime.strptime(
                        end, '%Y%m%dT%H%M%SZ'
                    ).replace(tzinfo=datetime.timezone.utc)

                    expanded_element = _expand(
                        element, copy.copy(item), start, end)
                    found_props.append(expanded_element)
                else:
                    found_props.append(element)
            else:
                not_found_props.append(element)

        assert item.href
        uri = pathutils.unstrip_path(
            posixpath.join(collection.path, item.href))
        multistatus.append(xml_item_response(
            base_prefix, uri, found_props=found_props,
            not_found_props=not_found_props, found_item=True))

    return client.MULTI_STATUS, multistatus


def _expand(
        element: ET.Element,
        item: radicale_item.Item,
        start: datetime.datetime,
        end: datetime.datetime,
) -> ET.Element:
    rruleset = None
    if hasattr(item.vobject_item.vevent, 'rrule'):
        rruleset = item.vobject_item.vevent.getrruleset()

    expanded_item = _make_vobject_expanded_item(item)

    if rruleset:
        recurrences = rruleset.between(start, end)

        expanded: vobject.base.Component = copy.copy(expanded_item.vobject_item)
        is_expanded_filled: bool = False

        for recurrence_dt in recurrences:
            recurrence_utc = recurrence_dt.astimezone(datetime.timezone.utc)

            vevent = copy.deepcopy(expanded.vevent)
            vevent.recurrence_id = ContentLine(
                name='RECURRENCE-ID',
                value=recurrence_utc.strftime('%Y%m%dT%H%M%SZ'), params={}
            )

            if is_expanded_filled is False:
                expanded.vevent = vevent
                is_expanded_filled = True
            else:
                expanded.add(vevent)

        element.text = expanded.serialize()
    else:
        element.text = expanded_item.vobject_item.serialize()

    return element


def _make_vobject_expanded_item(
        item: radicale_item.Item
) -> radicale_item.Item:
    # https://www.rfc-editor.org/rfc/rfc4791#section-9.6.5
    # The returned calendar components MUST NOT use recurrence
    #       properties (i.e., EXDATE, EXRULE, RDATE, and RRULE) and MUST NOT
    #       have reference to or include VTIMEZONE components.  Date and local
    #       time with reference to time zone information MUST be converted
    #       into date with UTC time.

    item = copy.copy(item)
    vevent = item.vobject_item.vevent

    start_utc = vevent.dtstart.value.astimezone(datetime.timezone.utc)
    vevent.dtstart = ContentLine(
        name='DTSTART',
        value=start_utc.strftime('%Y%m%dT%H%M%SZ'), params={})

    dt_end = getattr(vevent, 'dtend', None)
    if dt_end is not None:
        end_utc = dt_end.value.astimezone(datetime.timezone.utc)
        vevent.dtend = ContentLine(
            name='DTEND',
            value=end_utc.strftime('%Y%m%dT%H%M%SZ'), params={})

    timezones_to_remove = []
    for component in item.vobject_item.components():
        if component.name == 'VTIMEZONE':
            timezones_to_remove.append(component)

    for timezone in timezones_to_remove:
        item.vobject_item.remove(timezone)

    try:
        delattr(item.vobject_item.vevent, 'rrule')
        delattr(item.vobject_item.vevent, 'exdate')
        delattr(item.vobject_item.vevent, 'exrule')
        delattr(item.vobject_item.vevent, 'rdate')
    except AttributeError:
        pass

    return item


def xml_item_response(base_prefix: str, href: str,
                      found_props: Sequence[ET.Element] = (),
                      not_found_props: Sequence[ET.Element] = (),
                      found_item: bool = True) -> ET.Element:
    response = ET.Element(xmlutils.make_clark("D:response"))

    href_element = ET.Element(xmlutils.make_clark("D:href"))
    href_element.text = xmlutils.make_href(base_prefix, href)
    response.append(href_element)

    if found_item:
        for code, props in ((200, found_props), (404, not_found_props)):
            if props:
                propstat = ET.Element(xmlutils.make_clark("D:propstat"))
                status = ET.Element(xmlutils.make_clark("D:status"))
                status.text = xmlutils.make_response(code)
                prop_element = ET.Element(xmlutils.make_clark("D:prop"))
                for prop in props:
                    prop_element.append(prop)
                propstat.append(prop_element)
                propstat.append(status)
                response.append(propstat)
    else:
        status = ET.Element(xmlutils.make_clark("D:status"))
        status.text = xmlutils.make_response(404)
        response.append(status)

    return response


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]]:
    """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]:
        """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:
            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)
                multistatus.append(response)
                continue
            if name:
                # Reference is an item
                yield name
            else:
                # Reference is a collection
                collection_requested = True

    for name, item in collection.get_multi(get_names()):
        if not item:
            uri = pathutils.unstrip_path(posixpath.join(collection.path, name))
            response = xml_item_response(base_prefix, uri, found_item=False)
            multistatus.append(response)
        else:
            yield item, False
    if collection_requested:
        yield from collection.get_filtered(filters)


def test_filter(collection_tag: str, item: radicale_item.Item,
                filter_: ET.Element) -> bool:
    """Match an item against a filter."""
    if (collection_tag == "VCALENDAR" and
            filter_.tag != xmlutils.make_clark("C:%s" % filter_)):
        if len(filter_) == 0:
            return True
        if len(filter_) > 1:
            raise ValueError("Filter with %d children" % len(filter_))
        if filter_[0].tag != xmlutils.make_clark("C:comp-filter"):
            raise ValueError("Unexpected %r in filter" % filter_[0].tag)
        return radicale_filter.comp_match(item, filter_[0])
    if (collection_tag == "VADDRESSBOOK" and
            filter_.tag != xmlutils.make_clark("CR:%s" % filter_)):
        for child in filter_:
            if child.tag != xmlutils.make_clark("CR:prop-filter"):
                raise ValueError("Unexpected %r in filter" % child.tag)
        test = filter_.get("test", "anyof")
        if test == "anyof":
            return any(radicale_filter.prop_match(item.vobject_item, f, "CR")
                       for f in filter_)
        if test == "allof":
            return all(radicale_filter.prop_match(item.vobject_item, f, "CR")
                       for f in filter_)
        raise ValueError("Unsupported filter test: %r" % test)
    raise ValueError("Unsupported filter %r for %r" %
                     (filter_.tag, collection_tag))


class ApplicationPartReport(ApplicationBase):

    def do_REPORT(self, environ: types.WSGIEnviron, base_prefix: str,
                  path: str, user: str) -> types.WSGIResponse:
        """Manage REPORT request."""
        access = Access(self._rights, user, path)
        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 REPORT 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 contextlib.ExitStack() as lock_stack:
            lock_stack.enter_context(self._storage.acquire_lock("r", user))
            item = next(iter(self._storage.discover(path)), None)
            if not item:
                return httputils.NOT_FOUND
            if not access.check("r", item):
                return httputils.NOT_ALLOWED
            if isinstance(item, storage.BaseCollection):
                collection = item
            else:
                assert item.collection is not None
                collection = item.collection
            try:
                status, xml_answer = xml_report(
                    base_prefix, path, xml_content, collection, self._encoding,
                    lock_stack.close)
            except ValueError as e:
                logger.warning(
                    "Bad REPORT request on %r: %s", path, e, exc_info=True)
                return httputils.BAD_REQUEST
        headers = {"Content-Type": "text/xml; charset=%s" % self._encoding}
        return status, headers, self._xml_response(xml_answer)

1	# This file is part of Radicale - CalDAV and CardDAV server
2	# Copyright © 2008 Nicolas Kandel
3	# Copyright © 2008 Pascal Halter
4	# Copyright © 2008-2017 Guillaume Ayoub
5	# Copyright © 2017-2018 Unrud <unrud@outlook.com>
6	#
7	# This library is free software: you can redistribute it and/or modify
8	# it under the terms of the GNU General Public License as published by
9	# the Free Software Foundation, either version 3 of the License, or
10	# (at your option) any later version.
11	#
12	# This library is distributed in the hope that it will be useful,
13	# but WITHOUT ANY WARRANTY; without even the implied warranty of
14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15	# GNU General Public License for more details.
16	#
17	# You should have received a copy of the GNU General Public License
18	# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
19
20	import contextlib	24✔
21	import copy	24✔
22	import datetime	24✔
23	import posixpath	24✔
24	import socket	24✔
25	import xml.etree.ElementTree as ET	24✔
26	from http import client	24✔
27	from typing import (Callable, Iterable, Iterator, List, Optional, Sequence,	24✔
28	Tuple, Union)
29	from urllib.parse import unquote, urlparse	24✔
30
31	import vobject.base	24✔
32	from vobject.base import ContentLine	24✔
33
34	import radicale.item as radicale_item	24✔
35	from radicale import httputils, pathutils, storage, types, xmlutils	24✔
36	from radicale.app.base import Access, ApplicationBase	24✔
37	from radicale.item import filter as radicale_filter	24✔
38	from radicale.log import logger	24✔
39
40
41	def xml_report(base_prefix: str, path: str, xml_request: Optional[ET.Element],	24✔
42	collection: storage.BaseCollection, encoding: str,
43	unlock_storage_fn: Callable[[], None]
44	) -> Tuple[int, ET.Element]:
45	"""Read and answer REPORT requests.
46
47	Read rfc3253-3.6 for info.
48
49	"""
50	multistatus = ET.Element(xmlutils.make_clark("D:multistatus"))	24✔
51	if xml_request is None:	24!
UNCOV 52	return client.MULTI_STATUS, multistatus	×
53	root = xml_request	24✔
54	if root.tag in (xmlutils.make_clark("D:principal-search-property-set"),	24!
55	xmlutils.make_clark("D:principal-property-search"),
56	xmlutils.make_clark("D:expand-property")):
57	# We don't support searching for principals or indirect retrieving of
58	# properties, just return an empty result.
59	# InfCloud asks for expand-property reports (even if we don't announce
60	# support for them) and stops working if an error code is returned.
UNCOV 61	logger.warning("Unsupported REPORT method %r on %r requested",	×
62	xmlutils.make_human_tag(root.tag), path)
UNCOV 63	return client.MULTI_STATUS, multistatus	×
64	if (root.tag == xmlutils.make_clark("C:calendar-multiget") and	24!
65	collection.tag != "VCALENDAR" or
66	root.tag == xmlutils.make_clark("CR:addressbook-multiget") and
67	collection.tag != "VADDRESSBOOK" or
68	root.tag == xmlutils.make_clark("D:sync-collection") and
69	collection.tag not in ("VADDRESSBOOK", "VCALENDAR")):
UNCOV 70	logger.warning("Invalid REPORT method %r on %r requested",	×
71	xmlutils.make_human_tag(root.tag), path)
UNCOV 72	return client.FORBIDDEN, xmlutils.webdav_error("D:supported-report")	×
73
74	props: Union[ET.Element, List] = root.find(xmlutils.make_clark("D:prop")) or []	24✔
75
76	hreferences: Iterable[str]
77	if root.tag in (	24!
78	xmlutils.make_clark("C:calendar-multiget"),
79	xmlutils.make_clark("CR:addressbook-multiget")):
80	# Read rfc4791-7.9 for info
81	hreferences = set()	×
82	for href_element in root.findall(xmlutils.make_clark("D:href")):	×
83	temp_url_path = urlparse(href_element.text).path	×
84	assert isinstance(temp_url_path, str)	×
UNCOV 85	href_path = pathutils.sanitize_path(unquote(temp_url_path))	×
86	if (href_path + "/").startswith(base_prefix + "/"):	×
UNCOV 87	hreferences.add(href_path[len(base_prefix):])	×
88	else:
UNCOV 89	logger.warning("Skipping invalid path %r in REPORT request on "	×
90	"%r", href_path, path)
91	elif root.tag == xmlutils.make_clark("D:sync-collection"):	24✔
92	old_sync_token_element = root.find(	24✔
93	xmlutils.make_clark("D:sync-token"))
94	old_sync_token = ""	24✔
95	if old_sync_token_element is not None and old_sync_token_element.text:	24✔
96	old_sync_token = old_sync_token_element.text.strip()	24✔
97	logger.debug("Client provided sync token: %r", old_sync_token)	24✔
98	try:	24✔
99	sync_token, names = collection.sync(old_sync_token)	24✔
100	except ValueError as e:	24✔
101	# Invalid sync token
102	logger.warning("Client provided invalid sync token %r: %s",	24✔
103	old_sync_token, e, exc_info=True)
104	# client.CONFLICT doesn't work with some clients (e.g. InfCloud)
105	return (client.FORBIDDEN,	24✔
106	xmlutils.webdav_error("D:valid-sync-token"))
107	hreferences = (pathutils.unstrip_path(	24✔
108	posixpath.join(collection.path, n)) for n in names)
109	# Append current sync token to response
110	sync_token_element = ET.Element(xmlutils.make_clark("D:sync-token"))	24✔
111	sync_token_element.text = sync_token	24✔
112	multistatus.append(sync_token_element)	24✔
113	else:
114	hreferences = (path,)	24✔
115	filters = (	24✔
116	root.findall(xmlutils.make_clark("C:filter")) +
117	root.findall(xmlutils.make_clark("CR:filter")))
118
119	# Retrieve everything required for finishing the request.
120	retrieved_items = list(retrieve_items(	24✔
121	base_prefix, path, collection, hreferences, filters, multistatus))
122	collection_tag = collection.tag	24✔
123	# !!! Don't access storage after this !!!
124	unlock_storage_fn()	24✔
125
126	while retrieved_items:	24✔
127	# ``item.vobject_item`` might be accessed during filtering.
128	# Don't keep reference to ``item``, because VObject requires a lot of
129	# memory.
130	item, filters_matched = retrieved_items.pop(0)	24✔
131	if filters and not filters_matched:	24✔
132	try:	24✔
133	if not all(test_filter(collection_tag, item, filter_)	24✔
134	for filter_ in filters):
135	continue	24✔
136	except ValueError as e:	×
137	raise ValueError("Failed to filter item %r from %r: %s" %	×
138	(item.href, collection.path, e)) from e
UNCOV 139	except Exception as e:	×
UNCOV 140	raise RuntimeError("Failed to filter item %r from %r: %s" %	×
141	(item.href, collection.path, e)) from e
142
143	found_props = []	24✔
144	not_found_props = []	24✔
145
146	for prop in props:	24✔
147	element = ET.Element(prop.tag)	24✔
148	if prop.tag == xmlutils.make_clark("D:getetag"):	24✔
149	element.text = item.etag	24✔
150	found_props.append(element)	24✔
151	elif prop.tag == xmlutils.make_clark("D:getcontenttype"):	24✔
152	element.text = xmlutils.get_content_type(item, encoding)	24✔
153	found_props.append(element)	24✔
154	elif prop.tag in (	24!
155	xmlutils.make_clark("C:calendar-data"),
156	xmlutils.make_clark("CR:address-data")):
157	element.text = item.serialize()	24✔
158
159	expand = prop.find(xmlutils.make_clark("C:expand"))	24✔
160	if expand is not None:	24✔
161	start = expand.get('start')	24✔
162	end = expand.get('end')	24✔
163
164	if (start is None) or (end is None):	24!
UNCOV 165	return client.FORBIDDEN, \	×
166	xmlutils.webdav_error("C:expand")
167
168	start = datetime.datetime.strptime(	24✔
169	start, '%Y%m%dT%H%M%SZ'
170	).replace(tzinfo=datetime.timezone.utc)
171	end = datetime.datetime.strptime(	24✔
172	end, '%Y%m%dT%H%M%SZ'
173	).replace(tzinfo=datetime.timezone.utc)
174
175	expanded_element = _expand(	24✔
176	element, copy.copy(item), start, end)
177	found_props.append(expanded_element)	24✔
178	else:
179	found_props.append(element)	24✔
180	else:
UNCOV 181	not_found_props.append(element)	×
182
183	assert item.href	24✔
184	uri = pathutils.unstrip_path(	24✔
185	posixpath.join(collection.path, item.href))
186	multistatus.append(xml_item_response(	24✔
187	base_prefix, uri, found_props=found_props,
188	not_found_props=not_found_props, found_item=True))
189
190	return client.MULTI_STATUS, multistatus	24✔
191
192
193	def _expand(	24✔
194	element: ET.Element,
195	item: radicale_item.Item,
196	start: datetime.datetime,
197	end: datetime.datetime,
198	) -> ET.Element:
199	rruleset = None	24✔
200	if hasattr(item.vobject_item.vevent, 'rrule'):	24!
201	rruleset = item.vobject_item.vevent.getrruleset()	24✔
202
203	expanded_item = _make_vobject_expanded_item(item)	24✔
204
205	if rruleset:	24!
206	recurrences = rruleset.between(start, end)	24✔
207
208	expanded: vobject.base.Component = copy.copy(expanded_item.vobject_item)	24✔
209	is_expanded_filled: bool = False	24✔
210
211	for recurrence_dt in recurrences:	24✔
212	recurrence_utc = recurrence_dt.astimezone(datetime.timezone.utc)	24✔
213
214	vevent = copy.deepcopy(expanded.vevent)	24✔
215	vevent.recurrence_id = ContentLine(	24✔
216	name='RECURRENCE-ID',
217	value=recurrence_utc.strftime('%Y%m%dT%H%M%SZ'), params={}
218	)
219
220	if is_expanded_filled is False:	24✔
221	expanded.vevent = vevent	24✔
222	is_expanded_filled = True	24✔
223	else:
224	expanded.add(vevent)	24✔
225
226	element.text = expanded.serialize()	24✔
227	else:
UNCOV 228	element.text = expanded_item.vobject_item.serialize()	×
229
230	return element	24✔
231
232
233	def _make_vobject_expanded_item(	24✔
234	item: radicale_item.Item
235	) -> radicale_item.Item:
236	# https://www.rfc-editor.org/rfc/rfc4791#section-9.6.5
237	# The returned calendar components MUST NOT use recurrence
238	# properties (i.e., EXDATE, EXRULE, RDATE, and RRULE) and MUST NOT
239	# have reference to or include VTIMEZONE components. Date and local
240	# time with reference to time zone information MUST be converted
241	# into date with UTC time.
242
243	item = copy.copy(item)	24✔
244	vevent = item.vobject_item.vevent	24✔
245
246	start_utc = vevent.dtstart.value.astimezone(datetime.timezone.utc)	24✔
247	vevent.dtstart = ContentLine(	24✔
248	name='DTSTART',
249	value=start_utc.strftime('%Y%m%dT%H%M%SZ'), params={})
250
251	dt_end = getattr(vevent, 'dtend', None)	24✔
252	if dt_end is not None:	24!
UNCOV 253	end_utc = dt_end.value.astimezone(datetime.timezone.utc)	×
UNCOV 254	vevent.dtend = ContentLine(	×
255	name='DTEND',
256	value=end_utc.strftime('%Y%m%dT%H%M%SZ'), params={})
257
258	timezones_to_remove = []	24✔
259	for component in item.vobject_item.components():	24✔
260	if component.name == 'VTIMEZONE':	24✔
261	timezones_to_remove.append(component)	24✔
262
263	for timezone in timezones_to_remove:	24✔
264	item.vobject_item.remove(timezone)	24✔
265
266	try:	24✔
267	delattr(item.vobject_item.vevent, 'rrule')	24✔
268	delattr(item.vobject_item.vevent, 'exdate')	24✔
UNCOV 269	delattr(item.vobject_item.vevent, 'exrule')	×
UNCOV 270	delattr(item.vobject_item.vevent, 'rdate')	×
271	except AttributeError:	24✔
272	pass	24✔
273
274	return item	24✔
275
276
277	def xml_item_response(base_prefix: str, href: str,	24✔
278	found_props: Sequence[ET.Element] = (),
279	not_found_props: Sequence[ET.Element] = (),
280	found_item: bool = True) -> ET.Element:
281	response = ET.Element(xmlutils.make_clark("D:response"))	24✔
282
283	href_element = ET.Element(xmlutils.make_clark("D:href"))	24✔
284	href_element.text = xmlutils.make_href(base_prefix, href)	24✔
285	response.append(href_element)	24✔
286
287	if found_item:	24✔
288	for code, props in ((200, found_props), (404, not_found_props)):	24✔
289	if props:	24✔
290	propstat = ET.Element(xmlutils.make_clark("D:propstat"))	24✔
291	status = ET.Element(xmlutils.make_clark("D:status"))	24✔
292	status.text = xmlutils.make_response(code)	24✔
293	prop_element = ET.Element(xmlutils.make_clark("D:prop"))	24✔
294	for prop in props:	24✔
295	prop_element.append(prop)	24✔
296	propstat.append(prop_element)	24✔
297	propstat.append(status)	24✔
298	response.append(propstat)	24✔
299	else:
300	status = ET.Element(xmlutils.make_clark("D:status"))	24✔
301	status.text = xmlutils.make_response(404)	24✔
302	response.append(status)	24✔
303
304	return response	24✔
305
306
307	def retrieve_items(	24✔
308	base_prefix: str, path: str, collection: storage.BaseCollection,
309	hreferences: Iterable[str], filters: Sequence[ET.Element],
310	multistatus: ET.Element) -> Iterator[Tuple[radicale_item.Item, bool]]:
311	"""Retrieves all items that are referenced in ``hreferences`` from
312	``collection`` and adds 404 responses for missing and invalid items
313	to ``multistatus``."""
314	collection_requested = False	24✔
315
316	def get_names() -> Iterator[str]:	24✔
317	"""Extracts all names from references in ``hreferences`` and adds
318	404 responses for invalid references to ``multistatus``.
319	If the whole collections is referenced ``collection_requested``
320	gets set to ``True``."""
321	nonlocal collection_requested
322	for hreference in hreferences:	24✔
323	try:	24✔
324	name = pathutils.name_from_path(hreference, collection)	24✔
UNCOV 325	except ValueError as e:	×
UNCOV 326	logger.warning("Skipping invalid path %r in REPORT request on "	×
327	"%r: %s", hreference, path, e)
UNCOV 328	response = xml_item_response(base_prefix, hreference,	×
329	found_item=False)
UNCOV 330	multistatus.append(response)	×
UNCOV 331	continue	×
332	if name:	24✔
333	# Reference is an item
334	yield name	24✔
335	else:
336	# Reference is a collection
337	collection_requested = True	24✔
338
339	for name, item in collection.get_multi(get_names()):	24✔
340	if not item:	24✔
341	uri = pathutils.unstrip_path(posixpath.join(collection.path, name))	24✔
342	response = xml_item_response(base_prefix, uri, found_item=False)	24✔
343	multistatus.append(response)	24✔
344	else:
345	yield item, False	24✔
346	if collection_requested:	24✔
347	yield from collection.get_filtered(filters)	24✔
348
349
350	def test_filter(collection_tag: str, item: radicale_item.Item,	24✔
351	filter_: ET.Element) -> bool:
352	"""Match an item against a filter."""
353	if (collection_tag == "VCALENDAR" and	24✔
354	filter_.tag != xmlutils.make_clark("C:%s" % filter_)):
355	if len(filter_) == 0:	24!
UNCOV 356	return True	×
357	if len(filter_) > 1:	24!
UNCOV 358	raise ValueError("Filter with %d children" % len(filter_))	×
359	if filter_[0].tag != xmlutils.make_clark("C:comp-filter"):	24!
UNCOV 360	raise ValueError("Unexpected %r in filter" % filter_[0].tag)	×
361	return radicale_filter.comp_match(item, filter_[0])	24✔
362	if (collection_tag == "VADDRESSBOOK" and	24!
363	filter_.tag != xmlutils.make_clark("CR:%s" % filter_)):
364	for child in filter_:	24✔
365	if child.tag != xmlutils.make_clark("CR:prop-filter"):	24!
UNCOV 366	raise ValueError("Unexpected %r in filter" % child.tag)	×
367	test = filter_.get("test", "anyof")	24✔
368	if test == "anyof":	24✔
369	return any(radicale_filter.prop_match(item.vobject_item, f, "CR")	24✔
370	for f in filter_)
371	if test == "allof":	24!
372	return all(radicale_filter.prop_match(item.vobject_item, f, "CR")	24✔
373	for f in filter_)
UNCOV 374	raise ValueError("Unsupported filter test: %r" % test)	×
375	raise ValueError("Unsupported filter %r for %r" %	9✔
376	(filter_.tag, collection_tag))
377
378
379	class ApplicationPartReport(ApplicationBase):	24✔
380
381	def do_REPORT(self, environ: types.WSGIEnviron, base_prefix: str,	24✔
382	path: str, user: str) -> types.WSGIResponse:
383	"""Manage REPORT request."""
384	access = Access(self._rights, user, path)	24✔
385	if not access.check("r"):	24!
UNCOV 386	return httputils.NOT_ALLOWED	×
387	try:	24✔
388	xml_content = self._read_xml_request_body(environ)	24✔
UNCOV 389	except RuntimeError as e:	×
UNCOV 390	logger.warning("Bad REPORT request on %r: %s", path, e,	×
391	exc_info=True)
UNCOV 392	return httputils.BAD_REQUEST	×
UNCOV 393	except socket.timeout:	×
UNCOV 394	logger.debug("Client timed out", exc_info=True)	×
UNCOV 395	return httputils.REQUEST_TIMEOUT	×
396	with contextlib.ExitStack() as lock_stack:	24!
397	lock_stack.enter_context(self._storage.acquire_lock("r", user))	24✔
398	item = next(iter(self._storage.discover(path)), None)	24✔
399	if not item:	24!
UNCOV 400	return httputils.NOT_FOUND	×
401	if not access.check("r", item):	24!
UNCOV 402	return httputils.NOT_ALLOWED	×
403	if isinstance(item, storage.BaseCollection):	24✔
404	collection = item	24✔
405	else:
406	assert item.collection is not None	24✔
407	collection = item.collection	24✔
408	try:	24✔
409	status, xml_answer = xml_report(	24✔
410	base_prefix, path, xml_content, collection, self._encoding,
411	lock_stack.close)
UNCOV 412	except ValueError as e:	×
UNCOV 413	logger.warning(	×
414	"Bad REPORT request on %r: %s", path, e, exc_info=True)
UNCOV 415	return httputils.BAD_REQUEST	×
416	headers = {"Content-Type": "text/xml; charset=%s" % self._encoding}	24✔
417	return status, headers, self._xml_response(xml_answer)	24✔

Kozea / Radicale / 8580040934

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous