107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from collections.abc import Mapping, Sequence
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
STANDARD_FIELDS = {
|
|
"name",
|
|
"msg",
|
|
"args",
|
|
"levelname",
|
|
"levelno",
|
|
"pathname",
|
|
"filename",
|
|
"module",
|
|
"exc_info",
|
|
"exc_text",
|
|
"stack_info",
|
|
"lineno",
|
|
"funcName",
|
|
"created",
|
|
"msecs",
|
|
"relativeCreated",
|
|
"thread",
|
|
"threadName",
|
|
"processName",
|
|
"process",
|
|
"taskName",
|
|
}
|
|
|
|
REDACTED = "[REDACTED]"
|
|
SENSITIVE_KEY = re.compile(
|
|
r"(?:password|passwd|secret|token|authorization|cookie|credential|api[_-]?key|"
|
|
r"access[_-]?key|private[_-]?key|master[_-]?key|csrf|session)",
|
|
re.IGNORECASE,
|
|
)
|
|
AUTH_VALUE = re.compile(r"(?i)\b(Bearer|Basic)\s+[A-Za-z0-9._~+/=-]+")
|
|
SECRET_ASSIGNMENT = re.compile(
|
|
r"(?i)(\b(?:password|passwd|secret|token|authorization|cookie|credential|"
|
|
r"api[_-]?key|access[_-]?key|private[_-]?key|master[_-]?key|csrf|session)\b\s*[=:]\s*)"
|
|
r"([^\s,;&]+)"
|
|
)
|
|
SECRET_QUERY = re.compile(
|
|
r"(?i)([?&](?:password|passwd|secret|token|authorization|credential|api[_-]?key|"
|
|
r"access[_-]?key|private[_-]?key|master[_-]?key|csrf|session)=)([^&#\s]+)"
|
|
)
|
|
AWS_ACCESS_KEY = re.compile(r"\b(?:AKIA|ASIA)[A-Z0-9]{16}\b")
|
|
|
|
|
|
def redact_text(value: str) -> str:
|
|
redacted = AUTH_VALUE.sub(lambda match: f"{match.group(1)} {REDACTED}", value)
|
|
redacted = SECRET_ASSIGNMENT.sub(lambda match: f"{match.group(1)}{REDACTED}", redacted)
|
|
redacted = SECRET_QUERY.sub(lambda match: f"{match.group(1)}{REDACTED}", redacted)
|
|
return AWS_ACCESS_KEY.sub(REDACTED, redacted)
|
|
|
|
|
|
def redact(value: Any, *, key: str | None = None) -> Any:
|
|
if key is not None and SENSITIVE_KEY.search(key):
|
|
return REDACTED
|
|
if isinstance(value, Mapping):
|
|
return {str(item_key): redact(item, key=str(item_key)) for item_key, item in value.items()}
|
|
if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
|
|
return [redact(item) for item in value]
|
|
if isinstance(value, str):
|
|
return redact_text(value)
|
|
return value
|
|
|
|
|
|
def _redacted_default(value: Any) -> str:
|
|
return redact_text(str(value))
|
|
|
|
|
|
class JsonFormatter(logging.Formatter):
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
payload: dict[str, Any] = {
|
|
"timestamp": datetime.now(UTC).isoformat(timespec="milliseconds"),
|
|
"level": record.levelname.lower(),
|
|
"message": redact_text(record.getMessage()),
|
|
"logger": record.name,
|
|
}
|
|
payload.update(
|
|
redact(
|
|
{
|
|
key: value
|
|
for key, value in record.__dict__.items()
|
|
if key not in STANDARD_FIELDS and not key.startswith("_")
|
|
}
|
|
)
|
|
)
|
|
if record.exc_info:
|
|
payload["exception"] = redact_text(self.formatException(record.exc_info))
|
|
return json.dumps(payload, ensure_ascii=False, default=_redacted_default)
|
|
|
|
|
|
def configure_logging(level: str) -> None:
|
|
handler = logging.StreamHandler()
|
|
handler.setFormatter(JsonFormatter())
|
|
root = logging.getLogger()
|
|
root.handlers.clear()
|
|
root.addHandler(handler)
|
|
root.setLevel(level.upper())
|
|
for logger_name in ("botocore", "boto3", "httpx", "httpcore"):
|
|
logging.getLogger(logger_name).setLevel(logging.WARNING)
|