1713995253

Build Type

push

github

Committed by Unrud

Commit Message

Extract httputils.serve_folder

Run Details

1453 of 1831 branches covered (79.36%)

Branch coverage included in aggregate %.

26 of 31 new or added lines in 2 files covered. (83.87%)

16 existing lines in 2 files now uncovered.

3336 of 3895 relevant lines covered (85.65%)

12.82 hits per line

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

84.95

/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 time
from http import client
from typing import List, Mapping, cast

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

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

Kozea / Radicale / 1713995253

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