254424173

Build Type

push

github

Committed by Unrud

Commit Message

Fix exceptions raised by decode_request

Run Details

1200 of 1549 branches covered (77.47%)

Branch coverage included in aggregate %.

3 of 4 new or added lines in 1 file covered. (75.0%)

30 existing lines in 2 files now uncovered.

2891 of 3438 relevant lines covered (84.09%)

15.12 hits per line

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

84.92

/radicale/app/__init__.py

# This file is part of Radicale Server - Calendar Server
# Copyright © 2008 Nicolas Kandel
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2019 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/>.

"""
Radicale WSGI application.

Can be used with an external WSGI server (see ``radicale.application()``) or
the built-in server (see ``radicale.server`` module).

"""

import base64
import datetime
import io
import logging
import posixpath
import pprint
import random
import time
import zlib
from http import client
from xml.etree import ElementTree as ET

import defusedxml.ElementTree as DefusedET
import pkg_resources

from radicale import (auth, httputils, log, pathutils, rights, storage, web,
                      xmlutils)
from radicale.app.delete import ApplicationDeleteMixin
from radicale.app.get import ApplicationGetMixin
from radicale.app.head import ApplicationHeadMixin
from radicale.app.mkcalendar import ApplicationMkcalendarMixin
from radicale.app.mkcol import ApplicationMkcolMixin
from radicale.app.move import ApplicationMoveMixin
from radicale.app.options import ApplicationOptionsMixin
from radicale.app.post import ApplicationPostMixin
from radicale.app.propfind import ApplicationPropfindMixin
from radicale.app.proppatch import ApplicationProppatchMixin
from radicale.app.put import ApplicationPutMixin
from radicale.app.report import ApplicationReportMixin
from radicale.log import logger

VERSION = pkg_resources.get_distribution("radicale").version


