diff --git a/README.md b/README.md index 128fdc68d..afb10e52e 100644 --- a/README.md +++ b/README.md @@ -193,6 +193,13 @@ To quickly get hands on LCS, we can run it using the default configurations prov ```bash uv sync --group dev --group ogxlibdev ``` + + If you plan to enable OpenTelemetry tracing with in-process PII redaction + (OTEL_TRACES_EXPORTER=redacting_otlp), also download the spaCy model: + ```bash + uv run python -m spacy download en_core_web_lg + ``` + 1. create OGX `run.yaml`. you can do this by running the local run generation script ```bash ./scripts/generate_local_run.sh diff --git a/deploy/lightspeed-stack/Containerfile b/deploy/lightspeed-stack/Containerfile index 6160daf5e..ce1ec1c09 100644 --- a/deploy/lightspeed-stack/Containerfile +++ b/deploy/lightspeed-stack/Containerfile @@ -39,7 +39,9 @@ COPY ${LSC_SOURCE_DIR}/pyproject.toml ${LSC_SOURCE_DIR}/LICENSE ${LSC_SOURCE_DIR COPY ${LSC_SOURCE_DIR}/providers/lightspeed_stack_providers /app-root/lightspeed_stack_providers COPY ${LSC_SOURCE_DIR}/providers/resources/external_providers /app-root/providers.d -# Bundle additional dependencies for library mode. +# Bundle additional dependencies for library mode and install the spaCy model for PII +# redaction. The model is baked into the image so it is available in disconnected +# environments where runtime downloads are not possible. # Source cachi2 environment for hermetic builds if available, otherwise use normal installation # cachi2.env has these env vars: # PIP_FIND_LINKS=/cachi2/output/deps/pip @@ -51,7 +53,8 @@ RUN if [ -f /cachi2/cachi2.env ]; then \ pip install --no-cache-dir --ignore-installed --no-index --find-links ${PIP_FIND_LINKS} --no-deps -r requirements.hashes.wheel.txt -r requirements.hashes.wheel.pypi.txt -r requirements.hashes.source.txt && \ pip check; \ else \ - uv sync --locked --no-dev --group ogxlibdev; \ + uv sync --locked --no-dev --group ogxlibdev && \ + uv run --with pip python -m spacy download en_core_web_lg; \ fi # Explicitly remove some packages to mitigate some CVEs diff --git a/docker-compose-library.yaml b/docker-compose-library.yaml index 0f29b33f8..1823deb4a 100755 --- a/docker-compose-library.yaml +++ b/docker-compose-library.yaml @@ -80,6 +80,10 @@ services: - OTEL_SERVICE_NAME=${OTEL_SERVICE_NAME:-lightspeed-stack-e2e} - OTEL_ANONYMIZATION_SECRET=${OTEL_ANONYMIZATION_SECRET:-lightspeed-stack-otel-anonymization-dev-default} - OTEL_SDK_DISABLED=${OTEL_SDK_DISABLED:-false} + # In-process PII redaction is enabled: OTEL_TRACES_EXPORTER=redacting_otlp + # routes spans through the RedactingSpanExporter before export. The spaCy + # model (en_core_web_lg) is baked into the image at build time. + - OTEL_TRACES_EXPORTER=${OTEL_TRACES_EXPORTER:-redacting_otlp} healthcheck: test: ["CMD", "curl", "-f", "http://localhost:8080/liveness"] interval: 10s # how often to run the check diff --git a/docker-compose.yaml b/docker-compose.yaml index 958f865b0..1f11f2740 100755 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -123,6 +123,10 @@ services: - OTEL_SERVICE_NAME=${OTEL_SERVICE_NAME:-lightspeed-stack-e2e} - OTEL_ANONYMIZATION_SECRET=${OTEL_ANONYMIZATION_SECRET:-lightspeed-stack-otel-anonymization-dev-default} - OTEL_SDK_DISABLED=${OTEL_SDK_DISABLED:-false} + # In-process PII redaction is enabled: OTEL_TRACES_EXPORTER=redacting_otlp + # routes spans through the RedactingSpanExporter before export. The spaCy + # model (en_core_web_lg) is baked into the image at build time. + - OTEL_TRACES_EXPORTER=${OTEL_TRACES_EXPORTER:-redacting_otlp} depends_on: ogx: condition: service_healthy @@ -229,7 +233,6 @@ services: retries: 3 start_period: 5s - volumes: ogx-storage: mock-tls-certs: diff --git a/docs/devel_doc/installation_linux.md b/docs/devel_doc/installation_linux.md index f724b9119..eb2c45dc0 100644 --- a/docs/devel_doc/installation_linux.md +++ b/docs/devel_doc/installation_linux.md @@ -15,6 +15,8 @@ 1. `cd service` 1. `uv info` -- should return no error 1. `uv install` +1. (Optional) If you plan to use OpenTelemetry tracing with in-process PII redaction: + `uv run python -m spacy download en_core_web_lg` Note: `pip` installs the `uv` binary into `~/.local/bin`, which is often missing from `$PATH` on fresh distributions. If the `uv` command is not found, please diff --git a/docs/devel_doc/installation_macos.md b/docs/devel_doc/installation_macos.md index a1ec0e62a..9ac4e12cd 100644 --- a/docs/devel_doc/installation_macos.md +++ b/docs/devel_doc/installation_macos.md @@ -15,3 +15,5 @@ `git clone https://github.com/lightspeed-core/lightspeed-stack` 1. `cd lightspeed-stack` 1. `uv sync` +1. (Optional) If you plan to use OpenTelemetry tracing with in-process PII redaction: + `uv run python -m spacy download en_core_web_lg` diff --git a/pyproject.toml b/pyproject.toml index fb78555ec..9490c8204 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -121,6 +121,11 @@ paths = ["tests/e2e/features"] [project.scripts] lightspeed-stack = "lightspeed_stack:main" +# In-process PII redaction span exporter. Enable by setting +# OTEL_TRACES_EXPORTER=redacting_otlp (auto-instrumentation picks it up). +[project.entry-points.opentelemetry_traces_exporter] +redacting_otlp = "observability.redacting_span_exporter:RedactingSpanExporter" + [project.urls] Homepage = "https://github.com/lightspeed-core/lightspeed-stack" Issues = "https://github.com/lightspeed-core/lightspeed-stack/issues" diff --git a/src/observability/redacting_span_exporter.py b/src/observability/redacting_span_exporter.py new file mode 100644 index 000000000..d5b113dc4 --- /dev/null +++ b/src/observability/redacting_span_exporter.py @@ -0,0 +1,183 @@ +"""Span exporter that redacts PII from span content before delegating to OTLP. + +In-process (producer-side) sanitization: redaction happens inside lightspeed-core, +before spans leave the process, so no unredacted span is ever put on the wire. + +It is wired via the ``opentelemetry_traces_exporter`` entry point, selected with +``OTEL_TRACES_EXPORTER=redacting_otlp``. Auto-instrumentation +(``opentelemetry-instrument``) instantiates it and wraps it in the +``BatchSpanProcessor``, so ``export()`` runs on the processor's background thread, +off the request path. The request/handler code is unchanged. + +Redaction is delegated to the vendored ``pii_redaction_engine`` (Presidio + spaCy). +""" + +from __future__ import annotations + +import os +from collections.abc import Sequence + +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( + OTLPSpanExporter as GRPCSpanExporter, +) +from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( + OTLPSpanExporter as HTTPSpanExporter, +) +from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult + +from log import get_logger +from pii_redaction_engine.anonymizer import PIIAnonymizer + +logger = get_logger(__name__) + +# Span attributes whose string values may carry PII (see utils.otel_tracing). +# Note: request.safety_identifier is intentionally excluded (it is a caller-supplied +# identifier for tracking, not user content). +CONTENT_ATTRIBUTES: tuple[str, ...] = ( + "request.input", + "response.output", + "rag.input", + "tool_calls", + "tool_results", + "feedback.comment", +) + + +def _default_otlp_exporter() -> SpanExporter: + """Return an OTLP span exporter matching the configured protocol. + + Matches the stock SDK's supported values exactly: ``grpc`` (default) or + ``http/protobuf``. Both read the standard ``OTEL_EXPORTER_OTLP_*`` environment + for endpoint, headers, and TLS. Any other value (e.g. ``http/json``, which + neither OTLP span exporter supports) is rejected rather than silently sending + protobuf. + + Returns: + A gRPC or HTTP OTLP span exporter. + + Raises: + ValueError: if the configured protocol is not ``grpc`` or ``http/protobuf``. + + """ + protocol = ( + ( + os.getenv("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL") + or os.getenv("OTEL_EXPORTER_OTLP_PROTOCOL") + or "grpc" + ) + .strip() + .lower() + ) + if protocol == "grpc": + return GRPCSpanExporter() + if protocol == "http/protobuf": + return HTTPSpanExporter() + raise ValueError( + f"unsupported OTEL_EXPORTER_OTLP_PROTOCOL {protocol!r}; " + "supported values are 'grpc' and 'http/protobuf'" + ) + + +class RedactingSpanExporter(SpanExporter): + """Wraps an OTLP span exporter, redacting content attributes before export. + + Attributes: + delegate: The downstream exporter that actually ships the (redacted) spans. + + """ + + def __init__( + self, + delegate: SpanExporter | None = None, + anonymizer: PIIAnonymizer | None = None, + ) -> None: + """Initialize the exporter. + + Args: + delegate: Downstream exporter. Defaults to an OTLP exporter matching + ``OTEL_EXPORTER_OTLP_PROTOCOL`` (gRPC or HTTP), which reads the + standard ``OTEL_EXPORTER_OTLP_*`` environment. + anonymizer: Redaction engine. Defaults to a ``PIIAnonymizer``, which + loads the spaCy model once at construction (off the request path). + Injectable for testing. + + """ + self._delegate = delegate or _default_otlp_exporter() + self._anonymizer = anonymizer or PIIAnonymizer() + logger.info( + "RedactingSpanExporter active; content attributes redacted in-process" + ) + + def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: + """Redact content attributes, then delegate the export. + + Fail closed: if redaction of a span raises, that span is dropped rather + than exported unredacted, so a redaction failure can never leak PII. Other + spans in the batch are still exported. + + Args: + spans: The spans to redact and export. + + Returns: + The delegate's export result, or ``SUCCESS`` if every span was dropped + (reporting failure would only make the batch processor retry the same + failing spans). + + """ + safe_spans: list[ReadableSpan] = [] + for span in spans: + try: + self._redact_span(span) + except Exception: # pylint: disable=broad-except # noqa: BLE001 + # Log the span name only, never the content being redacted. + logger.error( + "in-process redaction failed for span %r; dropping it to avoid " + "exporting unredacted PII", + span.name, + ) + continue + safe_spans.append(span) + + if not safe_spans: + return SpanExportResult.SUCCESS + return self._delegate.export(safe_spans) + + def _redact_span(self, span: ReadableSpan) -> None: + """Rewrite the content attributes of a span in place with redacted text. + + Args: + span: The span whose content attributes are redacted in place. + + Raises: + Exception: if the redaction engine fails. The caller drops the span + (fail closed) rather than exporting unredacted content. + + """ + attributes = span.attributes + if not attributes: + return + + new_attributes = dict(attributes) + changed = False + for key in CONTENT_ATTRIBUTES: + value = new_attributes.get(key) + if not isinstance(value, str) or not value: + continue + redacted = self._anonymizer.anonymize_text(value) + if redacted != value: + new_attributes[key] = redacted + changed = True + + if changed: + # ReadableSpan exposes attributes via self._attributes; replacing it + # with a plain dict is what the OTLP encoder reads at export time. + span._attributes = new_attributes # pylint: disable=protected-access + + def shutdown(self) -> None: + """Shut down the delegate exporter.""" + self._delegate.shutdown() + + def force_flush(self, timeout_millis: int = 30_000) -> bool: + """Force-flush the delegate exporter.""" + return self._delegate.force_flush(timeout_millis) diff --git a/tests/e2e/features/opentelemetry.feature b/tests/e2e/features/opentelemetry.feature index aba242621..770b76c42 100644 --- a/tests/e2e/features/opentelemetry.feature +++ b/tests/e2e/features/opentelemetry.feature @@ -25,3 +25,33 @@ Feature: OpenTelemetry observability tests """ Then The status code of the response is 200 And The OpenTelemetry service received data containing e2e-otel-delivery-marker + + + Scenario: PII in span content is redacted in-process before telemetry export + # One request carrying every default-redacted entity type. safety_identifier is + # intentionally not redacted (it is a caller-supplied tracking id, not user + # content), so it doubles as the delivery marker. + Given The system is in default state + When I use "responses" to ask question with authorization header + """ + { + "input": "Reach me at test.user@redhat.com or 212-555-0199, host db.example.com at 10.1.2.3; contact Barack Obama in Chicago regarding e2e-redaction-check", + "model": "{PROVIDER}/{MODEL}", + "stream": false, + "safety_identifier": "e2e-redaction-check" + } + """ + Then The status code of the response is 200 + And The OpenTelemetry service received data containing e2e-redaction-check + # Raw PII values must never reach the collector. + And The OpenTelemetry service did not receive data containing test.user@redhat.com + And The OpenTelemetry service did not receive data containing 212-555-0199 + And The OpenTelemetry service did not receive data containing db.example.com + And The OpenTelemetry service did not receive data containing 10.1.2.3 + # Each entity is replaced in-process with its typed marker. + And The OpenTelemetry service received data containing + And The OpenTelemetry service received data containing + And The OpenTelemetry service received data containing + And The OpenTelemetry service received data containing + And The OpenTelemetry service received data containing + And The OpenTelemetry service received data containing diff --git a/tests/e2e/features/steps/opentelemetry.py b/tests/e2e/features/steps/opentelemetry.py index 4d9f25939..6bb3a0753 100644 --- a/tests/e2e/features/steps/opentelemetry.py +++ b/tests/e2e/features/steps/opentelemetry.py @@ -85,3 +85,29 @@ def collector_received_data(context: Context, marker: str) -> None: f"Mock OTEL collector did not receive data containing {marker!r} " f"within {_DELIVERY_TIMEOUT_S:.0f}s" ) + + +@then("The OpenTelemetry service did not receive data containing {marker}") +def collector_did_not_receive_data(context: Context, marker: str) -> None: + """Assert the mock collector buffered telemetry that does NOT contain ``marker``. + + Used to verify in-process redaction: a known PII value sent in a query must be + absent from the exported telemetry. Ordering matters — a preceding "received + data containing ..." step should already have confirmed the span was delivered, + so a single check proves the value is absent (redacted) rather than merely + not-yet-delivered. + """ + assert getattr( + context, "otel_collector_ready", False + ), "The OpenTelemetry service must be started before asserting on delivery" + marker = marker.strip() + response = requests.get( + f"{MOCK_OTEL_CONTROL_BASE}/received", params={"contains": marker}, timeout=5 + ) + assert ( + response.status_code == 200 + ), f"Failed to query mock OTEL collector: HTTP {response.status_code}" + assert not response.json().get("found"), ( + f"Mock OTEL collector unexpectedly received data containing {marker!r} " + "(expected it to be redacted in-process)" + ) diff --git a/tests/e2e/mock_otel_collector/server.py b/tests/e2e/mock_otel_collector/server.py index 1e8bff0a5..612c430e6 100644 --- a/tests/e2e/mock_otel_collector/server.py +++ b/tests/e2e/mock_otel_collector/server.py @@ -31,6 +31,7 @@ from collections import deque from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any +from urllib.parse import unquote_plus # Resource bounds so a reachable caller cannot exhaust process memory by sending # large or repeated exports (the Compose service publishes port 4318). The @@ -94,7 +95,12 @@ def _handle_received_query(self, query: str) -> None: for pair in query.split("&"): key, sep, value = pair.partition("=") if sep and key == "contains": - contains = value + # Decode percent-encoding (and '+') so markers with reserved + # characters match. A client such as ``requests`` encodes + # ```` to ``%3CEMAIL_ADDRESS%3E`` and ``@`` to + # ``%40``; without decoding, the raw-byte search would never + # find the plaintext marker embedded in the OTLP payload. + contains = unquote_plus(value) break if contains is not None: diff --git a/tests/unit/observability/test_redacting_span_exporter.py b/tests/unit/observability/test_redacting_span_exporter.py new file mode 100644 index 000000000..3fff90588 --- /dev/null +++ b/tests/unit/observability/test_redacting_span_exporter.py @@ -0,0 +1,139 @@ +"""Unit tests for the in-process redacting span exporter.""" + +from collections.abc import Callable +from typing import Any + +import pytest +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( + OTLPSpanExporter as GRPCSpanExporter, +) +from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( + OTLPSpanExporter as HTTPSpanExporter, +) +from opentelemetry.sdk.trace.export import SpanExportResult +from pytest_mock import MockerFixture + +from observability.redacting_span_exporter import ( + RedactingSpanExporter, + _default_otlp_exporter, +) + + +class _FakeSpan: # pylint: disable=too-few-public-methods + """Minimal stand-in for a ReadableSpan: exposes .name, .attributes, ._attributes.""" + + def __init__(self, attributes: dict, name: str = "test.span") -> None: + self._attributes = dict(attributes) + self.name = name + + @property + def attributes(self) -> dict: + """Return the span's attribute mapping, as ReadableSpan does.""" + return self._attributes + + +def _make_exporter( + mocker: MockerFixture, + anonymize_side_effect: Callable[[str], str] | None = None, +) -> tuple[RedactingSpanExporter, Any, Any]: + """Build a RedactingSpanExporter with a mocked delegate and anonymizer.""" + delegate = mocker.Mock() + delegate.export.return_value = SpanExportResult.SUCCESS + anonymizer = mocker.Mock() + anonymizer.anonymize_text.side_effect = anonymize_side_effect or ( + lambda text: text.replace("sam@redhat.com", "") + ) + exporter = RedactingSpanExporter(delegate=delegate, anonymizer=anonymizer) + return exporter, delegate, anonymizer + + +def test_redacts_content_attributes_and_delegates(mocker: MockerFixture) -> None: + """Content attributes are redacted; other attributes and the span pass through.""" + exporter, delegate, _ = _make_exporter(mocker) + span = _FakeSpan( + { + "request.input": "sam@redhat.com asks about K8s", + "response.output": "hello sam@redhat.com", + "llm.model.id": "gpt-4o-mini", + } + ) + + result = exporter.export([span]) + + assert result is SpanExportResult.SUCCESS + assert span.attributes["request.input"] == " asks about K8s" + assert span.attributes["response.output"] == "hello " + assert span.attributes["llm.model.id"] == "gpt-4o-mini" + delegate.export.assert_called_once_with([span]) + + +def test_skips_missing_non_string_and_empty_values(mocker: MockerFixture) -> None: + """Empty, non-string, and absent content attributes are left alone.""" + exporter, _, anonymizer = _make_exporter(mocker) + span = _FakeSpan({"request.input": "", "rag.input": 123}) + + exporter.export([span]) + + anonymizer.anonymize_text.assert_not_called() + assert span.attributes == {"request.input": "", "rag.input": 123} + + +def test_fail_closed_drops_span_on_redaction_error(mocker: MockerFixture) -> None: + """A redaction error drops the span (fail closed): it is never exported.""" + + def boom(_text: str) -> str: + raise RuntimeError("anonymizer unavailable") + + exporter, delegate, _ = _make_exporter(mocker, anonymize_side_effect=boom) + span = _FakeSpan({"request.input": "sam@redhat.com"}) + + result = exporter.export([span]) + + delegate.export.assert_not_called() + assert result is SpanExportResult.SUCCESS + + +def test_fail_closed_drops_only_the_failing_span(mocker: MockerFixture) -> None: + """A span whose redaction fails is dropped; healthy spans still export.""" + + def selective(text: str) -> str: + if "boom" in text: + raise RuntimeError("anonymizer unavailable") + return text.replace("sam@redhat.com", "") + + exporter, delegate, _ = _make_exporter(mocker, anonymize_side_effect=selective) + good = _FakeSpan({"request.input": "contact sam@redhat.com"}) + bad = _FakeSpan({"request.input": "boom trigger"}) + + result = exporter.export([good, bad]) + + assert result is SpanExportResult.SUCCESS + delegate.export.assert_called_once_with([good]) + assert good.attributes["request.input"] == "contact " + + +def test_shutdown_and_force_flush_delegate(mocker: MockerFixture) -> None: + """shutdown and force_flush delegate to the downstream exporter.""" + exporter, delegate, _ = _make_exporter(mocker) + + exporter.shutdown() + delegate.shutdown.assert_called_once() + + exporter.force_flush(1234) + delegate.force_flush.assert_called_once_with(1234) + + +def test_default_exporter_defaults_to_grpc(monkeypatch: pytest.MonkeyPatch) -> None: + """With no protocol set, the default delegate is the gRPC OTLP exporter.""" + monkeypatch.delenv("OTEL_EXPORTER_OTLP_PROTOCOL", raising=False) + monkeypatch.delenv("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", raising=False) + assert isinstance(_default_otlp_exporter(), GRPCSpanExporter) + + +def test_default_exporter_honors_http_protocol( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf selects the HTTP OTLP exporter.""" + monkeypatch.delenv("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL", raising=False) + monkeypatch.setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf") + assert isinstance(_default_otlp_exporter(), HTTPSpanExporter)