Kozea / Radicale / 934

pending completion

Build # 934

Build Type

Pull #998

travis-ci

Committed by

web-flow

Commit Message

Add example using dummy database with read-only access

Pull Request Pull Request #998: Remove configparser dependency when constructing Application object

Run Details

1202 of 1545 branches covered (77.8%)

Branch coverage included in aggregate %.

22 of 23 new or added lines in 3 files covered. (95.65%)

1 existing line in 1 file now uncovered.

2950 of 3500 relevant lines covered (84.29%)

4.94 hits per line

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

83.78

/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 typing import Mapping, Optional
from xml.etree import ElementTree as ET

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, *, auth, rights, storage, web,
            auth_delay: float = 0,
            auth_realm="Radicale - Password Required",
            encoding="utf-8",
            headers: Optional[Mapping[str, str]] = None,
            internal_server: bool = False,
            mask_passwords: bool = True,
            max_content_length: int = 0
    ) -> None:
        """Initialize application.

        :param auth_delay: Incorrect authentication delay in seconds.
            Set this to number greater than 0 to enable random delay (up to
            ``auth_delay``) to avoid timing oracles and bruteforce attacks.
        :param auth_realm: Message displayed when a password is needed.
        :param encoding: Encode responses using given codec.
        :param headers: Extra headers to include in every response.
        :param internal_server: The internal server is used.
            Set this to ``True`` if application is running by
            :func:`radicale.server.serve`.
        :param max_content_length: If ``internal_server`` is ``True`` and
            value is greater than 0 - reject incoming request if
            Content-Length header is higher than ``max_content_length`` value.
        """

        super().__init__()

        self._auth = auth
        self._rights = rights
        self._storage = storage
        self._web = web

        self._auth_delay = auth_delay
        self._auth_realm = auth_realm
        self._encoding = encoding
        self._headers = headers or {}
        self._internal_server = internal_server
        self._mask_passwords = mask_passwords
        self._max_content_length = max_content_length

    @classmethod
    def from_config(cls, config):
        return cls(
            auth=auth.load(config),
            rights=rights.load(config),
            storage=storage.load(config),
            web=web.load(config),

            auth_delay=config.get("auth", "delay"),
            auth_realm=config.get("auth", "realm"),
            encoding=config.get("encoding", "request"),
            headers={
                key: config.get("headers", key)
                for key in config.options("headers")
            },
            internal_server=config.get("internal", "internal_server"),
            mask_passwords=config.get("logging", "mask_passwords"),
            max_content_length=config.get("server", "max_content_length"),
        )

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

        # Mask passwords
        authorization = request_environ.get("HTTP_AUTHORIZATION", "")
        if self._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, 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
            headers.update(self._headers)

            # 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())

        # 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._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
            if self._rights.authorized(user, principal_path, "W"):
                with self._storage.acquire_lock("r", user):
                    principal = next(
                        self._storage.discover(principal_path, depth="1"),
                        None)
                if not principal:
                    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._internal_server:
            # Verify content length
            content_length = int(environ.get("CONTENT_LENGTH") or 0)
            if content_length:
                max_content_length = self._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._auth_realm
            headers = dict(headers)
            headers.update({
                "WWW-Authenticate":
                "Basic realm=\"%s\"" % realm})

        return response(status, headers, answer)

    def access(self, user, path, 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
        if permissions and self._rights.authorized(user, path, permissions):
            return True
        if parent_permissions:
            parent_path = pathutils.unstrip_path(
                posixpath.dirname(pathutils.strip_path(path)), True)
            if self._rights.authorized(user, parent_path, parent_permissions):
                return True
        return False

    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 = ET.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, namespace, name,
                              status=httputils.WEBDAV_PRECONDITION_FAILED[0]):
        """Generate XML error response."""
        headers = {"Content-Type": "text/xml; charset=%s" % self._encoding}
        content = self.write_xml_content(
            xmlutils.webdav_error(namespace, name))
        return status, headers, content

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

Kozea / Radicale / 934

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