11997528648

Committed 24 Nov 2024 03:50PM UTC coverage: 78.038% (-0.2%) from 78.205%

Build # 11997528648

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1631 from pbiering/improve-hook

Improve storage hook

Run Details

1697 of 2292 branches covered (74.04%)

Branch coverage included in aggregate %.

8 of 15 new or added lines in 1 file covered. (53.33%)

8 existing lines in 1 file now uncovered.

3857 of 4825 relevant lines covered (79.94%)

15.41 hits per line

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

79.59

/radicale/app/__init__.py

# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2008 Nicolas Kandel
# Copyright © 2008 Pascal Halter
# Copyright © 2008-2017 Guillaume Ayoub
# Copyright © 2017-2019 Unrud <unrud@outlook.com>
# Copyright © 2024-2024 Peter Bieringer <pb@bieringer.de>
#
# 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 pprint
import random
import time
import zlib
from http import client
from typing import Iterable, List, Mapping, Tuple, Union

from radicale import config, httputils, log, pathutils, types
from radicale.app.base import ApplicationBase
from radicale.app.delete import ApplicationPartDelete
from radicale.app.get import ApplicationPartGet
from radicale.app.head import ApplicationPartHead
from radicale.app.mkcalendar import ApplicationPartMkcalendar
from radicale.app.mkcol import ApplicationPartMkcol
from radicale.app.move import ApplicationPartMove
from radicale.app.options import ApplicationPartOptions
from radicale.app.post import ApplicationPartPost
from radicale.app.propfind import ApplicationPartPropfind
from radicale.app.proppatch import ApplicationPartProppatch
from radicale.app.put import ApplicationPartPut
from radicale.app.report import ApplicationPartReport
from radicale.log import logger

# Combination of types.WSGIStartResponse and WSGI application return value
_IntermediateResponse = Tuple[str, List[Tuple[str, str]], Iterable[bytes]]


