8121184820

Committed 02 Mar 2024 07:10AM UTC coverage: 83.816% (-0.1%) from 83.925%

Build # 8121184820

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1365 from pbiering/fix-actions

fix for actions

Run Details

1556 of 1984 branches covered (78.43%)

Branch coverage included in aggregate %.

6 of 6 new or added lines in 6 files covered. (100.0%)

37 existing lines in 2 files now uncovered.

3478 of 4022 relevant lines covered (86.47%)

17.6 hits per line

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

68.87

/radicale/app/move.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 posixpath
import re
from http import client
from urllib.parse import urlparse

from radicale import httputils, pathutils, storage, types
from radicale.app.base import Access, ApplicationBase
from radicale.log import logger


def get_server_netloc(environ: types.WSGIEnviron, force_port: bool = False):
    if environ.get("HTTP_X_FORWARDED_HOST"):
        host = environ["HTTP_X_FORWARDED_HOST"]
        proto = environ.get("HTTP_X_FORWARDED_PROTO") or "http"
        port = "443" if proto == "https" else "80"
        port = environ["HTTP_X_FORWARDED_PORT"] or port
    else:
        host = environ.get("HTTP_HOST") or environ["SERVER_NAME"]
        proto = environ["wsgi.url_scheme"]
        port = environ["SERVER_PORT"]
    if (not force_port and port == ("443" if proto == "https" else "80") or
            re.search(r":\d+$", host)):
        return host
    return host + ":" + port


