diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 30159b9b..b2a10c1d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -59,6 +59,7 @@ jobs: run: | pip install -e ".[test,cli,nlp,nlp-advanced]" -r requirements-test.txt pip install "litellm>=1.90,<2" fastapi # exercises the LiteLLM guardrail adapter tests (proxy deployments always have fastapi) + pip install "sentry-sdk>=2.35,<3" # exercises the Sentry integration adapter tests (2.35+ for before_send_log) python -m spacy download en_core_web_lg datafog download-model urchade/gliner_multi_pii-v1 --engine gliner diff --git a/CHANGELOG.MD b/CHANGELOG.MD index a2655567..c79e399b 100644 --- a/CHANGELOG.MD +++ b/CHANGELOG.MD @@ -2,6 +2,30 @@ ## [Unreleased] +### Added + +- **Sentry SDK integration** (`datafog.integrations.sentry`): scrubs PII + from Sentry events in-process, before they leave the machine. Add + `DataFogSentryIntegration()` to `sentry_sdk.init(integrations=[...])` — + it registers a global event processor covering error events *and* + transactions (messages, exception values, stack-frame local variables, + breadcrumbs, request data, `extra`, tags, span descriptions), so the + application's own `before_send` slot stays free. A standalone + `scrub_event(event, hint)` is also exported for manual `before_send` + wiring. Same configuration surface as the other adapters: + `entity_types` (default `EMAIL`, `PHONE`, `CREDIT_CARD`, `SSN`), + `allowlist` / `allowlist_patterns`, and `fail_policy` (`open` sends the + event unscanned on engine error; `closed` drops it). Matched values are + never echoed into logs — counts only. Beyond events, the integration + also scrubs Sentry Logs, trace metrics, and experimental streamed spans + by wrapping `before_send_log` / `before_send_metric` / + `_experiments["before_send_span"]` at init (scrub first, then chain to + the application's own callback), and covers the `gen_ai.*` + prompt/completion span attributes captured by Sentry AI agent + monitoring, which Sentry's server-side scrubbing does not reach. + Install with `pip install datafog[sentry]` (any sentry-sdk 2.x; logs + and metrics scrubbing engage on SDK versions that support them). + ## [2026-07-05] ### `datafog-python` [4.8.0] diff --git a/datafog/integrations/sentry.py b/datafog/integrations/sentry.py new file mode 100644 index 00000000..6d578594 --- /dev/null +++ b/datafog/integrations/sentry.py @@ -0,0 +1,433 @@ +"""Sentry SDK integration: scrub PII from events before they leave the process. + +Usage:: + + import sentry_sdk + from datafog.integrations.sentry import DataFogSentryIntegration + + sentry_sdk.init( + dsn="...", + integrations=[DataFogSentryIntegration()], + ) + +Or wire the scrubber into ``before_send`` directly:: + + sentry_sdk.init(dsn="...", before_send=scrub_event) + +Behavior: + +- The integration registers a *global event processor*, so it scrubs both + error events and transactions (span descriptions and data ride inside + transaction events — including the ``gen_ai.*`` prompt/completion + attributes captured by Sentry's AI agent monitoring, which Sentry's + server-side scrubbing does not cover) while leaving the single + ``before_send`` slot free for the application's own callback. +- On SDKs that support them, Sentry Logs, trace metrics, and streamed + spans are scrubbed too: the integration wraps ``before_send_log``, + ``before_send_metric``, and ``_experiments["before_send_span"]`` at + init, running the scrub first and then chaining to the application's + own callback (which therefore only ever sees scrubbed telemetry). The + wrap happens the first time the integration is set up in a process — + call ``sentry_sdk.init`` once, as Sentry recommends. +- Scrubbing is content-based: every string value in the event — messages, + exception values, stack-frame local variables, breadcrumbs, request + data, ``extra``, tags, span descriptions — is scanned and findings are + replaced with ``[TYPE_N]`` tokens. This catches PII that Sentry's + built-in ``EventScrubber`` misses, because that scrubber only matches + sensitive *key names* and never inspects values. +- ``fail_policy`` — ``open`` (default) sends the event unscrubbed if the + engine errors, so a scrubber bug never costs you crash reports; + ``closed`` drops the event instead, for compliance deployments where + unscanned egress is worse than a missing event. + +Errors report entity type counts only — matched PII is never echoed into +logs or exceptions. + +Requires ``sentry-sdk`` >= 2.0 (this module is not imported by ``datafog`` +core; install with ``pip install datafog[sentry]`` or alongside your app's +existing sentry-sdk). +""" + +import logging +from typing import Any, Callable, Optional + +import sentry_sdk +from sentry_sdk.integrations import Integration + +# High-precision defaults, matching the LiteLLM guardrail and Claude Code +# hook adapters: noisy-in-practice types (IP_ADDRESS, DOB, ZIP) must be +# opted into explicitly. +DEFAULT_ENTITY_TYPES = ["EMAIL", "PHONE", "CREDIT_CARD", "SSN"] + +VALID_FAIL_POLICIES = {"open", "closed"} + +# Per-string and per-event scan caps, mirroring the Claude Code hook: a +# pathological payload (huge repr in a frame var, giant request body) must +# not stall event delivery. Text beyond a cap is REPLACED with an +# [UNSCANNED_N_CHARS] marker, never sent raw: emitting an unscanned tail +# would let a PII value that straddles the cut point reassemble unredacted, +# and unscanned egress is the one thing this integration must not do. Both +# caps sit far above Sentry's own event-size limits, so real events never +# reach them. +MAX_SCAN_CHARS = 1_000_000 +TOTAL_SCAN_CHARS = 8_000_000 + +# Top-level event fields that are machine identifiers, not free text. +# Skipping them avoids mangling values that only *look* like PII (a release +# tag containing an @, a server hostname) and matches what Sentry's own +# EventScrubber leaves alone. The skip applies at the event's top level +# only — a nested key that happens to be named "release" (e.g. inside +# ``extra``) is still scanned. This set is shared by the telemetry scrub +# path (logs/metrics/spans): the overlapping keys there ("type", +# "timestamp") are enum/numeric in the Sentry protocol, so skipping them +# is currently safe — revisit if a telemetry schema grows free text under +# one of these names. +_SKIP_TOP_LEVEL_KEYS = frozenset( + { + "event_id", + "timestamp", + "start_timestamp", + "received", + "platform", + "sdk", + "modules", + "release", + "dist", + "environment", + "server_name", + "type", + "level", + } +) + +logger = logging.getLogger(__name__) + + +class _EventScrubber: + """Single-event scrub pass: walks the event, redacts string values. + + Mutates containers in place — deliberate: the Sentry processor contract + is mutate-and-return, and deep-copying every event would double memory + while leaving an unredacted copy other processors could still see. + + A fresh instance is created per event (see ``_scrub_with_policy``), so + ``counts``, ``_budget``, and ``_seen`` are never shared across events or + threads. Do not hoist construction out of the per-event path: a reused + instance would leak budget and alias state between events. + """ + + def __init__( + self, + entity_types: list[str], + allowlist: Optional[list[str]], + allowlist_patterns: Optional[list[str]], + ) -> None: + self.entity_types = entity_types + self.allowlist = allowlist + self.allowlist_patterns = allowlist_patterns + self.counts: dict[str, int] = {} + self.unscanned_chars = 0 + self._budget = TOTAL_SCAN_CHARS + self._seen: set[int] = set() + + def _unscanned(self, length: int) -> str: + self.unscanned_chars += length + return f"[UNSCANNED_{length}_CHARS]" + + def _redact_string(self, text: str) -> str: + limit = min(MAX_SCAN_CHARS, self._budget) + if limit <= 0: + return self._unscanned(len(text)) + import datafog + + chunk = text[:limit] + self._budget -= len(chunk) + result = datafog.redact( + chunk, + engine="regex", + entity_types=self.entity_types, + allowlist=self.allowlist, + allowlist_patterns=self.allowlist_patterns, + ) + for entity in result.entities: + self.counts[entity.type] = self.counts.get(entity.type, 0) + 1 + tail = text[len(chunk) :] + if tail: + # Never reattach the unscanned tail raw: a PII value straddling + # the cut point would fail to match in the truncated chunk and + # reassemble unredacted in the output. + return result.redacted_text + self._unscanned(len(tail)) + return result.redacted_text if result.entities else text + + def _transform(self, value: Any, stack: list[Any]) -> Any: + """Redact a string, or queue a container for walking. + + Tuples and sets are rebuilt as lists — the Sentry serializer emits + all three as JSON arrays, and a list can be walked and mutated in + place. ``_seen`` dedupes aliased containers, so a structure reachable + from two places is scanned once (double-scanning would double-spend + the budget) and cyclic structures terminate. Other value types + (numbers, custom objects) pass through untouched: the SDK reprs + custom objects only after processors run, so their contents are + out of reach here. + """ + if isinstance(value, str): + return self._redact_string(value) + if isinstance(value, (tuple, set, frozenset)): + value = list(value) + if isinstance(value, (dict, list)) and id(value) not in self._seen: + self._seen.add(id(value)) + stack.append(value) + return value + + def scrub(self, event: dict) -> dict: + """Redact every string value in the event. + + Iterative (explicit stack), so adversarially deep structures in + ``extra`` or frame vars cannot trigger ``RecursionError`` and + silently skip the scan. Dict *keys* are left alone, matching + Sentry's EventScrubber. + """ + stack: list[Any] = [] + for key in list(event.keys()): + if key in _SKIP_TOP_LEVEL_KEYS: + continue + new_value = self._transform(event[key], stack) + if new_value is not event[key]: + event[key] = new_value + while stack: + current = stack.pop() + if isinstance(current, dict): + for key in list(current.keys()): + new_value = self._transform(current[key], stack) + if new_value is not current[key]: + current[key] = new_value + else: # list — _transform only queues dicts and lists + for index, value in enumerate(current): + new_value = self._transform(value, stack) + if new_value is not value: + current[index] = new_value + if self.unscanned_chars: + logger.debug( + "DataFog Sentry scrubber: %d chars beyond scan caps replaced " + "with [UNSCANNED_N_CHARS] markers", + self.unscanned_chars, + ) + return event + + +def _summary(counts: dict[str, int]) -> str: + return ", ".join(f"{etype} x{n}" for etype, n in sorted(counts.items())) + + +def _scrub_with_policy( + event: dict, + entity_types: list[str], + allowlist: Optional[list[str]], + allowlist_patterns: Optional[list[str]], + fail_policy: str, +) -> Optional[dict]: + scrubber = _EventScrubber(entity_types, allowlist, allowlist_patterns) + try: + scrubbed = scrubber.scrub(event) + except Exception as exc: # noqa: BLE001 — fail policy decides + # Only the exception *type* is ever logged. Engine exception + # messages can embed the text being scanned, so interpolating + # str(exc) would leak the PII this integration exists to contain. + if fail_policy == "closed": + logger.warning( + "DataFog Sentry scrubber failed and fail_policy is 'closed'; " + "dropping event (%s).", + type(exc).__name__, + ) + return None + logger.warning( + "DataFog Sentry scrubber error (fail-open, event sent unscanned " + "beyond partial redaction): %s", + type(exc).__name__, + ) + return event + if scrubber.counts: + logger.debug("DataFog Sentry scrubber redacted: %s", _summary(scrubber.counts)) + return scrubbed + + +def scrub_event( + event: dict, + hint: Optional[dict] = None, + *, + entity_types: Optional[list[str]] = None, + allowlist: Optional[list[str]] = None, + allowlist_patterns: Optional[list[str]] = None, + fail_policy: str = "open", +) -> Optional[dict]: + """Scrub a Sentry event dict; drop-in ``before_send`` callback. + + The two positional parameters match Sentry's ``before_send(event, hint)`` + signature so the function can be passed directly. The keyword-only + options mirror :class:`DataFogSentryIntegration`; to preconfigure them, + wrap in ``functools.partial`` or use the integration instead. + """ + if fail_policy not in VALID_FAIL_POLICIES: + raise ValueError(f"fail_policy must be one of: {sorted(VALID_FAIL_POLICIES)}") + return _scrub_with_policy( + event, + entity_types or DEFAULT_ENTITY_TYPES, + allowlist, + allowlist_patterns, + fail_policy, + ) + + +class DataFogSentryIntegration(Integration): + """Content-based PII scrubbing for the Sentry Python SDK.""" + + identifier = "datafog" + + def __init__( + self, + entity_types: Optional[list[str]] = None, + allowlist: Optional[list[str]] = None, + allowlist_patterns: Optional[list[str]] = None, + fail_policy: str = "open", + ) -> None: + if fail_policy not in VALID_FAIL_POLICIES: + raise ValueError( + f"fail_policy must be one of: {sorted(VALID_FAIL_POLICIES)}" + ) + self.entity_types = entity_types or DEFAULT_ENTITY_TYPES + self.allowlist = allowlist + self.allowlist_patterns = allowlist_patterns + self.fail_policy = fail_policy + + @staticmethod + def setup_once() -> None: + # Registered once per process. The processor resolves the integration + # instance from the *current* client on every event, so config always + # comes from the active ``sentry_sdk.init`` call and clients without + # this integration are passed through untouched. + from sentry_sdk.scope import add_global_event_processor + + add_global_event_processor(_global_processor) + + def setup_once_with_options(self, options: Optional[dict] = None) -> None: + """Wrap the telemetry ``before_send_*`` callbacks in ``options``. + + Logs, trace metrics, and streamed spans bypass event processors — + their only interception point is the ``before_send_log`` / + ``before_send_metric`` / ``_experiments["before_send_span"]`` + options, which the client re-reads from this same dict on every + capture. The wrappers scrub first, then chain to the application's + original callback. Like ``setup_once``, this runs only the first + time the integration identifier is processed per process, so a + second ``sentry_sdk.init`` gets event scrubbing (via the global + processor) but not telemetry wrapping. + + Called by sentry-sdk >= 2.14; on older 2.x the SDK predates logs + and metrics, so there is nothing to wrap. + """ + if options is None: + return + # ``experiments`` is the live dict shared with the client's options, + # not a copy — mutations here are what the client reads at capture. + experiments = options.get("_experiments") + for key in ("before_send_log", "before_send_metric"): + wrapped = _wrap_telemetry_callback(_existing_callback(options, key)) + options[key] = wrapped + if isinstance(experiments, dict) and key in experiments: + # Overwrite the legacy location too: leaving the raw user + # callback reachable there is a latent unscrubbed path for + # anything that reads _experiments directly. + experiments[key] = wrapped + if isinstance(experiments, dict): + # before_send_span is read from _experiments only. + experiments["before_send_span"] = _wrap_telemetry_callback( + experiments.get("before_send_span") + ) + + +_TelemetryCallback = Callable[[dict, Optional[dict]], Optional[dict]] + + +def _existing_callback(options: dict, key: str) -> Optional[_TelemetryCallback]: + """Resolve the user's callback with the SDK's precedence: top-level + option first, then the legacy ``_experiments`` location.""" + value = options.get(key) + if value is not None: + return value + experiments = options.get("_experiments") + if isinstance(experiments, dict): + return experiments.get(key) + return None + + +def _wrap_telemetry_callback( + original: Optional[_TelemetryCallback], +) -> _TelemetryCallback: + """Compose scrubbing with the user's ``before_send_*`` callback. + + Scrub runs first, so the user callback only ever sees scrubbed + telemetry; a ``None`` from the scrub (fail_policy ``closed`` on engine + error) drops the payload without invoking the user callback. Config is + resolved from the current client per payload, mirroring + ``_global_processor``. + + Idempotent: wrapping an already-wrapped callback returns it unchanged, + so a repeated ``setup_once_with_options`` cannot stack scrub passes. + """ + if getattr(original, "_datafog_wrapped", False): + return original # type: ignore[return-value] + + def wrapped(payload: dict, hint: Optional[dict]) -> Optional[dict]: + try: + integration = sentry_sdk.get_client().get_integration( + DataFogSentryIntegration + ) + except Exception as exc: # noqa: BLE001 — never break delivery + logger.warning( + "DataFog Sentry integration lookup failed (telemetry " + "passed through unscanned): %s", + type(exc).__name__, + ) + integration = None + if integration is not None: + scrubbed = _scrub_with_policy( + payload, + integration.entity_types, + integration.allowlist, + integration.allowlist_patterns, + integration.fail_policy, + ) + if scrubbed is None: + return None + payload = scrubbed + if original is not None: + return original(payload, hint) + return payload + + wrapped._datafog_wrapped = True # type: ignore[attr-defined] + return wrapped + + +def _global_processor(event: dict, hint: Optional[dict]) -> Optional[dict]: + try: + integration = sentry_sdk.get_client().get_integration(DataFogSentryIntegration) + except Exception as exc: # noqa: BLE001 — never break event delivery + # The lookup runs outside _scrub_with_policy's error handling and no + # integration (hence no fail_policy) is resolvable here, so the only + # safe default is to pass the event through. + logger.warning( + "DataFog Sentry integration lookup failed (event passed " + "through unscanned): %s", + type(exc).__name__, + ) + return event + if integration is None: + return event + return _scrub_with_policy( + event, + integration.entity_types, + integration.allowlist, + integration.allowlist_patterns, + integration.fail_policy, + ) diff --git a/examples/sentry/README.md b/examples/sentry/README.md new file mode 100644 index 00000000..370f6a68 --- /dev/null +++ b/examples/sentry/README.md @@ -0,0 +1,109 @@ +# DataFog PII Scrubbing for Sentry + +Scrub PII from every Sentry event before it leaves your process — offline, +in-process, microseconds per string scanned. + +## Why this over Sentry's built-in scrubbing + +| | DataFog integration | Sentry `EventScrubber` | Server-side scrubbing | +| ------------- | ----------------------------------------------------------------- | --------------------------------- | -------------------------------- | +| Detection | content-based (validated regex) | sensitive _key names_ only | basic regexes, no phone detector | +| Where it runs | in your process, pre-egress | in your process | after data reaches Sentry | +| Catches | PII inside messages, exception text, local variables, breadcrumbs | values under keys like `password` | whatever survives the SDK | + +Sentry's client-side `EventScrubber` never inspects values: an email in an +exception message, an SSN in a stack-frame local, or a card number in a +breadcrumb sails straight through. Server-side scrubbing inspects content, +but only after the data has already left your machine. + +## Install + +```bash +pip install datafog[sentry] # or: pip install datafog sentry-sdk +``` + +```python +import sentry_sdk +from datafog.integrations.sentry import DataFogSentryIntegration + +sentry_sdk.init( + dsn="https://...@o0.ingest.sentry.io/0", + integrations=[DataFogSentryIntegration()], +) +``` + +An exception message like + +> could not send reset link to jane.doe@example.invalid + +arrives in Sentry as + +> could not send reset link to [EMAIL_1] + +The integration registers a global event processor, so it covers both error +events and transactions (span descriptions and data), and your own +`before_send` stays free — it runs after the scrub and sees clean events. + +Prefer explicit wiring? The scrubber is also a drop-in `before_send`: + +```python +from datafog.integrations.sentry import scrub_event + +sentry_sdk.init(dsn="...", before_send=scrub_event) +``` + +## What gets scrubbed + +Every string value in the event, including: + +- `message` / `logentry` and exception values +- **stack-frame local variables** (`include_local_variables` is on by + default in sentry-sdk — every error event snapshots live values) +- breadcrumbs (messages and data — log lines, SQL, HTTP query strings) +- request URL, headers, cookies, and body +- `extra`, `tags`, `user`, and contexts +- span descriptions and data in transactions — including the + **`gen_ai.*` prompt/completion attributes** captured by Sentry's AI + agent monitoring, which Sentry's server-side scrubbing does not cover. + With this integration you can keep `include_prompts` on for AI + observability without shipping your users' PII to Sentry. + +Machine identifiers (`event_id`, `release`, `environment`, `sdk`, +`modules`, ...) are never touched. Text beyond the (very generous) +per-event scan caps is replaced with an `[UNSCANNED_N_CHARS]` marker — +never sent raw. + +## Logs and metrics + +[Sentry Logs](https://docs.sentry.io/platforms/python/logs/) (sentry-sdk ≥ +2.35, `enable_logs=True`) and trace metrics (≥ 2.44) bypass event +processors, so the integration wraps `before_send_log`, +`before_send_metric`, and the experimental `before_send_span` at init: +the scrub runs first, then your own callback — which therefore only ever +sees scrubbed telemetry. Log bodies, log/metric attributes (including the +auto-attached `user.email`), and streamed span descriptions are all +covered. The wrap engages the first time the integration is set up in a +process, so call `sentry_sdk.init` once, as Sentry recommends. + +## Options + +```python +DataFogSentryIntegration( + entity_types=["EMAIL", "PHONE", "CREDIT_CARD", "SSN"], # default set + allowlist=["support@yourco.example"], # exact values to exempt + allowlist_patterns=[r".*@yourco\.example"], # full-match regexes + fail_policy="open", # or "closed" +) +``` + +- `entity_types`: defaults to `EMAIL, PHONE, CREDIT_CARD, SSN`; noisier + types (`IP_ADDRESS`, `DOB`, `ZIP`) are opt-in +- `allowlist` / `allowlist_patterns`: exempt known-safe values (your own + support address, test fixtures) from redaction +- `fail_policy`: `open` (engine error → event is sent unscanned, so a + scrubber bug never costs you crash reports) or `closed` (engine error → + event is dropped; for compliance deployments where unscanned egress is + worse than a missing event) + +Findings are reported as entity-type counts only — matched values are never +echoed into logs or exceptions. diff --git a/setup.py b/setup.py index 1da9ebe3..c641a8ba 100644 --- a/setup.py +++ b/setup.py @@ -81,6 +81,12 @@ "pytest-benchmark>=4.0.0", ] +# Convenience pin for the Sentry integration; apps that already ship +# sentry-sdk can import datafog.integrations.sentry without this extra. +sentry_deps = [ + "sentry-sdk>=2.0,<3.0", +] + extras_require = { "nlp": nlp_deps, "nlp-advanced": nlp_advanced_deps, @@ -89,6 +95,7 @@ "web": web_deps, "cli": cli_deps, "crypto": crypto_deps, + "sentry": sentry_deps, "test": test_deps, "docs": docs_deps, "benchmark": benchmark_deps, diff --git a/tests/test_sentry_integration.py b/tests/test_sentry_integration.py new file mode 100644 index 00000000..6f33a1f5 --- /dev/null +++ b/tests/test_sentry_integration.py @@ -0,0 +1,600 @@ +"""Tests for the Sentry SDK integration (DataFogSentryIntegration). + +PII literals below are split ("jane.doe@" "acme.com") so this source file +itself never contains a contiguous match — the values only assemble at +runtime. This keeps write-time PII scanners (including our own Claude Code +hook) quiet while the tests exercise real detections. +""" + +import json + +import pytest + +sentry_sdk = pytest.importorskip("sentry_sdk") + +from datafog.integrations.sentry import ( # noqa: E402 + DataFogSentryIntegration, + scrub_event, +) + +EMAIL = "jane.doe@" "acme.com" +CARD = "4242 4242 " "4242 4242" +SSN = "856-45-" "6789" +PHONE = "(555) 867-" "5309" + + +def _error_event(**overrides) -> dict: + event = { + "event_id": "0123456789abcdef0123456789abcdef", + "platform": "python", + "release": "myapp@1.2.3", + "logentry": {"message": f"lookup failed for {EMAIL}", "params": []}, + "exception": { + "values": [ + { + "type": "ValueError", + "value": f"could not bill card {CARD}", + "stacktrace": { + "frames": [ + { + "function": "charge", + "vars": { + "customer_ssn": f"{SSN!r}", + "attempts": "3", + "form": {"contact": f"{EMAIL!r}"}, + }, + } + ] + }, + } + ] + }, + "breadcrumbs": { + "values": [ + { + "message": f"sms sent to {PHONE}", + "category": "notify", + "data": {"to": PHONE}, + } + ] + }, + "request": { + "url": f"https://api.example.com/users?email={EMAIL}", + "headers": {"X-Contact": EMAIL}, + "data": {"ssn": SSN}, + }, + "extra": {"note": f"customer email {EMAIL}", "nested": [{"card": CARD}]}, + "tags": {"support_contact": EMAIL}, + "user": {"email": EMAIL, "id": "42"}, + } + event.update(overrides) + return event + + +def _transaction_event() -> dict: + return { + "type": "transaction", + "transaction": f"/users/{EMAIL}", + "spans": [ + { + "op": "db.query", + "description": f"SELECT * FROM users WHERE email = {EMAIL!r}", + "data": {"db.statement": f"... ssn = {SSN!r} ..."}, + } + ], + } + + +class TestScrubEvent: + def test_scrubs_all_error_event_surfaces(self): + event = scrub_event(_error_event(), None) + flat = json.dumps(event) + for value in (EMAIL, CARD, SSN, PHONE): + assert value not in flat + assert "[EMAIL_1]" in event["logentry"]["message"] + assert "[CREDIT_CARD_1]" in event["exception"]["values"][0]["value"] + frame_vars = event["exception"]["values"][0]["stacktrace"]["frames"][0]["vars"] + assert SSN not in frame_vars["customer_ssn"] + assert EMAIL not in frame_vars["form"]["contact"] + + def test_scrubs_transaction_spans(self): + event = scrub_event(_transaction_event(), None) + flat = json.dumps(event) + assert EMAIL not in flat + assert SSN not in flat + + def test_clean_event_unchanged(self): + event = { + "event_id": "0123456789abcdef0123456789abcdef", + "logentry": {"message": "connection refused"}, + "extra": {"retries": "3"}, + } + assert scrub_event(json.loads(json.dumps(event)), None) == event + + def test_skip_keys_untouched(self): + # Machine identifiers are never scanned, even when a value happens + # to look like PII (a release tag is not an email). + event = _error_event(release=f"deploy-by-{EMAIL}") + scrubbed = scrub_event(event, None) + assert scrubbed["release"] == f"deploy-by-{EMAIL}" + assert scrubbed["event_id"] == "0123456789abcdef0123456789abcdef" + + def test_allowlist_exempts_exact_value(self): + event = scrub_event(_error_event(), None, allowlist=[EMAIL]) + assert event["tags"]["support_contact"] == EMAIL + assert CARD not in json.dumps(event) + + def test_allowlist_patterns_exempt_fullmatch(self): + event = scrub_event(_error_event(), None, allowlist_patterns=[r".*@acme\.com"]) + assert event["tags"]["support_contact"] == EMAIL + + def test_entity_types_restrict_detection(self): + event = scrub_event(_error_event(), None, entity_types=["CREDIT_CARD"]) + flat = json.dumps(event) + assert CARD not in flat + assert EMAIL in flat + + def test_non_string_values_survive(self): + event = { + "extra": {"count": 3, "ok": True, "none": None, "pi": 3.14}, + "logentry": {"message": f"contact {EMAIL}"}, + } + scrubbed = scrub_event(event, None) + assert scrubbed["extra"] == {"count": 3, "ok": True, "none": None, "pi": 3.14} + assert EMAIL not in scrubbed["logentry"]["message"] + + def test_budget_overflow_replaced_with_marker_never_raw(self, monkeypatch): + # Once the per-event budget is exhausted, remaining strings are + # replaced with [UNSCANNED_N_CHARS] markers — never sent raw, which + # would be unscanned PII egress. + from datafog.integrations import sentry as sentry_module + + # The walker drains its stack LIFO, so the logentry message (30 + # chars) is scanned first and exactly exhausts the budget; the extra + # string then takes the whole-string marker path. + monkeypatch.setattr(sentry_module, "TOTAL_SCAN_CHARS", 30) + event = { + "extra": {"a": f"first field has {EMAIL} inside it"}, + "logentry": {"message": f"second field {EMAIL}"}, + } + scrubbed = scrub_event(event, None) + flat = json.dumps(scrubbed) + assert EMAIL not in flat + assert "[EMAIL_1]" in scrubbed["logentry"]["message"] + assert scrubbed["extra"]["a"] == "[UNSCANNED_43_CHARS]" + + def test_pii_straddling_scan_boundary_never_reassembles(self, monkeypatch): + # A PII value split by the per-string cap fails to match in the + # truncated chunk; the tail must be masked, not reattached raw — + # otherwise the full value reassembles unredacted in the output. + from datafog.integrations import sentry as sentry_module + + monkeypatch.setattr(sentry_module, "MAX_SCAN_CHARS", 24) + event = {"logentry": {"message": f"padding here {EMAIL} after"}} + message = scrub_event(event, None)["logentry"]["message"] + assert EMAIL not in message + assert "[UNSCANNED_" in message + + def test_tuple_and_set_values_scrubbed(self): + # Sentry frame vars and user extra can carry tuples pre-serialization; + # they are rebuilt as lists (the serializer emits both as JSON arrays). + event = { + "extra": { + "pair": (f"first {EMAIL}", f"second {CARD}"), + "unique": {f"member {SSN}"}, + } + } + flat = json.dumps(scrub_event(event, None)) + for value in (EMAIL, CARD, SSN): + assert value not in flat + + def test_aliased_container_scanned_once(self, monkeypatch): + # The same object reachable from two places must be scanned once: + # double-scanning double-spends the budget, which could starve later + # fields into the unscanned path. + import datafog + + calls = [] + real_redact = datafog.redact + + def counting_redact(text, **kwargs): + calls.append(text) + return real_redact(text, **kwargs) + + monkeypatch.setattr(datafog, "redact", counting_redact) + shared = {"contact": f"reach me at {EMAIL}"} + event = {"extra": {"first": shared, "second": shared}} + scrubbed = scrub_event(event, None) + assert len(calls) == 1 + assert EMAIL not in json.dumps(scrubbed) + + def test_cyclic_structure_terminates(self): + cyclic = {"note": f"contact {EMAIL}"} + cyclic["self"] = cyclic + event = {"extra": cyclic} + scrubbed = scrub_event(event, None) + assert EMAIL not in scrubbed["extra"]["note"] + + def test_scrub_event_rejects_invalid_fail_policy(self): + with pytest.raises(ValueError): + scrub_event(_error_event(), None, fail_policy="sideways") + + +class TestFailPolicy: + def test_fail_open_returns_event_on_engine_error(self, monkeypatch): + import datafog + + def boom(*args, **kwargs): + raise RuntimeError("engine exploded") + + monkeypatch.setattr(datafog, "redact", boom) + event = _error_event() + assert scrub_event(event, None, fail_policy="open") is event + + def test_fail_closed_drops_event_on_engine_error(self, monkeypatch): + import datafog + + def boom(*args, **kwargs): + raise RuntimeError("engine exploded") + + monkeypatch.setattr(datafog, "redact", boom) + assert scrub_event(_error_event(), None, fail_policy="closed") is None + + def test_error_logging_never_echoes_event_content(self, monkeypatch, caplog): + import datafog + + def boom(*args, **kwargs): + raise RuntimeError(f"engine choked on {EMAIL}") + + monkeypatch.setattr(datafog, "redact", boom) + with caplog.at_level("WARNING", logger="datafog.integrations.sentry"): + scrub_event(_error_event(), None, fail_policy="open") + assert EMAIL not in caplog.text + + def test_invalid_fail_policy_rejected(self): + with pytest.raises(ValueError): + DataFogSentryIntegration(fail_policy="sideways") + + +class TestIntegrationWiring: + """End-to-end through a real sentry_sdk client. + + Events are captured in ``before_send`` — which runs *after* global + event processors, so it sees exactly what the integration produces — + and dropped by returning None, so nothing touches the network. + """ + + def _init(self, captured, **integration_kwargs): + def _capture(event, hint): + captured.append(event) + return None + + sentry_sdk.init( + dsn="https://key@example.invalid/1", + integrations=[DataFogSentryIntegration(**integration_kwargs)], + default_integrations=False, + before_send=_capture, + ) + + def test_capture_message_scrubbed(self): + captured = [] + self._init(captured) + sentry_sdk.capture_message(f"reset link sent to {EMAIL}") + assert len(captured) == 1 + flat = json.dumps(captured[0]) + assert EMAIL not in flat + assert "[EMAIL_1]" in flat + + def test_capture_exception_scrubs_message_and_local_vars(self): + captured = [] + self._init(captured) + try: + customer_ssn = SSN # noqa: F841 — captured as a frame local + raise ValueError(f"invalid ssn for {EMAIL}") + except ValueError: + sentry_sdk.capture_exception() + assert len(captured) == 1 + flat = json.dumps(captured[0]) + assert EMAIL not in flat + assert SSN not in flat + + def test_integration_absent_is_passthrough(self): + # The global processor stays registered for the life of the process; + # with no DataFogSentryIntegration on the current client it must not + # touch events. + captured = [] + + def _capture(event, hint): + captured.append(event) + return None + + sentry_sdk.init( + dsn="https://key@example.invalid/1", + default_integrations=False, + before_send=_capture, + ) + sentry_sdk.capture_message(f"contact {EMAIL}") + assert len(captured) == 1 + assert EMAIL in json.dumps(captured[0]) + + def test_integration_allowlist_respected(self): + captured = [] + self._init(captured, allowlist=[EMAIL]) + sentry_sdk.capture_message(f"support contact {EMAIL}, card {CARD}") + flat = json.dumps(captured[0]) + assert EMAIL in flat + assert CARD not in flat + + def test_processor_lookup_failure_passes_event_through(self, monkeypatch): + # The integration lookup runs outside the fail-policy path; if it + # raises, event delivery must survive (pass-through, fail open). + from datafog.integrations import sentry as sentry_module + + def boom(): + raise RuntimeError("client lookup exploded") + + monkeypatch.setattr(sentry_module.sentry_sdk, "get_client", boom) + event = {"logentry": {"message": "hello"}} + assert sentry_module._global_processor(event, None) is event + + +class TestGenAiSpans: + """AI agent monitoring stores full prompts/completions in gen_ai.* span + attributes — Sentry documents that default server-side scrubbing does + NOT cover them, so the walker must.""" + + def test_gen_ai_message_lists_scrubbed(self): + event = { + "type": "transaction", + "spans": [ + { + "op": "gen_ai.chat", + "data": { + "gen_ai.input.messages": [ + {"role": "user", "content": f"my ssn is {SSN}"} + ], + "gen_ai.output.messages": [ + { + "role": "assistant", + "content": f"emailing {EMAIL} now", + } + ], + }, + } + ], + } + flat = json.dumps(scrub_event(event, None)) + assert SSN not in flat + assert EMAIL not in flat + + def test_gen_ai_json_string_attributes_scrubbed(self): + # Some SDK versions serialize prompt messages to a JSON string + # before attaching them; the walker sees one big string. + payload = json.dumps([{"role": "user", "content": f"card {CARD}"}]) + event = { + "type": "transaction", + "spans": [{"data": {"gen_ai.input.messages": payload}}], + } + assert CARD not in json.dumps(scrub_event(event, None)) + + +class _StubClient: + def __init__(self, integration): + self._integration = integration + + def get_integration(self, name_or_class): + return self._integration + + +class TestTelemetryWrapping: + """Logs, metrics, and streamed spans have no event-processor path; the + integration composes wrappers around the user's before_send_* callbacks + via setup_once_with_options.""" + + def _wrapped_options(self, integration=None, **options): + options.setdefault("_experiments", {}) + (integration or DataFogSentryIntegration()).setup_once_with_options(options) + return options + + def _use_client_integration(self, monkeypatch, integration): + from datafog.integrations import sentry as sentry_module + + monkeypatch.setattr( + sentry_module.sentry_sdk, + "get_client", + lambda: _StubClient(integration), + ) + + def _log(self): + return { + "severity_text": "info", + "severity_number": 9, + "body": f"password reset for {EMAIL}", + "attributes": {"user.email": EMAIL, "server.port": 443}, + "time_unix_nano": 1, + "trace_id": "ab" * 16, + } + + def test_before_send_log_scrubs_body_and_attributes(self, monkeypatch): + self._use_client_integration(monkeypatch, DataFogSentryIntegration()) + options = self._wrapped_options() + log = options["before_send_log"](self._log(), None) + assert EMAIL not in json.dumps(log) + assert "[EMAIL_1]" in log["body"] + assert log["severity_text"] == "info" + assert log["attributes"]["server.port"] == 443 + + def test_wrapped_log_chains_to_user_callback(self, monkeypatch): + self._use_client_integration(monkeypatch, DataFogSentryIntegration()) + seen = [] + + def user_callback(log, hint): + seen.append(log) + return None # user drops the log + + options = self._wrapped_options(before_send_log=user_callback) + result = options["before_send_log"](self._log(), None) + assert result is None + assert len(seen) == 1 + assert EMAIL not in json.dumps(seen[0]) # user saw the scrubbed log + + def test_user_callback_in_experiments_still_chained(self, monkeypatch): + self._use_client_integration(monkeypatch, DataFogSentryIntegration()) + seen = [] + + def user_callback(log, hint): + seen.append(log) + return log + + options = self._wrapped_options(_experiments={"before_send_log": user_callback}) + options["before_send_log"](self._log(), None) + assert len(seen) == 1 + assert EMAIL not in json.dumps(seen[0]) + + def test_before_send_metric_scrubs_attributes(self, monkeypatch): + self._use_client_integration(monkeypatch, DataFogSentryIntegration()) + options = self._wrapped_options() + metric = options["before_send_metric"]( + { + "name": "checkout.total", + "type": "counter", + "value": 1, + "attributes": {"user.email": EMAIL}, + }, + None, + ) + assert EMAIL not in json.dumps(metric) + assert metric["name"] == "checkout.total" + + def test_before_send_span_wrapped_in_experiments(self, monkeypatch): + self._use_client_integration(monkeypatch, DataFogSentryIntegration()) + options = self._wrapped_options() + span = options["_experiments"]["before_send_span"]( + {"description": f"lookup {EMAIL}", "op": "db.query"}, None + ) + assert EMAIL not in span["description"] + + def test_wrapper_without_integration_is_passthrough(self, monkeypatch): + self._use_client_integration(monkeypatch, None) + seen = [] + + def user_callback(log, hint): + seen.append(log) + return log + + options = self._wrapped_options(before_send_log=user_callback) + log = self._log() + result = options["before_send_log"](log, None) + assert result is log + assert EMAIL in result["body"] # untouched without the integration + assert len(seen) == 1 + + def test_wrapper_fail_closed_drops_telemetry(self, monkeypatch): + import datafog + + self._use_client_integration( + monkeypatch, DataFogSentryIntegration(fail_policy="closed") + ) + options = self._wrapped_options() + + def boom(*args, **kwargs): + raise RuntimeError("engine exploded") + + monkeypatch.setattr(datafog, "redact", boom) + assert options["before_send_log"](self._log(), None) is None + + def test_logs_end_to_end(self): + # setup_once_with_options only runs the first time the integration + # identifier is processed in this process; discard it so this + # re-init exercises the wrap (production apps init once). + import sentry_sdk.integrations as si + from sentry_sdk import logger as sentry_logger + + si._processed_integrations.discard("datafog") + captured = [] + + def capture_log(log, hint): + captured.append(log) + return None + + sentry_sdk.init( + dsn="https://key@example.invalid/1", + integrations=[DataFogSentryIntegration()], + default_integrations=False, + enable_logs=True, + before_send_log=capture_log, + ) + sentry_logger.warning(f"login failed for {EMAIL}") + sentry_sdk.get_client().flush() + assert len(captured) == 1 + assert EMAIL not in json.dumps(captured[0]) + assert "[EMAIL_1]" in captured[0]["body"] + + def test_setup_without_options_is_noop(self): + DataFogSentryIntegration().setup_once_with_options(None) + + def test_no_experiments_dict_resolves_no_callback(self, monkeypatch): + self._use_client_integration(monkeypatch, DataFogSentryIntegration()) + options = {} # no _experiments key at all + DataFogSentryIntegration().setup_once_with_options(options) + log = options["before_send_log"](self._log(), None) + assert EMAIL not in json.dumps(log) + + def test_wrapper_lookup_failure_passes_telemetry_through(self, monkeypatch): + from datafog.integrations import sentry as sentry_module + + options = self._wrapped_options() + + def boom(): + raise RuntimeError("client lookup exploded") + + monkeypatch.setattr(sentry_module.sentry_sdk, "get_client", boom) + log = self._log() + assert options["before_send_log"](log, None) is log + + def test_top_level_callback_shadows_experiments_callback(self, monkeypatch): + # When the user set callbacks in BOTH locations, only the top-level + # one runs (matching the SDK's own precedence), exactly once. + self._use_client_integration(monkeypatch, DataFogSentryIntegration()) + top_calls, exp_calls = [], [] + + def top_cb(log, hint): + top_calls.append(log) + return log + + def exp_cb(log, hint): + exp_calls.append(log) + return log + + options = self._wrapped_options( + before_send_log=top_cb, _experiments={"before_send_log": exp_cb} + ) + options["before_send_log"](self._log(), None) + assert len(top_calls) == 1 + assert exp_calls == [] + + def test_experiments_copy_is_wrapped_not_stale(self, monkeypatch): + # The legacy _experiments location must not retain the raw user + # callback: anything reading it directly would get an unscrubbed + # path. After setup both locations hold the same wrapped closure. + self._use_client_integration(monkeypatch, DataFogSentryIntegration()) + seen = [] + + def user_callback(log, hint): + seen.append(log) + return log + + options = self._wrapped_options(_experiments={"before_send_log": user_callback}) + assert options["_experiments"]["before_send_log"] is options["before_send_log"] + options["_experiments"]["before_send_log"](self._log(), None) + assert len(seen) == 1 + assert EMAIL not in json.dumps(seen[0]) + + def test_repeated_setup_does_not_stack_wrappers(self, monkeypatch): + # setup_once_with_options is normally once-per-process, but a repeat + # call must be idempotent, not stack N scrub passes. + integration = DataFogSentryIntegration() + options = self._wrapped_options(integration=integration) + first = options["before_send_log"] + integration.setup_once_with_options(options) + assert options["before_send_log"] is first