2091153640

Build # 2091153640

Build Type

push

github

Committed by Unrud

Commit Message

Replace pkg_resources with importlib for Python >= 3.9

Run Details

1705 of 2075 branches covered (82.17%)

Branch coverage included in aggregate %.

42 of 50 new or added lines in 4 files covered. (84.0%)

1 existing line in 1 file now uncovered.

3437 of 3924 relevant lines covered (87.59%)

16.09 hits per line

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

80.3

/radicale/httputils.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-2018 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/>.

"""
Helper functions for HTTP.

"""

import contextlib
import os
import pathlib
import sys
import time
from http import client
from typing import List, Mapping, Union, cast

from radicale import config, pathutils, types
from radicale.log import logger

if sys.version_info < (3, 9):
    import pkg_resources

    _TRAVERSABLE_LIKE_TYPE = pathlib.Path
else:
    import importlib.abc
    from importlib import resources

    _TRAVERSABLE_LIKE_TYPE = Union[importlib.abc.Traversable, pathlib.Path]

NOT_ALLOWED: types.WSGIResponse = (
    client.FORBIDDEN, (("Content-Type", "text/plain"),),
    "Access to the requested resource forbidden.")
FORBIDDEN: types.WSGIResponse = (
    client.FORBIDDEN, (("Content-Type", "text/plain"),),
    "Action on the requested resource refused.")
BAD_REQUEST: types.WSGIResponse = (
    client.BAD_REQUEST, (("Content-Type", "text/plain"),), "Bad Request")
NOT_FOUND: types.WSGIResponse = (
    client.NOT_FOUND, (("Content-Type", "text/plain"),),
    "The requested resource could not be found.")
CONFLICT: types.WSGIResponse = (
    client.CONFLICT, (("Content-Type", "text/plain"),),
    "Conflict in the request.")
METHOD_NOT_ALLOWED: types.WSGIResponse = (
    client.METHOD_NOT_ALLOWED, (("Content-Type", "text/plain"),),
    "The method is not allowed on the requested resource.")
PRECONDITION_FAILED: types.WSGIResponse = (
    client.PRECONDITION_FAILED,
    (("Content-Type", "text/plain"),), "Precondition failed.")
REQUEST_TIMEOUT: types.WSGIResponse = (
    client.REQUEST_TIMEOUT, (("Content-Type", "text/plain"),),
    "Connection timed out.")
REQUEST_ENTITY_TOO_LARGE: types.WSGIResponse = (
    client.REQUEST_ENTITY_TOO_LARGE, (("Content-Type", "text/plain"),),
    "Request body too large.")
REMOTE_DESTINATION: types.WSGIResponse = (
    client.BAD_GATEWAY, (("Content-Type", "text/plain"),),
    "Remote destination not supported.")
DIRECTORY_LISTING: types.WSGIResponse = (
    client.FORBIDDEN, (("Content-Type", "text/plain"),),
    "Directory listings are not supported.")
INTERNAL_SERVER_ERROR: types.WSGIResponse = (
    client.INTERNAL_SERVER_ERROR, (("Content-Type", "text/plain"),),
    "A server error occurred.  Please contact the administrator.")

DAV_HEADERS: str = "1, 2, 3, calendar-access, addressbook, extended-mkcol"

MIMETYPES: Mapping[str, str] = {
    ".css": "text/css",
    ".eot": "application/vnd.ms-fontobject",
    ".gif": "image/gif",
    ".html": "text/html",
    ".js": "application/javascript",
    ".manifest": "text/cache-manifest",
    ".png": "image/png",
    ".svg": "image/svg+xml",
    ".ttf": "application/font-sfnt",
    ".txt": "text/plain",
    ".woff": "application/font-woff",
    ".woff2": "font/woff2",
    ".xml": "text/xml"}
FALLBACK_MIMETYPE: str = "application/octet-stream"


def decode_request(configuration: "config.Configuration",
                   environ: types.WSGIEnviron, text: bytes) -> str:
    """Try to magically decode ``text`` according to given ``environ``."""
    # List of charsets to try
    charsets: List[str] = []

    # 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(cast(str, configuration.get("encoding", "request")))
    # Then append various fallbacks
    charsets.append("utf-8")
    charsets.append("iso8859-1")
    # Remove duplicates
    for i, s in reversed(list(enumerate(charsets))):
        if s in charsets[:i]:
            del charsets[i]

    # Try to decode
    for charset in charsets:
        with contextlib.suppress(UnicodeDecodeError):
            return text.decode(charset)
    raise UnicodeDecodeError("decode_request", text, 0, len(text),
                             "all codecs failed [%s]" % ", ".join(charsets))


