9422153004

Committed 07 Jun 2024 07:27PM UTC coverage: 80.834% (-0.1%) from 80.973%

Build # 9422153004

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1512 from pbiering/warn-default-config

Warn about default config and no active user authentication

Run Details

1668 of 2277 branches covered (73.25%)

Branch coverage included in aggregate %.

10 of 17 new or added lines in 3 files covered. (58.82%)

2 existing lines in 1 file now uncovered.

3587 of 4224 relevant lines covered (84.92%)

15.71 hits per line

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

79.06

/radicale/__main__.py

# This file is part of Radicale - CalDAV and CardDAV server
# Copyright © 2011-2017 Guillaume Ayoub
# Copyright © 2017-2022 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/>.

"""
Radicale executable module.

This module can be executed from a command line with ``$python -m radicale``.
Uses the built-in WSGI server.

"""

import argparse
import contextlib
import os
import signal
import socket
import sys
from types import FrameType
from typing import List, Optional, cast

from radicale import VERSION, config, log, server, storage, types
from radicale.log import logger


def run() -> None:
    """Run Radicale as a standalone server."""
    exit_signal_numbers = [signal.SIGTERM, signal.SIGINT]
    if sys.platform == "win32":
        exit_signal_numbers.append(signal.SIGBREAK)
    else:
        exit_signal_numbers.append(signal.SIGHUP)
        exit_signal_numbers.append(signal.SIGQUIT)

    # Raise SystemExit when signal arrives to run cleanup code
    # (like destructors, try-finish etc.), otherwise the process exits
    # without running any of them
    def exit_signal_handler(signal_number: int,
                            stack_frame: Optional[FrameType]) -> None:
        sys.exit(1)
    for signal_number in exit_signal_numbers:
        signal.signal(signal_number, exit_signal_handler)

    log.setup()

    # Get command-line arguments
    # Configuration options are stored in dest with format "c:SECTION:OPTION"
    parser = argparse.ArgumentParser(
        prog="radicale", usage="%(prog)s [OPTIONS]", allow_abbrev=False)

    parser.add_argument("--version", action="version", version=VERSION)
    parser.add_argument("--verify-storage", action="store_true",
                        help="check the storage for errors and exit")
    parser.add_argument("-C", "--config",
                        help="use specific configuration files", nargs="*")
    parser.add_argument("-D", "--debug", action="store_const", const="debug",
                        dest="c:logging:level", default=argparse.SUPPRESS,
                        help="print debug information")

    for section, section_data in config.DEFAULT_CONFIG_SCHEMA.items():
        if section.startswith("_"):
            continue
        assert ":" not in section  # check field separator
        assert "-" not in section and "_" not in section  # not implemented
        group_description = None
        if section_data.get("_allow_extra"):
            group_description = "additional options allowed"
            if section == "headers":
                group_description += " (e.g. --headers-Pragma=no-cache)"
        elif "type" in section_data:
            group_description = "backend specific options omitted"
        group = parser.add_argument_group(section, group_description)
        for option, data in section_data.items():
            if option.startswith("_"):
                continue
            kwargs = data.copy()
            long_name = "--%s-%s" % (section, option.replace("_", "-"))
            args: List[str] = list(kwargs.pop("aliases", ()))
            args.append(long_name)
            kwargs["dest"] = "c:%s:%s" % (section, option)
            kwargs["metavar"] = "VALUE"
            kwargs["default"] = argparse.SUPPRESS
            del kwargs["value"]
            with contextlib.suppress(KeyError):
                del kwargs["internal"]

            if kwargs["type"] == bool:
                del kwargs["type"]
                opposite_args = list(kwargs.pop("opposite_aliases", ()))
                opposite_args.append("--no%s" % long_name[1:])
                group.add_argument(*args, nargs="?", const="True", **kwargs)
                # Opposite argument
                kwargs["help"] = "do not %s (opposite of %s)" % (
                    kwargs["help"], long_name)
                group.add_argument(*opposite_args, action="store_const",
                                   const="False", **kwargs)
            else:
                del kwargs["type"]
                group.add_argument(*args, **kwargs)

    args_ns, remaining_args = parser.parse_known_args()
    unrecognized_args = []
    while remaining_args:
        arg = remaining_args.pop(0)
        for section, data in config.DEFAULT_CONFIG_SCHEMA.items():
            if "type" not in data and not data.get("_allow_extra"):
                continue
            prefix = "--%s-" % section
            if arg.startswith(prefix):
                arg = arg[len(prefix):]
                break
        else:
            unrecognized_args.append(arg)
            continue
        value = ""
        if "=" in arg:
            arg, value = arg.split("=", maxsplit=1)
        elif remaining_args and not remaining_args[0].startswith("-"):
            value = remaining_args.pop(0)
        option = arg
        if not data.get("_allow_extra"):  # preserve dash in HTTP header names
            option = option.replace("-", "_")
        vars(args_ns)["c:%s:%s" % (section, option)] = value
    if unrecognized_args:
        parser.error("unrecognized arguments: %s" %
                     " ".join(unrecognized_args))

    # Preliminary configure logging
    with contextlib.suppress(ValueError):
        log.set_level(config.DEFAULT_CONFIG_SCHEMA["logging"]["level"]["type"](
            vars(args_ns).get("c:logging:level", "")))

    # Update Radicale configuration according to arguments
    arguments_config: types.MUTABLE_CONFIG = {}
    for key, value in vars(args_ns).items():
        if key.startswith("c:"):
            _, section, option = key.split(":", maxsplit=2)
            arguments_config[section] = arguments_config.get(section, {})
            arguments_config[section][option] = value

    try:
        configuration = config.load(config.parse_compound_paths(
            config.DEFAULT_CONFIG_PATH,
            os.environ.get("RADICALE_CONFIG"),
            os.pathsep.join(args_ns.config) if args_ns.config is not None
            else None))
        if arguments_config:
            configuration.update(arguments_config, "command line arguments")
    except Exception as e:
        logger.critical("Invalid configuration: %s", e, exc_info=True)
        sys.exit(1)

    # Configure logging
    log.set_level(cast(str, configuration.get("logging", "level")))

    # Log configuration after logger is configured
    default_config_active = True
    for source, miss in configuration.sources():
        logger.info("%s %s", "Skipped missing/unreadable" if miss else "Loaded", source)
        if not miss and source != "default config":
            default_config_active = False

    if default_config_active:
        logger.warn("%s", "No config file found/readable - only default config is active")

    if args_ns.verify_storage:
        logger.info("Verifying storage")
        try:
            storage_ = storage.load(configuration)
            with storage_.acquire_lock("r"):
                if not storage_.verify():
                    logger.critical("Storage verifcation failed")
                    sys.exit(1)
        except Exception as e:
            logger.critical("An exception occurred during storage "
                            "verification: %s", e, exc_info=True)
            sys.exit(1)
        return

    # Create a socket pair to notify the server of program shutdown
    shutdown_socket, shutdown_socket_out = socket.socketpair()

    # Shutdown server when signal arrives
    def shutdown_signal_handler(signal_number: int,
                                stack_frame: Optional[FrameType]) -> None:
        shutdown_socket.close()
    for signal_number in exit_signal_numbers:
        signal.signal(signal_number, shutdown_signal_handler)

    try:
        server.serve(configuration, shutdown_socket_out)
    except Exception as e:
        logger.critical("An exception occurred during server startup: %s", e,
                        exc_info=False)
        sys.exit(1)


