import logging import re import sys import time from typing import Callable from .chat.base import MASK, mask_secret LOG_FORMAT = "%(levelname)s %(name)s: %(message)s" FALLBACK_LEVEL = logging.INFO QUIET_LOGGERS = ("httpx", "httpcore") SECRET_CACHE_SECONDS = 30.0 MIN_SECRET_LEN = 4 HANDLER_MARK = "_checkbot_handler" # Bot-Token in Telegram-URLs und Bearer-Header, auch wenn der Wert nicht (mehr) im SettingsStore steht. BOT_URL_RE = re.compile(r"(/bot)[^/\s\"']+(?=/)") BEARER_RE = re.compile(r"(Bearer\s+)\S+", re.IGNORECASE) SecretProvider = Callable[[], list[str]] def mask_patterns(text: str) -> str: return BEARER_RE.sub(rf"\g<1>{MASK}", BOT_URL_RE.sub(rf"\g<1>{MASK}", text)) def parse_level(raw: str | None) -> tuple[int, bool]: name = (raw or "").strip().upper() if not name: return FALLBACK_LEVEL, True level = logging.getLevelNamesMapping().get(name) if isinstance(level, int) and level != logging.NOTSET: return level, True return FALLBACK_LEVEL, False class SecretMaskFilter(logging.Filter): def __init__(self, provider: SecretProvider, ttl: float = SECRET_CACHE_SECONDS, clock: Callable[[], float] = time.monotonic): super().__init__() self._provider = provider self._ttl = ttl self._clock = clock self._secrets: list[str] = [] self._loaded_at: float | None = None def _current_secrets(self) -> list[str]: now = self._clock() if self._loaded_at is None or now - self._loaded_at >= self._ttl: self._loaded_at = now try: values = [v for v in self._provider() if isinstance(v, str) and len(v) >= MIN_SECRET_LEN] except Exception: values = self._secrets self._secrets = sorted(values, key=len, reverse=True) return self._secrets def _mask(self, text: str) -> str: for secret in self._current_secrets(): text = mask_secret(text, secret) return mask_patterns(text) def filter(self, record: logging.LogRecord) -> bool: try: record.msg = self._mask(record.getMessage()) record.args = None if record.exc_info and not record.exc_text: record.exc_text = logging.Formatter().formatException(record.exc_info) if record.exc_text: record.exc_text = self._mask(record.exc_text) if record.stack_info: record.stack_info = self._mask(record.stack_info) except Exception: record.msg, record.args = "Log-Meldung wegen Maskierungsfehler unterdrückt", None record.exc_info = record.exc_text = record.stack_info = None return True def setup_logging(raw_level: str | None, provider: SecretProvider) -> None: level, valid = parse_level(raw_level) root = logging.getLogger() for handler in list(root.handlers): if getattr(handler, HANDLER_MARK, False): root.removeHandler(handler) handler = logging.StreamHandler(sys.stderr) setattr(handler, HANDLER_MARK, True) handler.setFormatter(logging.Formatter(LOG_FORMAT)) handler.addFilter(SecretMaskFilter(provider)) root.addHandler(handler) root.setLevel(level) # httpx loggt Request-URLs auf INFO, die bei Telegram das Bot-Token enthalten. for name in QUIET_LOGGERS: logging.getLogger(name).setLevel(logging.WARNING) if not valid: logging.getLogger(__name__).warning("Ungültiger Wert für CHECKBOT_LOG_LEVEL: %r, verwende INFO", raw_level)