10077791997

Committed 24 Jul 2024 01:40PM UTC coverage: 80.601% (+0.05%) from 80.552%

Build # 10077791997

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1547 from deronnax/fix-misspellings

fix misspellings

Run Details

1689 of 2315 branches covered (72.96%)

Branch coverage included in aggregate %.

3 of 4 new or added lines in 2 files covered. (75.0%)

1 existing line in 1 file now uncovered.

3625 of 4278 relevant lines covered (84.74%)

15.67 hits per line

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

80.83

/radicale/app/propfind.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 collections
import itertools
import posixpath
import socket
import xml.etree.ElementTree as ET
from http import client
from typing import Dict, Iterable, Iterator, List, Optional, Sequence, Tuple

from radicale import httputils, pathutils, rights, storage, types, 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]],
                 user: str, encoding: str) -> 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"))

    for item, permission in allowed_items:
        write = permission == "w"
        multistatus.append(xml_propfind_response(
            base_prefix, path, item, props, user, encoding, write=write,
            allprop=allprop, propname=propname))

    return multistatus


def xml_propfind_response(
        base_prefix: str, path: str, item: types.CollectionOrItem,
        props: Sequence[str], user: str, encoding: str, write: bool = False,
        propname: bool = False, allprop: bool = False) -> 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"))
    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 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"))
                props.append(xmlutils.make_clark("D:sync-token"))
            if collection.tag == "VCALENDAR":
                props.append(xmlutils.make_clark("CS:getctag"))
                props.append(
                    xmlutils.make_clark("C:supported-calendar-component-set"))

            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)
            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"]
                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("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)
                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 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:
                reports.append("D:sync-collection")
                if collection.tag == "VADDRESSBOOK":
                    reports.append("CR:addressbook-multiget")
                    reports.append("CR:addressbook-query")
                elif collection.tag == "VCALENDAR":
                    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:
                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 is_collection:
            if tag == xmlutils.make_clark("D:getcontenttype"):
                if is_leaf:
                    element.text = xmlutils.MIMETYPES[
                        collection.tag]
                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":
                        child_element = ET.Element(
                            xmlutils.make_clark("CR:addressbook"))
                        element.append(child_element)
                    elif collection.tag == "VCALENDAR":
                        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 displayname is not None:
                    element.text = displayname
                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:
                    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 not displayname and is_leaf:
                    displayname = collection.path
                if displayname is not None:
                    element.text = displayname
                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 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)
        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 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) -> types.WSGIResponse:
        """Manage PROPFIND 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 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):
            items_iter = iter(self._storage.discover(
                path, environ.get("HTTP_DEPTH", "0")))
            # 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)
            allowed_items = self._collect_allowed_items(items_iter, user)
            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)
            if xml_answer is None:
                return httputils.NOT_ALLOWED
            return client.MULTI_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 collections	19✔
21	import itertools	19✔
22	import posixpath	19✔
23	import socket	19✔
24	import xml.etree.ElementTree as ET	19✔
25	from http import client	19✔
26	from typing import Dict, Iterable, Iterator, List, Optional, Sequence, Tuple	19✔
27
28	from radicale import httputils, pathutils, rights, storage, types, xmlutils	19✔
29	from radicale.app.base import Access, ApplicationBase	19✔
30	from radicale.log import logger	19✔
31
32
33	def xml_propfind(base_prefix: str, path: str,	19✔
34	xml_request: Optional[ET.Element],
35	allowed_items: Iterable[Tuple[types.CollectionOrItem, str]],
36	user: str, encoding: str) -> Optional[ET.Element]:
37	"""Read and answer PROPFIND requests.
38
39	Read rfc4918-9.1 for info.
40
41	The collections parameter is a list of collections that are to be included
42	in the output.
43
44	"""
45	# A client may choose not to submit a request body. An empty PROPFIND
46	# request body MUST be treated as if it were an 'allprop' request.
47	top_element = (xml_request[0] if xml_request is not None else	19✔
48	ET.Element(xmlutils.make_clark("D:allprop")))
49
50	props: List[str] = []	19✔
51	allprop = False	19✔
52	propname = False	19✔
53	if top_element.tag == xmlutils.make_clark("D:allprop"):	19✔
54	allprop = True	19✔
55	elif top_element.tag == xmlutils.make_clark("D:propname"):	19✔
56	propname = True	19✔
57	elif top_element.tag == xmlutils.make_clark("D:prop"):	19!
58	props.extend(prop.tag for prop in top_element)	19✔
59
60	if xmlutils.make_clark("D:current-user-principal") in props and not user:	19✔
61	# Ask for authentication
62	# Returning the DAV:unauthenticated pseudo-principal as specified in
63	# RFC 5397 doesn't seem to work with DAVx5.
64	return None	19✔
65
66	# Writing answer
67	multistatus = ET.Element(xmlutils.make_clark("D:multistatus"))	19✔
68
69	for item, permission in allowed_items:	19!
70	write = permission == "w"	19✔
71	multistatus.append(xml_propfind_response(	19✔
72	base_prefix, path, item, props, user, encoding, write=write,
73	allprop=allprop, propname=propname))
74
75	return multistatus	19✔
76
77
78	def xml_propfind_response(	19✔
79	base_prefix: str, path: str, item: types.CollectionOrItem,
80	props: Sequence[str], user: str, encoding: str, write: bool = False,
81	propname: bool = False, allprop: bool = False) -> ET.Element:
82	"""Build and return a PROPFIND response."""
83	if propname and allprop or (props and (propname or allprop)):	19!
84	raise ValueError("Only use one of props, propname and allprops")	×
85
86	if isinstance(item, storage.BaseCollection):	19✔
87	is_collection = True	19✔
88	is_leaf = item.tag in ("VADDRESSBOOK", "VCALENDAR", "VSUBSCRIBED")	19✔
89	collection = item	19✔
90	# Some clients expect collections to end with `/`
91	uri = pathutils.unstrip_path(item.path, True)	19✔
92	else:
93	is_collection = is_leaf = False	19✔
94	assert item.collection is not None	19✔
95	assert item.href	19✔
96	collection = item.collection	19✔
97	uri = pathutils.unstrip_path(posixpath.join(	19✔
98	collection.path, item.href))
99	response = ET.Element(xmlutils.make_clark("D:response"))	19✔
100	href = ET.Element(xmlutils.make_clark("D:href"))	19✔
101	href.text = xmlutils.make_href(base_prefix, uri)	19✔
102	response.append(href)	19✔
103
104	if propname or allprop:	19!
105	props = []	19✔
106	# Should list all properties that can be retrieved by the code below
107	props.append(xmlutils.make_clark("D:principal-collection-set"))	19✔
108	props.append(xmlutils.make_clark("D:current-user-principal"))	19✔
109	props.append(xmlutils.make_clark("D:current-user-privilege-set"))	19✔
110	props.append(xmlutils.make_clark("D:supported-report-set"))	19✔
111	props.append(xmlutils.make_clark("D:resourcetype"))	19✔
112	props.append(xmlutils.make_clark("D:owner"))	19✔
113
114	if is_collection and collection.is_principal:	19✔
115	props.append(xmlutils.make_clark("C:calendar-user-address-set"))	19✔
116	props.append(xmlutils.make_clark("D:principal-URL"))	19✔
117	props.append(xmlutils.make_clark("CR:addressbook-home-set"))	19✔
118	props.append(xmlutils.make_clark("C:calendar-home-set"))	19✔
119
120	if not is_collection or is_leaf:	19✔
121	props.append(xmlutils.make_clark("D:getetag"))	19✔
122	props.append(xmlutils.make_clark("D:getlastmodified"))	19✔
123	props.append(xmlutils.make_clark("D:getcontenttype"))	19✔
124	props.append(xmlutils.make_clark("D:getcontentlength"))	19✔
125
126	if is_collection:	19!
127	if is_leaf:	19✔
128	props.append(xmlutils.make_clark("D:displayname"))	19✔
129	props.append(xmlutils.make_clark("D:sync-token"))	19✔
130	if collection.tag == "VCALENDAR":	19✔
131	props.append(xmlutils.make_clark("CS:getctag"))	19✔
132	props.append(	19✔
133	xmlutils.make_clark("C:supported-calendar-component-set"))
134
135	meta = collection.get_meta()	19✔
136	for tag in meta:	19!
137	if tag == "tag":	19✔
138	continue	19✔
139	clark_tag = xmlutils.make_clark(tag)	19✔
140	if clark_tag not in props:	19!
141	props.append(clark_tag)	19✔
142
143	responses: Dict[int, List[ET.Element]] = collections.defaultdict(list)	19✔
144	if propname:	19✔
145	for tag in props:	19!
146	responses[200].append(ET.Element(tag))	19✔
147	props = []	19✔
148	for tag in props:	19✔
149	element = ET.Element(tag)	19✔
150	is404 = False	19✔
151	if tag == xmlutils.make_clark("D:getetag"):	19✔
152	if not is_collection or is_leaf:	19!
153	element.text = item.etag	19✔
154	else:
155	is404 = True	×
156	elif tag == xmlutils.make_clark("D:getlastmodified"):	19✔
157	if not is_collection or is_leaf:	19!
158	element.text = item.last_modified	19✔
159	else:
160	is404 = True	×
161	elif tag == xmlutils.make_clark("D:principal-collection-set"):	19✔
162	child_element = ET.Element(xmlutils.make_clark("D:href"))	19✔
163	child_element.text = xmlutils.make_href(base_prefix, "/")	19✔
164	element.append(child_element)	19✔
165	elif (tag in (xmlutils.make_clark("C:calendar-user-address-set"),	19✔
166	xmlutils.make_clark("D:principal-URL"),
167	xmlutils.make_clark("CR:addressbook-home-set"),
168	xmlutils.make_clark("C:calendar-home-set")) and
169	is_collection and collection.is_principal):
170	child_element = ET.Element(xmlutils.make_clark("D:href"))	19✔
171	child_element.text = xmlutils.make_href(base_prefix, path)	19✔
172	element.append(child_element)	19✔
173	elif tag == xmlutils.make_clark("C:supported-calendar-component-set"):	19✔
174	human_tag = xmlutils.make_human_tag(tag)	19✔
175	if is_collection and is_leaf:	19!
176	components_text = collection.get_meta(human_tag)	19✔
177	if components_text:	19!
178	components = components_text.split(",")	×
179	else:
180	components = ["VTODO", "VEVENT", "VJOURNAL"]	19✔
181	for component in components:	19✔
182	comp = ET.Element(xmlutils.make_clark("C:comp"))	19✔
183	comp.set("name", component)	19✔
184	element.append(comp)	19✔
185	else:
186	is404 = True	×
187	elif tag == xmlutils.make_clark("D:current-user-principal"):	19✔
188	if user:	19✔
189	child_element = ET.Element(xmlutils.make_clark("D:href"))	19✔
190	child_element.text = xmlutils.make_href(	19✔
191	base_prefix, "/%s/" % user)
192	element.append(child_element)	19✔
193	else:
194	element.append(ET.Element(	19✔
195	xmlutils.make_clark("D:unauthenticated")))
196	elif tag == xmlutils.make_clark("D:current-user-privilege-set"):	19✔
197	privileges = ["D:read"]	19✔
198	if write:	19✔
199	privileges.append("D:all")	19✔
200	privileges.append("D:write")	19✔
201	privileges.append("D:write-properties")	19✔
202	privileges.append("D:write-content")	19✔
203	for human_tag in privileges:	19✔
204	privilege = ET.Element(xmlutils.make_clark("D:privilege"))	19✔
205	privilege.append(ET.Element(	19✔
206	xmlutils.make_clark(human_tag)))
207	element.append(privilege)	19✔
208	elif tag == xmlutils.make_clark("D:supported-report-set"):	19✔
209	# These 3 reports are not implemented
210	reports = ["D:expand-property",	19✔
211	"D:principal-search-property-set",
212	"D:principal-property-search"]
213	if is_collection and is_leaf:	19✔
214	reports.append("D:sync-collection")	19✔
215	if collection.tag == "VADDRESSBOOK":	19!
216	reports.append("CR:addressbook-multiget")	×
217	reports.append("CR:addressbook-query")	×
218	elif collection.tag == "VCALENDAR":	19!
219	reports.append("C:calendar-multiget")	19✔
220	reports.append("C:calendar-query")	19✔
221	for human_tag in reports:	19✔
222	supported_report = ET.Element(	19✔
223	xmlutils.make_clark("D:supported-report"))
224	report_element = ET.Element(xmlutils.make_clark("D:report"))	19✔
225	report_element.append(	19✔
226	ET.Element(xmlutils.make_clark(human_tag)))
227	supported_report.append(report_element)	19✔
228	element.append(supported_report)	19✔
229	elif tag == xmlutils.make_clark("D:getcontentlength"):	19✔
230	if not is_collection or is_leaf:	19!
231	element.text = str(len(item.serialize().encode(encoding)))	19✔
232	else:
233	is404 = True	×
234	elif tag == xmlutils.make_clark("D:owner"):	19✔
235	# return empty elment, if no owner available (rfc3744-5.1)
236	if collection.owner:	19✔
237	child_element = ET.Element(xmlutils.make_clark("D:href"))	19✔
238	child_element.text = xmlutils.make_href(	19✔
239	base_prefix, "/%s/" % collection.owner)
240	element.append(child_element)	19✔
241	elif is_collection:	19✔
242	if tag == xmlutils.make_clark("D:getcontenttype"):	19✔
243	if is_leaf:	19!
244	element.text = xmlutils.MIMETYPES[	19✔
245	collection.tag]
246	else:
247	is404 = True	×
248	elif tag == xmlutils.make_clark("D:resourcetype"):	19✔
249	if collection.is_principal:	19✔
250	child_element = ET.Element(	19✔
251	xmlutils.make_clark("D:principal"))
252	element.append(child_element)	19✔
253	if is_leaf:	19✔
254	if collection.tag == "VADDRESSBOOK":	19!
255	child_element = ET.Element(	×
256	xmlutils.make_clark("CR:addressbook"))
257	element.append(child_element)	×
258	elif collection.tag == "VCALENDAR":	19!
259	child_element = ET.Element(	19✔
260	xmlutils.make_clark("C:calendar"))
261	element.append(child_element)	19✔
262	elif collection.tag == "VSUBSCRIBED":	×
263	child_element = ET.Element(	×
264	xmlutils.make_clark("CS:subscribed"))
265	element.append(child_element)	×
266	child_element = ET.Element(xmlutils.make_clark("D:collection"))	19✔
267	element.append(child_element)	19✔
268	elif tag == xmlutils.make_clark("RADICALE:displayname"):	19!
269	# Only for internal use by the web interface
270	displayname = collection.get_meta("D:displayname")	×
271	if displayname is not None:	×
272	element.text = displayname	×
273	else:
274	is404 = True	×
275	elif tag == xmlutils.make_clark("RADICALE:getcontentcount"):	19!
276	# Only for internal use by the web interface
277	if isinstance(item, storage.BaseCollection) and not collection.is_principal:	×
278	element.text = str(sum(1 for x in item.get_all()))	×
279	else:
280	is404 = True	×
281	elif tag == xmlutils.make_clark("D:displayname"):	19✔
282	displayname = collection.get_meta("D:displayname")	19✔
283	if not displayname and is_leaf:	19!
284	displayname = collection.path	19✔
285	if displayname is not None:	19!
286	element.text = displayname	19✔
287	else:
288	is404 = True	×
289	elif tag == xmlutils.make_clark("CS:getctag"):	19✔
290	if is_leaf:	19!
291	element.text = collection.etag	19✔
292	else:
293	is404 = True	×
294	elif tag == xmlutils.make_clark("D:sync-token"):	19✔
295	if is_leaf:	19!
296	element.text, _ = collection.sync()	19✔
297	else:
298	is404 = True	×
299	elif tag == xmlutils.make_clark("CS:source"):	19!
300	if is_leaf:	×
301	child_element = ET.Element(xmlutils.make_clark("D:href"))	×
302	child_element.text = collection.get_meta('CS:source')	×
303	element.append(child_element)	×
304	else:
305	is404 = True	×
306	else:
307	human_tag = xmlutils.make_human_tag(tag)	19✔
308	tag_text = collection.get_meta(human_tag)	19✔
309	if tag_text is not None:	19✔
310	element.text = tag_text	19✔
311	else:
312	is404 = True	19✔
313	# Not for collections
314	elif tag == xmlutils.make_clark("D:getcontenttype"):	19✔
315	assert not isinstance(item, storage.BaseCollection)	19✔
316	element.text = xmlutils.get_content_type(item, encoding)	19✔
317	elif tag == xmlutils.make_clark("D:resourcetype"):	19!
318	# resourcetype must be returned empty for non-collection elements
319	pass	19✔
320	else:
321	is404 = True	×
322
323	responses[404 if is404 else 200].append(element)	19✔
324
325	for status_code, children in responses.items():	19!
326	if not children:	19!
UNCOV 327	continue	×
328	propstat = ET.Element(xmlutils.make_clark("D:propstat"))	19✔
329	response.append(propstat)	19✔
330	prop = ET.Element(xmlutils.make_clark("D:prop"))	19✔
331	prop.extend(children)	19✔
332	propstat.append(prop)	19✔
333	status = ET.Element(xmlutils.make_clark("D:status"))	19✔
334	status.text = xmlutils.make_response(status_code)	19✔
335	propstat.append(status)	19✔
336
337	return response	19✔
338
339
340	class ApplicationPartPropfind(ApplicationBase):	19✔
341
342	def _collect_allowed_items(	19✔
343	self, items: Iterable[types.CollectionOrItem], user: str
344	) -> Iterator[Tuple[types.CollectionOrItem, str]]:
345	"""Get items from request that user is allowed to access."""
346	for item in items:	19✔
347	if isinstance(item, storage.BaseCollection):	19✔
348	path = pathutils.unstrip_path(item.path, True)	19✔
349	if item.tag:	19✔
350	permissions = rights.intersect(	19✔
351	self._rights.authorization(user, path), "rw")
352	target = "collection with tag %r" % item.path	19✔
353	else:
354	permissions = rights.intersect(	19✔
355	self._rights.authorization(user, path), "RW")
356	target = "collection %r" % item.path	19✔
357	else:
358	assert item.collection is not None	19✔
359	path = pathutils.unstrip_path(item.collection.path, True)	19✔
360	permissions = rights.intersect(	19✔
361	self._rights.authorization(user, path), "rw")
362	target = "item %r from %r" % (item.href, item.collection.path)	19✔
363	if rights.intersect(permissions, "Ww"):	19✔
364	permission = "w"	19✔
365	status = "write"	19✔
366	elif rights.intersect(permissions, "Rr"):	19!
367	permission = "r"	19✔
368	status = "read"	19✔
369	else:
370	permission = ""	×
371	status = "NO"	×
372	logger.debug(	19✔
373	"%s has %s access to %s",
374	repr(user) if user else "anonymous user", status, target)
375	if permission:	19!
376	yield item, permission	19✔
377
378	def do_PROPFIND(self, environ: types.WSGIEnviron, base_prefix: str,	19✔
379	path: str, user: str) -> types.WSGIResponse:
380	"""Manage PROPFIND request."""
381	access = Access(self._rights, user, path)	19✔
382	if not access.check("r"):	19!
383	return httputils.NOT_ALLOWED	19✔
384	try:	19✔
385	xml_content = self._read_xml_request_body(environ)	19✔
386	except RuntimeError as e:	×
387	logger.warning(	×
388	"Bad PROPFIND request on %r: %s", path, e, exc_info=True)
389	return httputils.BAD_REQUEST	×
390	except socket.timeout:	×
391	logger.debug("Client timed out", exc_info=True)	×
392	return httputils.REQUEST_TIMEOUT	×
393	with self._storage.acquire_lock("r", user):	19✔
394	items_iter = iter(self._storage.discover(	19✔
395	path, environ.get("HTTP_DEPTH", "0")))
396	# take root item for rights checking
397	item = next(items_iter, None)	19✔
398	if not item:	19!
399	return httputils.NOT_FOUND	19✔
400	if not access.check("r", item):	19!
401	return httputils.NOT_ALLOWED	×
402	# put item back
403	items_iter = itertools.chain([item], items_iter)	19✔
404	allowed_items = self._collect_allowed_items(items_iter, user)	19✔
405	headers = {"DAV": httputils.DAV_HEADERS,	19✔
406	"Content-Type": "text/xml; charset=%s" % self._encoding}
407	xml_answer = xml_propfind(base_prefix, path, xml_content,	19✔
408	allowed_items, user, self._encoding)
409	if xml_answer is None:	19!
410	return httputils.NOT_ALLOWED	19✔
411	return client.MULTI_STATUS, headers, self._xml_response(xml_answer)	19✔

Kozea / Radicale / 10077791997

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