8271917321

Committed 13 Mar 2024 09:40PM UTC coverage: 83.493% (-0.07%) from 83.565%

Build # 8271917321

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1435 from pbiering/fix-1313

Fix 1313 by rework server listen code

Run Details

1559 of 1996 branches covered (78.11%)

Branch coverage included in aggregate %.

7 of 12 new or added lines in 1 file covered. (58.33%)

1 existing line in 1 file now uncovered.

3489 of 4050 relevant lines covered (86.15%)

20.04 hits per line

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

79.93

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

"""
Built-in WSGI server.

"""

import http
import select
import socket
import socketserver
import ssl
import sys
import wsgiref.simple_server
from typing import (Any, Callable, Dict, List, MutableMapping, Optional, Set,
                    Tuple, Union)
from urllib.parse import unquote

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

COMPAT_EAI_ADDRFAMILY: int
if hasattr(socket, "EAI_ADDRFAMILY"):
    COMPAT_EAI_ADDRFAMILY = socket.EAI_ADDRFAMILY  # type:ignore[attr-defined]
elif hasattr(socket, "EAI_NONAME"):
    # Windows and BSD don't have a special error code for this
    COMPAT_EAI_ADDRFAMILY = socket.EAI_NONAME
COMPAT_EAI_NODATA: int
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
COMPAT_IPPROTO_IPV6: int
if hasattr(socket, "IPPROTO_IPV6"):
    COMPAT_IPPROTO_IPV6 = socket.IPPROTO_IPV6
elif sys.platform == "win32":
    # HACK: https://bugs.python.org/issue29515
    COMPAT_IPPROTO_IPV6 = 41


# IPv4 (host, port) and IPv6 (host, port, flowinfo, scopeid)
ADDRESS_TYPE = Union[Tuple[Union[str, bytes, bytearray], int],
                     Tuple[str, int, int, int]]