class Application(ApplicationPartDelete, ApplicationPartHead,
                  ApplicationPartGet, ApplicationPartMkcalendar,
                  ApplicationPartMkcol, ApplicationPartMove,
                  ApplicationPartOptions, ApplicationPartPropfind,
                  ApplicationPartProppatch, ApplicationPartPost,
                  ApplicationPartPut, ApplicationPartReport, ApplicationBase):
    """WSGI application."""

    _mask_passwords: bool
    _auth_delay: float
    _internal_server: bool
    _max_content_length: int
    _auth_realm: str
    _extra_headers: Mapping[str, str]
    _permit_delete_collection: bool
    _permit_overwrite_collection: bool

    def __init__(self, configuration: config.Configuration) -> None:
        """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__(configuration)
        self._mask_passwords = configuration.get("logging", "mask_passwords")
        self._bad_put_request_content = configuration.get("logging", "bad_put_request_content")
        self._request_header_on_debug = configuration.get("logging", "request_header_on_debug")
        self._response_content_on_debug = configuration.get("logging", "response_content_on_debug")
        self._auth_delay = configuration.get("auth", "delay")
        self._internal_server = configuration.get("server", "_internal_server")
        self._max_content_length = configuration.get(
            "server", "max_content_length")
        self._auth_realm = configuration.get("auth", "realm")
        self._permit_delete_collection = configuration.get("rights", "permit_delete_collection")
        logger.info("permit delete of collection: %s", self._permit_delete_collection)
        self._permit_overwrite_collection = configuration.get("rights", "permit_overwrite_collection")
        logger.info("permit overwrite of collection: %s", self._permit_overwrite_collection)
        self._extra_headers = dict()
        for key in self.configuration.options("headers"):
            self._extra_headers[key] = configuration.get("headers", key)

    def _scrub_headers(self, environ: types.WSGIEnviron) -> types.WSGIEnviron:
        """Mask passwords and cookies."""
        headers = dict(environ)
        if (self._mask_passwords and
                headers.get("HTTP_AUTHORIZATION", "").startswith("Basic")):
            headers["HTTP_AUTHORIZATION"] = "Basic **masked**"
        if headers.get("HTTP_COOKIE"):
            headers["HTTP_COOKIE"] = "**masked**"
        return headers

    def __call__(self, environ: types.WSGIEnviron, start_response:
                 types.WSGIStartResponse) -> Iterable[bytes]:
        with log.register_stream(environ["wsgi.errors"]):
            try:
                status_text, headers, answers = self._handle_request(environ)
            except Exception as e:
                logger.error("An exception occurred during %s request on %r: "
                             "%s", environ.get("REQUEST_METHOD", "unknown"),
                             environ.get("PATH_INFO", ""), e, exc_info=True)
                # Make minimal response
                status, raw_headers, raw_answer = (
                    httputils.INTERNAL_SERVER_ERROR)
                assert isinstance(raw_answer, str)
                answer = raw_answer.encode("ascii")
                status_text = "%d %s" % (
                    status, client.responses.get(status, "Unknown"))
                headers = [*raw_headers, ("Content-Length", str(len(answer)))]
                answers = [answer]
            start_response(status_text, headers)
        if environ.get("REQUEST_METHOD") == "HEAD":
            return []
        return answers

    def _handle_request(self, environ: types.WSGIEnviron
                        ) -> _IntermediateResponse:
        time_begin = datetime.datetime.now()
        request_method = environ["REQUEST_METHOD"].upper()
        unsafe_path = environ.get("PATH_INFO", "")

        """Manage a request."""
        def response(status: int, headers: types.WSGIResponseHeaders,
                     answer: Union[None, str, bytes]) -> _IntermediateResponse:
            """Helper to create response from internal types.WSGIResponse"""
            headers = dict(headers)
            # Set content length
            answers = []
            if answer is not None:
                if isinstance(answer, str):
                    if self._response_content_on_debug:
                        logger.debug("Response content:\n%s", answer)
                    else:
                        logger.debug("Response content: suppressed by config/option [logging] response_content_on_debug")
                    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))
                answers.append(answer)

            # Add extra headers set in configuration
            headers.update(self._extra_headers)

            # Start response
            time_end = datetime.datetime.now()
            status_text = "%d %s" % (
                status, client.responses.get(status, "Unknown"))
            logger.info("%s response status for %r%s in %.3f seconds: %s",
                        request_method, unsafe_path, depthinfo,
                        (time_end - time_begin).total_seconds(), status_text)
            # Return response content
            return status_text, list(headers.items()), answers

        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 = "%s (forwarded for %r)" % (
                remote_host, environ["HTTP_X_FORWARDED_FOR"])
        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"]
        logger.info("%s request for %r%s received from %s%s",
                    request_method, unsafe_path, depthinfo,
                    remote_host, remote_useragent)
        if self._request_header_on_debug:
            logger.debug("Request header:\n%s",
                         pprint.pformat(self._scrub_headers(environ)))
        else:
            logger.debug("Request header: suppressed by config/option [logging] request_header_on_debug")

        # SCRIPT_NAME is already removed from PATH_INFO, according to the
        # WSGI specification.
        # Reverse proxies can overwrite SCRIPT_NAME with X-SCRIPT-NAME header
        base_prefix_src = ("HTTP_X_SCRIPT_NAME" if "HTTP_X_SCRIPT_NAME" in
                           environ else "SCRIPT_NAME")
        base_prefix = environ.get(base_prefix_src, "")
        if base_prefix and base_prefix[0] != "/":
            logger.error("Base prefix (from %s) must start with '/': %r",
                         base_prefix_src, base_prefix)
            if base_prefix_src == "HTTP_X_SCRIPT_NAME":
                return response(*httputils.BAD_REQUEST)
            return response(*httputils.INTERNAL_SERVER_ERROR)
        if base_prefix.endswith("/"):
            logger.warning("Base prefix (from %s) must not end with '/': %r",
                           base_prefix_src, base_prefix)
            base_prefix = base_prefix.rstrip("/")
        logger.debug("Base prefix (from %s): %r", base_prefix_src, 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(unsafe_path)
        logger.debug("Sanitized path: %r", path)

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

        # Redirect all "…/.well-known/{caldav,carddav}" paths to "/".
        # This shouldn't be necessary but some clients like TbSync require it.
        # Status must be MOVED PERMANENTLY using FOUND causes problems
        if (path.rstrip("/").endswith("/.well-known/caldav") or
                path.rstrip("/").endswith("/.well-known/carddav")):
            return response(*httputils.redirect(
                base_prefix + "/", client.MOVED_PERMANENTLY))
        # Return NOT FOUND for all other paths containing ".well-known"
        if path.endswith("/.well-known") or "/.well-known/" in path:
            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 self.configuration.get("auth", "type") == "ldap":
            try:
                logger.debug("Groups %r", ",".join(self._auth._ldap_groups))
                self._rights._user_groups = self._auth._ldap_groups
            except AttributeError:
                pass
        if user and login == user:
            logger.info("Successful login: %r", user)
        elif user:
            logger.info("Successful login: %r -> %r", login, user)
        elif login:
            logger.warning("Failed login attempt from %s: %r",
                           remote_host, login)
            # Random delay to avoid timing oracles and bruteforce attacks
            if self._auth_delay > 0:
                random_delay = self._auth_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(iter(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:
                            new_coll = self._storage.create_collection(principal_path)
                            if new_coll:
                                jsn_coll = self.configuration.get("storage", "predefined_collections")
                                for (name_coll, props) in jsn_coll.items():
                                    try:
                                        self._storage.create_collection(principal_path + name_coll, props=props)
                                    except ValueError as e:
                                        logger.warning("Failed to create predefined collection %r: %s", name_coll, e)
                        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:
                if (self._max_content_length > 0 and
                        content_length > self._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
            headers = dict(headers)
            headers.update({
                "WWW-Authenticate":
                "Basic realm=\"%s\"" % self._auth_realm})

        return response(status, headers, answer)

1	# This file is part of Radicale - CalDAV and CardDAV server
2	# Copyright © 2008 Nicolas Kandel
3	# Copyright © 2008 Pascal Halter
4	# Copyright © 2008-2017 Guillaume Ayoub
5	# Copyright © 2017-2019 Unrud <unrud@outlook.com>
6	# Copyright © 2024-2024 Peter Bieringer <pb@bieringer.de>
7	#
8	# This library is free software: you can redistribute it and/or modify
9	# it under the terms of the GNU General Public License as published by
10	# the Free Software Foundation, either version 3 of the License, or
11	# (at your option) any later version.
12	#
13	# This library is distributed in the hope that it will be useful,
14	# but WITHOUT ANY WARRANTY; without even the implied warranty of
15	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16	# GNU General Public License for more details.
17	#
18	# You should have received a copy of the GNU General Public License
19	# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
20
21	"""	3✔
22	Radicale WSGI application.
23
24	Can be used with an external WSGI server (see ``radicale.application()``) or
25	the built-in server (see ``radicale.server`` module).
26
27	"""
28
29	import base64	20✔
30	import datetime	20✔
31	import pprint	20✔
32	import random	20✔
33	import time	20✔
34	import zlib	20✔
35	from http import client	20✔
36	from typing import Iterable, List, Mapping, Tuple, Union	20✔
37
38	from radicale import config, httputils, log, pathutils, types	20✔
39	from radicale.app.base import ApplicationBase	20✔
40	from radicale.app.delete import ApplicationPartDelete	20✔
41	from radicale.app.get import ApplicationPartGet	20✔
42	from radicale.app.head import ApplicationPartHead	20✔
43	from radicale.app.mkcalendar import ApplicationPartMkcalendar	20✔
44	from radicale.app.mkcol import ApplicationPartMkcol	20✔
45	from radicale.app.move import ApplicationPartMove	20✔
46	from radicale.app.options import ApplicationPartOptions	20✔
47	from radicale.app.post import ApplicationPartPost	20✔
48	from radicale.app.propfind import ApplicationPartPropfind	20✔
49	from radicale.app.proppatch import ApplicationPartProppatch	20✔
50	from radicale.app.put import ApplicationPartPut	20✔
51	from radicale.app.report import ApplicationPartReport	20✔
52	from radicale.log import logger	20✔
53
54	# Combination of types.WSGIStartResponse and WSGI application return value
55	_IntermediateResponse = Tuple[str, List[Tuple[str, str]], Iterable[bytes]]	20✔
56
57
58	class Application(ApplicationPartDelete, ApplicationPartHead,	20✔
59	ApplicationPartGet, ApplicationPartMkcalendar,
60	ApplicationPartMkcol, ApplicationPartMove,
61	ApplicationPartOptions, ApplicationPartPropfind,
62	ApplicationPartProppatch, ApplicationPartPost,
63	ApplicationPartPut, ApplicationPartReport, ApplicationBase):
64	"""WSGI application."""
65
66	_mask_passwords: bool	20✔
67	_auth_delay: float	20✔
68	_internal_server: bool	20✔
69	_max_content_length: int	20✔
70	_auth_realm: str	20✔
71	_extra_headers: Mapping[str, str]	20✔
72	_permit_delete_collection: bool	20✔
73	_permit_overwrite_collection: bool	20✔
74
75	def __init__(self, configuration: config.Configuration) -> None:	20✔
76	"""Initialize Application.
77
78	``configuration`` see ``radicale.config`` module.
79	The ``configuration`` must not change during the lifetime of
80	this object, it is kept as an internal reference.
81
82	"""
83	super().__init__(configuration)	20✔
84	self._mask_passwords = configuration.get("logging", "mask_passwords")	20✔
85	self._bad_put_request_content = configuration.get("logging", "bad_put_request_content")	20✔
86	self._request_header_on_debug = configuration.get("logging", "request_header_on_debug")	20✔
87	self._response_content_on_debug = configuration.get("logging", "response_content_on_debug")	20✔
88	self._auth_delay = configuration.get("auth", "delay")	20✔
89	self._internal_server = configuration.get("server", "_internal_server")	20✔
90	self._max_content_length = configuration.get(	20✔
91	"server", "max_content_length")
92	self._auth_realm = configuration.get("auth", "realm")	20✔
93	self._permit_delete_collection = configuration.get("rights", "permit_delete_collection")	20✔
94	logger.info("permit delete of collection: %s", self._permit_delete_collection)	20✔
95	self._permit_overwrite_collection = configuration.get("rights", "permit_overwrite_collection")	20✔
96	logger.info("permit overwrite of collection: %s", self._permit_overwrite_collection)	20✔
97	self._extra_headers = dict()	20✔
98	for key in self.configuration.options("headers"):	20✔
99	self._extra_headers[key] = configuration.get("headers", key)	20✔
100
101	def _scrub_headers(self, environ: types.WSGIEnviron) -> types.WSGIEnviron:	20✔
102	"""Mask passwords and cookies."""
103	headers = dict(environ)	×
104	if (self._mask_passwords and	×
105	headers.get("HTTP_AUTHORIZATION", "").startswith("Basic")):
106	headers["HTTP_AUTHORIZATION"] = "Basic masked"	×
107	if headers.get("HTTP_COOKIE"):	×
108	headers["HTTP_COOKIE"] = "masked"	×
109	return headers	×
110
111	def __call__(self, environ: types.WSGIEnviron, start_response:	20✔
112	types.WSGIStartResponse) -> Iterable[bytes]:
113	with log.register_stream(environ["wsgi.errors"]):	20✔
114	try:	20✔
115	status_text, headers, answers = self._handle_request(environ)	20✔
UNCOV 116	except Exception as e:	×
UNCOV 117	logger.error("An exception occurred during %s request on %r: "	×
118	"%s", environ.get("REQUEST_METHOD", "unknown"),
119	environ.get("PATH_INFO", ""), e, exc_info=True)
120	# Make minimal response
UNCOV 121	status, raw_headers, raw_answer = (	×
122	httputils.INTERNAL_SERVER_ERROR)
UNCOV 123	assert isinstance(raw_answer, str)	×
UNCOV 124	answer = raw_answer.encode("ascii")	×
UNCOV 125	status_text = "%d %s" % (	×
126	status, client.responses.get(status, "Unknown"))
UNCOV 127	headers = [*raw_headers, ("Content-Length", str(len(answer)))]	×
UNCOV 128	answers = [answer]	×
129	start_response(status_text, headers)	20✔
130	if environ.get("REQUEST_METHOD") == "HEAD":	20✔
131	return []	20✔
132	return answers	20✔
133
134	def _handle_request(self, environ: types.WSGIEnviron	20✔
135	) -> _IntermediateResponse:
136	time_begin = datetime.datetime.now()	20✔
137	request_method = environ["REQUEST_METHOD"].upper()	20✔
138	unsafe_path = environ.get("PATH_INFO", "")	20✔
139
140	"""Manage a request."""	12✔
141	def response(status: int, headers: types.WSGIResponseHeaders,	20✔
142	answer: Union[None, str, bytes]) -> _IntermediateResponse:
143	"""Helper to create response from internal types.WSGIResponse"""
144	headers = dict(headers)	20✔
145	# Set content length
146	answers = []	20✔
147	if answer is not None:	20✔
148	if isinstance(answer, str):	20✔
149	if self._response_content_on_debug:	20!
150	logger.debug("Response content:\n%s", answer)	×
151	else:
152	logger.debug("Response content: suppressed by config/option [logging] response_content_on_debug")	20✔
153	headers["Content-Type"] += "; charset=%s" % self._encoding	20✔
154	answer = answer.encode(self._encoding)	20✔
155	accept_encoding = [	20✔
156	encoding.strip() for encoding in
157	environ.get("HTTP_ACCEPT_ENCODING", "").split(",")
158	if encoding.strip()]
159
160	if "gzip" in accept_encoding:	20!
161	zcomp = zlib.compressobj(wbits=16 + zlib.MAX_WBITS)	×
162	answer = zcomp.compress(answer) + zcomp.flush()	×
163	headers["Content-Encoding"] = "gzip"	×
164
165	headers["Content-Length"] = str(len(answer))	20✔
166	answers.append(answer)	20✔
167
168	# Add extra headers set in configuration
169	headers.update(self._extra_headers)	20✔
170
171	# Start response
172	time_end = datetime.datetime.now()	20✔
173	status_text = "%d %s" % (	20✔
174	status, client.responses.get(status, "Unknown"))
175	logger.info("%s response status for %r%s in %.3f seconds: %s",	20✔
176	request_method, unsafe_path, depthinfo,
177	(time_end - time_begin).total_seconds(), status_text)
178	# Return response content
179	return status_text, list(headers.items()), answers	20✔
180
181	remote_host = "unknown"	20✔
182	if environ.get("REMOTE_HOST"):	20!
183	remote_host = repr(environ["REMOTE_HOST"])	×
184	elif environ.get("REMOTE_ADDR"):	20✔
185	remote_host = environ["REMOTE_ADDR"]	20✔
186	if environ.get("HTTP_X_FORWARDED_FOR"):	20!
187	remote_host = "%s (forwarded for %r)" % (	×
188	remote_host, environ["HTTP_X_FORWARDED_FOR"])
189	remote_useragent = ""	20✔
190	if environ.get("HTTP_USER_AGENT"):	20✔
191	remote_useragent = " using %r" % environ["HTTP_USER_AGENT"]	20✔
192	depthinfo = ""	20✔
193	if environ.get("HTTP_DEPTH"):	20✔
194	depthinfo = " with depth %r" % environ["HTTP_DEPTH"]	20✔
195	logger.info("%s request for %r%s received from %s%s",	20✔
196	request_method, unsafe_path, depthinfo,
197	remote_host, remote_useragent)
198	if self._request_header_on_debug:	20!
199	logger.debug("Request header:\n%s",	×
200	pprint.pformat(self._scrub_headers(environ)))
201	else:
202	logger.debug("Request header: suppressed by config/option [logging] request_header_on_debug")	20✔
203
204	# SCRIPT_NAME is already removed from PATH_INFO, according to the
205	# WSGI specification.
206	# Reverse proxies can overwrite SCRIPT_NAME with X-SCRIPT-NAME header
207	base_prefix_src = ("HTTP_X_SCRIPT_NAME" if "HTTP_X_SCRIPT_NAME" in	20✔
208	environ else "SCRIPT_NAME")
209	base_prefix = environ.get(base_prefix_src, "")	20✔
210	if base_prefix and base_prefix[0] != "/":	20✔
211	logger.error("Base prefix (from %s) must start with '/': %r",	20✔
212	base_prefix_src, base_prefix)
213	if base_prefix_src == "HTTP_X_SCRIPT_NAME":	20✔
214	return response(*httputils.BAD_REQUEST)	20✔
215	return response(*httputils.INTERNAL_SERVER_ERROR)	20✔
216	if base_prefix.endswith("/"):	20✔
217	logger.warning("Base prefix (from %s) must not end with '/': %r",	20✔
218	base_prefix_src, base_prefix)
219	base_prefix = base_prefix.rstrip("/")	20✔
220	logger.debug("Base prefix (from %s): %r", base_prefix_src, base_prefix)	20✔
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(unsafe_path)	20✔
224	logger.debug("Sanitized path: %r", path)	20✔
225
226	# Get function corresponding to method
227	function = getattr(self, "do_%s" % request_method, None)	20✔
228	if not function:	20!
229	return response(*httputils.METHOD_NOT_ALLOWED)	×
230
231	# Redirect all "…/.well-known/{caldav,carddav}" paths to "/".
232	# This shouldn't be necessary but some clients like TbSync require it.
233	# Status must be MOVED PERMANENTLY using FOUND causes problems
234	if (path.rstrip("/").endswith("/.well-known/caldav") or	20✔
235	path.rstrip("/").endswith("/.well-known/carddav")):
236	return response(*httputils.redirect(	20✔
237	base_prefix + "/", client.MOVED_PERMANENTLY))
238	# Return NOT FOUND for all other paths containing ".well-known"
239	if path.endswith("/.well-known") or "/.well-known/" in path:	20✔
240	return response(*httputils.NOT_FOUND)	20✔
241
242	# Ask authentication backend to check rights
243	login = password = ""	20✔
244	external_login = self._auth.get_external_login(environ)	20✔
245	authorization = environ.get("HTTP_AUTHORIZATION", "")	20✔
246	if external_login:	20✔
247	login, password = external_login	20✔
248	login, password = login or "", password or ""	20✔
249	elif authorization.startswith("Basic"):	20✔
250	authorization = authorization[len("Basic"):].strip()	20✔
251	login, password = httputils.decode_request(	20✔
252	self.configuration, environ, base64.b64decode(
253	authorization.encode("ascii"))).split(":", 1)
254
255	user = self._auth.login(login, password) or "" if login else ""	20✔
256	if self.configuration.get("auth", "type") == "ldap":	20!
257	try:	×
258	logger.debug("Groups %r", ",".join(self._auth._ldap_groups))	×
259	self._rights._user_groups = self._auth._ldap_groups	×
260	except AttributeError:	×
261	pass	×
262	if user and login == user:	20✔
263	logger.info("Successful login: %r", user)	20✔
264	elif user:	20✔
265	logger.info("Successful login: %r -> %r", login, user)	20✔
266	elif login:	20✔
267	logger.warning("Failed login attempt from %s: %r",	20✔
268	remote_host, login)
269	# Random delay to avoid timing oracles and bruteforce attacks
270	if self._auth_delay > 0:	20!
271	random_delay = self._auth_delay * (0.5 + random.random())	20✔
272	logger.debug("Sleeping %.3f seconds", random_delay)	20✔
273	time.sleep(random_delay)	20✔
274
275	if user and not pathutils.is_safe_path_component(user):	20!
276	# Prevent usernames like "user/calendar.ics"
277	logger.info("Refused unsafe username: %r", user)	×
278	user = ""	×
279
280	# Create principal collection
281	if user:	20✔
282	principal_path = "/%s/" % user	20✔
283	with self._storage.acquire_lock("r", user):	20✔
284	principal = next(iter(self._storage.discover(	20✔
285	principal_path, depth="1")), None)
286	if not principal:	20✔
287	if "W" in self._rights.authorization(user, principal_path):	20!
288	with self._storage.acquire_lock("w", user):	20✔
289	try:	20✔
290	new_coll = self._storage.create_collection(principal_path)	20✔
291	if new_coll:	20!
292	jsn_coll = self.configuration.get("storage", "predefined_collections")	20✔
293	for (name_coll, props) in jsn_coll.items():	20!
294	try:	×
295	self._storage.create_collection(principal_path + name_coll, props=props)	×
296	except ValueError as e:	×
297	logger.warning("Failed to create predefined collection %r: %s", name_coll, e)	×
298	except ValueError as e:	×
299	logger.warning("Failed to create principal "	×
300	"collection %r: %s", user, e)
301	user = ""	×
302	else:
303	logger.warning("Access to principal path %r denied by "	×
304	"rights backend", principal_path)
305
306	if self._internal_server:	20✔
307	# Verify content length
308	content_length = int(environ.get("CONTENT_LENGTH") or 0)	20✔
309	if content_length:	20!
310	if (self._max_content_length > 0 and	×
311	content_length > self._max_content_length):
312	logger.info("Request body too large: %d", content_length)	×
313	return response(*httputils.REQUEST_ENTITY_TOO_LARGE)	×
314
315	if not login or user:	20✔
316	status, headers, answer = function(	20✔
317	environ, base_prefix, path, user)
318	if (status, headers, answer) == httputils.NOT_ALLOWED:	20✔
319	logger.info("Access to %r denied for %s", path,	20✔
320	repr(user) if user else "anonymous user")
321	else:
322	status, headers, answer = httputils.NOT_ALLOWED	20✔
323
324	if ((status, headers, answer) == httputils.NOT_ALLOWED and not user and	20✔
325	not external_login):
326	# Unknown or unauthorized user
327	logger.debug("Asking client for authentication")	20✔
328	status = client.UNAUTHORIZED	20✔
329	headers = dict(headers)	20✔
330	headers.update({	20✔
331	"WWW-Authenticate":
332	"Basic realm=\"%s\"" % self._auth_realm})
333
334	return response(status, headers, answer)	20✔

Kozea / Radicale / 11997528648

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