File size: 3,538 Bytes
638cad0 | 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 | from __future__ import annotations
import re
from typing import Any
REDACTED_SECRET = "[REDACTED]"
REDACTED_EMAIL = "[REDACTED_EMAIL]"
_EMAIL_RE = re.compile(
r"(?<![A-Za-z0-9.!#$%&'*+/=?^_`{|}~-])"
r"[A-Za-z0-9.!#$%&'*+/=?^_`{|}~-]+@"
r"[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?"
r"(?:\.[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?)+"
)
_BEARER_RE = re.compile(r"(?i)(\bBearer\s+)[^\s,;\"']+")
_JWT_RE = re.compile(r"\beyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\b")
_QUERY_SECRET_RE = re.compile(
r"(?i)([?&](?:access_token|refresh_token|id_token|session_token|token|api_key|apikey|key|password|secret)=)[^&#\s]+"
)
_TEXT_SECRET_FIELD_RE = re.compile(
r"(?i)((?:\\?[\"'])(?:access_token|refresh_token|id_token|session_token|api_key|apikey|password|client_secret|"
r"authorization|cookie|cf_clearance)(?:\\?[\"'])\s*:\s*(?:\\?[\"']))([^\"'\\\s]+)((?:\\?[\"']))"
)
_PERSONAL_FIELD_KEYS = {
"account_email",
"_account_email",
"user_email",
"email",
}
_SECRET_FIELD_KEYS = {
"access_token",
"refresh_token",
"id_token",
"session_token",
"api_key",
"apikey",
"authorization",
"proxy_authorization",
"password",
"client_secret",
"secret",
"cookie",
"set_cookie",
"cf_clearance",
}
def _normalized_key(value: object) -> str:
return str(value or "").strip().lower().replace("-", "_")
def _is_secret_key(key: object) -> bool:
normalized = _normalized_key(key)
return (
normalized in _SECRET_FIELD_KEYS
or normalized.endswith("_access_token")
or normalized.endswith("_refresh_token")
or normalized.endswith("_session_token")
or normalized.endswith("_api_key")
or normalized.endswith("_password")
or normalized.endswith("_secret")
)
def sanitize_sensitive_text(value: object) -> str:
"""Redact credentials and account emails embedded in arbitrary text."""
text = str(value or "")
if not text:
return text
text = _BEARER_RE.sub(lambda match: f"{match.group(1)}{REDACTED_SECRET}", text)
text = _JWT_RE.sub(REDACTED_SECRET, text)
text = _QUERY_SECRET_RE.sub(lambda match: f"{match.group(1)}{REDACTED_SECRET}", text)
text = _TEXT_SECRET_FIELD_RE.sub(
lambda match: f"{match.group(1)}{REDACTED_SECRET}{match.group(3)}",
text,
)
return _EMAIL_RE.sub(REDACTED_EMAIL, text)
def sanitize_sensitive_data(value: Any, *, drop_personal_fields: bool = False) -> Any:
"""Recursively sanitize JSON-like data without mutating the input value."""
if isinstance(value, dict):
sanitized: dict[Any, Any] = {}
for key, item in value.items():
normalized = _normalized_key(key)
if normalized in _PERSONAL_FIELD_KEYS:
if not drop_personal_fields:
sanitized[key] = REDACTED_EMAIL
continue
if _is_secret_key(key):
sanitized[key] = REDACTED_SECRET
continue
sanitized[key] = sanitize_sensitive_data(item, drop_personal_fields=drop_personal_fields)
return sanitized
if isinstance(value, list):
return [sanitize_sensitive_data(item, drop_personal_fields=drop_personal_fields) for item in value]
if isinstance(value, tuple):
return tuple(sanitize_sensitive_data(item, drop_personal_fields=drop_personal_fields) for item in value)
if isinstance(value, str):
return sanitize_sensitive_text(value)
return value
|