| 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 |
|
|
