feat: add privacy-safe structured logging
This commit is contained in:
@@ -0,0 +1,48 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
from datetime import datetime, timezone
|
||||
|
||||
|
||||
SAFE_FIELDS = (
|
||||
"request_id", "method", "path", "status_code", "duration_ms",
|
||||
"component", "event", "result", "rows_seen", "rows_created",
|
||||
"rows_updated", "not_modified", "interval_seconds", "error_type",
|
||||
)
|
||||
SENSITIVE = re.compile(
|
||||
r"(?i)(bearer\s+)[^\s]+|((?:token|password|secret|authorization)\s*[=:]\s*)[^\s,;]+"
|
||||
)
|
||||
|
||||
|
||||
def redact(value: str) -> str:
|
||||
return SENSITIVE.sub(lambda match: f"{match.group(1) or match.group(2)}[REDACTED]", value)
|
||||
|
||||
|
||||
class JsonFormatter(logging.Formatter):
|
||||
def format(self, record: logging.LogRecord) -> str:
|
||||
payload: dict[str, object] = {
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
"level": record.levelname.lower(),
|
||||
"logger": record.name,
|
||||
"message": redact(record.getMessage()),
|
||||
}
|
||||
for field in SAFE_FIELDS:
|
||||
value = getattr(record, field, None)
|
||||
if value is not None:
|
||||
payload[field] = value
|
||||
if record.exc_info and "error_type" not in payload:
|
||||
payload["error_type"] = record.exc_info[0].__name__
|
||||
return json.dumps(payload, ensure_ascii=False, separators=(",", ":"), default=str)
|
||||
|
||||
|
||||
def configure_logging(level: str = "INFO") -> None:
|
||||
root = logging.getLogger()
|
||||
if not any(getattr(handler, "_rf4_json", False) for handler in root.handlers):
|
||||
handler = logging.StreamHandler()
|
||||
handler.setFormatter(JsonFormatter())
|
||||
handler._rf4_json = True # type: ignore[attr-defined]
|
||||
root.handlers.clear()
|
||||
root.addHandler(handler)
|
||||
root.setLevel(getattr(logging, level.upper(), logging.INFO))
|
||||
Reference in New Issue
Block a user