"""Secret redaction for logs, exceptions and persisted job messages.""" from __future__ import annotations import re from typing import Any _PARAM_RE = re.compile(r"(?i)(userid|api_key|apikey|token|password|secret)=([^&\s\"']+)") def redact(text: Any, secrets: list[str] | None = None) -> str: s = str(text) for v in secrets or []: if v: s = s.replace(v, "") s = _PARAM_RE.sub(lambda m: f"{m.group(1)}=", s) return s class RedactingFilter: """structlog processor: redact every string value in the event dict.""" def __init__(self, secrets: list[str]): self.secrets = secrets def __call__(self, logger, method, event_dict): # type: ignore[no-untyped-def] for k, v in list(event_dict.items()): if isinstance(v, str): event_dict[k] = redact(v, self.secrets) return event_dict