class Application(
        ApplicationDeleteMixin, ApplicationGetMixin, ApplicationHeadMixin,
        ApplicationMkcalendarMixin, ApplicationMkcolMixin,
        ApplicationMoveMixin, ApplicationOptionsMixin,
        ApplicationPropfindMixin, ApplicationProppatchMixin,
        ApplicationPostMixin, ApplicationPutMixin,
        ApplicationReportMixin):

    """WSGI application."""

    def __init__(self, configuration):
        """Initialize Application.

        ``configuration`` see ``radicale.config`` module.
        The ``configuration`` must not change during the lifetime of
        this object, it is kept as an internal reference.

        """
        super().__init__()
        self.configuration = configuration
        self._auth = auth.load(configuration)
        self._storage = storage.load(configuration)
        self._rights = rights.load(configuration)
        self._web = web.load(configuration)
        self._encoding = configuration.get("encoding", "request")

    def _headers_log(self, environ):
        """Sanitize headers for logging."""
        request_environ = dict(environ)

        # Mask passwords
        mask_passwords = self.configuration.get("logging", "mask_passwords")
        authorization = request_environ.get("HTTP_AUTHORIZATION", "")
        if mask_passwords and authorization.startswith("Basic"):
            request_environ["HTTP_AUTHORIZATION"] = "Basic **masked**"
        if request_environ.get("HTTP_COOKIE"):
            request_environ["HTTP_COOKIE"] = "**masked**"

        return request_environ

    def __call__(self, environ, start_response):
        with log.register_stream(environ["wsgi.errors"]):
            try:
                status, headers, answers = self._handle_request(environ)
            except Exception as e:
                try:
                    method = str(environ["REQUEST_METHOD"])
                except Exception:
                    method = "unknown"
                try:
                    path = str(environ.get("PATH_INFO", ""))
                except Exception:
                    path = ""
                logger.error("An exception occurred during %s request on %r: "
                             "%s", method, path, e, exc_info=True)
                status, headers, answer = httputils.INTERNAL_SERVER_ERROR
                answer = answer.encode("ascii")
                status = "%d %s" % (
                    status.value, client.responses.get(status, "Unknown"))
                headers = [
                    ("Content-Length", str(len(answer)))] + list(headers)
                answers = [answer]
            start_response(status, headers)
        return answers

    def _handle_request(self, environ):
        """Manage a request."""
        def response(status, headers=(), answer=None):
            headers = dict(headers)
            # Set content length
            if answer:
                if hasattr(answer, "encode"):
                    logger.debug("Response content:\n%s", answer)
                    headers["Content-Type"] += "; charset=%s" % self._encoding
                    answer = answer.encode(self._encoding)
                accept_encoding = [
                    encoding.strip() for encoding in
                    environ.get("HTTP_ACCEPT_ENCODING", "").split(",")
                    if encoding.strip()]

                if "gzip" in accept_encoding:
                    zcomp = zlib.compressobj(wbits=16 + zlib.MAX_WBITS)
                    answer = zcomp.compress(answer) + zcomp.flush()
                    headers["Content-Encoding"] = "gzip"

                headers["Content-Length"] = str(len(answer))

            # Add extra headers set in configuration
            for key in self.configuration.options("headers"):
                headers[key] = self.configuration.get("headers", key)

            # Start response
            time_end = datetime.datetime.now()
            status = "%d %s" % (
                status, client.responses.get(status, "Unknown"))
            logger.info(
                "%s response status for %r%s in %.3f seconds: %s",
                environ["REQUEST_METHOD"], environ.get("PATH_INFO", ""),
                depthinfo, (time_end - time_begin).total_seconds(), status)
            # Return response content
            return status, list(headers.items()), [answer] if answer else []

        remote_host = "unknown"
        if environ.get("REMOTE_HOST"):
            remote_host = repr(environ["REMOTE_HOST"])
        elif environ.get("REMOTE_ADDR"):
            remote_host = environ["REMOTE_ADDR"]
        if environ.get("HTTP_X_FORWARDED_FOR"):
            remote_host = "%r (forwarded by %s)" % (
                environ["HTTP_X_FORWARDED_FOR"], remote_host)
        remote_useragent = ""
        if environ.get("HTTP_USER_AGENT"):
            remote_useragent = " using %r" % environ["HTTP_USER_AGENT"]
        depthinfo = ""
        if environ.get("HTTP_DEPTH"):
            depthinfo = " with depth %r" % environ["HTTP_DEPTH"]
        time_begin = datetime.datetime.now()
        logger.info(
            "%s request for %r%s received from %s%s",
            environ["REQUEST_METHOD"], environ.get("PATH_INFO", ""), depthinfo,
            remote_host, remote_useragent)
        headers = pprint.pformat(self._headers_log(environ))
        logger.debug("Request headers:\n%s", headers)

        # Let reverse proxies overwrite SCRIPT_NAME
        if "HTTP_X_SCRIPT_NAME" in environ:
            # script_name must be removed from PATH_INFO by the client.
            unsafe_base_prefix = environ["HTTP_X_SCRIPT_NAME"]
            logger.debug("Script name overwritten by client: %r",
                         unsafe_base_prefix)
        else:
            # SCRIPT_NAME is already removed from PATH_INFO, according to the
            # WSGI specification.
            unsafe_base_prefix = environ.get("SCRIPT_NAME", "")
        # Sanitize base prefix
        base_prefix = pathutils.sanitize_path(unsafe_base_prefix).rstrip("/")
        logger.debug("Sanitized script name: %r", base_prefix)
        # Sanitize request URI (a WSGI server indicates with an empty path,
        # that the URL targets the application root without a trailing slash)
        path = pathutils.sanitize_path(environ.get("PATH_INFO", ""))
        logger.debug("Sanitized path: %r", path)

        # Get function corresponding to method
        function = getattr(
            self, "do_%s" % environ["REQUEST_METHOD"].upper(), None)
        if not function:
            return response(*httputils.METHOD_NOT_ALLOWED)

        # If "/.well-known" is not available, clients query "/"
        if path == "/.well-known" or path.startswith("/.well-known/"):
            return response(*httputils.NOT_FOUND)

        # Ask authentication backend to check rights
        login = password = ""
        external_login = self._auth.get_external_login(environ)
        authorization = environ.get("HTTP_AUTHORIZATION", "")
        if external_login:
            login, password = external_login
            login, password = login or "", password or ""
        elif authorization.startswith("Basic"):
            authorization = authorization[len("Basic"):].strip()
            login, password = httputils.decode_request(
                self.configuration, environ, base64.b64decode(
                    authorization.encode("ascii"))).split(":", 1)

        user = self._auth.login(login, password) or "" if login else ""
        if user and login == user:
            logger.info("Successful login: %r", user)
        elif user:
            logger.info("Successful login: %r -> %r", login, user)
        elif login:
            logger.info("Failed login attempt: %r", login)
            # Random delay to avoid timing oracles and bruteforce attacks
            delay = self.configuration.get("auth", "delay")
            if delay > 0:
                random_delay = delay * (0.5 + random.random())
                logger.debug("Sleeping %.3f seconds", random_delay)
                time.sleep(random_delay)

        if user and not pathutils.is_safe_path_component(user):
            # Prevent usernames like "user/calendar.ics"
            logger.info("Refused unsafe username: %r", user)
            user = ""

        # Create principal collection
        if user:
            principal_path = "/%s/" % user
            with self._storage.acquire_lock("r", user):
                principal = next(self._storage.discover(
                    principal_path, depth="1"), None)
            if not principal:
                if "W" in self._rights.authorization(user, principal_path):
                    with self._storage.acquire_lock("w", user):
                        try:
                            self._storage.create_collection(principal_path)
                        except ValueError as e:
                            logger.warning("Failed to create principal "
                                           "collection %r: %s", user, e)
                            user = ""
                else:
                    logger.warning("Access to principal path %r denied by "
                                   "rights backend", principal_path)

        if self.configuration.get("server", "_internal_server"):
            # Verify content length
            content_length = int(environ.get("CONTENT_LENGTH") or 0)
            if content_length:
                max_content_length = self.configuration.get(
                    "server", "max_content_length")
                if max_content_length and content_length > max_content_length:
                    logger.info("Request body too large: %d", content_length)
                    return response(*httputils.REQUEST_ENTITY_TOO_LARGE)

        if not login or user:
            status, headers, answer = function(
                environ, base_prefix, path, user)
            if (status, headers, answer) == httputils.NOT_ALLOWED:
                logger.info("Access to %r denied for %s", path,
                            repr(user) if user else "anonymous user")
        else:
            status, headers, answer = httputils.NOT_ALLOWED

        if ((status, headers, answer) == httputils.NOT_ALLOWED and not user and
                not external_login):
            # Unknown or unauthorized user
            logger.debug("Asking client for authentication")
            status = client.UNAUTHORIZED
            realm = self.configuration.get("auth", "realm")
            headers = dict(headers)
            headers.update({
                "WWW-Authenticate":
                "Basic realm=\"%s\"" % realm})

        return response(status, headers, answer)

    def _read_xml_request_body(self, environ):
        content = httputils.decode_request(
            self.configuration, environ,
            httputils.read_raw_request_body(self.configuration, environ))
        if not content:
            return None
        try:
            xml_content = DefusedET.fromstring(content)
        except ET.ParseError as e:
            logger.debug("Request content (Invalid XML):\n%s", content)
            raise RuntimeError("Failed to parse XML: %s" % e) from e
        if logger.isEnabledFor(logging.DEBUG):
            logger.debug("Request content:\n%s",
                         xmlutils.pretty_xml(xml_content))
        return xml_content

    def _xml_response(self, xml_content):
        if logger.isEnabledFor(logging.DEBUG):
            logger.debug("Response content:\n%s",
                         xmlutils.pretty_xml(xml_content))
        f = io.BytesIO()
        ET.ElementTree(xml_content).write(f, encoding=self._encoding,
                                          xml_declaration=True)
        return f.getvalue()

    def _webdav_error_response(self, status, human_tag):
        """Generate XML error response."""
        headers = {"Content-Type": "text/xml; charset=%s" % self._encoding}
        content = self._xml_response(xmlutils.webdav_error(human_tag))
        return status, headers, content


