import re
import json
from typing import Any, Dict, Optional
from opentelemetry.sdk.trace import SpanProcessor
from opentelemetry.sdk.trace.export import SpanExporter
from opentelemetry.sdk.trace import ReadableSpan
from opentelemetry.trace import Span
class PIIRedactingSpanProcessor(SpanProcessor):
def __init__(self, exporter: SpanExporter, pii_patterns: Optional[Dict[str, str]] = None):
self._exporter = exporter
self._default_patterns = {
"email": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
"phone": r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b",
"ssn": r"\b\d{3}-\d{2}-\d{4}\b",
"credit_card": r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b",
"ip_address": r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b",
"date_of_birth": r"\b\d{2}[-/]\d{2}[-/]\d{4}\b",
}
self._patterns = {**self._default_patterns, **(pii_patterns or {})}
self._compiled_patterns = {
name: re.compile(pattern) for name, pattern in self._patterns.items()
}
def _redact_string(self, value: str) -> str:
redacted = value
for pattern_name, pattern in self._compiled_patterns.items():
redacted = pattern.sub(f"[REDACTED_{pattern_name.upper()}]", redacted)
return redacted
def _redact_value(self, value: Any) -> Any:
if isinstance(value, str):
try:
json_obj = json.loads(value)
return json.dumps(self._redact_value(json_obj))
except json.JSONDecodeError:
return self._redact_string(value)
elif isinstance(value, dict):
return {k: self._redact_value(v) for k, v in value.items()}
elif isinstance(value, list):
return [self._redact_value(item) for item in value]
elif isinstance(value, (int, float, bool, type(None))):
return value
return self._redact_string(str(value))
def on_start(self, span: Span, parent_context=None):
pass
def on_end(self, span: ReadableSpan):
redacted_attributes = {}
for key, value in span.attributes.items():
if key in {"service.name", "telemetry.sdk.name", "telemetry.sdk.version"}:
redacted_attributes[key] = value
continue
redacted_attributes[key] = self._redact_value(value)
# Create redacted span copy and export
redacted_span = ReadableSpan(
name=self._redact_string(span.name),
context=span.get_span_context(),
parent=span.parent,
resource=span.resource,
attributes=redacted_attributes,
events=span.events,
links=span.links,
kind=span.kind,
status=span.status,
start_time=span.start_time,
end_time=span.end_time,
instrumentation_scope=span.instrumentation_scope,
)
self._exporter.export([redacted_span])
def shutdown(self):
self._exporter.shutdown()
def force_flush(self, timeout_millis: int = 30000):
self._exporter.force_flush(timeout_millis)