132 lines
4.5 KiB
Python
132 lines
4.5 KiB
Python
"""Logging configuration: console + daily-rotating file, with API-key masking."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from logging.handlers import TimedRotatingFileHandler
|
|
from pathlib import Path
|
|
|
|
from us2kr.config import Settings
|
|
|
|
LOG_FILE_NAME = "us2kr.log"
|
|
LOG_FORMAT = "%(asctime)s %(levelname)-7s %(name)s: %(message)s"
|
|
_HANDLER_TAG = "_us2kr_handler"
|
|
|
|
# MiniMax keys (sk-...), Telegram bot tokens (123456:ABC...), and secrets passed as URL query parameters
|
|
# (FRED sends api_key=<32 hex>; requests/urllib3 embed the full URL in HTTPError/ConnectionError messages,
|
|
# which then land in tracebacks via log.exception). The parameter name is kept so the log stays readable.
|
|
SECRET_RE = re.compile(
|
|
r"(sk-[A-Za-z0-9_-]{6,}|\d{6,}:[A-Za-z0-9_-]{20,})"
|
|
r"|(?i:([?&](?:api_key|apikey|api-key|key|token|access_token)=))[^&\s\"'<>]+"
|
|
)
|
|
MASK = "***"
|
|
|
|
# Exact secret values registered by setup_logging(); replaced verbatim whatever their shape.
|
|
_SECRET_VALUES: tuple[str, ...] = ()
|
|
_MIN_SECRET_LEN = 8
|
|
|
|
NOISY_LOGGERS = (
|
|
"httpx",
|
|
"httpx2", # transport used by openai>=3 (logs request URLs at INFO)
|
|
"httpcore",
|
|
"apscheduler.executors",
|
|
"apscheduler.executors.default",
|
|
"yfinance",
|
|
"peewee",
|
|
"urllib3",
|
|
)
|
|
|
|
|
|
def register_secret_values(*values: str | None) -> None:
|
|
"""Register configured secret values so they are masked even when they match no known shape.
|
|
|
|
Calling with no arguments clears the registry. Longest values first so a value that contains
|
|
another registered value leaves no residue.
|
|
"""
|
|
global _SECRET_VALUES
|
|
_SECRET_VALUES = tuple(
|
|
sorted({v for v in values if v and len(v) >= _MIN_SECRET_LEN}, key=len, reverse=True)
|
|
)
|
|
|
|
|
|
def _sub(m: re.Match[str]) -> str:
|
|
prefix = m.group(2)
|
|
return f"{prefix}{MASK}" if prefix else MASK
|
|
|
|
|
|
def mask_secrets(text: str) -> str:
|
|
"""Replace anything that looks like an API key / bot token / URL query secret with ``***``."""
|
|
for value in _SECRET_VALUES:
|
|
text = text.replace(value, MASK)
|
|
return SECRET_RE.sub(_sub, text)
|
|
|
|
|
|
def _mask_value(value: object) -> object:
|
|
return mask_secrets(value) if isinstance(value, str) else value
|
|
|
|
|
|
class SecretMaskingFilter(logging.Filter):
|
|
"""Masks secrets in ``record.msg`` and ``record.args`` before formatting."""
|
|
|
|
def filter(self, record: logging.LogRecord) -> bool:
|
|
if isinstance(record.msg, str):
|
|
record.msg = mask_secrets(record.msg)
|
|
args = record.args
|
|
if isinstance(args, tuple):
|
|
record.args = tuple(_mask_value(a) for a in args)
|
|
elif isinstance(args, dict):
|
|
record.args = {k: _mask_value(v) for k, v in args.items()}
|
|
return True
|
|
|
|
|
|
class SecretMaskingFormatter(logging.Formatter):
|
|
"""Second line of defence: masks the fully formatted line (covers tracebacks too)."""
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
return mask_secrets(super().format(record))
|
|
|
|
|
|
def _remove_our_handlers(root: logging.Logger) -> None:
|
|
for handler in list(root.handlers):
|
|
if getattr(handler, _HANDLER_TAG, False):
|
|
root.removeHandler(handler)
|
|
handler.close()
|
|
|
|
|
|
def setup_logging(settings: Settings) -> None:
|
|
"""Configure the root logger. Safe to call more than once (handlers are replaced, not duplicated)."""
|
|
level_name = (settings.log_level or "INFO").upper()
|
|
level = logging.getLevelNamesMapping().get(level_name, logging.INFO)
|
|
|
|
root = logging.getLogger()
|
|
_remove_our_handlers(root)
|
|
root.setLevel(level)
|
|
|
|
register_secret_values(settings.minimax_api_key, settings.telegram_bot_token, settings.fred_api_key)
|
|
formatter = SecretMaskingFormatter(LOG_FORMAT)
|
|
mask_filter = SecretMaskingFilter()
|
|
|
|
console = logging.StreamHandler()
|
|
handlers: list[logging.Handler] = [console]
|
|
|
|
log_dir = Path(settings.log_dir)
|
|
try:
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
file_handler = TimedRotatingFileHandler(
|
|
log_dir / LOG_FILE_NAME, when="midnight", backupCount=14, encoding="utf-8"
|
|
)
|
|
handlers.append(file_handler)
|
|
except OSError as exc: # unwritable log dir must not prevent start-up
|
|
logging.getLogger(__name__).warning("file logging disabled (%s): %s", log_dir, exc)
|
|
|
|
for handler in handlers:
|
|
handler.setLevel(level)
|
|
handler.setFormatter(formatter)
|
|
handler.addFilter(mask_filter)
|
|
setattr(handler, _HANDLER_TAG, True)
|
|
root.addHandler(handler)
|
|
|
|
for name in NOISY_LOGGERS:
|
|
logging.getLogger(name).setLevel(logging.WARNING)
|