if __name__ == "__main__":
    run()

1	# This file is part of Radicale - CalDAV and CardDAV server
2	# Copyright © 2011-2017 Guillaume Ayoub
3	# Copyright © 2017-2022 Unrud <unrud@outlook.com>
4	# Copyright © 2024-2024 Peter Bieringer <pb@bieringer.de>
5	#
6	# This library is free software: you can redistribute it and/or modify
7	# it under the terms of the GNU General Public License as published by
8	# the Free Software Foundation, either version 3 of the License, or
9	# (at your option) any later version.
10	#
11	# This library is distributed in the hope that it will be useful,
12	# but WITHOUT ANY WARRANTY; without even the implied warranty of
13	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14	# GNU General Public License for more details.
15	#
16	# You should have received a copy of the GNU General Public License
17	# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
18
19	"""
20	Radicale executable module.
21
22	This module can be executed from a command line with ``$python -m radicale``.
23	Uses the built-in WSGI server.
24
25	"""
26
27	import argparse	14✔
28	import contextlib	14✔
29	import os	14✔
30	import signal	14✔
31	import socket	14✔
32	import sys	14✔
33	from types import FrameType	14✔
34	from typing import List, Optional, cast	14✔
35
36	from radicale import VERSION, config, log, server, storage, types	14✔
37	from radicale.log import logger	14✔
38
39
40	def run() -> None:	14✔
41	"""Run Radicale as a standalone server."""
42	exit_signal_numbers = [signal.SIGTERM, signal.SIGINT]	14✔
43	if sys.platform == "win32":	14!
44	exit_signal_numbers.append(signal.SIGBREAK)	×
45	else:
46	exit_signal_numbers.append(signal.SIGHUP)	14✔
47	exit_signal_numbers.append(signal.SIGQUIT)	14✔
48
49	# Raise SystemExit when signal arrives to run cleanup code
50	# (like destructors, try-finish etc.), otherwise the process exits
51	# without running any of them
52	def exit_signal_handler(signal_number: int,	14✔
53	stack_frame: Optional[FrameType]) -> None:
54	sys.exit(1)	×
55	for signal_number in exit_signal_numbers:	14✔
56	signal.signal(signal_number, exit_signal_handler)	14✔
57
58	log.setup()	14✔
59
60	# Get command-line arguments
61	# Configuration options are stored in dest with format "c:SECTION:OPTION"
62	parser = argparse.ArgumentParser(	14✔
63	prog="radicale", usage="%(prog)s [OPTIONS]", allow_abbrev=False)
64
65	parser.add_argument("--version", action="version", version=VERSION)	14✔
66	parser.add_argument("--verify-storage", action="store_true",	14✔
67	help="check the storage for errors and exit")
68	parser.add_argument("-C", "--config",	14✔
69	help="use specific configuration files", nargs="*")
70	parser.add_argument("-D", "--debug", action="store_const", const="debug",	14✔
71	dest="c:logging:level", default=argparse.SUPPRESS,
72	help="print debug information")
73
74	for section, section_data in config.DEFAULT_CONFIG_SCHEMA.items():	14✔
75	if section.startswith("_"):	14!
76	continue	×
77	assert ":" not in section # check field separator	14✔
78	assert "-" not in section and "_" not in section # not implemented	14✔
79	group_description = None	14✔
80	if section_data.get("_allow_extra"):	14✔
81	group_description = "additional options allowed"	14✔
82	if section == "headers":	14!
83	group_description += " (e.g. --headers-Pragma=no-cache)"	14✔
84	elif "type" in section_data:	14✔
85	group_description = "backend specific options omitted"	14✔
86	group = parser.add_argument_group(section, group_description)	14✔
87	for option, data in section_data.items():	14✔
88	if option.startswith("_"):	14✔
89	continue	14✔
90	kwargs = data.copy()	14✔
91	long_name = "--%s-%s" % (section, option.replace("_", "-"))	14✔
92	args: List[str] = list(kwargs.pop("aliases", ()))	14✔
93	args.append(long_name)	14✔
94	kwargs["dest"] = "c:%s:%s" % (section, option)	14✔
95	kwargs["metavar"] = "VALUE"	14✔
96	kwargs["default"] = argparse.SUPPRESS	14✔
97	del kwargs["value"]	14✔
98	with contextlib.suppress(KeyError):	14✔
99	del kwargs["internal"]	14✔
100
101	if kwargs["type"] == bool:	14✔
102	del kwargs["type"]	14✔
103	opposite_args = list(kwargs.pop("opposite_aliases", ()))	14✔
104	opposite_args.append("--no%s" % long_name[1:])	14✔
105	group.add_argument(args, nargs="?", const="True", *kwargs)	14✔
106	# Opposite argument
107	kwargs["help"] = "do not %s (opposite of %s)" % (	14✔
108	kwargs["help"], long_name)
109	group.add_argument(*opposite_args, action="store_const",	14✔
110	const="False", **kwargs)
111	else:
112	del kwargs["type"]	14✔
113	group.add_argument(args, *kwargs)	14✔
114
115	args_ns, remaining_args = parser.parse_known_args()	14✔
116	unrecognized_args = []	14✔
117	while remaining_args:	14✔
118	arg = remaining_args.pop(0)	14✔
119	for section, data in config.DEFAULT_CONFIG_SCHEMA.items():	14!
120	if "type" not in data and not data.get("_allow_extra"):	14✔
121	continue	14✔
122	prefix = "--%s-" % section	14✔
123	if arg.startswith(prefix):	14✔
124	arg = arg[len(prefix):]	14✔
125	break	14✔
126	else:
127	unrecognized_args.append(arg)	×
128	continue	×
129	value = ""	14✔
130	if "=" in arg:	14✔
131	arg, value = arg.split("=", maxsplit=1)	14✔
132	elif remaining_args and not remaining_args[0].startswith("-"):	14!
133	value = remaining_args.pop(0)	14✔
134	option = arg	14✔
135	if not data.get("_allow_extra"): # preserve dash in HTTP header names	14!
136	option = option.replace("-", "_")	×
137	vars(args_ns)["c:%s:%s" % (section, option)] = value	14✔
138	if unrecognized_args:	14!
139	parser.error("unrecognized arguments: %s" %	×
140	" ".join(unrecognized_args))
141
142	# Preliminary configure logging
143	with contextlib.suppress(ValueError):	14✔
144	log.set_level(config.DEFAULT_CONFIG_SCHEMA["logging"]["level"]["type"](	14✔
145	vars(args_ns).get("c:logging:level", "")))
146
147	# Update Radicale configuration according to arguments
148	arguments_config: types.MUTABLE_CONFIG = {}	14✔
149	for key, value in vars(args_ns).items():	14✔
150	if key.startswith("c:"):	14✔
151	_, section, option = key.split(":", maxsplit=2)	14✔
152	arguments_config[section] = arguments_config.get(section, {})	14✔
153	arguments_config[section][option] = value	14✔
154
155	try:	14✔
156	configuration = config.load(config.parse_compound_paths(	14✔
157	config.DEFAULT_CONFIG_PATH,
158	os.environ.get("RADICALE_CONFIG"),
159	os.pathsep.join(args_ns.config) if args_ns.config is not None
160	else None))
161	if arguments_config:	14!
162	configuration.update(arguments_config, "command line arguments")	14✔
163	except Exception as e:	×
164	logger.critical("Invalid configuration: %s", e, exc_info=True)	×
165	sys.exit(1)	×
166
167	# Configure logging
168	log.set_level(cast(str, configuration.get("logging", "level")))	14✔
169
170	# Log configuration after logger is configured
171	default_config_active = True	14✔
172	for source, miss in configuration.sources():	14✔
173	logger.info("%s %s", "Skipped missing/unreadable" if miss else "Loaded", source)	14✔
174	if not miss and source != "default config":	14✔
175	default_config_active = False	14✔
176
177	if default_config_active:	14!
NEW 178	logger.warn("%s", "No config file found/readable - only default config is active")	×
179
180	if args_ns.verify_storage:	14!
181	logger.info("Verifying storage")	×
182	try:	×
183	storage_ = storage.load(configuration)	×
184	with storage_.acquire_lock("r"):	×
185	if not storage_.verify():	×
186	logger.critical("Storage verifcation failed")	×
187	sys.exit(1)	×
188	except Exception as e:	×
189	logger.critical("An exception occurred during storage "	×
190	"verification: %s", e, exc_info=True)
191	sys.exit(1)	×
192	return	×
193
194	# Create a socket pair to notify the server of program shutdown
195	shutdown_socket, shutdown_socket_out = socket.socketpair()	14✔
196
197	# Shutdown server when signal arrives
198	def shutdown_signal_handler(signal_number: int,	14✔
199	stack_frame: Optional[FrameType]) -> None:
200	shutdown_socket.close()	14✔
201	for signal_number in exit_signal_numbers:	14✔
202	signal.signal(signal_number, shutdown_signal_handler)	14✔
203
204	try:	14✔
205	server.serve(configuration, shutdown_socket_out)	14✔
206	except Exception as e:	×
207	logger.critical("An exception occurred during server startup: %s", e,	×
208	exc_info=False)
209	sys.exit(1)	×
210
211
212	if __name__ == "__main__":
213	run()

Kozea / Radicale / 9422153004

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