"""JSON logs on stdout with a request id (AD-18).""" from __future__ import annotations import json import logging import re from contextvars import ContextVar from datetime import UTC, datetime request_id_var: ContextVar[str] = ContextVar("request_id", default="-") # Attributes set by logging itself; everything else passed via `extra` is emitted. _STANDARD = set(logging.LogRecord("", 0, "", 0, "", (), None).__dict__) | {"message", "asctime"} # One-time links carry their token in the path (AD-9); a log line must never hold it. _LINK_TOKEN = re.compile(r"(/email/(?:confirm|undo)/)[^/\s\"']+") def redact(text: str) -> str: return _LINK_TOKEN.sub(r"\1", text) class RedactLinkTokens(logging.Filter): """Blank the token of a one-time link in a record's message.""" def filter(self, record: logging.LogRecord) -> bool: message = record.getMessage() redacted = redact(message) if redacted != message: record.msg, record.args = redacted, () return True class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: payload: dict[str, object] = { "ts": datetime.fromtimestamp(record.created, UTC).isoformat(timespec="milliseconds"), "level": record.levelname, "logger": record.name, "message": record.getMessage(), "request_id": getattr(record, "request_id", None) or request_id_var.get(), } for key, value in record.__dict__.items(): if key in _STANDARD or key in ("request_id", "request"): continue if value is None or isinstance(value, str | int | float | bool): payload[key] = value if record.exc_info: payload["exception"] = self.formatException(record.exc_info) return json.dumps(payload, default=str) def build_logging(level: str) -> dict: return { "version": 1, "disable_existing_loggers": False, "formatters": {"json": {"()": "portal.config.logging.JsonFormatter"}}, "filters": {"redact_link_tokens": {"()": "portal.config.logging.RedactLinkTokens"}}, "handlers": { "stdout": { "class": "logging.StreamHandler", "stream": "ext://sys.stdout", "formatter": "json", "filters": ["redact_link_tokens"], } }, "loggers": {"django.request": {"level": "ERROR"}}, "root": {"handlers": ["stdout"], "level": level}, }