"""Logging to stdout as JSON lines, with secrets masked.""" from __future__ import annotations import json import logging import re import sys from datetime import UTC, datetime from typing import Any REDACT = re.compile(r"(?i)(password|pwd|secret)=\S+") def redact(text: str) -> str: return REDACT.sub(lambda m: f"{m.group(1)}=***", text) class RedactFilter(logging.Filter): """Masks password=..., pwd=..., secret=... in the rendered message.""" def filter(self, record: logging.LogRecord) -> bool: try: msg = record.getMessage() except Exception: msg = str(record.msg) record.msg = redact(msg) record.args = None return True class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: out: dict[str, Any] = { "ts": datetime.fromtimestamp(record.created, UTC).isoformat(timespec="milliseconds"), "level": record.levelname, "logger": record.name, "msg": redact(record.getMessage()), } extra = getattr(record, "ps", None) if isinstance(extra, dict): out.update({k: redact(v) if isinstance(v, str) else v for k, v in extra.items()}) if record.exc_info: out["exc"] = redact(self.formatException(record.exc_info)) # Redact the fields, not the JSON line: the pattern would eat closing quotes return json.dumps(out, default=str, ensure_ascii=False) def setup(level: str = "INFO") -> None: """Configure the root logger once; safe to call again.""" root = logging.getLogger() for h in list(root.handlers): if getattr(h, "_ps", False): root.removeHandler(h) handler = logging.StreamHandler(sys.stdout) handler.setFormatter(JsonFormatter()) handler.addFilter(RedactFilter()) handler._ps = True # type: ignore[attr-defined] root.addHandler(handler) root.setLevel(level) # uvicorn's access log is replaced by our request log logging.getLogger("uvicorn.access").disabled = True