1730294139

Build Type

push

github

Committed by Unrud

Commit Message

Error if SCRIPT_NAME ends with '/'

Run Details

1457 of 1835 branches covered (79.4%)

Branch coverage included in aggregate %.

5 of 5 new or added lines in 1 file covered. (100.0%)

8 existing lines in 1 file now uncovered.

3344 of 3903 relevant lines covered (85.68%)

12.82 hits per line

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

88.07

/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>
#
# 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]

    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._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._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):
                    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))
                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)
        logger.debug("Request headers:\n%s",
                     pprint.pformat(self._scrub_headers(environ)))

        # 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.endswith("/"):
            logger.error("Base prefix (from %s) must not end 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)
        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)

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

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

        user = self._auth.login(login, password) or "" if login else ""
        if user and login == user:
            logger.info("Successful login: %r", user)
        elif user:
            logger.info("Successful login: %r -> %r", login, user)
        elif login:
            logger.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:
                            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:
                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	#
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	"""	15✔
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	15✔
29	import datetime	15✔
30	import pprint	15✔
31	import random	15✔
32	import time	15✔
33	import zlib	15✔
34	from http import client	15✔
35	from typing import Iterable, List, Mapping, Tuple, Union	15✔
36
37	from radicale import config, httputils, log, pathutils, types	15✔
38	from radicale.app.base import ApplicationBase	15✔
39	from radicale.app.delete import ApplicationPartDelete	15✔
40	from radicale.app.get import ApplicationPartGet	15✔
41	from radicale.app.head import ApplicationPartHead	15✔
42	from radicale.app.mkcalendar import ApplicationPartMkcalendar	15✔
43	from radicale.app.mkcol import ApplicationPartMkcol	15✔
44	from radicale.app.move import ApplicationPartMove	15✔
45	from radicale.app.options import ApplicationPartOptions	15✔
46	from radicale.app.post import ApplicationPartPost	15✔
47	from radicale.app.propfind import ApplicationPartPropfind	15✔
48	from radicale.app.proppatch import ApplicationPartProppatch	15✔
49	from radicale.app.put import ApplicationPartPut	15✔
50	from radicale.app.report import ApplicationPartReport	15✔
51	from radicale.log import logger	15✔
52
53	# Combination of types.WSGIStartResponse and WSGI application return value
54	_IntermediateResponse = Tuple[str, List[Tuple[str, str]], Iterable[bytes]]	15✔
55
56
57	class Application(ApplicationPartDelete, ApplicationPartHead,	15✔
58	ApplicationPartGet, ApplicationPartMkcalendar,
59	ApplicationPartMkcol, ApplicationPartMove,
60	ApplicationPartOptions, ApplicationPartPropfind,
61	ApplicationPartProppatch, ApplicationPartPost,
62	ApplicationPartPut, ApplicationPartReport, ApplicationBase):
63	"""WSGI application."""
64
65	_mask_passwords: bool	15✔
66	_auth_delay: float	15✔
67	_internal_server: bool	15✔
68	_max_content_length: int	15✔
69	_auth_realm: str	15✔
70	_extra_headers: Mapping[str, str]	15✔
71
72	def __init__(self, configuration: config.Configuration) -> None:	15✔
73	"""Initialize Application.
74
75	``configuration`` see ``radicale.config`` module.
76	The ``configuration`` must not change during the lifetime of
77	this object, it is kept as an internal reference.
78
79	"""
80	super().__init__(configuration)	15✔
81	self._mask_passwords = configuration.get("logging", "mask_passwords")	15✔
82	self._auth_delay = configuration.get("auth", "delay")	15✔
83	self._internal_server = configuration.get("server", "_internal_server")	15✔
84	self._max_content_length = configuration.get(	15✔
85	"server", "max_content_length")
86	self._auth_realm = configuration.get("auth", "realm")	15✔
87	self._extra_headers = dict()	15✔
88	for key in self.configuration.options("headers"):	15✔
89	self._extra_headers[key] = configuration.get("headers", key)	15✔
90
91	def _scrub_headers(self, environ: types.WSGIEnviron) -> types.WSGIEnviron:	15✔
92	"""Mask passwords and cookies."""
93	headers = dict(environ)	15✔
94	if (self._mask_passwords and	15✔
95	headers.get("HTTP_AUTHORIZATION", "").startswith("Basic")):
96	headers["HTTP_AUTHORIZATION"] = "Basic masked"	15✔
97	if headers.get("HTTP_COOKIE"):	15!
98	headers["HTTP_COOKIE"] = "masked"	×
99	return headers	15✔
100
101	def __call__(self, environ: types.WSGIEnviron, start_response:	15✔
102	types.WSGIStartResponse) -> Iterable[bytes]:
103	with log.register_stream(environ["wsgi.errors"]):	15✔
104	try:	15✔
105	status_text, headers, answers = self._handle_request(environ)	15✔
106	except Exception as e:	15✔
107	logger.error("An exception occurred during %s request on %r: "	15✔
108	"%s", environ.get("REQUEST_METHOD", "unknown"),
109	environ.get("PATH_INFO", ""), e, exc_info=True)
110	# Make minimal response
111	status, raw_headers, raw_answer = (	15✔
112	httputils.INTERNAL_SERVER_ERROR)
113	assert isinstance(raw_answer, str)	15✔
114	answer = raw_answer.encode("ascii")	15✔
115	status_text = "%d %s" % (	15✔
116	status, client.responses.get(status, "Unknown"))
117	headers = [*raw_headers, ("Content-Length", str(len(answer)))]	15✔
118	answers = [answer]	15✔
119	start_response(status_text, headers)	15✔
120	if environ.get("REQUEST_METHOD") == "HEAD":	15✔
121	return []	15✔
122	return answers	15✔
123
124	def _handle_request(self, environ: types.WSGIEnviron	15✔
125	) -> _IntermediateResponse:
126	time_begin = datetime.datetime.now()	15✔
127	request_method = environ["REQUEST_METHOD"].upper()	15✔
128	unsafe_path = environ.get("PATH_INFO", "")	15✔
129
130	"""Manage a request."""
131	def response(status: int, headers: types.WSGIResponseHeaders,	15✔
132	answer: Union[None, str, bytes]) -> _IntermediateResponse:
133	"""Helper to create response from internal types.WSGIResponse"""
134	headers = dict(headers)	15✔
135	# Set content length
136	answers = []	15✔
137	if answer is not None:	15✔
138	if isinstance(answer, str):	15✔
139	logger.debug("Response content:\n%s", answer)	15✔
140	headers["Content-Type"] += "; charset=%s" % self._encoding	15✔
141	answer = answer.encode(self._encoding)	15✔
142	accept_encoding = [	15✔
143	encoding.strip() for encoding in
144	environ.get("HTTP_ACCEPT_ENCODING", "").split(",")
145	if encoding.strip()]
146
147	if "gzip" in accept_encoding:	15!
148	zcomp = zlib.compressobj(wbits=16 + zlib.MAX_WBITS)	×
149	answer = zcomp.compress(answer) + zcomp.flush()	×
150	headers["Content-Encoding"] = "gzip"	×
151
152	headers["Content-Length"] = str(len(answer))	15✔
153	answers.append(answer)	15✔
154
155	# Add extra headers set in configuration
156	headers.update(self._extra_headers)	15✔
157
158	# Start response
159	time_end = datetime.datetime.now()	15✔
160	status_text = "%d %s" % (	15✔
161	status, client.responses.get(status, "Unknown"))
162	logger.info("%s response status for %r%s in %.3f seconds: %s",	15✔
163	request_method, unsafe_path, depthinfo,
164	(time_end - time_begin).total_seconds(), status_text)
165	# Return response content
166	return status_text, list(headers.items()), answers	15✔
167
168	remote_host = "unknown"	15✔
169	if environ.get("REMOTE_HOST"):	15!
170	remote_host = repr(environ["REMOTE_HOST"])	×
171	elif environ.get("REMOTE_ADDR"):	15✔
172	remote_host = environ["REMOTE_ADDR"]	15✔
173	if environ.get("HTTP_X_FORWARDED_FOR"):	15!
174	remote_host = "%s (forwarded for %r)" % (	×
175	remote_host, environ["HTTP_X_FORWARDED_FOR"])
176	remote_useragent = ""	15✔
177	if environ.get("HTTP_USER_AGENT"):	15✔
178	remote_useragent = " using %r" % environ["HTTP_USER_AGENT"]	15✔
179	depthinfo = ""	15✔
180	if environ.get("HTTP_DEPTH"):	15✔
181	depthinfo = " with depth %r" % environ["HTTP_DEPTH"]	15✔
182	logger.info("%s request for %r%s received from %s%s",	15✔
183	request_method, unsafe_path, depthinfo,
184	remote_host, remote_useragent)
185	logger.debug("Request headers:\n%s",	15✔
186	pprint.pformat(self._scrub_headers(environ)))
187
188	# SCRIPT_NAME is already removed from PATH_INFO, according to the
189	# WSGI specification.
190	# Reverse proxies can overwrite SCRIPT_NAME with X-SCRIPT-NAME header
191	base_prefix_src = ("HTTP_X_SCRIPT_NAME" if "HTTP_X_SCRIPT_NAME" in	15✔
192	environ else "SCRIPT_NAME")
193	base_prefix = environ.get(base_prefix_src, "")	15✔
194	if base_prefix.endswith("/"):	15✔
195	logger.error("Base prefix (from %s) must not end with '/': %r",	15✔
196	base_prefix_src, base_prefix)
197	if base_prefix_src == "HTTP_X_SCRIPT_NAME":	15✔
198	return response(*httputils.BAD_REQUEST)	15✔
199	return response(*httputils.INTERNAL_SERVER_ERROR)	15✔
200	logger.debug("Base prefix (from %s): %r", base_prefix_src, base_prefix)	15✔
201	# Sanitize request URI (a WSGI server indicates with an empty path,
202	# that the URL targets the application root without a trailing slash)
203	path = pathutils.sanitize_path(unsafe_path)	15✔
204	logger.debug("Sanitized path: %r", path)	15✔
205
206	# Get function corresponding to method
207	function = getattr(self, "do_%s" % request_method, None)	15✔
208	if not function:	15!
UNCOV 209	return response(*httputils.METHOD_NOT_ALLOWED)	×
210
211	# If "/.well-known" is not available, clients query "/"
212	if path == "/.well-known" or path.startswith("/.well-known/"):	15✔
213	return response(*httputils.NOT_FOUND)	15✔
214
215	# Ask authentication backend to check rights
216	login = password = ""	15✔
217	external_login = self._auth.get_external_login(environ)	15✔
218	authorization = environ.get("HTTP_AUTHORIZATION", "")	15✔
219	if external_login:	15✔
220	login, password = external_login	15✔
221	login, password = login or "", password or ""	15✔
222	elif authorization.startswith("Basic"):	15✔
223	authorization = authorization[len("Basic"):].strip()	15✔
224	login, password = httputils.decode_request(	15✔
225	self.configuration, environ, base64.b64decode(
226	authorization.encode("ascii"))).split(":", 1)
227
228	user = self._auth.login(login, password) or "" if login else ""	15✔
229	if user and login == user:	15✔
230	logger.info("Successful login: %r", user)	15✔
231	elif user:	15!
UNCOV 232	logger.info("Successful login: %r -> %r", login, user)	×
233	elif login:	15✔
234	logger.warning("Failed login attempt from %s: %r",	15✔
235	remote_host, login)
236	# Random delay to avoid timing oracles and bruteforce attacks
237	if self._auth_delay > 0:	15!
238	random_delay = self._auth_delay * (0.5 + random.random())	15✔
239	logger.debug("Sleeping %.3f seconds", random_delay)	15✔
240	time.sleep(random_delay)	15✔
241
242	if user and not pathutils.is_safe_path_component(user):	15!
243	# Prevent usernames like "user/calendar.ics"
244	logger.info("Refused unsafe username: %r", user)	×
UNCOV 245	user = ""	×
246
247	# Create principal collection
248	if user:	15✔
249	principal_path = "/%s/" % user	15✔
250	with self._storage.acquire_lock("r", user):	15✔
251	principal = next(iter(self._storage.discover(	15✔
252	principal_path, depth="1")), None)
253	if not principal:	15✔
254	if "W" in self._rights.authorization(user, principal_path):	15!
255	with self._storage.acquire_lock("w", user):	15✔
256	try:	15✔
257	self._storage.create_collection(principal_path)	15✔
258	except ValueError as e:	×
UNCOV 259	logger.warning("Failed to create principal "	×
260	"collection %r: %s", user, e)
UNCOV 261	user = ""	×
262	else:
UNCOV 263	logger.warning("Access to principal path %r denied by "	×
264	"rights backend", principal_path)
265
266	if self._internal_server:	15✔
267	# Verify content length
268	content_length = int(environ.get("CONTENT_LENGTH") or 0)	15✔
269	if content_length:	15!
UNCOV 270	if (self._max_content_length > 0 and	×
271	content_length > self._max_content_length):
272	logger.info("Request body too large: %d", content_length)	×
UNCOV 273	return response(*httputils.REQUEST_ENTITY_TOO_LARGE)	×
274
275	if not login or user:	15✔
276	status, headers, answer = function(	15✔
277	environ, base_prefix, path, user)
278	if (status, headers, answer) == httputils.NOT_ALLOWED:	15✔
279	logger.info("Access to %r denied for %s", path,	15✔
280	repr(user) if user else "anonymous user")
281	else:
282	status, headers, answer = httputils.NOT_ALLOWED	15✔
283
284	if ((status, headers, answer) == httputils.NOT_ALLOWED and not user and	15✔
285	not external_login):
286	# Unknown or unauthorized user
287	logger.debug("Asking client for authentication")	15✔
288	status = client.UNAUTHORIZED	15✔
289	headers = dict(headers)	15✔
290	headers.update({	15✔
291	"WWW-Authenticate":
292	"Basic realm=\"%s\"" % self._auth_realm})
293
294	return response(status, headers, answer)	15✔

Kozea / Radicale / 1730294139

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