class Access:
    """Helper class to check access rights of an item"""

    def __init__(self, rights, user, path):
        self._rights = rights
        self.user = user
        self.path = path
        self.parent_path = pathutils.unstrip_path(
            posixpath.dirname(pathutils.strip_path(path)), True)
        self.permissions = self._rights.authorization(self.user, self.path)
        self._parent_permissions = None

    @property
    def parent_permissions(self):
        if self.path == self.parent_path:
            return self.permissions
        if self._parent_permissions is None:
            self._parent_permissions = self._rights.authorization(
                self.user, self.parent_path)
        return self._parent_permissions

    def check(self, permission, item=None):
        if permission not in "rw":
            raise ValueError("Invalid permission argument: %r" % permission)
        if not item:
            permissions = permission + permission.upper()
            parent_permissions = permission
        elif isinstance(item, storage.BaseCollection):
            if item.get_meta("tag"):
                permissions = permission
            else:
                permissions = permission.upper()
            parent_permissions = ""
        else:
            permissions = ""
            parent_permissions = permission
        return bool(rights.intersect(self.permissions, permissions) or (
            self.path != self.parent_path and
            rights.intersect(self.parent_permissions, parent_permissions)))

1	# This file is part of Radicale Server - Calendar Server
2	# Copyright © 2008 Nicolas Kandel
3	# Copyright © 2008 Pascal Halter
4	# Copyright © 2008-2017 Guillaume Ayoub
5	# Copyright © 2017-2019 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	"""	18✔
21	Radicale WSGI application.
22
23	Can be used with an external WSGI server (see ``radicale.application()``) or
24	the built-in server (see ``radicale.server`` module).
25
26	"""
27
28	import base64	18✔
29	import datetime	18✔
30	import io	18✔
31	import logging	18✔
32	import posixpath	18✔
33	import pprint	18✔
34	import random	18✔
35	import time	18✔
36	import zlib	18✔
37	from http import client	18✔
38	from xml.etree import ElementTree as ET	18✔
39
40	import defusedxml.ElementTree as DefusedET	18✔
41	import pkg_resources	18✔
42
43	from radicale import (auth, httputils, log, pathutils, rights, storage, web,	18✔
44	xmlutils)
45	from radicale.app.delete import ApplicationDeleteMixin	18✔
46	from radicale.app.get import ApplicationGetMixin	18✔
47	from radicale.app.head import ApplicationHeadMixin	18✔
48	from radicale.app.mkcalendar import ApplicationMkcalendarMixin	18✔
49	from radicale.app.mkcol import ApplicationMkcolMixin	18✔
50	from radicale.app.move import ApplicationMoveMixin	18✔
51	from radicale.app.options import ApplicationOptionsMixin	18✔
52	from radicale.app.post import ApplicationPostMixin	18✔
53	from radicale.app.propfind import ApplicationPropfindMixin	18✔
54	from radicale.app.proppatch import ApplicationProppatchMixin	18✔
55	from radicale.app.put import ApplicationPutMixin	18✔
56	from radicale.app.report import ApplicationReportMixin	18✔
57	from radicale.log import logger	18✔
58
59	VERSION = pkg_resources.get_distribution("radicale").version	18✔
60
61
62	class Application(	18✔
63	ApplicationDeleteMixin, ApplicationGetMixin, ApplicationHeadMixin,
64	ApplicationMkcalendarMixin, ApplicationMkcolMixin,
65	ApplicationMoveMixin, ApplicationOptionsMixin,
66	ApplicationPropfindMixin, ApplicationProppatchMixin,
67	ApplicationPostMixin, ApplicationPutMixin,
68	ApplicationReportMixin):
69
70	"""WSGI application."""
71
72	def __init__(self, configuration):	18✔
73	"""Initialize Application.
74
75	``configuration`` see ``radicale.config`` module.
76	The ``configuration`` must not change during the lifetime of
77	this object, it is kept as an internal reference.
78
79	"""
80	super().__init__()	18✔
81	self.configuration = configuration	18✔
82	self._auth = auth.load(configuration)	18✔
83	self._storage = storage.load(configuration)	18✔
84	self._rights = rights.load(configuration)	18✔
85	self._web = web.load(configuration)	18✔
86	self._encoding = configuration.get("encoding", "request")	18✔
87
88	def _headers_log(self, environ):	18✔
89	"""Sanitize headers for logging."""
90	request_environ = dict(environ)	18✔
91
92	# Mask passwords
93	mask_passwords = self.configuration.get("logging", "mask_passwords")	18✔
94	authorization = request_environ.get("HTTP_AUTHORIZATION", "")	18✔
95	if mask_passwords and authorization.startswith("Basic"):	18✔
96	request_environ["HTTP_AUTHORIZATION"] = "Basic masked"	18✔
97	if request_environ.get("HTTP_COOKIE"):	18!
98	request_environ["HTTP_COOKIE"] = "masked"	×
99
100	return request_environ	18✔
101
102	def __call__(self, environ, start_response):	18✔
103	with log.register_stream(environ["wsgi.errors"]):	18✔
104	try:	18✔
105	status, headers, answers = self._handle_request(environ)	18✔
106	except Exception as e:	18✔
107	try:	18✔
108	method = str(environ["REQUEST_METHOD"])	18✔
UNCOV 109	except Exception:	×
110	method = "unknown"	×
111	try:	18✔
112	path = str(environ.get("PATH_INFO", ""))	18✔
UNCOV 113	except Exception:	×
UNCOV 114	path = ""	×
115	logger.error("An exception occurred during %s request on %r: "	18✔
116	"%s", method, path, e, exc_info=True)
117	status, headers, answer = httputils.INTERNAL_SERVER_ERROR	18✔
118	answer = answer.encode("ascii")	18✔
119	status = "%d %s" % (	18✔
120	status.value, client.responses.get(status, "Unknown"))
121	headers = [	18✔
122	("Content-Length", str(len(answer)))] + list(headers)
123	answers = [answer]	18✔
124	start_response(status, headers)	18✔
125	return answers	18✔
126
127	def _handle_request(self, environ):	18✔
128	"""Manage a request."""
129	def response(status, headers=(), answer=None):	18✔
130	headers = dict(headers)	18✔
131	# Set content length
132	if answer:	18✔
133	if hasattr(answer, "encode"):	18✔
134	logger.debug("Response content:\n%s", answer)	18✔
135	headers["Content-Type"] += "; charset=%s" % self._encoding	18✔
136	answer = answer.encode(self._encoding)	18✔
137	accept_encoding = [	18✔
138	encoding.strip() for encoding in
139	environ.get("HTTP_ACCEPT_ENCODING", "").split(",")
140	if encoding.strip()]
141
142	if "gzip" in accept_encoding:	18!
UNCOV 143	zcomp = zlib.compressobj(wbits=16 + zlib.MAX_WBITS)	×
UNCOV 144	answer = zcomp.compress(answer) + zcomp.flush()	×
UNCOV 145	headers["Content-Encoding"] = "gzip"	×
146
147	headers["Content-Length"] = str(len(answer))	18✔
148
149	# Add extra headers set in configuration
150	for key in self.configuration.options("headers"):	18✔
151	headers[key] = self.configuration.get("headers", key)	18✔
152
153	# Start response
154	time_end = datetime.datetime.now()	18✔
155	status = "%d %s" % (	18✔
156	status, client.responses.get(status, "Unknown"))
157	logger.info(	18✔
158	"%s response status for %r%s in %.3f seconds: %s",
159	environ["REQUEST_METHOD"], environ.get("PATH_INFO", ""),
160	depthinfo, (time_end - time_begin).total_seconds(), status)
161	# Return response content
162	return status, list(headers.items()), [answer] if answer else []	18✔
163
164	remote_host = "unknown"	18✔
165	if environ.get("REMOTE_HOST"):	18!
UNCOV 166	remote_host = repr(environ["REMOTE_HOST"])	×
167	elif environ.get("REMOTE_ADDR"):	18✔
168	remote_host = environ["REMOTE_ADDR"]	18✔
169	if environ.get("HTTP_X_FORWARDED_FOR"):	18!
UNCOV 170	remote_host = "%r (forwarded by %s)" % (	×
171	environ["HTTP_X_FORWARDED_FOR"], remote_host)
172	remote_useragent = ""	18✔
173	if environ.get("HTTP_USER_AGENT"):	18✔
174	remote_useragent = " using %r" % environ["HTTP_USER_AGENT"]	18✔
175	depthinfo = ""	18✔
176	if environ.get("HTTP_DEPTH"):	18✔
177	depthinfo = " with depth %r" % environ["HTTP_DEPTH"]	18✔
178	time_begin = datetime.datetime.now()	18✔
179	logger.info(	18✔
180	"%s request for %r%s received from %s%s",
181	environ["REQUEST_METHOD"], environ.get("PATH_INFO", ""), depthinfo,
182	remote_host, remote_useragent)
183	headers = pprint.pformat(self._headers_log(environ))	18✔
184	logger.debug("Request headers:\n%s", headers)	18✔
185
186	# Let reverse proxies overwrite SCRIPT_NAME
187	if "HTTP_X_SCRIPT_NAME" in environ:	18!
188	# script_name must be removed from PATH_INFO by the client.
UNCOV 189	unsafe_base_prefix = environ["HTTP_X_SCRIPT_NAME"]	×
190	logger.debug("Script name overwritten by client: %r",	×
191	unsafe_base_prefix)
192	else:
193	# SCRIPT_NAME is already removed from PATH_INFO, according to the
194	# WSGI specification.
195	unsafe_base_prefix = environ.get("SCRIPT_NAME", "")	18✔
196	# Sanitize base prefix
197	base_prefix = pathutils.sanitize_path(unsafe_base_prefix).rstrip("/")	18✔
198	logger.debug("Sanitized script name: %r", base_prefix)	18✔
199	# Sanitize request URI (a WSGI server indicates with an empty path,
200	# that the URL targets the application root without a trailing slash)
201	path = pathutils.sanitize_path(environ.get("PATH_INFO", ""))	18✔
202	logger.debug("Sanitized path: %r", path)	18✔
203
204	# Get function corresponding to method
205	function = getattr(	18✔
206	self, "do_%s" % environ["REQUEST_METHOD"].upper(), None)
207	if not function:	18!
UNCOV 208	return response(*httputils.METHOD_NOT_ALLOWED)	×
209
210	# If "/.well-known" is not available, clients query "/"
211	if path == "/.well-known" or path.startswith("/.well-known/"):	18✔
212	return response(*httputils.NOT_FOUND)	18✔
213
214	# Ask authentication backend to check rights
215	login = password = ""	18✔
216	external_login = self._auth.get_external_login(environ)	18✔
217	authorization = environ.get("HTTP_AUTHORIZATION", "")	18✔
218	if external_login:	18✔
219	login, password = external_login	18✔
220	login, password = login or "", password or ""	18✔
221	elif authorization.startswith("Basic"):	18✔
222	authorization = authorization[len("Basic"):].strip()	18✔
223	login, password = httputils.decode_request(	18✔
224	self.configuration, environ, base64.b64decode(
225	authorization.encode("ascii"))).split(":", 1)
226
227	user = self._auth.login(login, password) or "" if login else ""	18✔
228	if user and login == user:	18✔
229	logger.info("Successful login: %r", user)	18✔
230	elif user:	18!
UNCOV 231	logger.info("Successful login: %r -> %r", login, user)	×
232	elif login:	18✔
233	logger.info("Failed login attempt: %r", login)	18✔
234	# Random delay to avoid timing oracles and bruteforce attacks
235	delay = self.configuration.get("auth", "delay")	18✔
236	if delay > 0:	18!
237	random_delay = delay * (0.5 + random.random())	18✔
238	logger.debug("Sleeping %.3f seconds", random_delay)	18✔
239	time.sleep(random_delay)	18✔
240
241	if user and not pathutils.is_safe_path_component(user):	18!
242	# Prevent usernames like "user/calendar.ics"
UNCOV 243	logger.info("Refused unsafe username: %r", user)	×
UNCOV 244	user = ""	×
245
246	# Create principal collection
247	if user:	18✔
248	principal_path = "/%s/" % user	18✔
249	with self._storage.acquire_lock("r", user):	18✔
250	principal = next(self._storage.discover(	18✔
251	principal_path, depth="1"), None)
252	if not principal:	18✔
253	if "W" in self._rights.authorization(user, principal_path):	18!
254	with self._storage.acquire_lock("w", user):	18✔
255	try:	18✔
256	self._storage.create_collection(principal_path)	18✔
UNCOV 257	except ValueError as e:	×
UNCOV 258	logger.warning("Failed to create principal "	×
259	"collection %r: %s", user, e)
UNCOV 260	user = ""	×
261	else:
UNCOV 262	logger.warning("Access to principal path %r denied by "	×
263	"rights backend", principal_path)
264
265	if self.configuration.get("server", "_internal_server"):	18✔
266	# Verify content length
267	content_length = int(environ.get("CONTENT_LENGTH") or 0)	18✔
268	if content_length:	18!
UNCOV 269	max_content_length = self.configuration.get(	×
270	"server", "max_content_length")
UNCOV 271	if max_content_length and content_length > max_content_length:	×
UNCOV 272	logger.info("Request body too large: %d", content_length)	×
UNCOV 273	return response(*httputils.REQUEST_ENTITY_TOO_LARGE)	×
274
275	if not login or user:	18✔
276	status, headers, answer = function(	18✔
277	environ, base_prefix, path, user)
278	if (status, headers, answer) == httputils.NOT_ALLOWED:	18✔
279	logger.info("Access to %r denied for %s", path,	18✔
280	repr(user) if user else "anonymous user")
281	else:
282	status, headers, answer = httputils.NOT_ALLOWED	18✔
283
284	if ((status, headers, answer) == httputils.NOT_ALLOWED and not user and	18✔
285	not external_login):
286	# Unknown or unauthorized user
287	logger.debug("Asking client for authentication")	18✔
288	status = client.UNAUTHORIZED	18✔
289	realm = self.configuration.get("auth", "realm")	18✔
290	headers = dict(headers)	18✔
291	headers.update({	18✔
292	"WWW-Authenticate":
293	"Basic realm=\"%s\"" % realm})
294
295	return response(status, headers, answer)	18✔
296
297	def _read_xml_request_body(self, environ):	18✔
298	content = httputils.decode_request(	18✔
299	self.configuration, environ,
300	httputils.read_raw_request_body(self.configuration, environ))
301	if not content:	18✔
302	return None	18✔
303	try:	18✔
304	xml_content = DefusedET.fromstring(content)	18✔
UNCOV 305	except ET.ParseError as e:	×
UNCOV 306	logger.debug("Request content (Invalid XML):\n%s", content)	×
UNCOV 307	raise RuntimeError("Failed to parse XML: %s" % e) from e	×
308	if logger.isEnabledFor(logging.DEBUG):	18!
309	logger.debug("Request content:\n%s",	18✔
310	xmlutils.pretty_xml(xml_content))
311	return xml_content	18✔
312
313	def _xml_response(self, xml_content):	18✔
314	if logger.isEnabledFor(logging.DEBUG):	18!
315	logger.debug("Response content:\n%s",	18✔
316	xmlutils.pretty_xml(xml_content))
317	f = io.BytesIO()	18✔
318	ET.ElementTree(xml_content).write(f, encoding=self._encoding,	18✔
319	xml_declaration=True)
320	return f.getvalue()	18✔
321
322	def _webdav_error_response(self, status, human_tag):	18✔
323	"""Generate XML error response."""
324	headers = {"Content-Type": "text/xml; charset=%s" % self._encoding}	18✔
325	content = self._xml_response(xmlutils.webdav_error(human_tag))	18✔
326	return status, headers, content	18✔
327
328
329	class Access:	18✔
330	"""Helper class to check access rights of an item"""
331
332	def __init__(self, rights, user, path):	18✔
333	self._rights = rights	18✔
334	self.user = user	18✔
335	self.path = path	18✔
336	self.parent_path = pathutils.unstrip_path(	18✔
337	posixpath.dirname(pathutils.strip_path(path)), True)
338	self.permissions = self._rights.authorization(self.user, self.path)	18✔
339	self._parent_permissions = None	18✔
340
341	@property	18✔
342	def parent_permissions(self):	6✔
343	if self.path == self.parent_path:	18!
UNCOV 344	return self.permissions	×
345	if self._parent_permissions is None:	18✔
346	self._parent_permissions = self._rights.authorization(	18✔
347	self.user, self.parent_path)
348	return self._parent_permissions	18✔
349
350	def check(self, permission, item=None):	18✔
351	if permission not in "rw":	18!
UNCOV 352	raise ValueError("Invalid permission argument: %r" % permission)	×
353	if not item:	18✔
354	permissions = permission + permission.upper()	18✔
355	parent_permissions = permission	18✔
356	elif isinstance(item, storage.BaseCollection):	18✔
357	if item.get_meta("tag"):	18✔
358	permissions = permission	18✔
359	else:
360	permissions = permission.upper()	18✔
361	parent_permissions = ""	18✔
362	else:
363	permissions = ""	18✔
364	parent_permissions = permission	18✔
365	return bool(rights.intersect(self.permissions, permissions) or (	18✔
366	self.path != self.parent_path and
367	rights.intersect(self.parent_permissions, parent_permissions)))

Kozea / Radicale / 254424173

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