Spaces:
Running
Running
File size: 3,759 Bytes
fba6023 e1104b3 fba6023 e1104b3 fba6023 e1104b3 fba6023 e1104b3 fba6023 e1104b3 fba6023 e1104b3 fba6023 e1104b3 fba6023 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | from __future__ import annotations
import contextvars
import json
import logging
import re
import sys
from datetime import datetime, timezone
from typing import Any, TextIO
from app.core.config import get_settings
request_id_context: contextvars.ContextVar[str] = contextvars.ContextVar("request_id", default="-")
class JsonFormatter(logging.Formatter):
"""One-line structured JSON logs suitable for container log collectors."""
_standard = set(logging.makeLogRecord({}).__dict__) | {"message", "asctime"}
_sensitive_keys = frozenset(
{
"access_token",
"refresh_token",
"id_token",
"authorization",
"client_secret",
"secret",
"password",
"api_key",
"credential",
"cookie",
}
)
_bearer = re.compile(r"(?i)\bbearer\s+[A-Za-z0-9._~+/=-]+")
_assigned_secret = re.compile(
r"(?i)(access[_-]?token|refresh[_-]?token|id[_-]?token|client[_-]?secret|"
r"authorization|api[_-]?key|password|secret|credential)"
r"([\"']?\s*[:=]\s*[\"']?)([^\"'\s,&}]+)"
)
def format(self, record: logging.LogRecord) -> str:
payload: dict[str, Any] = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"level": record.levelname,
"logger": record.name,
"message": self._redact_text(record.getMessage()),
"request_id": getattr(record, "request_id", request_id_context.get()),
}
for key, value in record.__dict__.items():
if key not in self._standard and not key.startswith("_"):
payload[key] = self._json_safe(value, key=key)
if record.exc_info:
payload["exception"] = self._redact_text(
self.formatException(record.exc_info)
)
return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
@classmethod
def _json_safe(cls, value: Any, *, key: str | None = None) -> Any:
if key is not None and cls._is_sensitive_key(key):
return "[REDACTED]"
if isinstance(value, dict):
return {
str(item_key): cls._json_safe(item, key=str(item_key))
for item_key, item in value.items()
}
if isinstance(value, (list, tuple, set)):
return [cls._json_safe(item) for item in value]
if isinstance(value, str):
return cls._redact_text(value)
try:
json.dumps(value)
return value
except (TypeError, ValueError):
return cls._redact_text(str(value))
@classmethod
def _is_sensitive_key(cls, key: str) -> bool:
normalized = key.strip().lower().replace("-", "_")
return any(part in normalized for part in cls._sensitive_keys)
@classmethod
def _redact_text(cls, value: str) -> str:
redacted = cls._bearer.sub("Bearer [REDACTED]", value)
return cls._assigned_secret.sub(
lambda match: f"{match.group(1)}{match.group(2)}[REDACTED]",
redacted,
)
def configure_logging(stream: TextIO | None = None) -> None:
"""Configure structured logging, optionally targeting a stdio-safe stream."""
settings = get_settings()
handler = logging.StreamHandler(stream or sys.stdout)
handler.setFormatter(JsonFormatter())
root = logging.getLogger()
root.handlers.clear()
root.addHandler(handler)
root.setLevel(settings.log_level)
for name in ("uvicorn.access", "uvicorn.error"):
logging.getLogger(name).handlers.clear()
logging.getLogger(name).propagate = True
def get_logger(name: str) -> logging.Logger:
return logging.getLogger(name)
|