def read_raw_request_body(configuration: "config.Configuration",
                          environ: types.WSGIEnviron) -> bytes:
    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_request_body(configuration: "config.Configuration",
                      environ: types.WSGIEnviron) -> str:
    content = decode_request(configuration, environ,
                             read_raw_request_body(configuration, environ))
    logger.debug("Request content:\n%s", content)
    return content


def redirect(location: str, status: int = client.FOUND) -> types.WSGIResponse:
    return (status,
            {"Location": location, "Content-Type": "text/plain"},
            "Redirected to %s" % location)


def _serve_traversable(
        traversable: _TRAVERSABLE_LIKE_TYPE, base_prefix: str, path: str,
        path_prefix: str, index_file: str, mimetypes: Mapping[str, str],
        fallback_mimetype: str) -> types.WSGIResponse:
    if path != path_prefix and not path.startswith(path_prefix):
        raise ValueError("path must start with path_prefix: %r --> %r" %
                         (path_prefix, path))
    assert pathutils.sanitize_path(path) == path
    parts_path = path[len(path_prefix):].strip('/')
    parts = parts_path.split("/") if parts_path else []
    for part in parts:
        if not pathutils.is_safe_filesystem_path_component(part):
            logger.debug("Web content with unsafe path %r requested", path)
            return NOT_FOUND
        if (not traversable.is_dir() or
                all(part != entry.name for entry in traversable.iterdir())):
            return NOT_FOUND
        traversable = traversable.joinpath(part)
    if traversable.is_dir():
        if not path.endswith("/"):
            return redirect(base_prefix + path + "/")
        if not index_file:
            return NOT_FOUND
        traversable = traversable.joinpath(index_file)
    if not traversable.is_file():
        return NOT_FOUND
    content_type = MIMETYPES.get(
        os.path.splitext(traversable.name)[1].lower(), FALLBACK_MIMETYPE)
    headers = {"Content-Type": content_type}
    if isinstance(traversable, pathlib.Path):
        headers["Last-Modified"] = time.strftime(
            "%a, %d %b %Y %H:%M:%S GMT",
            time.gmtime(traversable.stat().st_mtime))
    answer = traversable.read_bytes()
    return client.OK, headers, answer


def serve_resource(
        package: str, resource: str, base_prefix: str, path: str,
        path_prefix: str = "/.web", index_file: str = "index.html",
        mimetypes: Mapping[str, str] = MIMETYPES,
        fallback_mimetype: str = FALLBACK_MIMETYPE) -> types.WSGIResponse:
    if sys.version_info < (3, 9):
        traversable = pathlib.Path(
            pkg_resources.resource_filename(package, resource))
    else:
        traversable = resources.files(package).joinpath(resource)
    return _serve_traversable(traversable, base_prefix, path, path_prefix,
                              index_file, mimetypes, fallback_mimetype)


