324815672

Build Type

push

github

Committed by Unrud

Commit Message

Use renameat2 on Linux for atomic exchanging of files

Run Details

1224 of 1587 branches covered (77.13%)

Branch coverage included in aggregate %.

22 of 37 new or added lines in 2 files covered. (59.46%)

21 existing lines in 2 files now uncovered.

2930 of 3505 relevant lines covered (83.59%)

12.53 hits per line

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

76.31

/radicale/server.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/>.

"""
Built-in WSGI server.

"""

import errno
import os
import select
import socket
import socketserver
import ssl
import sys
import wsgiref.simple_server
from urllib.parse import unquote

from radicale import Application, config
from radicale.log import logger

if hasattr(socket, "EAI_ADDRFAMILY"):
    COMPAT_EAI_ADDRFAMILY = socket.EAI_ADDRFAMILY
elif hasattr(socket, "EAI_NONAME"):
    # Windows and BSD don't have a special error code for this
    COMPAT_EAI_ADDRFAMILY = socket.EAI_NONAME
if hasattr(socket, "EAI_NODATA"):
    COMPAT_EAI_NODATA = socket.EAI_NODATA
elif hasattr(socket, "EAI_NONAME"):
    # Windows and BSD don't have a special error code for this
    COMPAT_EAI_NODATA = socket.EAI_NONAME
if hasattr(socket, "IPPROTO_IPV6"):
    COMPAT_IPPROTO_IPV6 = socket.IPPROTO_IPV6
elif os.name == "nt":
    # Workaround: https://bugs.python.org/issue29515
    COMPAT_IPPROTO_IPV6 = 41


def format_address(address):
    return "[%s]:%d" % address[:2]


