mi
48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from collections.abc import Mapping
|
|
from typing import Any
|
|
|
|
REDACTED = "[REDACTED]"
|
|
_SENSITIVE_KEY = re.compile(
|
|
r"(authorization|cookie|password|passwd|secret|token|api[_-]?key|"
|
|
r"database[_-]?url|redis[_-]?url|dsn|callback[_-]?url)",
|
|
re.IGNORECASE,
|
|
)
|
|
_URI_USERINFO = re.compile(r"(?P<scheme>[a-z][a-z0-9+.-]*://)[^/@\s]+@", re.IGNORECASE)
|
|
_QUERY_SECRET = re.compile(
|
|
r"(?P<prefix>[?&](?:token|access_token|api_key|key|secret|password)=)[^&#\s]+",
|
|
re.IGNORECASE,
|
|
)
|
|
_AUTH_VALUE = re.compile(r"\b(Bearer|Basic)\s+[A-Za-z0-9._~+/=-]+", re.IGNORECASE)
|
|
|
|
|
|
def sanitize_text(value: str) -> str:
|
|
value = _URI_USERINFO.sub(r"\g<scheme>[REDACTED]@", value)
|
|
value = _QUERY_SECRET.sub(r"\g<prefix>[REDACTED]", value)
|
|
return _AUTH_VALUE.sub(r"\1 [REDACTED]", value)
|
|
|
|
|
|
def sanitize_value(value: Any) -> Any:
|
|
if isinstance(value, str):
|
|
return sanitize_text(value)
|
|
if isinstance(value, Mapping):
|
|
return {
|
|
str(key): REDACTED if _SENSITIVE_KEY.search(str(key)) else sanitize_value(item)
|
|
for key, item in value.items()
|
|
}
|
|
if isinstance(value, list):
|
|
return [sanitize_value(item) for item in value]
|
|
if isinstance(value, tuple):
|
|
return tuple(sanitize_value(item) for item in value)
|
|
return value
|
|
|
|
|
|
def redact_event(
|
|
_logger: Any,
|
|
_method_name: str,
|
|
event_dict: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
return sanitize_value(event_dict)
|