Coverage for an_website/main.py: 81.070%
243 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-12-14 14:44 +0000
1# This program is free software: you can redistribute it and/or modify
2# it under the terms of the GNU Affero General Public License as
3# published by the Free Software Foundation, either version 3 of the
4# License, or (at your option) any later version.
5#
6# This program is distributed in the hope that it will be useful,
7# but WITHOUT ANY WARRANTY; without even the implied warranty of
8# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
9# GNU Affero General Public License for more details.
10#
11# You should have received a copy of the GNU Affero General Public License
12# along with this program. If not, see <https://www.gnu.org/licenses/>.
13# pylint: disable=import-private-name, too-many-lines
15"""
16The website of the AN.
18Loads config and modules and starts Tornado.
19"""
21from __future__ import annotations
23import asyncio
24import atexit
25import importlib
26import logging
27import os
28import platform
29import signal
30import ssl
31import sys
32import threading
33import time
34import types
35import uuid
36from asyncio import AbstractEventLoop
37from asyncio.runners import _cancel_all_tasks # type: ignore[attr-defined]
38from base64 import b64encode
39from collections.abc import Callable, Iterable, Mapping, MutableSequence
40from configparser import ConfigParser
41from functools import partial
42from hashlib import sha256
43from multiprocessing.process import _children # type: ignore[attr-defined]
44from pathlib import Path
45from socket import socket
46from typing import Any, Final, TypedDict, TypeGuard, cast
47from warnings import catch_warnings, simplefilter
48from zoneinfo import ZoneInfo
50import regex
51from Crypto.Hash import RIPEMD160
52from ecs_logging import StdlibFormatter
53from elasticapm.contrib.tornado import ElasticAPM
54from redis.asyncio import (
55 BlockingConnectionPool,
56 Redis,
57 SSLConnection,
58 UnixDomainSocketConnection,
59)
60from setproctitle import setproctitle
61from tornado.httpserver import HTTPServer
62from tornado.log import LogFormatter
63from tornado.netutil import bind_sockets, bind_unix_socket
64from tornado.process import fork_processes, task_id
65from tornado.web import Application, RedirectHandler
66from typed_stream import Stream
68from . import (
69 CA_BUNDLE_PATH,
70 DIR,
71 EVENT_SHUTDOWN,
72 NAME,
73 TEMPLATES_DIR,
74 UPTIME,
75 VERSION,
76 pytest_is_running,
77)
78from .contact.contact import apply_contact_stuff_to_app
79from .utils import background_tasks, static_file_handling
80from .utils.base_request_handler import BaseRequestHandler, request_ctx_var
81from .utils.better_config_parser import BetterConfigParser
82from .utils.elasticsearch_setup import setup_elasticsearch
83from .utils.logging import WebhookFormatter, WebhookHandler
84from .utils.request_handler import NotFoundHandler
85from .utils.static_file_from_traversable import TraversableStaticFileHandler
86from .utils.template_loader import TemplateLoader
87from .utils.utils import (
88 ArgparseNamespace,
89 Handler,
90 ModuleInfo,
91 Permission,
92 Timer,
93 create_argument_parser,
94 geoip,
95 get_arguments_without_help,
96 time_function,
97)
99try:
100 import perf8 # type: ignore[import, unused-ignore]
101except ModuleNotFoundError:
102 perf8 = None # pylint: disable=invalid-name
104IGNORED_MODULES: Final[set[str]] = {
105 "patches",
106 "static",
107 "templates",
108} | (set() if sys.flags.dev_mode or pytest_is_running() else {"example"})
110LOGGER: Final = logging.getLogger(__name__)
113# add all the information from the packages to a list
114# this calls the get_module_info function in every file
115# files and dirs starting with '_' get ignored
116def get_module_infos() -> str | tuple[ModuleInfo, ...]:
117 """Import the modules and return the loaded module infos in a tuple."""
118 module_infos: list[ModuleInfo] = []
119 loaded_modules: list[str] = []
120 errors: list[str] = []
122 for potential_module in DIR.iterdir():
123 if (
124 potential_module.name.startswith("_")
125 or potential_module.name in IGNORED_MODULES
126 or not potential_module.is_dir()
127 ):
128 continue
130 _module_infos = get_module_infos_from_module(
131 potential_module.name, errors, ignore_not_found=True
132 )
133 if _module_infos:
134 module_infos.extend(_module_infos)
135 loaded_modules.append(potential_module.name)
136 LOGGER.debug(
137 (
138 "Found module_infos in %s.__init__.py, "
139 "not searching in other modules in the package."
140 ),
141 potential_module,
142 )
143 continue
145 if f"{potential_module.name}.*" in IGNORED_MODULES:
146 continue
148 for potential_file in potential_module.iterdir():
149 module_name = f"{potential_module.name}.{potential_file.name[:-3]}"
150 if (
151 not potential_file.name.endswith(".py")
152 or module_name in IGNORED_MODULES
153 or potential_file.name.startswith("_")
154 ):
155 continue
156 _module_infos = get_module_infos_from_module(module_name, errors)
157 if _module_infos:
158 module_infos.extend(_module_infos)
159 loaded_modules.append(module_name)
161 if len(errors) > 0:
162 if sys.flags.dev_mode:
163 # exit to make sure it gets fixed
164 return "\n".join(errors)
165 # don't exit in production to keep stuff running
166 LOGGER.error("\n".join(errors))
168 LOGGER.info(
169 "Loaded %d modules: '%s'",
170 len(loaded_modules),
171 "', '".join(loaded_modules),
172 )
174 LOGGER.info(
175 "Ignored %d modules: '%s'",
176 len(IGNORED_MODULES),
177 "', '".join(IGNORED_MODULES),
178 )
180 sort_module_infos(module_infos)
182 # make module_infos immutable so it never changes
183 return tuple(module_infos)
186def get_module_infos_from_module(
187 module_name: str,
188 errors: MutableSequence[str], # gets modified
189 ignore_not_found: bool = False,
190) -> None | list[ModuleInfo]:
191 """Get the module infos based on a module."""
192 import_timer = Timer()
193 module = importlib.import_module(
194 f".{module_name}",
195 package="an_website",
196 )
197 if import_timer.stop() > 0.1:
198 LOGGER.warning(
199 "Import of %s took %ss. That's affecting the startup time.",
200 module_name,
201 import_timer.get(),
202 )
204 module_infos: list[ModuleInfo] = []
206 has_get_module_info = "get_module_info" in dir(module)
207 has_get_module_infos = "get_module_infos" in dir(module)
209 if not (has_get_module_info or has_get_module_infos):
210 if ignore_not_found:
211 return None
212 errors.append(
213 f"{module_name} has no 'get_module_info' and no 'get_module_infos' "
214 "method. Please add at least one of the methods or add "
215 f"'{module_name.rsplit('.', 1)[0]}.*' or {module_name!r} to "
216 "IGNORED_MODULES."
217 )
218 return None
220 if has_get_module_info and isinstance(
221 module_info := module.get_module_info(),
222 ModuleInfo,
223 ):
224 module_infos.append(module_info)
225 elif has_get_module_info:
226 errors.append(
227 f"'get_module_info' in {module_name} does not return ModuleInfo. "
228 "Please fix the returned value."
229 )
231 if not has_get_module_infos:
232 return module_infos or None
234 _module_infos = module.get_module_infos()
236 if not isinstance(_module_infos, Iterable):
237 errors.append(
238 f"'get_module_infos' in {module_name} does not return an Iterable. "
239 "Please fix the returned value."
240 )
241 return module_infos or None
243 for _module_info in _module_infos:
244 if isinstance(_module_info, ModuleInfo):
245 module_infos.append(_module_info)
246 else:
247 errors.append(
248 f"'get_module_infos' in {module_name} did return an Iterable "
249 f"with an element of type {type(_module_info)}. "
250 "Please fix the returned value."
251 )
253 return module_infos or None
256def sort_module_infos(module_infos: list[ModuleInfo]) -> None:
257 """Sort a list of module info and move the main page to the top."""
258 # sort it so the order makes sense
259 module_infos.sort()
261 # move the main page to the top
262 for i, info in enumerate(module_infos):
263 if info.path == "/":
264 module_infos.insert(0, module_infos.pop(i))
265 break
268def get_all_handlers(module_infos: Iterable[ModuleInfo]) -> list[Handler]:
269 """
270 Parse the module information and return the handlers in a tuple.
272 If a handler has only 2 elements a dict with title and description
273 gets added. This information is gotten from the module info.
274 """
275 handler: Handler | list[Any]
276 handlers: list[Handler] = static_file_handling.get_handlers()
278 # add all the normal handlers
279 for module_info in module_infos:
280 for handler in module_info.handlers:
281 handler = list(handler) # pylint: disable=redefined-loop-name
282 # if the handler is a request handler from us
283 # and not a built-in like StaticFileHandler & RedirectHandler
284 if issubclass(handler[1], BaseRequestHandler):
285 if len(handler) == 2:
286 # set "default_title" or "default_description" to False so
287 # that module_info.name & module_info.description get used
288 handler.append(
289 {
290 "default_title": False,
291 "default_description": False,
292 "module_info": module_info,
293 }
294 )
295 else:
296 handler[2]["module_info"] = module_info
297 handlers.append(tuple(handler))
299 # redirect handler, to make finding APIs easier
300 handlers.append((r"/(.+)/api/*", RedirectHandler, {"url": "/api/{0}"}))
302 handlers.append(
303 (
304 r"(?i)/\.well-known/(.*)",
305 TraversableStaticFileHandler,
306 {
307 "root": Path(".well-known"),
308 "headers": (("Access-Control-Allow-Origin", "*"),),
309 },
310 )
311 )
313 LOGGER.debug("Loaded %d handlers", len(handlers))
315 return handlers
318def ignore_modules(config: BetterConfigParser) -> None:
319 """Read ignored modules from the config."""
320 IGNORED_MODULES.update(
321 config.getset("GENERAL", "IGNORED_MODULES", fallback=set())
322 )
325def get_normed_paths_from_module_infos(
326 module_infos: Iterable[ModuleInfo],
327) -> dict[str, str]:
328 """Get all paths from the module infos."""
330 def tuple_has_no_none(
331 value: tuple[str | None, str | None],
332 ) -> TypeGuard[tuple[str, str]]:
333 return None not in value
335 def info_to_paths(info: ModuleInfo) -> Stream[tuple[str, str]]:
336 return (
337 Stream(((info.path, info.path),))
338 .chain(
339 info.aliases.items()
340 if isinstance(info.aliases, Mapping)
341 else ((alias, info.path) for alias in info.aliases)
342 )
343 .chain(
344 Stream(info.sub_pages)
345 .map(lambda sub_info: sub_info.path)
346 .filter()
347 .map(lambda path: (path, path))
348 )
349 .filter(tuple_has_no_none)
350 )
352 return (
353 Stream(module_infos)
354 .flat_map(info_to_paths)
355 .filter(lambda p: p[0].startswith("/"))
356 .map(lambda p: (p[0].strip("/").lower(), p[1]))
357 .filter(lambda p: p[0])
358 .collect(dict)
359 )
362def make_app(config: ConfigParser) -> str | Application:
363 """Create the Tornado application and return it."""
364 module_infos, duration = time_function(get_module_infos)
365 if isinstance(module_infos, str):
366 return module_infos
367 if duration > 1:
368 LOGGER.warning(
369 "Getting the module infos took %ss. That's probably too long.",
370 duration,
371 )
372 handlers = get_all_handlers(module_infos)
373 return Application(
374 handlers,
375 MODULE_INFOS=module_infos,
376 SHOW_HAMBURGER_MENU=not Stream(module_infos)
377 .exclude(lambda info: info.hidden)
378 .filter(lambda info: info.path)
379 .empty(),
380 NORMED_PATHS=get_normed_paths_from_module_infos(module_infos),
381 HANDLERS=handlers,
382 # General settings
383 autoreload=False,
384 debug=sys.flags.dev_mode,
385 default_handler_class=NotFoundHandler,
386 compress_response=config.getboolean(
387 "GENERAL", "COMPRESS_RESPONSE", fallback=False
388 ),
389 websocket_ping_interval=10,
390 # Template settings
391 template_loader=TemplateLoader(
392 root=TEMPLATES_DIR, whitespace="oneline"
393 ),
394 )
397def apply_config_to_app(app: Application, config: BetterConfigParser) -> None:
398 """Apply the config (from the config.ini file) to the application."""
399 app.settings["CONFIG"] = config
401 app.settings["cookie_secret"] = config.get(
402 "GENERAL", "COOKIE_SECRET", fallback="xyzzy"
403 )
405 app.settings["CRAWLER_SECRET"] = config.get(
406 "APP_SEARCH", "CRAWLER_SECRET", fallback=None
407 )
409 app.settings["DOMAIN"] = config.get("GENERAL", "DOMAIN", fallback=None)
411 app.settings["ELASTICSEARCH_PREFIX"] = config.get(
412 "ELASTICSEARCH", "PREFIX", fallback=NAME
413 )
415 app.settings["HSTS"] = config.getboolean("TLS", "HSTS", fallback=False)
417 app.settings["NETCUP"] = config.getboolean(
418 "GENERAL", "NETCUP", fallback=False
419 )
421 onion_address = config.get("GENERAL", "ONION_ADDRESS", fallback=None)
422 app.settings["ONION_ADDRESS"] = onion_address
423 if onion_address is None:
424 app.settings["ONION_PROTOCOL"] = None
425 else:
426 app.settings["ONION_PROTOCOL"] = onion_address.split("://")[0]
428 app.settings["RATELIMITS"] = config.getboolean(
429 "GENERAL",
430 "RATELIMITS",
431 fallback=config.getboolean("REDIS", "ENABLED", fallback=False),
432 )
434 app.settings["REDIS_PREFIX"] = config.get("REDIS", "PREFIX", fallback=NAME)
436 app.settings["REPORTING"] = config.getboolean(
437 "REPORTING", "ENABLED", fallback=True
438 )
440 app.settings["REPORTING_BUILTIN"] = config.getboolean(
441 "REPORTING", "BUILTIN", fallback=sys.flags.dev_mode
442 )
444 app.settings["REPORTING_ENDPOINT"] = config.get(
445 "REPORTING",
446 "ENDPOINT",
447 fallback=(
448 "/api/reports"
449 if app.settings["REPORTING_BUILTIN"]
450 else "https://asozial.org/api/reports"
451 ),
452 )
454 app.settings["TRUSTED_API_SECRETS"] = {
455 key_perms[0]: Permission(
456 int(key_perms[1])
457 if len(key_perms) > 1
458 else (1 << len(Permission)) - 1 # should be all permissions
459 )
460 for secret in config.getset(
461 "GENERAL", "TRUSTED_API_SECRETS", fallback={"xyzzy"}
462 )
463 if (key_perms := [part.strip() for part in secret.split("=")])
464 if key_perms[0]
465 }
467 app.settings["AUTH_TOKEN_SECRET"] = config.get(
468 "GENERAL", "AUTH_TOKEN_SECRET", fallback=None
469 )
470 if not app.settings["AUTH_TOKEN_SECRET"]:
471 node = uuid.getnode().to_bytes(6, "big")
472 secret = RIPEMD160.new(node).digest().decode("BRAILLE")
473 LOGGER.warning(
474 "AUTH_TOKEN_SECRET is unset, implicitly setting it to %r",
475 secret,
476 )
477 app.settings["AUTH_TOKEN_SECRET"] = secret
479 app.settings["UNDER_ATTACK"] = config.getboolean(
480 "GENERAL", "UNDER_ATTACK", fallback=False
481 )
483 apply_contact_stuff_to_app(app, config)
486def get_ssl_context( # pragma: no cover
487 config: ConfigParser,
488) -> None | ssl.SSLContext:
489 """Create SSL context and configure using the config."""
490 if config.getboolean("TLS", "ENABLED", fallback=False):
491 ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
492 ssl_ctx.load_cert_chain(
493 config.get("TLS", "CERTFILE"),
494 config.get("TLS", "KEYFILE", fallback=None),
495 config.get("TLS", "PASSWORD", fallback=None),
496 )
497 return ssl_ctx
498 return None
501def setup_logging( # pragma: no cover
502 config: ConfigParser,
503 force: bool = False,
504) -> None:
505 """Setup logging.""" # noqa: D401
506 root_logger = logging.getLogger()
508 if root_logger.handlers:
509 if not force:
510 return
511 for handler in root_logger.handlers[:]:
512 root_logger.removeHandler(handler)
513 handler.close()
515 debug = config.getboolean("LOGGING", "DEBUG", fallback=sys.flags.dev_mode)
517 logging.captureWarnings(True)
519 root_logger.setLevel(logging.DEBUG if debug else logging.INFO)
520 logging.getLogger("tornado.curl_httpclient").setLevel(logging.INFO)
521 logging.getLogger("elasticsearch").setLevel(logging.INFO)
523 stream_handler = logging.StreamHandler()
524 if sys.flags.dev_mode:
525 spam = regex.sub(r"%\((end_)?color\)s", "", LogFormatter.DEFAULT_FORMAT)
526 formatter = logging.Formatter(spam, LogFormatter.DEFAULT_DATE_FORMAT)
527 else:
528 formatter = LogFormatter()
529 stream_handler.setFormatter(formatter)
530 root_logger.addHandler(stream_handler)
532 if path := config.get("LOGGING", "PATH", fallback=None):
533 os.makedirs(path, 0o755, True)
534 file_handler = logging.handlers.TimedRotatingFileHandler(
535 os.path.join(path, f"{NAME}.log"),
536 encoding="UTF-8",
537 when="midnight",
538 backupCount=30,
539 utc=True,
540 )
541 file_handler.setFormatter(StdlibFormatter())
542 root_logger.addHandler(file_handler)
545class WebhookLoggingOptions: # pylint: disable=too-few-public-methods
546 """Webhook logging options."""
548 __slots__ = (
549 "url",
550 "content_type",
551 "body_format",
552 "timestamp_format",
553 "timestamp_timezone",
554 "escape_message",
555 "max_message_length",
556 )
558 url: str | None
559 content_type: str
560 body_format: str
561 timestamp_format: str | None
562 timestamp_timezone: str | None
563 escape_message: bool
564 max_message_length: int | None
566 def __init__(self, config: ConfigParser) -> None:
567 """Initialize Webhook logging options."""
568 self.url = config.get("LOGGING", "WEBHOOK_URL", fallback=None)
569 self.content_type = config.get(
570 "LOGGING",
571 "WEBHOOK_CONTENT_TYPE",
572 fallback="application/json",
573 )
574 spam = regex.sub(r"%\((end_)?color\)s", "", LogFormatter.DEFAULT_FORMAT)
575 self.body_format = config.get(
576 "LOGGING",
577 "WEBHOOK_BODY_FORMAT",
578 fallback='{"text":"' + spam + '"}',
579 )
580 self.timestamp_format = config.get(
581 "LOGGING",
582 "WEBHOOK_TIMESTAMP_FORMAT",
583 fallback=None,
584 )
585 self.timestamp_timezone = config.get(
586 "LOGGING", "WEBHOOK_TIMESTAMP_TIMEZONE", fallback=None
587 )
588 self.escape_message = config.getboolean(
589 "LOGGING",
590 "WEBHOOK_ESCAPE_MESSAGE",
591 fallback=True,
592 )
593 self.max_message_length = config.getint(
594 "LOGGING", "WEBHOOK_MAX_MESSAGE_LENGTH", fallback=None
595 )
598def setup_webhook_logging( # pragma: no cover
599 options: WebhookLoggingOptions,
600 loop: asyncio.AbstractEventLoop,
601) -> None:
602 """Setup Webhook logging.""" # noqa: D401
603 if not options.url:
604 return
606 LOGGER.info("Setting up Webhook logging")
608 root_logger = logging.getLogger()
610 webhook_content_type = options.content_type
611 webhook_handler = WebhookHandler(
612 logging.ERROR,
613 loop=loop,
614 url=options.url,
615 content_type=webhook_content_type,
616 )
617 formatter = WebhookFormatter(
618 options.body_format,
619 options.timestamp_format,
620 )
621 formatter.timezone = (
622 None
623 if options.timestamp_format is None
624 else ZoneInfo(options.timestamp_format)
625 )
626 formatter.escape_message = options.escape_message
627 formatter.max_message_length = options.max_message_length
628 formatter.get_context_line = lambda _: (
629 f"Request: {request}"
630 if (request := request_ctx_var.get(None))
631 else None
632 )
633 webhook_handler.setFormatter(formatter)
634 root_logger.addHandler(webhook_handler)
636 info_handler = WebhookHandler(
637 logging.INFO,
638 loop=loop,
639 url=options.url,
640 content_type=webhook_content_type,
641 )
642 info_handler.setFormatter(formatter)
643 logging.getLogger("an_website.quotes.create").addHandler(info_handler)
646def setup_apm(app: Application) -> None: # pragma: no cover
647 """Setup APM.""" # noqa: D401
648 config: BetterConfigParser = app.settings["CONFIG"]
649 app.settings["ELASTIC_APM"] = {
650 "ENABLED": config.getboolean("ELASTIC_APM", "ENABLED", fallback=False),
651 "SERVER_URL": config.get(
652 "ELASTIC_APM", "SERVER_URL", fallback="http://localhost:8200"
653 ),
654 "SECRET_TOKEN": config.get(
655 "ELASTIC_APM", "SECRET_TOKEN", fallback=None
656 ),
657 "API_KEY": config.get("ELASTIC_APM", "API_KEY", fallback=None),
658 "VERIFY_SERVER_CERT": config.getboolean(
659 "ELASTIC_APM", "VERIFY_SERVER_CERT", fallback=True
660 ),
661 "USE_CERTIFI": True, # doesn't actually use certifi
662 "SERVICE_NAME": NAME.removesuffix("-dev"),
663 "SERVICE_VERSION": VERSION,
664 "ENVIRONMENT": (
665 "production" if not sys.flags.dev_mode else "development"
666 ),
667 "DEBUG": True,
668 "CAPTURE_BODY": "errors",
669 "TRANSACTION_IGNORE_URLS": [
670 "/api/ping",
671 "/static/*",
672 "/favicon.png",
673 ],
674 "TRANSACTIONS_IGNORE_PATTERNS": ["^OPTIONS "],
675 "PROCESSORS": [
676 "an_website.utils.utils.apm_anonymization_processor",
677 "elasticapm.processors.sanitize_stacktrace_locals",
678 "elasticapm.processors.sanitize_http_request_cookies",
679 "elasticapm.processors.sanitize_http_headers",
680 "elasticapm.processors.sanitize_http_wsgi_env",
681 "elasticapm.processors.sanitize_http_request_body",
682 ],
683 "RUM_SERVER_URL": config.get(
684 "ELASTIC_APM", "RUM_SERVER_URL", fallback=None
685 ),
686 "RUM_SERVER_URL_PREFIX": config.get(
687 "ELASTIC_APM", "RUM_SERVER_URL_PREFIX", fallback=None
688 ),
689 }
691 script_options = [
692 f"serviceName:{app.settings['ELASTIC_APM']['SERVICE_NAME']!r}",
693 f"serviceVersion:{app.settings['ELASTIC_APM']['SERVICE_VERSION']!r}",
694 f"environment:{app.settings['ELASTIC_APM']['ENVIRONMENT']!r}",
695 ]
697 rum_server_url = app.settings["ELASTIC_APM"]["RUM_SERVER_URL"]
699 if rum_server_url is None:
700 script_options.append(
701 f"serverUrl:{app.settings['ELASTIC_APM']['SERVER_URL']!r}"
702 )
703 elif rum_server_url:
704 script_options.append(f"serverUrl:{rum_server_url!r}")
705 else:
706 script_options.append("serverUrl:window.location.origin")
708 if app.settings["ELASTIC_APM"]["RUM_SERVER_URL_PREFIX"]:
709 script_options.append(
710 f"serverUrlPrefix:{app.settings['ELASTIC_APM']['RUM_SERVER_URL_PREFIX']!r}"
711 )
713 app.settings["ELASTIC_APM"]["INLINE_SCRIPT"] = (
714 "elasticApm.init({" + ",".join(script_options) + "})"
715 )
717 app.settings["ELASTIC_APM"]["INLINE_SCRIPT_HASH"] = b64encode(
718 sha256(
719 app.settings["ELASTIC_APM"]["INLINE_SCRIPT"].encode("ASCII")
720 ).digest()
721 ).decode("ASCII")
723 if app.settings["ELASTIC_APM"]["ENABLED"]:
724 app.settings["ELASTIC_APM"]["CLIENT"] = ElasticAPM(app).client
727def setup_app_search(app: Application) -> None: # pragma: no cover
728 """Setup Elastic App Search.""" # noqa: D401
729 with catch_warnings():
730 simplefilter("ignore", DeprecationWarning)
731 # pylint: disable-next=import-outside-toplevel
732 from elastic_enterprise_search import ( # type: ignore[import-untyped]
733 AppSearch,
734 )
736 config: BetterConfigParser = app.settings["CONFIG"]
737 host = config.get("APP_SEARCH", "HOST", fallback=None)
738 key = config.get("APP_SEARCH", "SEARCH_KEY", fallback=None)
739 verify_certs = config.getboolean(
740 "APP_SEARCH", "VERIFY_CERTS", fallback=True
741 )
742 app.settings["APP_SEARCH"] = (
743 AppSearch(
744 host,
745 bearer_auth=key,
746 verify_certs=verify_certs,
747 ca_certs=CA_BUNDLE_PATH,
748 )
749 if host
750 else None
751 )
752 app.settings["APP_SEARCH_HOST"] = host
753 app.settings["APP_SEARCH_KEY"] = key
754 app.settings["APP_SEARCH_ENGINE"] = config.get(
755 "APP_SEARCH", "ENGINE_NAME", fallback=NAME.removesuffix("-dev")
756 )
759def setup_redis(app: Application) -> None | Redis[str]:
760 """Setup Redis.""" # noqa: D401
761 config: BetterConfigParser = app.settings["CONFIG"]
763 class Kwargs(TypedDict, total=False):
764 """Kwargs of BlockingConnectionPool constructor."""
766 db: int
767 username: None | str
768 password: None | str
769 retry_on_timeout: bool
770 connection_class: type[UnixDomainSocketConnection] | type[SSLConnection]
771 path: str
772 host: str
773 port: int
774 ssl_ca_certs: str
775 ssl_keyfile: None | str
776 ssl_certfile: None | str
777 ssl_check_hostname: bool
778 ssl_cert_reqs: str
780 kwargs: Kwargs = {
781 "db": config.getint("REDIS", "DB", fallback=0),
782 "username": config.get("REDIS", "USERNAME", fallback=None),
783 "password": config.get("REDIS", "PASSWORD", fallback=None),
784 "retry_on_timeout": config.getboolean(
785 "REDIS", "RETRY_ON_TIMEOUT", fallback=False
786 ),
787 }
788 redis_ssl_kwargs: Kwargs = {
789 "connection_class": SSLConnection,
790 "ssl_ca_certs": CA_BUNDLE_PATH,
791 "ssl_keyfile": config.get("REDIS", "SSL_KEYFILE", fallback=None),
792 "ssl_certfile": config.get("REDIS", "SSL_CERTFILE", fallback=None),
793 "ssl_cert_reqs": config.get(
794 "REDIS", "SSL_CERT_REQS", fallback="required"
795 ),
796 "ssl_check_hostname": config.getboolean(
797 "REDIS", "SSL_CHECK_HOSTNAME", fallback=False
798 ),
799 }
800 redis_host_port_kwargs: Kwargs = {
801 "host": config.get("REDIS", "HOST", fallback="localhost"),
802 "port": config.getint("REDIS", "PORT", fallback=6379),
803 }
804 redis_use_ssl = config.getboolean("REDIS", "SSL", fallback=False)
805 redis_unix_socket_path = config.get(
806 "REDIS", "UNIX_SOCKET_PATH", fallback=None
807 )
809 if redis_unix_socket_path is not None:
810 if redis_use_ssl:
811 LOGGER.warning(
812 "SSL is enabled for Redis, but a UNIX socket is used"
813 )
814 if config.has_option("REDIS", "HOST"):
815 LOGGER.warning(
816 "A host is configured for Redis, but a UNIX socket is used"
817 )
818 if config.has_option("REDIS", "PORT"):
819 LOGGER.warning(
820 "A port is configured for Redis, but a UNIX socket is used"
821 )
822 kwargs.update(
823 {
824 "connection_class": UnixDomainSocketConnection,
825 "path": redis_unix_socket_path,
826 }
827 )
828 else:
829 kwargs.update(redis_host_port_kwargs)
830 if redis_use_ssl:
831 kwargs.update(redis_ssl_kwargs)
833 if not config.getboolean("REDIS", "ENABLED", fallback=False):
834 app.settings["REDIS"] = None
835 return None
836 connection_pool = BlockingConnectionPool(
837 client_name=NAME,
838 decode_responses=True,
839 **kwargs,
840 )
841 redis = cast("Redis[str]", Redis(connection_pool=connection_pool))
842 app.settings["REDIS"] = redis
843 return redis
846def signal_handler( # noqa: D103 # pragma: no cover
847 signalnum: int, frame: None | types.FrameType
848) -> None:
849 # pylint: disable=unused-argument, missing-function-docstring
850 if signalnum in {signal.SIGINT, signal.SIGTERM}:
851 EVENT_SHUTDOWN.set()
852 if signalnum == getattr(signal, "SIGHUP", None):
853 EVENT_SHUTDOWN.set()
856def install_signal_handler() -> None: # pragma: no cover
857 """Install the signal handler."""
858 signal.signal(signal.SIGINT, signal_handler)
859 signal.signal(signal.SIGTERM, signal_handler)
860 if hasattr(signal, "SIGHUP"):
861 signal.signal(signal.SIGHUP, signal_handler)
864def supervise(loop: AbstractEventLoop) -> None:
865 """Supervise."""
866 while foobarbaz := background_tasks.HEARTBEAT: # pylint: disable=while-used
867 if time.monotonic() - foobarbaz >= 10:
868 worker = task_id()
869 pid = os.getpid()
871 task = asyncio.current_task(loop)
872 request = task.get_context().get(request_ctx_var) if task else None
874 LOGGER.fatal(
875 "Heartbeat timed out for worker %s (pid %d), "
876 "current request: %s, current task: %s",
877 worker,
878 pid,
879 request,
880 task,
881 )
882 atexit._run_exitfuncs() # pylint: disable=protected-access
883 os.abort()
884 time.sleep(1)
887def main( # noqa: C901 # pragma: no cover
888 config: BetterConfigParser | None = None,
889) -> int | str:
890 """
891 Start everything.
893 This is the main function that is called when running this program.
894 """
895 # pylint: disable=too-complex, too-many-branches
896 # pylint: disable=too-many-locals, too-many-statements
897 setproctitle(NAME)
899 install_signal_handler()
901 parser = create_argument_parser()
902 args, _ = parser.parse_known_args(
903 get_arguments_without_help(), ArgparseNamespace()
904 )
906 if args.version:
907 print("Version:", VERSION)
908 if args.verbose:
909 # pylint: disable-next=import-outside-toplevel
910 from .version.version import (
911 get_file_hashes,
912 get_hash_of_file_hashes,
913 )
915 print()
916 print("Hash der Datei-Hashes:")
917 print(get_hash_of_file_hashes())
919 if args.verbose > 1:
920 print()
921 print("Datei-Hashes:")
922 print(get_file_hashes())
924 return 0
926 config = config or BetterConfigParser.from_path(*args.config)
927 assert config is not None
928 config.add_override_argument_parser(parser)
930 setup_logging(config)
932 LOGGER.info("Starting %s %s", NAME, VERSION)
934 if platform.system() == "Windows":
935 LOGGER.warning(
936 "Running %s on Windows is not officially supported",
937 NAME.removesuffix("-dev"),
938 )
940 ignore_modules(config)
941 app = make_app(config)
942 if isinstance(app, str):
943 return app
945 apply_config_to_app(app, config)
946 setup_elasticsearch(app)
947 setup_app_search(app)
948 setup_redis(app)
949 setup_apm(app)
951 behind_proxy = config.getboolean("GENERAL", "BEHIND_PROXY", fallback=False)
953 server = HTTPServer(
954 app,
955 body_timeout=3600,
956 decompress_request=True,
957 max_body_size=1_000_000_000,
958 ssl_options=get_ssl_context(config),
959 xheaders=behind_proxy,
960 )
962 socket_factories: list[Callable[[], Iterable[socket]]] = []
964 port = config.getint("GENERAL", "PORT", fallback=None)
966 if port:
967 socket_factories.append(
968 partial(
969 bind_sockets,
970 port,
971 "localhost" if behind_proxy else "",
972 )
973 )
975 unix_socket_path = config.get(
976 "GENERAL",
977 "UNIX_SOCKET_PATH",
978 fallback=None,
979 )
981 if unix_socket_path:
982 os.makedirs(unix_socket_path, 0o755, True)
983 socket_factories.append(
984 lambda: (
985 bind_unix_socket(
986 os.path.join(unix_socket_path, f"{NAME}.sock"),
987 mode=0o666,
988 ),
989 )
990 )
992 processes = config.getint(
993 "GENERAL",
994 "PROCESSES",
995 fallback=hasattr(os, "fork") * (2 if sys.flags.dev_mode else -1),
996 )
998 if processes < 0:
999 processes = (
1000 os.process_cpu_count() # type: ignore[attr-defined]
1001 if sys.version_info >= (3, 13)
1002 else os.cpu_count()
1003 ) or 0
1005 worker: None | int = None
1007 run_supervisor_thread = config.getboolean(
1008 "GENERAL", "SUPERVISE", fallback=False
1009 )
1010 elasticsearch_is_enabled = config.getboolean(
1011 "ELASTICSEARCH", "ENABLED", fallback=False
1012 )
1013 redis_is_enabled = config.getboolean("REDIS", "ENABLED", fallback=False)
1014 webhook_logging_options = WebhookLoggingOptions(config)
1015 # all config options should be read before forking
1016 if args.save_config_to:
1017 with open(args.save_config_to, "w", encoding="UTF-8") as file:
1018 config.write(file)
1019 config.set_all_options_should_be_parsed()
1020 del config
1021 # show help message if --help is given (after reading config, before forking)
1022 parser.parse_args()
1024 if not socket_factories:
1025 LOGGER.warning("No sockets configured")
1026 return 0
1028 # create sockets after checking for --help
1029 sockets: list[socket] = (
1030 Stream(socket_factories).flat_map(lambda fun: fun()).collect(list)
1031 )
1033 UPTIME.reset()
1034 main_pid = os.getpid()
1036 if processes:
1037 setproctitle(f"{NAME} - Master")
1039 worker = fork_processes(processes)
1041 setproctitle(f"{NAME} - Worker {worker}")
1043 # yeet all children (there should be none, but do it regardless, just in case)
1044 _children.clear()
1046 if "an_website.quotes" in sys.modules:
1047 from .quotes.utils import ( # pylint: disable=import-outside-toplevel
1048 AUTHORS_CACHE,
1049 QUOTES_CACHE,
1050 WRONG_QUOTES_CACHE,
1051 )
1053 del AUTHORS_CACHE.control.created_by_ultra # type: ignore[attr-defined]
1054 del QUOTES_CACHE.control.created_by_ultra # type: ignore[attr-defined]
1055 del WRONG_QUOTES_CACHE.control.created_by_ultra # type: ignore[attr-defined]
1056 del (geoip.__kwdefaults__ or {})["caches"].control.created_by_ultra
1058 if unix_socket_path:
1059 sockets.append(
1060 bind_unix_socket(
1061 os.path.join(unix_socket_path, f"{NAME}.{worker}.sock"),
1062 mode=0o666,
1063 )
1064 )
1066 # get loop after forking
1067 # if not forking allow loop to be set in advance by external code
1068 loop: None | asyncio.AbstractEventLoop
1069 try:
1070 with catch_warnings(): # TODO: remove after dropping support for 3.13
1071 simplefilter("ignore", DeprecationWarning)
1072 loop = asyncio.get_event_loop()
1073 if loop.is_closed():
1074 loop = None
1075 except RuntimeError:
1076 loop = None
1078 if loop is None:
1079 loop = asyncio.new_event_loop()
1080 asyncio.set_event_loop(loop)
1082 if sys.version_info >= (3, 13) and not loop.get_task_factory():
1083 loop.set_task_factory(asyncio.eager_task_factory)
1085 if perf8 and "PERF8" in os.environ:
1086 loop.run_until_complete(perf8.enable())
1088 setup_webhook_logging(webhook_logging_options, loop)
1090 server.add_sockets(sockets)
1092 tasks = background_tasks.start_background_tasks( # noqa: F841
1093 module_infos=app.settings["MODULE_INFOS"],
1094 loop=loop,
1095 main_pid=main_pid,
1096 app=app,
1097 processes=processes,
1098 elasticsearch_is_enabled=elasticsearch_is_enabled,
1099 redis_is_enabled=redis_is_enabled,
1100 worker=worker,
1101 )
1103 if run_supervisor_thread:
1104 background_tasks.HEARTBEAT = time.monotonic()
1105 threading.Thread(
1106 target=supervise, args=(loop,), name="supervisor", daemon=True
1107 ).start()
1109 try:
1110 loop.run_forever()
1111 EVENT_SHUTDOWN.set()
1112 finally:
1113 try: # pylint: disable=too-many-try-statements
1114 server.stop()
1115 loop.run_until_complete(asyncio.sleep(1))
1116 loop.run_until_complete(server.close_all_connections())
1117 if perf8 and "PERF8" in os.environ:
1118 loop.run_until_complete(perf8.disable())
1119 if redis := app.settings.get("REDIS"):
1120 loop.run_until_complete(
1121 redis.aclose(close_connection_pool=True)
1122 )
1123 if elasticsearch := app.settings.get("ELASTICSEARCH"):
1124 loop.run_until_complete(elasticsearch.close())
1125 finally:
1126 try:
1127 _cancel_all_tasks(loop)
1128 loop.run_until_complete(loop.shutdown_asyncgens())
1129 loop.run_until_complete(loop.shutdown_default_executor())
1130 finally:
1131 loop.close()
1132 background_tasks.HEARTBEAT = 0
1134 return len(tasks)