def serve_folder(
        folder: str, base_prefix: str, path: str,
        path_prefix: str = "/.web", index_file: str = "index.html",
        mimetypes: Mapping[str, str] = MIMETYPES,
        fallback_mimetype: str = FALLBACK_MIMETYPE) -> types.WSGIResponse:
    # deprecated: use `serve_resource` instead
    traversable = pathlib.Path(folder)
    return _serve_traversable(traversable, base_prefix, path, path_prefix,
                              index_file, mimetypes, fallback_mimetype)

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-2018 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	"""	19✔
21	Helper functions for HTTP.
22
23	"""
24
25	import contextlib	19✔
26	import os	19✔
27	import pathlib	19✔
28	import sys	19✔
29	import time	19✔
30	from http import client	19✔
31	from typing import List, Mapping, Union, cast	19✔
32
33	from radicale import config, pathutils, types	19✔
34	from radicale.log import logger	19✔
35
36	if sys.version_info < (3, 9):	19✔
37	import pkg_resources	13✔
38
39	_TRAVERSABLE_LIKE_TYPE = pathlib.Path	13✔
40	else:
41	import importlib.abc	6✔
42	from importlib import resources	6✔
43
44	_TRAVERSABLE_LIKE_TYPE = Union[importlib.abc.Traversable, pathlib.Path]	6✔
45
46	NOT_ALLOWED: types.WSGIResponse = (	19✔
47	client.FORBIDDEN, (("Content-Type", "text/plain"),),
48	"Access to the requested resource forbidden.")
49	FORBIDDEN: types.WSGIResponse = (	19✔
50	client.FORBIDDEN, (("Content-Type", "text/plain"),),
51	"Action on the requested resource refused.")
52	BAD_REQUEST: types.WSGIResponse = (	19✔
53	client.BAD_REQUEST, (("Content-Type", "text/plain"),), "Bad Request")
54	NOT_FOUND: types.WSGIResponse = (	19✔
55	client.NOT_FOUND, (("Content-Type", "text/plain"),),
56	"The requested resource could not be found.")
57	CONFLICT: types.WSGIResponse = (	19✔
58	client.CONFLICT, (("Content-Type", "text/plain"),),
59	"Conflict in the request.")
60	METHOD_NOT_ALLOWED: types.WSGIResponse = (	19✔
61	client.METHOD_NOT_ALLOWED, (("Content-Type", "text/plain"),),
62	"The method is not allowed on the requested resource.")
63	PRECONDITION_FAILED: types.WSGIResponse = (	19✔
64	client.PRECONDITION_FAILED,
65	(("Content-Type", "text/plain"),), "Precondition failed.")
66	REQUEST_TIMEOUT: types.WSGIResponse = (	19✔
67	client.REQUEST_TIMEOUT, (("Content-Type", "text/plain"),),
68	"Connection timed out.")
69	REQUEST_ENTITY_TOO_LARGE: types.WSGIResponse = (	19✔
70	client.REQUEST_ENTITY_TOO_LARGE, (("Content-Type", "text/plain"),),
71	"Request body too large.")
72	REMOTE_DESTINATION: types.WSGIResponse = (	19✔
73	client.BAD_GATEWAY, (("Content-Type", "text/plain"),),
74	"Remote destination not supported.")
75	DIRECTORY_LISTING: types.WSGIResponse = (	19✔
76	client.FORBIDDEN, (("Content-Type", "text/plain"),),
77	"Directory listings are not supported.")
78	INTERNAL_SERVER_ERROR: types.WSGIResponse = (	19✔
79	client.INTERNAL_SERVER_ERROR, (("Content-Type", "text/plain"),),
80	"A server error occurred. Please contact the administrator.")
81
82	DAV_HEADERS: str = "1, 2, 3, calendar-access, addressbook, extended-mkcol"	19✔
83
84	MIMETYPES: Mapping[str, str] = {	19✔
85	".css": "text/css",
86	".eot": "application/vnd.ms-fontobject",
87	".gif": "image/gif",
88	".html": "text/html",
89	".js": "application/javascript",
90	".manifest": "text/cache-manifest",
91	".png": "image/png",
92	".svg": "image/svg+xml",
93	".ttf": "application/font-sfnt",
94	".txt": "text/plain",
95	".woff": "application/font-woff",
96	".woff2": "font/woff2",
97	".xml": "text/xml"}
98	FALLBACK_MIMETYPE: str = "application/octet-stream"	19✔
99
100
101	def decode_request(configuration: "config.Configuration",	19✔
102	environ: types.WSGIEnviron, text: bytes) -> str:
103	"""Try to magically decode ``text`` according to given ``environ``."""
104	# List of charsets to try
105	charsets: List[str] = []	19✔
106
107	# First append content charset given in the request
108	content_type = environ.get("CONTENT_TYPE")	19✔
109	if content_type and "charset=" in content_type:	19!
110	charsets.append(	×
111	content_type.split("charset=")[1].split(";")[0].strip())
112	# Then append default Radicale charset
113	charsets.append(cast(str, configuration.get("encoding", "request")))	19✔
114	# Then append various fallbacks
115	charsets.append("utf-8")	19✔
116	charsets.append("iso8859-1")	19✔
117	# Remove duplicates
118	for i, s in reversed(list(enumerate(charsets))):	19✔
119	if s in charsets[:i]:	19✔
120	del charsets[i]	19✔
121
122	# Try to decode
123	for charset in charsets:	19!
124	with contextlib.suppress(UnicodeDecodeError):	19✔
125	return text.decode(charset)	19✔
126	raise UnicodeDecodeError("decode_request", text, 0, len(text),	3✔
127	"all codecs failed [%s]" % ", ".join(charsets))
128
129
130	def read_raw_request_body(configuration: "config.Configuration",	19✔
131	environ: types.WSGIEnviron) -> bytes:
132	content_length = int(environ.get("CONTENT_LENGTH") or 0)	19✔
133	if not content_length:	19✔
134	return b""	19✔
135	content = environ["wsgi.input"].read(content_length)	19✔
136	if len(content) < content_length:	19!
137	raise RuntimeError("Request body too short: %d" % len(content))	×
138	return content	19✔
139
140
141	def read_request_body(configuration: "config.Configuration",	19✔
142	environ: types.WSGIEnviron) -> str:
143	content = decode_request(configuration, environ,	19✔
144	read_raw_request_body(configuration, environ))
145	logger.debug("Request content:\n%s", content)	19✔
146	return content	19✔
147
148
149	def redirect(location: str, status: int = client.FOUND) -> types.WSGIResponse:	19✔
150	return (status,	19✔
151	{"Location": location, "Content-Type": "text/plain"},
152	"Redirected to %s" % location)
153
154
155	def _serve_traversable(	19✔
156	traversable: _TRAVERSABLE_LIKE_TYPE, base_prefix: str, path: str,
157	path_prefix: str, index_file: str, mimetypes: Mapping[str, str],
158	fallback_mimetype: str) -> types.WSGIResponse:
159	if path != path_prefix and not path.startswith(path_prefix):	19!
160	raise ValueError("path must start with path_prefix: %r --> %r" %	×
161	(path_prefix, path))
162	assert pathutils.sanitize_path(path) == path	19✔
163	parts_path = path[len(path_prefix):].strip('/')	19✔
164	parts = parts_path.split("/") if parts_path else []	19✔
165	for part in parts:	19!
NEW 166	if not pathutils.is_safe_filesystem_path_component(part):	×
NEW 167	logger.debug("Web content with unsafe path %r requested", path)	×
NEW 168	return NOT_FOUND	×
NEW 169	if (not traversable.is_dir() or	×
170	all(part != entry.name for entry in traversable.iterdir())):
NEW 171	return NOT_FOUND	×
NEW 172	traversable = traversable.joinpath(part)	×
173	if traversable.is_dir():	19!
174	if not path.endswith("/"):	19✔
175	return redirect(base_prefix + path + "/")	19✔
176	if not index_file:	19!
NEW 177	return NOT_FOUND	×
178	traversable = traversable.joinpath(index_file)	19✔
179	if not traversable.is_file():	19!
UNCOV 180	return NOT_FOUND	×
181	content_type = MIMETYPES.get(	19✔
182	os.path.splitext(traversable.name)[1].lower(), FALLBACK_MIMETYPE)
183	headers = {"Content-Type": content_type}	19✔
184	if isinstance(traversable, pathlib.Path):	19!
185	headers["Last-Modified"] = time.strftime(	19✔
186	"%a, %d %b %Y %H:%M:%S GMT",
187	time.gmtime(traversable.stat().st_mtime))
188	answer = traversable.read_bytes()	19✔
189	return client.OK, headers, answer	19✔
190
191
192	def serve_resource(	19✔
193	package: str, resource: str, base_prefix: str, path: str,
194	path_prefix: str = "/.web", index_file: str = "index.html",
195	mimetypes: Mapping[str, str] = MIMETYPES,
196	fallback_mimetype: str = FALLBACK_MIMETYPE) -> types.WSGIResponse:
197	if sys.version_info < (3, 9):	19✔
198	traversable = pathlib.Path(	13✔
199	pkg_resources.resource_filename(package, resource))
200	else:
201	traversable = resources.files(package).joinpath(resource)	6✔
202	return _serve_traversable(traversable, base_prefix, path, path_prefix,	19✔
203	index_file, mimetypes, fallback_mimetype)
204
205
206	def serve_folder(	19✔
207	folder: str, base_prefix: str, path: str,
208	path_prefix: str = "/.web", index_file: str = "index.html",
209	mimetypes: Mapping[str, str] = MIMETYPES,
210	fallback_mimetype: str = FALLBACK_MIMETYPE) -> types.WSGIResponse:
211	# deprecated: use `serve_resource` instead
NEW 212	traversable = pathlib.Path(folder)	×
213	return _serve_traversable(traversable, base_prefix, path, path_prefix,	3✔
214	index_file, mimetypes, fallback_mimetype)

Kozea / Radicale / 2091153640

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