"""Sentry data scrubbing utilities for removing sensitive information from logs and events.""" import json import re from re import Match from typing import Any class SentryScrubber: """A class for scrubbing sensitive data from Sentry events and general data structures.""" def __init__( self, sensitive_fields: list[str], scrub_marker: str = "[Scrubbed]", max_depth: int = 10, ): """Initialize the SentryScrubber. Args: sensitive_fields: List of field names to scrub (case-insensitive) scrub_marker: String to replace sensitive data with max_depth: Maximum recursion depth for scrubbing nested structures """ self.sensitive_fields = {field.lower() for field in sensitive_fields} self.scrub_marker = scrub_marker self.max_depth = max_depth # This pattern will match JSON objects with one level of nesting only. For example: # Will match: # {"key": "value"} # {"outer": {"inner": "value"}, "other": "data"} # {"a": {"b": "c"}, "d": "e", "f": {"g": "h"}} # Will NOT match: # {"level1": {"level2": {"level3": "value"}}} (more than one level deep) # Nested arrays with objects # Complex nested structures self.json_pattern = re.compile(r"\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}") def _scrub_json_match(self, match: Match[str]) -> str: """Scrub sensitive data from a JSON match.""" json_str = match.group(0) try: data = json.loads(json_str) scrubbed_data = self._scrub_sensitive_data(data) return json.dumps(scrubbed_data) except (json.JSONDecodeError, TypeError, UnicodeDecodeError): try: json_like_str = json_str.replace("'", '"') data = json.loads(json_like_str) scrubbed_data = self._scrub_sensitive_data(data) result = json.dumps(scrubbed_data) return result.replace('"', "'") except (json.JSONDecodeError, TypeError, UnicodeDecodeError): return str(json_str) def _scrub_sensitive_data_from_text(self, text: str) -> str: """Scrub sensitive data from text content (like error messages).""" try: return self.json_pattern.sub(self._scrub_json_match, text) except re.error: return text def _scrub_sensitive_data(self, data: Any, current_depth: int | None = None) -> Any: """Recursively scrub sensitive data from any data structure.""" if current_depth is None: current_depth = self.max_depth if current_depth < 0: return data if isinstance(data, dict): scrubbed = {} for key, value in data.items(): if isinstance(key, str) and key.lower() in self.sensitive_fields: scrubbed[key] = self.scrub_marker else: scrubbed[key] = self._scrub_sensitive_data(value, current_depth - 1) return scrubbed if isinstance(data, list): return [ self._scrub_sensitive_data(item, current_depth - 1) for item in data ] if isinstance(data, str): return self._scrub_sensitive_data_from_text(data) return data def before_send_handler( self, event: dict[str, Any] | None, hint: Any ) -> dict[str, Any] | None: """Sentry before_send callback to scrub sensitive data from events.""" if event is None: return None scrubbed_event = {} for key, value in event.items(): scrubbed_event[key] = self._scrub_sensitive_data(value) return scrubbed_event def scrub_data(self, data: Any) -> Any: """Public method to scrub sensitive data from any data structure.""" return self._scrub_sensitive_data(data)