"""Small JSON logger that redacts configured secret values.""" from __future__ import annotations import json import logging from datetime import datetime, timezone from typing import Iterable class RedactingJsonFormatter(logging.Formatter): def __init__(self, secrets: Iterable[str]): super().__init__() self._secrets = tuple( sorted((secret for secret in secrets if secret), key=len, reverse=True) ) def format(self, record: logging.LogRecord) -> str: payload: dict[str, object] = { "timestamp": datetime.fromtimestamp( record.created, timezone.utc ).isoformat(), "level": record.levelname.lower(), "logger": record.name, "event": self._redact(record.getMessage()), } for name in ( "client_id", "role", "error_type", "backup_name", "attempt", "service", "health", "error_code", ): if hasattr(record, name): payload[name] = self._redact(str(getattr(record, name))) if record.exc_info: payload["exception_type"] = record.exc_info[0].__name__ return json.dumps(payload, sort_keys=True, separators=(",", ":")) def _redact(self, value: str) -> str: for secret in self._secrets: value = value.replace(secret, "[REDACTED]") return value def configure_logging(secrets: Iterable[str]) -> None: handler = logging.StreamHandler() handler.setFormatter(RedactingJsonFormatter(secrets)) root = logging.getLogger() root.handlers.clear() root.addHandler(handler) root.setLevel(logging.INFO) logging.getLogger("websockets").setLevel(logging.WARNING)