class ApplicationPartMove(ApplicationBase):

    def do_MOVE(self, environ: types.WSGIEnviron, base_prefix: str,
                path: str, user: str) -> types.WSGIResponse:
        """Manage MOVE request."""
        raw_dest = environ.get("HTTP_DESTINATION", "")
        to_url = urlparse(raw_dest)
        to_netloc_with_port = to_url.netloc
        if to_url.port is None:
            to_netloc_with_port += (":443" if to_url.scheme == "https"
                                    else ":80")
        if to_netloc_with_port != get_server_netloc(environ, force_port=True):
            logger.info("Unsupported destination address: %r", raw_dest)
            # Remote destination server, not supported
            return httputils.REMOTE_DESTINATION
        access = Access(self._rights, user, path)
        if not access.check("w"):
            return httputils.NOT_ALLOWED
        to_path = pathutils.sanitize_path(to_url.path)
        if not (to_path + "/").startswith(base_prefix + "/"):
            logger.warning("Destination %r from MOVE request on %r doesn't "
                           "start with base prefix", to_path, path)
            return httputils.NOT_ALLOWED
        to_path = to_path[len(base_prefix):]
        to_access = Access(self._rights, user, to_path)
        if not to_access.check("w"):
            return httputils.NOT_ALLOWED

        with self._storage.acquire_lock("w", user):
            item = next(iter(self._storage.discover(path)), None)
            if not item:
                return httputils.NOT_FOUND
            if (not access.check("w", item) or
                    not to_access.check("w", item)):
                return httputils.NOT_ALLOWED
            if isinstance(item, storage.BaseCollection):
                # TODO: support moving collections
                return httputils.METHOD_NOT_ALLOWED

            to_item = next(iter(self._storage.discover(to_path)), None)
            if isinstance(to_item, storage.BaseCollection):
                return httputils.FORBIDDEN
            to_parent_path = pathutils.unstrip_path(
                posixpath.dirname(pathutils.strip_path(to_path)), True)
            to_collection = next(iter(
                self._storage.discover(to_parent_path)), None)
            if not to_collection:
                return httputils.CONFLICT
            assert isinstance(to_collection, storage.BaseCollection)
            assert item.collection is not None
            collection_tag = item.collection.tag
            if not collection_tag or collection_tag != to_collection.tag:
                return httputils.FORBIDDEN
            if to_item and environ.get("HTTP_OVERWRITE", "F") != "T":
                return httputils.PRECONDITION_FAILED
            if (to_item and item.uid != to_item.uid or
                    not to_item and
                    to_collection.path != item.collection.path and
                    to_collection.has_uid(item.uid)):
                return self._webdav_error_response(
                    client.CONFLICT, "%s:no-uid-conflict" % (
                        "C" if collection_tag == "VCALENDAR" else "CR"))
            to_href = posixpath.basename(pathutils.strip_path(to_path))
            try:
                self._storage.move(item, to_collection, to_href)
            except ValueError as e:
                logger.warning(
                    "Bad MOVE request on %r: %s", path, e, exc_info=True)
                return httputils.BAD_REQUEST
            return client.NO_CONTENT if to_item else client.CREATED, {}, None

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 posixpath	21✔
21	import re	21✔
22	from http import client	21✔
23	from urllib.parse import urlparse	21✔
24
25	from radicale import httputils, pathutils, storage, types	21✔
26	from radicale.app.base import Access, ApplicationBase	21✔
27	from radicale.log import logger	21✔
28
29
30	def get_server_netloc(environ: types.WSGIEnviron, force_port: bool = False):	21✔
31	if environ.get("HTTP_X_FORWARDED_HOST"):	21!
32	host = environ["HTTP_X_FORWARDED_HOST"]	×
33	proto = environ.get("HTTP_X_FORWARDED_PROTO") or "http"	×
34	port = "443" if proto == "https" else "80"	×
UNCOV 35	port = environ["HTTP_X_FORWARDED_PORT"] or port	×
36	else:
37	host = environ.get("HTTP_HOST") or environ["SERVER_NAME"]	21✔
38	proto = environ["wsgi.url_scheme"]	21✔
39	port = environ["SERVER_PORT"]	21✔
40	if (not force_port and port == ("443" if proto == "https" else "80") or	21!
41	re.search(r":\d+$", host)):
UNCOV 42	return host	×
43	return host + ":" + port	21✔
44
45
46	class ApplicationPartMove(ApplicationBase):	21✔
47
48	def do_MOVE(self, environ: types.WSGIEnviron, base_prefix: str,	21✔
49	path: str, user: str) -> types.WSGIResponse:
50	"""Manage MOVE request."""
51	raw_dest = environ.get("HTTP_DESTINATION", "")	21✔
52	to_url = urlparse(raw_dest)	21✔
53	to_netloc_with_port = to_url.netloc	21✔
54	if to_url.port is None:	21!
55	to_netloc_with_port += (":443" if to_url.scheme == "https"	21✔
56	else ":80")
57	if to_netloc_with_port != get_server_netloc(environ, force_port=True):	21!
UNCOV 58	logger.info("Unsupported destination address: %r", raw_dest)	×
59	# Remote destination server, not supported
UNCOV 60	return httputils.REMOTE_DESTINATION	×
61	access = Access(self._rights, user, path)	21✔
62	if not access.check("w"):	21!
UNCOV 63	return httputils.NOT_ALLOWED	×
64	to_path = pathutils.sanitize_path(to_url.path)	21✔
65	if not (to_path + "/").startswith(base_prefix + "/"):	21!
UNCOV 66	logger.warning("Destination %r from MOVE request on %r doesn't "	×
67	"start with base prefix", to_path, path)
UNCOV 68	return httputils.NOT_ALLOWED	×
69	to_path = to_path[len(base_prefix):]	21✔
70	to_access = Access(self._rights, user, to_path)	21✔
71	if not to_access.check("w"):	21!
UNCOV 72	return httputils.NOT_ALLOWED	×
73
74	with self._storage.acquire_lock("w", user):	21✔
75	item = next(iter(self._storage.discover(path)), None)	21✔
76	if not item:	21!
UNCOV 77	return httputils.NOT_FOUND	×
78	if (not access.check("w", item) or	21!
79	not to_access.check("w", item)):
UNCOV 80	return httputils.NOT_ALLOWED	×
81	if isinstance(item, storage.BaseCollection):	21!
82	# TODO: support moving collections
UNCOV 83	return httputils.METHOD_NOT_ALLOWED	×
84
85	to_item = next(iter(self._storage.discover(to_path)), None)	21✔
86	if isinstance(to_item, storage.BaseCollection):	21!
UNCOV 87	return httputils.FORBIDDEN	×
88	to_parent_path = pathutils.unstrip_path(	21✔
89	posixpath.dirname(pathutils.strip_path(to_path)), True)
90	to_collection = next(iter(	21✔
91	self._storage.discover(to_parent_path)), None)
92	if not to_collection:	21!
UNCOV 93	return httputils.CONFLICT	×
94	assert isinstance(to_collection, storage.BaseCollection)	21✔
95	assert item.collection is not None	21✔
96	collection_tag = item.collection.tag	21✔
97	if not collection_tag or collection_tag != to_collection.tag:	21!
UNCOV 98	return httputils.FORBIDDEN	×
99	if to_item and environ.get("HTTP_OVERWRITE", "F") != "T":	21✔
100	return httputils.PRECONDITION_FAILED	21✔
101	if (to_item and item.uid != to_item.uid or	21✔
102	not to_item and
103	to_collection.path != item.collection.path and
104	to_collection.has_uid(item.uid)):
105	return self._webdav_error_response(	21✔
106	client.CONFLICT, "%s:no-uid-conflict" % (
107	"C" if collection_tag == "VCALENDAR" else "CR"))
108	to_href = posixpath.basename(pathutils.strip_path(to_path))	21✔
109	try:	21✔
110	self._storage.move(item, to_collection, to_href)	21✔
111	except ValueError as e:	×
UNCOV 112	logger.warning(	×
113	"Bad MOVE request on %r: %s", path, e, exc_info=True)
UNCOV 114	return httputils.BAD_REQUEST	×
115	return client.NO_CONTENT if to_item else client.CREATED, {}, None	21✔

Kozea / Radicale / 8121184820

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