def format_address(address: ADDRESS_TYPE) -> str:
    host, port, *_ = address
    if not isinstance(host, str):
        raise NotImplementedError("Unsupported address format: %r" %
                                  (address,))
    return "[%s]:%d" % (host, port)


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

    configuration: config.Configuration
    worker_sockets: Set[socket.socket]
    _timeout: float

    # We wait for child threads ourself (ThreadingMixIn)
    block_on_close: bool = False
    daemon_threads: bool = True

    def __init__(self, configuration: config.Configuration, family: int,
                 address: Tuple[str, int], RequestHandlerClass:
                 Callable[..., http.server.BaseHTTPRequestHandler]) -> None:
        self.configuration = configuration
        self.address_family = family
        super().__init__(address, RequestHandlerClass)
        self.worker_sockets = set()
        self._timeout = configuration.get("server", "timeout")

    def server_bind(self) -> None:
        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(  # type:ignore[override]
            self) -> Tuple[socket.socket, Tuple[ADDRESS_TYPE, socket.socket]]:
        # Set timeout for client
        request: socket.socket
        client_address: ADDRESS_TYPE
        request, client_address = super().get_request()  # type:ignore[misc]
        if self._timeout > 0:
            request.settimeout(self._timeout)
        worker_socket, worker_socket_out = socket.socketpair()
        self.worker_sockets.add(worker_socket_out)
        # HACK: Forward `worker_socket` via `client_address` return value
        # to worker thread.
        # The super class calls `verify_request`, `process_request` and
        # `handle_error` with modified `client_address` value.
        return request, (client_address, worker_socket)

    def verify_request(  # type:ignore[override]
            self, request: socket.socket, client_address_and_socket:
            Tuple[ADDRESS_TYPE, socket.socket]) -> bool:
        return True

    def process_request(  # type:ignore[override]
            self, request: socket.socket, client_address_and_socket:
            Tuple[ADDRESS_TYPE, socket.socket]) -> None:
        # HACK: Super class calls `finish_request` in new thread with
        # `client_address_and_socket`
        return super().process_request(
            request, client_address_and_socket)  # type:ignore[arg-type]

    def finish_request(  # type:ignore[override]
            self, request: socket.socket, client_address_and_socket:
            Tuple[ADDRESS_TYPE, socket.socket]) -> None:
        # HACK: Unpack `client_address_and_socket` and call super class
        # `finish_request` with original `client_address`
        client_address, worker_socket = client_address_and_socket
        try:
            return self.finish_request_locked(request, client_address)
        finally:
            worker_socket.close()

    def finish_request_locked(self, request: socket.socket,
                              client_address: ADDRESS_TYPE) -> None:
        return super().finish_request(
            request, client_address)  # type:ignore[arg-type]

    def handle_error(  # type:ignore[override]
            self, request: socket.socket,
            client_address_or_client_address_and_socket:
            Union[ADDRESS_TYPE, Tuple[ADDRESS_TYPE, socket.socket]]) -> None:
        # HACK: This method can be called with the modified
        # `client_address_and_socket` or the original `client_address` value
        e = sys.exc_info()[1]
        assert e is not None
        if isinstance(e, 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) -> None:
        super().server_bind()
        # Wrap the TCP socket in an SSL socket
        certfile: str = self.configuration.get("server", "certificate")
        keyfile: str = self.configuration.get("server", "key")
        cafile: str = 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(  # type:ignore[override]
            self, request: ssl.SSLSocket, client_address: ADDRESS_TYPE
            ) -> None:
        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)  # type:ignore[attr-defined]
            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: MutableMapping[str, str] = {}

    def log_exception(self, exc_info) -> None:
        logger.error("An exception occurred during request: %s",
                     exc_info[1], exc_info=exc_info)  # type:ignore[arg-type]


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

    # HACK: Assigned in `socketserver.StreamRequestHandler`
    connection: socket.socket

    def log_request(self, code: Union[int, str] = "-",
                    size: Union[int, str] = "-") -> None:
        pass  # Disable request logging.

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

    def get_environ(self) -> Dict[str, Any]:
        env = super().get_environ()
        if isinstance(self.connection, ssl.SSLSocket):
            # 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) -> None:
        """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  # type:ignore[attr-defined]
        app = self.server.get_app()  # type:ignore[attr-defined]
        handler.run(app)


def serve(configuration: config.Configuration,
          shutdown_socket: Optional[socket.socket] = None) -> 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: bool = configuration.get("server", "ssl")
    server_class = ParallelHTTPSServer if use_ssl else ParallelHTTPServer
    application = Application(configuration)
    servers = {}
    try:
        hosts: List[Tuple[str, int]] = configuration.get("server", "hosts")
        for AddressPort in hosts:
            # retrieve IPv4/IPv6 address of address
            try:
                getaddrinfo = socket.getaddrinfo(AddressPort[0], AddressPort[1], 0, socket.SOCK_STREAM, socket.IPPROTO_TCP)
            except OSError as e:
                logger.warn("cannot retrieve IPv4 or IPv6 address of '%s': %s" % (format_address(AddressPort), e))
                continue
            logger.debug("getaddrinfo of '%s': %s" % (format_address(AddressPort), getaddrinfo))
            for (AddressFamily, SocketKind, SocketProto, SocketFlags, SocketAddress) in getaddrinfo:
                logger.debug("try to create server socket on '%s'" % (format_address(SocketAddress)))
                try:
                    server = server_class(configuration, AddressFamily, (SocketAddress[0], SocketAddress[1]), RequestHandler)
                except OSError as e:
                    logger.warn("cannot create server socket on '%s': %s" % (format_address(SocketAddress), e))
                    continue
                servers[server.socket] = server
                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 sys.platform == "win32":
            # Fallback to busy waiting. (select(...) blocks SIGINT on Windows.)
            select_timeout = 1.0
        max_connections: int = configuration.get("server", "max_connections")
        logger.info("Radicale server ready")
        while True:
            rlist: List[socket.socket] = []
            # Wait for finished clients
            for server in servers.values():
                rlist.extend(server.worker_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)
            rset = set(rlist)
            if shutdown_socket in rset:
                logger.info("Stopping Radicale")
                break
            for server in servers.values():
                finished_sockets = server.worker_sockets.intersection(rset)
                for s in finished_sockets:
                    s.close()
                    server.worker_sockets.remove(s)
                    rset.remove(s)
                if finished_sockets:
                    server.service_actions()
            if rset:
                active_server = servers.get(rset.pop())
                if active_server:
                    active_server.handle_request()
    finally:
        # Wait for clients to finish and close servers
        for server in servers.values():
            for s in server.worker_sockets:
                s.recv(1)
                s.close()
            server.server_close()

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	"""	24✔
22	Built-in WSGI server.
23
24	"""
25
26	import http	24✔
27	import select	24✔
28	import socket	24✔
29	import socketserver	24✔
30	import ssl	24✔
31	import sys	24✔
32	import wsgiref.simple_server	24✔
33	from typing import (Any, Callable, Dict, List, MutableMapping, Optional, Set,	24✔
34	Tuple, Union)
35	from urllib.parse import unquote	24✔
36
37	from radicale import Application, config	24✔
38	from radicale.log import logger	24✔
39
40	COMPAT_EAI_ADDRFAMILY: int	24✔
41	if hasattr(socket, "EAI_ADDRFAMILY"):	24✔
42	COMPAT_EAI_ADDRFAMILY = socket.EAI_ADDRFAMILY # type:ignore[attr-defined]	18✔
43	elif hasattr(socket, "EAI_NONAME"):	6!
44	# Windows and BSD don't have a special error code for this
45	COMPAT_EAI_ADDRFAMILY = socket.EAI_NONAME	6✔
46	COMPAT_EAI_NODATA: int	24✔
47	if hasattr(socket, "EAI_NODATA"):	24!
48	COMPAT_EAI_NODATA = socket.EAI_NODATA	24✔
49	elif hasattr(socket, "EAI_NONAME"):	×
50	# Windows and BSD don't have a special error code for this
51	COMPAT_EAI_NODATA = socket.EAI_NONAME	×
52	COMPAT_IPPROTO_IPV6: int	24✔
53	if hasattr(socket, "IPPROTO_IPV6"):	24✔
54	COMPAT_IPPROTO_IPV6 = socket.IPPROTO_IPV6	23✔
55	elif sys.platform == "win32":	1!
56	# HACK: https://bugs.python.org/issue29515
57	COMPAT_IPPROTO_IPV6 = 41	1✔
58
59
60	# IPv4 (host, port) and IPv6 (host, port, flowinfo, scopeid)
61	ADDRESS_TYPE = Union[Tuple[Union[str, bytes, bytearray], int],	24✔
62	Tuple[str, int, int, int]]
63
64
65	def format_address(address: ADDRESS_TYPE) -> str:	24✔
66	host, port, *_ = address	24✔
67	if not isinstance(host, str):	24✔
68	raise NotImplementedError("Unsupported address format: %r" %
69	(address,))
70	return "[%s]:%d" % (host, port)	24✔
71
72
73	class ParallelHTTPServer(socketserver.ThreadingMixIn,	24✔
74	wsgiref.simple_server.WSGIServer):
75
76	configuration: config.Configuration	24✔
77	worker_sockets: Set[socket.socket]	24✔
78	_timeout: float	24✔
79
80	# We wait for child threads ourself (ThreadingMixIn)
81	block_on_close: bool = False	24✔
82	daemon_threads: bool = True	24✔
83
84	def __init__(self, configuration: config.Configuration, family: int,	24✔
85	address: Tuple[str, int], RequestHandlerClass:
86	Callable[..., http.server.BaseHTTPRequestHandler]) -> None:
87	self.configuration = configuration	24✔
88	self.address_family = family	24✔
89	super().__init__(address, RequestHandlerClass)	24✔
90	self.worker_sockets = set()	24✔
91	self._timeout = configuration.get("server", "timeout")	24✔
92
93	def server_bind(self) -> None:	24✔
94	if self.address_family == socket.AF_INET6:	24✔
95	# Only allow IPv6 connections to the IPv6 socket
96	self.socket.setsockopt(COMPAT_IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)	24✔
97	super().server_bind()	24✔
98
99	def get_request( # type:ignore[override]	24✔
100	self) -> Tuple[socket.socket, Tuple[ADDRESS_TYPE, socket.socket]]:
101	# Set timeout for client
102	request: socket.socket
103	client_address: ADDRESS_TYPE
104	request, client_address = super().get_request() # type:ignore[misc]	24✔
105	if self._timeout > 0:	24!
106	request.settimeout(self._timeout)	24✔
107	worker_socket, worker_socket_out = socket.socketpair()	24✔
108	self.worker_sockets.add(worker_socket_out)	24✔
109	# HACK: Forward `worker_socket` via `client_address` return value
110	# to worker thread.
111	# The super class calls `verify_request`, `process_request` and
112	# `handle_error` with modified `client_address` value.
113	return request, (client_address, worker_socket)	24✔
114
115	def verify_request( # type:ignore[override]	24✔
116	self, request: socket.socket, client_address_and_socket:
117	Tuple[ADDRESS_TYPE, socket.socket]) -> bool:
118	return True	24✔
119
120	def process_request( # type:ignore[override]	24✔
121	self, request: socket.socket, client_address_and_socket:
122	Tuple[ADDRESS_TYPE, socket.socket]) -> None:
123	# HACK: Super class calls `finish_request` in new thread with
124	# `client_address_and_socket`
125	return super().process_request(	24✔
126	request, client_address_and_socket) # type:ignore[arg-type]
127
128	def finish_request( # type:ignore[override]	24✔
129	self, request: socket.socket, client_address_and_socket:
130	Tuple[ADDRESS_TYPE, socket.socket]) -> None:
131	# HACK: Unpack `client_address_and_socket` and call super class
132	# `finish_request` with original `client_address`
133	client_address, worker_socket = client_address_and_socket	24✔
134	try:	24✔
135	return self.finish_request_locked(request, client_address)	24✔
136	finally:
137	worker_socket.close()	24✔
138
139	def finish_request_locked(self, request: socket.socket,	24✔
140	client_address: ADDRESS_TYPE) -> None:
141	return super().finish_request(	24✔
142	request, client_address) # type:ignore[arg-type]
143
144	def handle_error( # type:ignore[override]	24✔
145	self, request: socket.socket,
146	client_address_or_client_address_and_socket:
147	Union[ADDRESS_TYPE, Tuple[ADDRESS_TYPE, socket.socket]]) -> None:
148	# HACK: This method can be called with the modified
149	# `client_address_and_socket` or the original `client_address` value
150	e = sys.exc_info()[1]	×
151	assert e is not None	×
152	if isinstance(e, socket.timeout):	×
153	logger.info("Client timed out", exc_info=True)	×
154	else:
155	logger.error("An exception occurred during request: %s",	9✔
156	sys.exc_info()[1], exc_info=True)
157
158
159	class ParallelHTTPSServer(ParallelHTTPServer):	24✔
160
161	def server_bind(self) -> None:	24✔
162	super().server_bind()	24✔
163	# Wrap the TCP socket in an SSL socket
164	certfile: str = self.configuration.get("server", "certificate")	24✔
165	keyfile: str = self.configuration.get("server", "key")	24✔
166	cafile: str = self.configuration.get("server", "certificate_authority")	24✔
167	# Test if the files can be read
168	for name, filename in [("certificate", certfile), ("key", keyfile),	24✔
169	("certificate_authority", cafile)]:
170	type_name = config.DEFAULT_CONFIG_SCHEMA["server"][name][	24✔
171	"type"].__name__
172	source = self.configuration.get_source("server", name)	24✔
173	if name == "certificate_authority" and not filename:	24✔
174	continue	24✔
175	try:	24✔
176	open(filename, "r").close()	24✔
177	except OSError as e:	×
178	raise RuntimeError(	×
179	"Invalid %s value for option %r in section %r in %s: %r "
180	"(%s)" % (type_name, name, "server", source, filename,
181	e)) from e
182	context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)	24✔
183	context.load_cert_chain(certfile=certfile, keyfile=keyfile)	24✔
184	if cafile:	24!
185	context.load_verify_locations(cafile=cafile)	×
186	context.verify_mode = ssl.CERT_REQUIRED	×
187	self.socket = context.wrap_socket(	24✔
188	self.socket, server_side=True, do_handshake_on_connect=False)
189
190	def finish_request_locked( # type:ignore[override]	24✔
191	self, request: ssl.SSLSocket, client_address: ADDRESS_TYPE
192	) -> None:
193	try:	24✔
194	try:	24✔
195	request.do_handshake()	24✔
196	except socket.timeout:	×
197	raise	×
198	except Exception as e:	×
199	raise RuntimeError("SSL handshake failed: %s" % e) from e	×
200	except Exception:	×
201	try:	×
202	self.handle_error(request, client_address)	×
203	finally:
204	self.shutdown_request(request) # type:ignore[attr-defined]	×
205	return	×
206	return super().finish_request_locked(request, client_address)	24✔
207
208
209	class ServerHandler(wsgiref.simple_server.ServerHandler):	24✔
210
211	# Don't pollute WSGI environ with OS environment
212	os_environ: MutableMapping[str, str] = {}	24✔
213
214	def log_exception(self, exc_info) -> None:	24✔
215	logger.error("An exception occurred during request: %s",	9✔
216	exc_info[1], exc_info=exc_info) # type:ignore[arg-type]
217
218
219	class RequestHandler(wsgiref.simple_server.WSGIRequestHandler):	24✔
220	"""HTTP requests handler."""
221
222	# HACK: Assigned in `socketserver.StreamRequestHandler`
223	connection: socket.socket	24✔
224
225	def log_request(self, code: Union[int, str] = "-",	24✔
226	size: Union[int, str] = "-") -> None:
227	pass # Disable request logging.	24✔
228
229	def log_error(self, format_: str, *args: Any) -> None:	24✔
230	logger.error("An error occurred during request: %s", format_ % args)	×
231
232	def get_environ(self) -> Dict[str, Any]:	24✔
233	env = super().get_environ()	24✔
234	if isinstance(self.connection, ssl.SSLSocket):	24✔
235	# The certificate can be evaluated by the auth module
236	env["REMOTE_CERTIFICATE"] = self.connection.getpeercert()	24✔
237	# Parent class only tries latin1 encoding
238	env["PATH_INFO"] = unquote(self.path.split("?", 1)[0])	24✔
239	return env	24✔
240
241	def handle(self) -> None:	24✔
242	"""Copy of WSGIRequestHandler.handle with different ServerHandler"""
243
244	self.raw_requestline = self.rfile.readline(65537)	24✔
245	if len(self.raw_requestline) > 65536:	24!
246	self.requestline = ""	×
247	self.request_version = ""	×
248	self.command = ""	×
249	self.send_error(414)	×
250	return	×
251
252	if not self.parse_request():	24!
253	return	×
254
255	handler = ServerHandler(	24✔
256	self.rfile, self.wfile, self.get_stderr(), self.get_environ()
257	)
258	handler.request_handler = self # type:ignore[attr-defined]	24✔
259	app = self.server.get_app() # type:ignore[attr-defined]	24✔
260	handler.run(app)	24✔
261
262
263	def serve(configuration: config.Configuration,	24✔
264	shutdown_socket: Optional[socket.socket] = None) -> None:
265	"""Serve radicale from configuration.
266
267	`shutdown_socket` can be used to gracefully shutdown the server.
268	The socket can be created with `socket.socketpair()`, when the other socket
269	gets closed the server stops accepting new requests by clients and the
270	function returns after all active requests are finished.
271
272	"""
273
274	logger.info("Starting Radicale")	24✔
275	# Copy configuration before modifying
276	configuration = configuration.copy()	24✔
277	configuration.update({"server": {"_internal_server": "True"}}, "server",	24✔
278	privileged=True)
279
280	use_ssl: bool = configuration.get("server", "ssl")	24✔
281	server_class = ParallelHTTPSServer if use_ssl else ParallelHTTPServer	24✔
282	application = Application(configuration)	24✔
283	servers = {}	24✔
284	try:	24✔
285	hosts: List[Tuple[str, int]] = configuration.get("server", "hosts")	24✔
286	for AddressPort in hosts:	24✔
287	# retrieve IPv4/IPv6 address of address
288	try:	24✔
289	getaddrinfo = socket.getaddrinfo(AddressPort[0], AddressPort[1], 0, socket.SOCK_STREAM, socket.IPPROTO_TCP)	24✔
NEW 290	except OSError as e:	×
NEW 291	logger.warn("cannot retrieve IPv4 or IPv6 address of '%s': %s" % (format_address(AddressPort), e))	×
NEW 292	continue	×
293	logger.debug("getaddrinfo of '%s': %s" % (format_address(AddressPort), getaddrinfo))	24✔
294	for (AddressFamily, SocketKind, SocketProto, SocketFlags, SocketAddress) in getaddrinfo:	24✔
295	logger.debug("try to create server socket on '%s'" % (format_address(SocketAddress)))	24✔
296	try:	24✔
297	server = server_class(configuration, AddressFamily, (SocketAddress[0], SocketAddress[1]), RequestHandler)	24✔
UNCOV 298	except OSError as e:	×
NEW 299	logger.warn("cannot create server socket on '%s': %s" % (format_address(SocketAddress), e))	×
NEW 300	continue	×
301	servers[server.socket] = server	24✔
302	server.set_app(application)	24✔
303	logger.info("Listening on %r%s",	24✔
304	format_address(server.server_address),
305	" with SSL" if use_ssl else "")
306	if not servers:	24!
307	raise RuntimeError("No servers started")	×
308
309	# Mainloop
310	select_timeout = None	24✔
311	if sys.platform == "win32":	24✔
312	# Fallback to busy waiting. (select(...) blocks SIGINT on Windows.)
313	select_timeout = 1.0	6✔
314	max_connections: int = configuration.get("server", "max_connections")	24✔
315	logger.info("Radicale server ready")	24✔
316	while True:	14✔
317	rlist: List[socket.socket] = []	24✔
318	# Wait for finished clients
319	for server in servers.values():	24✔
320	rlist.extend(server.worker_sockets)	24✔
321	# Accept new connections if max_connections is not reached
322	if max_connections <= 0 or len(rlist) < max_connections:	24!
323	rlist.extend(servers)	24✔
324	# Use socket to get notified of program shutdown
325	if shutdown_socket is not None:	24!
326	rlist.append(shutdown_socket)	24✔
327	rlist, _, _ = select.select(rlist, [], [], select_timeout)	24✔
328	rset = set(rlist)	24✔
329	if shutdown_socket in rset:	24✔
330	logger.info("Stopping Radicale")	24✔
331	break	24✔
332	for server in servers.values():	24✔
333	finished_sockets = server.worker_sockets.intersection(rset)	24✔
334	for s in finished_sockets:	24✔
335	s.close()	24✔
336	server.worker_sockets.remove(s)	24✔
337	rset.remove(s)	24✔
338	if finished_sockets:	24✔
339	server.service_actions()	24✔
340	if rset:	24✔
341	active_server = servers.get(rset.pop())	24✔
342	if active_server:	24!
343	active_server.handle_request()	24✔
344	finally:
345	# Wait for clients to finish and close servers
346	for server in servers.values():	24✔
347	for s in server.worker_sockets:	24!
348	s.recv(1)	×
349	s.close()	×
350	server.server_close()	24✔

Kozea / Radicale / 8271917321

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