class ParallelHTTPServer(socketserver.ThreadingMixIn,
                         wsgiref.simple_server.WSGIServer):

    # We wait for child threads ourself
    block_on_close = False
    daemon_threads = True

    def __init__(self, configuration, family, address, RequestHandlerClass):
        self.configuration = configuration
        self.address_family = family
        super().__init__(address, RequestHandlerClass)
        self.client_sockets = set()

    def server_bind(self):
        if self.address_family == socket.AF_INET6:
            # Only allow IPv6 connections to the IPv6 socket
            self.socket.setsockopt(COMPAT_IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
        super().server_bind()

    def get_request(self):
        # Set timeout for client
        request, client_address = super().get_request()
        timeout = self.configuration.get("server", "timeout")
        if timeout:
            request.settimeout(timeout)
        client_socket, client_socket_out = socket.socketpair()
        self.client_sockets.add(client_socket_out)
        return request, (*client_address, client_socket)

    def finish_request_locked(self, request, client_address):
        return super().finish_request(request, client_address)

    def finish_request(self, request, client_address):
        *client_address, client_socket = client_address
        client_address = tuple(client_address)
        try:
            return self.finish_request_locked(request, client_address)
        finally:
            client_socket.close()

    def handle_error(self, request, client_address):
        if issubclass(sys.exc_info()[0], socket.timeout):
            logger.info("Client timed out", exc_info=True)
        else:
            logger.error("An exception occurred during request: %s",
                         sys.exc_info()[1], exc_info=True)


class ParallelHTTPSServer(ParallelHTTPServer):

    def server_bind(self):
        super().server_bind()
        # Wrap the TCP socket in an SSL socket
        certfile = self.configuration.get("server", "certificate")
        keyfile = self.configuration.get("server", "key")
        cafile = self.configuration.get("server", "certificate_authority")
        # Test if the files can be read
        for name, filename in [("certificate", certfile), ("key", keyfile),
                               ("certificate_authority", cafile)]:
            type_name = config.DEFAULT_CONFIG_SCHEMA["server"][name][
                "type"].__name__
            source = self.configuration.get_source("server", name)
            if name == "certificate_authority" and not filename:
                continue
            try:
                open(filename, "r").close()
            except OSError as e:
                raise RuntimeError(
                    "Invalid %s value for option %r in section %r in %s: %r "
                    "(%s)" % (type_name, name, "server", source, filename,
                              e)) from e
        context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
        context.load_cert_chain(certfile=certfile, keyfile=keyfile)
        if cafile:
            context.load_verify_locations(cafile=cafile)
            context.verify_mode = ssl.CERT_REQUIRED
        self.socket = context.wrap_socket(
            self.socket, server_side=True, do_handshake_on_connect=False)

    def finish_request_locked(self, request, client_address):
        try:
            try:
                request.do_handshake()
            except socket.timeout:
                raise
            except Exception as e:
                raise RuntimeError("SSL handshake failed: %s" % e) from e
        except Exception:
            try:
                self.handle_error(request, client_address)
            finally:
                self.shutdown_request(request)
            return
        return super().finish_request_locked(request, client_address)


class ServerHandler(wsgiref.simple_server.ServerHandler):

    # Don't pollute WSGI environ with OS environment
    os_environ = {}

    def log_exception(self, exc_info):
        logger.error("An exception occurred during request: %s",
                     exc_info[1], exc_info=exc_info)


class RequestHandler(wsgiref.simple_server.WSGIRequestHandler):
    """HTTP requests handler."""

    def log_request(self, code="-", size="-"):
        pass  # Disable request logging.

    def log_error(self, format_, *args):
        logger.error("An error occurred during request: %s", format_ % args)

    def get_environ(self):
        env = super().get_environ()
        if hasattr(self.connection, "getpeercert"):
            # The certificate can be evaluated by the auth module
            env["REMOTE_CERTIFICATE"] = self.connection.getpeercert()
        # Parent class only tries latin1 encoding
        env["PATH_INFO"] = unquote(self.path.split("?", 1)[0])
        return env

    def handle(self):
        """Copy of WSGIRequestHandler.handle with different ServerHandler"""

        self.raw_requestline = self.rfile.readline(65537)
        if len(self.raw_requestline) > 65536:
            self.requestline = ""
            self.request_version = ""
            self.command = ""
            self.send_error(414)
            return

        if not self.parse_request():
            return

        handler = ServerHandler(
            self.rfile, self.wfile, self.get_stderr(), self.get_environ()
        )
        handler.request_handler = self
        handler.run(self.server.get_app())


def serve(configuration, shutdown_socket=None):
    """Serve radicale from configuration.

    `shutdown_socket` can be used to gracefully shutdown the server.
    The socket can be created with `socket.socketpair()`, when the other socket
    gets closed the server stops accepting new requests by clients and the
    function returns after all active requests are finished.

    """

    logger.info("Starting Radicale")
    # Copy configuration before modifying
    configuration = configuration.copy()
    configuration.update({"server": {"_internal_server": "True"}}, "server",
                         privileged=True)

    use_ssl = configuration.get("server", "ssl")
    server_class = ParallelHTTPSServer if use_ssl else ParallelHTTPServer
    application = Application(configuration)
    servers = {}
    try:
        for address in configuration.get("server", "hosts"):
            # Try to bind sockets for IPv4 and IPv6
            possible_families = (socket.AF_INET, socket.AF_INET6)
            bind_ok = False
            for i, family in enumerate(possible_families):
                is_last = i == len(possible_families) - 1
                try:
                    server = server_class(configuration, family, address,
                                          RequestHandler)
                except OSError as e:
                    # Ignore unsupported families (only one must work)
                    if ((bind_ok or not is_last) and (
                            isinstance(e, socket.gaierror) and (
                                # Hostname does not exist or doesn't have
                                # address for address family
                                # macOS: IPv6 address for INET address family
                                e.errno == socket.EAI_NONAME or
                                # Address not for address family
                                e.errno == COMPAT_EAI_ADDRFAMILY or
                                e.errno == COMPAT_EAI_NODATA) or
                            # Workaround for PyPy
                            str(e) == "address family mismatched" or
                            # Address family not available (e.g. IPv6 disabled)
                            # macOS: IPv4 address for INET6 address family with
                            #        IPV6_V6ONLY set
                            e.errno == errno.EADDRNOTAVAIL or
                            # Address family not supported
                            e.errno == errno.EAFNOSUPPORT or
                            # Protocol not supported
                            e.errno == errno.EPROTONOSUPPORT)):
                        continue
                    raise RuntimeError("Failed to start server %r: %s" % (
                                           format_address(address), e)) from e
                servers[server.socket] = server
                bind_ok = True
                server.set_app(application)
                logger.info("Listening on %r%s",
                            format_address(server.server_address),
                            " with SSL" if use_ssl else "")
        if not servers:
            raise RuntimeError("No servers started")

        # Mainloop
        select_timeout = None
        if os.name == "nt":
            # Fallback to busy waiting. (select(...) blocks SIGINT on Windows.)
            select_timeout = 1.0
        max_connections = configuration.get("server", "max_connections")
        logger.info("Radicale server ready")
        while True:
            rlist = []
            # Wait for finished clients
            for server in servers.values():
                rlist.extend(server.client_sockets)
            # Accept new connections if max_connections is not reached
            if max_connections <= 0 or len(rlist) < max_connections:
                rlist.extend(servers)
            # Use socket to get notified of program shutdown
            if shutdown_socket is not None:
                rlist.append(shutdown_socket)
            rlist, _, _ = select.select(rlist, [], [], select_timeout)
            rlist = set(rlist)
            if shutdown_socket in rlist:
                logger.info("Stopping Radicale")
                break
            for server in servers.values():
                finished_sockets = server.client_sockets.intersection(rlist)
                for s in finished_sockets:
                    s.close()
                    server.client_sockets.remove(s)
                    rlist.remove(s)
                if finished_sockets:
                    server.service_actions()
            if rlist:
                server = servers.get(rlist.pop())
                if server:
                    server.handle_request()
    finally:
        # Wait for clients to finish and close servers
        for server in servers.values():
            for s in server.client_sockets:
                s.recv(1)
                s.close()
            server.server_close()

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	"""	15✔
21	Built-in WSGI server.
22
23	"""
24
25	import errno	15✔
26	import os	15✔
27	import select	15✔
28	import socket	15✔
29	import socketserver	15✔
30	import ssl	15✔
31	import sys	15✔
32	import wsgiref.simple_server	15✔
33	from urllib.parse import unquote	15✔
34
35	from radicale import Application, config	15✔
36	from radicale.log import logger	15✔
37
38	if hasattr(socket, "EAI_ADDRFAMILY"):	15!
39	COMPAT_EAI_ADDRFAMILY = socket.EAI_ADDRFAMILY	15✔
40	elif hasattr(socket, "EAI_NONAME"):	×
41	# Windows and BSD don't have a special error code for this
42	COMPAT_EAI_ADDRFAMILY = socket.EAI_NONAME	×
43	if hasattr(socket, "EAI_NODATA"):	15!
44	COMPAT_EAI_NODATA = socket.EAI_NODATA	15✔
45	elif hasattr(socket, "EAI_NONAME"):	×
46	# Windows and BSD don't have a special error code for this
47	COMPAT_EAI_NODATA = socket.EAI_NONAME	×
48	if hasattr(socket, "IPPROTO_IPV6"):	15!
49	COMPAT_IPPROTO_IPV6 = socket.IPPROTO_IPV6	15✔
50	elif os.name == "nt":	×
51	# Workaround: https://bugs.python.org/issue29515
52	COMPAT_IPPROTO_IPV6 = 41	×
53
54
55	def format_address(address):	15✔
56	return "[%s]:%d" % address[:2]	15✔
57
58
59	class ParallelHTTPServer(socketserver.ThreadingMixIn,	15✔
60	wsgiref.simple_server.WSGIServer):
61
62	# We wait for child threads ourself
63	block_on_close = False	15✔
64	daemon_threads = True	15✔
65
66	def __init__(self, configuration, family, address, RequestHandlerClass):	15✔
67	self.configuration = configuration	15✔
68	self.address_family = family	15✔
69	super().__init__(address, RequestHandlerClass)	15✔
70	self.client_sockets = set()	15✔
71
72	def server_bind(self):	15✔
73	if self.address_family == socket.AF_INET6:	15✔
74	# Only allow IPv6 connections to the IPv6 socket
75	self.socket.setsockopt(COMPAT_IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)	15✔
76	super().server_bind()	15✔
77
78	def get_request(self):	15✔
79	# Set timeout for client
80	request, client_address = super().get_request()	15✔
81	timeout = self.configuration.get("server", "timeout")	15✔
82	if timeout:	15!
83	request.settimeout(timeout)	15✔
84	client_socket, client_socket_out = socket.socketpair()	15✔
85	self.client_sockets.add(client_socket_out)	15✔
86	return request, (*client_address, client_socket)	15✔
87
88	def finish_request_locked(self, request, client_address):	15✔
89	return super().finish_request(request, client_address)	15✔
90
91	def finish_request(self, request, client_address):	15✔
92	*client_address, client_socket = client_address	15✔
93	client_address = tuple(client_address)	15✔
94	try:	15✔
95	return self.finish_request_locked(request, client_address)	15✔
96	finally:
97	client_socket.close()	15✔
98
99	def handle_error(self, request, client_address):	15✔
100	if issubclass(sys.exc_info()[0], socket.timeout):	×
101	logger.info("Client timed out", exc_info=True)	×
102	else:
103	logger.error("An exception occurred during request: %s",	×
104	sys.exc_info()[1], exc_info=True)
105
106
107	class ParallelHTTPSServer(ParallelHTTPServer):	15✔
108
109	def server_bind(self):	15✔
110	super().server_bind()	15✔
111	# Wrap the TCP socket in an SSL socket
112	certfile = self.configuration.get("server", "certificate")	15✔
113	keyfile = self.configuration.get("server", "key")	15✔
114	cafile = self.configuration.get("server", "certificate_authority")	15✔
115	# Test if the files can be read
116	for name, filename in [("certificate", certfile), ("key", keyfile),	15✔
117	("certificate_authority", cafile)]:
118	type_name = config.DEFAULT_CONFIG_SCHEMA["server"][name][	15✔
119	"type"].__name__
120	source = self.configuration.get_source("server", name)	15✔
121	if name == "certificate_authority" and not filename:	15✔
122	continue	15✔
123	try:	15✔
124	open(filename, "r").close()	15✔
125	except OSError as e:	×
126	raise RuntimeError(	×
127	"Invalid %s value for option %r in section %r in %s: %r "
128	"(%s)" % (type_name, name, "server", source, filename,
129	e)) from e
130	context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)	15✔
131	context.load_cert_chain(certfile=certfile, keyfile=keyfile)	15✔
132	if cafile:	15!
133	context.load_verify_locations(cafile=cafile)	×
134	context.verify_mode = ssl.CERT_REQUIRED	×
135	self.socket = context.wrap_socket(	15✔
136	self.socket, server_side=True, do_handshake_on_connect=False)
137
138	def finish_request_locked(self, request, client_address):	15✔
139	try:	15✔
140	try:	15✔
141	request.do_handshake()	15✔
142	except socket.timeout:	×
143	raise	×
144	except Exception as e:	×
145	raise RuntimeError("SSL handshake failed: %s" % e) from e	×
146	except Exception:	×
147	try:	×
148	self.handle_error(request, client_address)	×
149	finally:
150	self.shutdown_request(request)	×
151	return	×
152	return super().finish_request_locked(request, client_address)	15✔
153
154
155	class ServerHandler(wsgiref.simple_server.ServerHandler):	15✔
156
157	# Don't pollute WSGI environ with OS environment
158	os_environ = {}	15✔
159
160	def log_exception(self, exc_info):	15✔
161	logger.error("An exception occurred during request: %s",	×
162	exc_info[1], exc_info=exc_info)
163
164
165	class RequestHandler(wsgiref.simple_server.WSGIRequestHandler):	15✔
166	"""HTTP requests handler."""
167
168	def log_request(self, code="-", size="-"):	15✔
169	pass # Disable request logging.	15✔
170
171	def log_error(self, format_, *args):	15✔
172	logger.error("An error occurred during request: %s", format_ % args)	×
173
174	def get_environ(self):	15✔
175	env = super().get_environ()	15✔
176	if hasattr(self.connection, "getpeercert"):	15✔
177	# The certificate can be evaluated by the auth module
178	env["REMOTE_CERTIFICATE"] = self.connection.getpeercert()	15✔
179	# Parent class only tries latin1 encoding
180	env["PATH_INFO"] = unquote(self.path.split("?", 1)[0])	15✔
181	return env	15✔
182
183	def handle(self):	15✔
184	"""Copy of WSGIRequestHandler.handle with different ServerHandler"""
185
186	self.raw_requestline = self.rfile.readline(65537)	15✔
187	if len(self.raw_requestline) > 65536:	15!
188	self.requestline = ""	×
189	self.request_version = ""	×
190	self.command = ""	×
191	self.send_error(414)	×
192	return	×
193
194	if not self.parse_request():	15!
195	return	×
196
197	handler = ServerHandler(	15✔
198	self.rfile, self.wfile, self.get_stderr(), self.get_environ()
199	)
200	handler.request_handler = self	15✔
201	handler.run(self.server.get_app())	15✔
202
203
204	def serve(configuration, shutdown_socket=None):	15✔
205	"""Serve radicale from configuration.
206
207	`shutdown_socket` can be used to gracefully shutdown the server.
208	The socket can be created with `socket.socketpair()`, when the other socket
209	gets closed the server stops accepting new requests by clients and the
210	function returns after all active requests are finished.
211
212	"""
213
214	logger.info("Starting Radicale")	15✔
215	# Copy configuration before modifying
216	configuration = configuration.copy()	15✔
217	configuration.update({"server": {"_internal_server": "True"}}, "server",	15✔
218	privileged=True)
219
220	use_ssl = configuration.get("server", "ssl")	15✔
221	server_class = ParallelHTTPSServer if use_ssl else ParallelHTTPServer	15✔
222	application = Application(configuration)	15✔
223	servers = {}	15✔
224	try:	15✔
225	for address in configuration.get("server", "hosts"):	15✔
226	# Try to bind sockets for IPv4 and IPv6
227	possible_families = (socket.AF_INET, socket.AF_INET6)	15✔
228	bind_ok = False	15✔
229	for i, family in enumerate(possible_families):	15✔
230	is_last = i == len(possible_families) - 1	15✔
231	try:	15✔
232	server = server_class(configuration, family, address,	15✔
233	RequestHandler)
234	except OSError as e:	15✔
235	# Ignore unsupported families (only one must work)
236	if ((bind_ok or not is_last) and (	15!
237	isinstance(e, socket.gaierror) and (
238	# Hostname does not exist or doesn't have
239	# address for address family
240	# macOS: IPv6 address for INET address family
241	e.errno == socket.EAI_NONAME or
242	# Address not for address family
243	e.errno == COMPAT_EAI_ADDRFAMILY or
244	e.errno == COMPAT_EAI_NODATA) or
245	# Workaround for PyPy
246	str(e) == "address family mismatched" or
247	# Address family not available (e.g. IPv6 disabled)
248	# macOS: IPv4 address for INET6 address family with
249	# IPV6_V6ONLY set
250	e.errno == errno.EADDRNOTAVAIL or
251	# Address family not supported
252	e.errno == errno.EAFNOSUPPORT or
253	# Protocol not supported
254	e.errno == errno.EPROTONOSUPPORT)):
255	continue	15✔
256	raise RuntimeError("Failed to start server %r: %s" % (	×
257	format_address(address), e)) from e
258	servers[server.socket] = server	15✔
259	bind_ok = True	15✔
260	server.set_app(application)	15✔
261	logger.info("Listening on %r%s",	15✔
262	format_address(server.server_address),
263	" with SSL" if use_ssl else "")
264	if not servers:	15!
265	raise RuntimeError("No servers started")	×
266
267	# Mainloop
268	select_timeout = None	15✔
269	if os.name == "nt":	15!
270	# Fallback to busy waiting. (select(...) blocks SIGINT on Windows.)
271	select_timeout = 1.0	×
272	max_connections = configuration.get("server", "max_connections")	15✔
273	logger.info("Radicale server ready")	15✔
274	while True:	9✔
275	rlist = []	15✔
276	# Wait for finished clients
277	for server in servers.values():	15✔
278	rlist.extend(server.client_sockets)	15✔
279	# Accept new connections if max_connections is not reached
280	if max_connections <= 0 or len(rlist) < max_connections:	15!
281	rlist.extend(servers)	15✔
282	# Use socket to get notified of program shutdown
283	if shutdown_socket is not None:	15!
284	rlist.append(shutdown_socket)	15✔
285	rlist, _, _ = select.select(rlist, [], [], select_timeout)	15✔
286	rlist = set(rlist)	15✔
287	if shutdown_socket in rlist:	15✔
288	logger.info("Stopping Radicale")	15✔
289	break	15✔
290	for server in servers.values():	15✔
291	finished_sockets = server.client_sockets.intersection(rlist)	15✔
292	for s in finished_sockets:	15✔
293	s.close()	15✔
294	server.client_sockets.remove(s)	15✔
295	rlist.remove(s)	15✔
296	if finished_sockets:	15✔
297	server.service_actions()	15✔
298	if rlist:	15✔
299	server = servers.get(rlist.pop())	15✔
300	if server:	15!
301	server.handle_request()	15✔
302	finally:
303	# Wait for clients to finish and close servers
304	for server in servers.values():	15✔
305	for s in server.client_sockets:	15!
UNCOV 306	s.recv(1)	×
UNCOV 307	s.close()	×
308	server.server_close()	15✔

Kozea / Radicale / 324815672

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