96 lines
3.5 KiB
Python
96 lines
3.5 KiB
Python
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)
|