251538896

Build Type

push

github

Committed by Unrud

Commit Message

Respond 405 for unsupported methods instead of server error

Run Details

1193 of 1542 branches covered (77.37%)

Branch coverage included in aggregate %.

2 of 3 new or added lines in 1 file covered. (66.67%)

2 existing lines in 1 file now uncovered.

2877 of 3424 relevant lines covered (84.02%)

15.12 hits per line

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

84.02

/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.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,
        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 _decode(self, text, environ):
        """Try to magically decode ``text`` according to given ``environ``."""
        # List of charsets to try
        charsets = []

        # First append content charset given in the request
        content_type = environ.get("CONTENT_TYPE")
        if content_type and "charset=" in content_type:
            charsets.append(
                content_type.split("charset=")[1].split(";")[0].strip())
        # Then append default Radicale charset
        charsets.append(self._encoding)
        # Then append various fallbacks
        charsets.append("utf-8")
        charsets.append("iso8859-1")

        # Try to decode
        for charset in charsets:
            try:
                return text.decode(charset)
            except UnicodeDecodeError:
                pass
        raise UnicodeDecodeError

    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 = self._decode(base64.b64decode(
                authorization.encode("ascii")), environ).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_raw_content(self, environ):
        content_length = int(environ.get("CONTENT_LENGTH") or 0)
        if not content_length:
            return b""
        content = environ["wsgi.input"].read(content_length)
        if len(content) < content_length:
            raise RuntimeError("Request body too short: %d" % len(content))
        return content

    def _read_content(self, environ):
        content = self._decode(self._read_raw_content(environ), environ)
        logger.debug("Request content:\n%s", content)
        return content

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

Kozea / Radicale / 251538896

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