63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
 
|
"""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
|