From 6e942b16ef512cebf0c2ba2db19006df5dcb6b29 Mon Sep 17 00:00:00 2001 From: kadekillary Date: Mon, 13 Jul 2026 07:36:52 -0700 Subject: [PATCH 1/3] [AAI-236] Mask exported third party spans --- docs/API_REFERENCE.md | 1 + docs/CONFIGURATION.md | 27 +++ docs/TRACING.md | 29 +++ lib/langfuse.rb | 1 + lib/langfuse/config.rb | 44 ++-- lib/langfuse/masking_exporter.rb | 217 ++++++++++++++++++ lib/langfuse/otel_setup.rb | 16 +- spec/langfuse/config_spec.rb | 31 +++ spec/langfuse/masking_exporter_spec.rb | 293 +++++++++++++++++++++++++ spec/langfuse/otel_setup_spec.rb | 61 +++++ 10 files changed, 699 insertions(+), 21 deletions(-) create mode 100644 lib/langfuse/masking_exporter.rb create mode 100644 spec/langfuse/masking_exporter_spec.rb diff --git a/docs/API_REFERENCE.md b/docs/API_REFERENCE.md index b6a2107..14cb5e5 100644 --- a/docs/API_REFERENCE.md +++ b/docs/API_REFERENCE.md @@ -58,6 +58,7 @@ Block receives a `Langfuse::Config` object with these properties: | `release` | String | No | `nil` (or `ENV["LANGFUSE_RELEASE"]` / common CI commit SHA env) | Default release identifier | | `should_export_span` | `#call` | No | `nil` | Span export filter callback | | `mask` | `#call` | No | `nil` | Mask callable for input/output/metadata (receives `data:` keyword) | +| `mask_otel_spans` | `#call` | No | `nil` | Export-stage span masking hook (receives `spans:` keyword; see [CONFIGURATION.md](CONFIGURATION.md#mask_otel_spans)) | **Example:** diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 4d06e41..1437b5f 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -393,6 +393,32 @@ config.mask = lambda { |data:| See [TRACING.md](TRACING.md#masking) for usage patterns and behavior details. +#### `mask_otel_spans` + +- **Type:** `#call` (Proc, Lambda, or any object responding to `call`) or `nil` +- **Default:** `nil` (export-stage masking disabled) +- **Description:** Export-stage masking hook applied to every span batch exported to Langfuse, including spans created by third-party instrumentations. Receives a `spans:` keyword argument — a frozen Hash mapping stable span identifiers (`":"`) to immutable `Langfuse::OtelSpanSnapshot` values (trace/span IDs, name, instrumentation scope name, frozen span attributes, frozen resource attributes). Return `nil` to export the batch unchanged, or a sparse Hash of patches keyed by the same identifiers. Each patch is a Hash with optional `:delete` (Array of attribute keys) and `:set` (Hash of replacement attributes); deletes run before sets, so a replacement wins when a key appears in both. + +```ruby +config.mask_otel_spans = lambda { |spans:| + spans.filter_map { |id, snapshot| + next unless snapshot.attributes.key?("gen_ai.prompt") + + [id, { set: { "gen_ai.prompt" => "[REDACTED]" } }] + }.to_h +} +``` + +Fail-closed behavior: + +- A hook exception or an invalid top-level result drops the whole Langfuse export batch. +- A malformed per-span patch drops that span from the Langfuse export. +- An invalid replacement attribute value omits that attribute rather than exporting its original value. + +Only the copy exported to Langfuse is transformed. Any other OpenTelemetry exporter receives the original, unmasked spans — Langfuse masking does not protect other telemetry backends. The hook is synchronous and runs on the batch export thread; it must not rely on request context, the current span, async work, or network calls. + +See [TRACING.md](TRACING.md#masking) for how the two masking boundaries relate. + ## Tracing Behavior and OpenTelemetry Ownership There are three states worth documenting. @@ -623,6 +649,7 @@ Validation rules: - `prompt_cache_observer` must respond to `#call` (if set) - `should_export_span` must respond to `#call` (if set) - `mask` must respond to `#call` (if set) +- `mask_otel_spans` must respond to `#call` (if set) ## Accessing Current Configuration diff --git a/docs/TRACING.md b/docs/TRACING.md index c46a323..8761787 100644 --- a/docs/TRACING.md +++ b/docs/TRACING.md @@ -316,3 +316,32 @@ end ``` Masking applies to observation `input`, `output`, and `metadata`. The full configuration contract is in [CONFIGURATION.md](CONFIGURATION.md#mask). + +### Export-stage masking with `mask_otel_spans` + +`mask` runs while the SDK creates Langfuse-owned attributes; it never sees the raw +attributes of third-party spans (e.g. `gen_ai.*` attributes set by an OpenAI or +LangChain instrumentation). To transform those, configure the export-stage hook: + +```ruby +Langfuse.configure do |config| + config.mask_otel_spans = lambda { |spans:| + spans.filter_map { |id, snapshot| + next unless snapshot.attributes.key?("gen_ai.prompt") + + [id, { delete: ["gen_ai.completion"], set: { "gen_ai.prompt" => "[REDACTED]" } }] + }.to_h + } +end +``` + +The hook runs after `should_export_span` selects spans and just before the batch +is handed to the Langfuse OTLP exporter, so it sees exactly the spans this client +will export — Langfuse-owned and third-party alike. Returning `nil` exports the +batch unchanged. Errors fail closed: the batch (or the offending span) is dropped +rather than exported unmasked. + +Only the Langfuse export copy is transformed. If you also send telemetry to +another OpenTelemetry backend, that backend receives the original spans and needs +its own masking. The two hooks are independent; applications may configure either +or both. The full contract is in [CONFIGURATION.md](CONFIGURATION.md#mask_otel_spans). diff --git a/lib/langfuse.rb b/lib/langfuse.rb index 5e5e28f..4102a39 100644 --- a/lib/langfuse.rb +++ b/lib/langfuse.rb @@ -52,6 +52,7 @@ class UnauthorizedError < ApiError; end require_relative "langfuse/sampling" require_relative "langfuse/otel_setup" require_relative "langfuse/masking" +require_relative "langfuse/masking_exporter" require_relative "langfuse/otel_attributes" require_relative "langfuse/propagation" require_relative "langfuse/span_processor" diff --git a/lib/langfuse/config.rb b/lib/langfuse/config.rb index 3d98c15..d0f4a80 100644 --- a/lib/langfuse/config.rb +++ b/lib/langfuse/config.rb @@ -86,8 +86,26 @@ class Config # @return [#call, nil] Mask callable applied to input, output, and metadata before serialization. # Receives `data:` keyword argument. nil disables masking. + # This is a creation-time hook for Langfuse-owned attributes; it never sees + # raw third-party span attributes. See {#mask_otel_spans} for those. attr_accessor :mask + # @return [#call, nil] Export-stage masking hook for spans exported to Langfuse. + # Receives a `spans:` keyword argument — a frozen Hash mapping stable span + # identifiers (`":"`) to {OtelSpanSnapshot} + # values covering every span in one export batch, including spans created + # by third-party instrumentations. Returns nil to export the batch + # unchanged, or a sparse Hash of patches keyed by the same identifiers. + # Each patch is a Hash with optional +:delete+ (Array of attribute keys) + # and +:set+ (Hash of replacement attributes); deletes run before sets. + # Only the copy exported to Langfuse is transformed — any other + # OpenTelemetry exporter receives the original, unmasked spans, so + # Langfuse masking does not protect other telemetry backends. + # The hook is synchronous and must not rely on request context, the + # current span, async work, or network calls. Exceptions and invalid + # results fail closed by dropping the Langfuse export batch. + attr_accessor :mask_otel_spans + # @return [String] Default Langfuse API base URL DEFAULT_BASE_URL = "https://cloud.langfuse.com" @@ -193,10 +211,11 @@ def validate! validate_swr_config! validate_cache_backend! - validate_prompt_cache_observer! validate_sample_rate! - validate_should_export_span! - validate_mask! + validate_callable!(prompt_cache_observer, "prompt_cache_observer") + validate_callable!(should_export_span, "should_export_span") + validate_callable!(mask, "mask") + validate_callable!(mask_otel_spans, "mask_otel_spans") end # rubocop:enable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity @@ -242,6 +261,7 @@ def initialize_tracing_defaults self.sample_rate = env_value("LANGFUSE_SAMPLE_RATE") || DEFAULT_SAMPLE_RATE @should_export_span = nil @mask = nil + @mask_otel_spans = nil end def validate_cache_backend! @@ -252,10 +272,10 @@ def validate_cache_backend! "cache_backend must be one of #{valid_backends.inspect}, got #{cache_backend.inspect}" end - def validate_prompt_cache_observer! - return if prompt_cache_observer.nil? || prompt_cache_observer.respond_to?(:call) + def validate_callable!(value, name) + return if value.nil? || value.respond_to?(:call) - raise ConfigurationError, "prompt_cache_observer must respond to #call" + raise ConfigurationError, "#{name} must respond to #call" end def validate_swr_config! @@ -296,18 +316,6 @@ def validate_sample_rate! raise ConfigurationError, "sample_rate must be between 0.0 and 1.0" end - def validate_mask! - return if mask.nil? || mask.respond_to?(:call) - - raise ConfigurationError, "mask must respond to #call" - end - - def validate_should_export_span! - return if should_export_span.nil? || should_export_span.respond_to?(:call) - - raise ConfigurationError, "should_export_span must respond to #call" - end - def detect_release_from_ci_env COMMON_RELEASE_ENV_KEYS.each do |key| value = env_value(key) diff --git a/lib/langfuse/masking_exporter.rb b/lib/langfuse/masking_exporter.rb new file mode 100644 index 0000000..95e1613 --- /dev/null +++ b/lib/langfuse/masking_exporter.rb @@ -0,0 +1,217 @@ +# frozen_string_literal: true + +require "opentelemetry/sdk" + +module Langfuse + # Immutable snapshot of one span in a Langfuse export batch. + # + # Passed to the {Config#mask_otel_spans} hook as the values of the +spans:+ + # mapping. Exposes only the fields needed to make a masking decision; the + # mutable OpenTelemetry SDK span objects are never handed to the hook. + # + # @!attribute [r] trace_id + # @return [String] 32-hex-character lowercase trace ID + # @!attribute [r] span_id + # @return [String] 16-hex-character lowercase span ID + # @!attribute [r] name + # @return [String] span name + # @!attribute [r] scope_name + # @return [String, nil] instrumentation scope name (e.g. "ruby-openai") + # @!attribute [r] attributes + # @return [Hash] frozen span attributes + # @!attribute [r] resource_attributes + # @return [Hash] frozen resource attributes + OtelSpanSnapshot = Data.define( + :trace_id, :span_id, :name, :scope_name, :attributes, :resource_attributes + ) + + # Export-stage masking wrapper around the Langfuse OTLP exporter. + # + # Applies the {Config#mask_otel_spans} hook to every export batch after + # +should_export_span+ has selected spans and the batch processor has built + # the batch. Only the copy exported to Langfuse is transformed; original + # span data (and therefore any other OpenTelemetry exporter) is untouched. + # + # Fail-closed behavior: + # - A hook exception or an invalid top-level result drops the whole batch. + # - A malformed per-span patch drops that span from the Langfuse export. + # - An invalid replacement attribute value omits that attribute entirely + # rather than exporting its original value. + # + # @api private + class MaskingExporter + SUCCESS = OpenTelemetry::SDK::Trace::Export::SUCCESS + FAILURE = OpenTelemetry::SDK::Trace::Export::FAILURE + private_constant :SUCCESS, :FAILURE + + # Allowed keys of one sparse span patch. + PATCH_KEYS = %i[delete set].freeze + private_constant :PATCH_KEYS + + # @param delegate [#export, #force_flush, #shutdown] Langfuse OTLP exporter + # @param hook [#call] The configured mask_otel_spans callable + # @param logger [Logger] + def initialize(delegate:, hook:, logger:) + @delegate = delegate + @hook = hook + @logger = logger + end + + # Mask the batch and delegate export. Drops the whole batch (fail closed) + # when the hook raises or returns an invalid result. + # + # @param span_data [Enumerable] + # @param timeout [Numeric, nil] + # @return [Integer] OpenTelemetry export result code + def export(span_data, timeout: nil) + batch = span_data.to_a + masked = mask_batch(batch) + return FAILURE unless masked + + @delegate.export(masked, timeout: timeout) + end + + # @return [Object] delegate's force_flush result + def force_flush(timeout: nil) + @delegate.force_flush(timeout: timeout) + end + + # @return [Object] delegate's shutdown result + def shutdown(timeout: nil) + @delegate.shutdown(timeout: timeout) + end + + private + + # @return [Array, nil] the masked batch, or nil to drop it + def mask_batch(batch) + patches = @hook.call(spans: snapshot_batch(batch)) + return batch if patches.nil? + + unless patches.is_a?(Hash) + @logger.error( + "Langfuse mask_otel_spans returned #{patches.class} instead of nil or a Hash " \ + "of patches; dropping the Langfuse export batch" + ) + return nil + end + + apply_patches(batch, patches) + rescue StandardError => e + @logger.error("Langfuse mask_otel_spans raised #{e.class}: #{e.message}; dropping the Langfuse export batch") + nil + end + + def snapshot_batch(batch) + batch.to_h { |span| [span_identifier(span), snapshot(span)] }.freeze + end + + # Stable identifier the hook uses to key sparse patches. + def span_identifier(span) + "#{span.hex_trace_id}:#{span.hex_span_id}" + end + + def snapshot(span) + OtelSpanSnapshot.new( + trace_id: span.hex_trace_id, + span_id: span.hex_span_id, + name: span.name, + scope_name: span.instrumentation_scope&.name, + attributes: frozen_attributes(span.attributes), + resource_attributes: frozen_attributes(span.resource&.attribute_enumerator&.to_h) + ) + end + + # Copies mutable values so freezing the snapshot never freezes objects + # still referenced by the original span data. + def frozen_attributes(attributes) + (attributes || {}).transform_values { |value| frozen_copy(value) }.freeze + end + + def frozen_copy(value) + case value + when String, Array then value.dup.freeze + else value + end + end + + # Unpatched spans are passed through untouched; patched spans are cloned + # so the originals stay pristine for any other exporter. + def apply_patches(batch, patches) + batch.filter_map do |span| + patch = patches[span_identifier(span)] + next span unless patch + + apply_patch(span, patch) + end + end + + # @return [SpanData, nil] the cloned masked span, or nil to drop it + def apply_patch(span, patch) + unless valid_patch?(patch) + @logger.error( + "Langfuse mask_otel_spans produced a malformed patch for span '#{span.name}'; " \ + "dropping the span from the Langfuse export" + ) + return nil + end + + clone_with_attributes(span, patched_attributes(span, patch)) + end + + def valid_patch?(patch) + patch.is_a?(Hash) && + (patch.keys - PATCH_KEYS).empty? && + (patch[:delete].nil? || patch[:delete].is_a?(Array)) && + (patch[:set].nil? || patch[:set].is_a?(Hash)) + end + + # Deletes run before sets so a replacement wins when a key appears in both. + def patched_attributes(span, patch) + attributes = (span.attributes || {}).dup + Array(patch[:delete]).each { |key| attributes.delete(key.to_s) } + (patch[:set] || {}).each do |key, value| + apply_replacement(attributes, key.to_s, value) + end + attributes.freeze + end + + # Invalid replacements delete the key: never export the original value of + # an attribute the hook intended to overwrite. The value itself is never + # logged. + def apply_replacement(attributes, key, value) + if valid_attribute_value?(value) + attributes[key] = value + else + attributes.delete(key) + @logger.warn( + "Langfuse mask_otel_spans replacement for attribute '#{key}' is not a valid " \ + "OpenTelemetry attribute value (#{value.class}); omitting the attribute" + ) + end + end + + def valid_attribute_value?(value) + case value + when String, Integer, Float, true, false then true + when Array then homogeneous_scalar_array?(value) + else false + end + end + + def homogeneous_scalar_array?(array) + return true if array.empty? + # Booleans mix TrueClass and FalseClass but form one OpenTelemetry type. + return array.all? { |element| [true, false].include?(element) } if [true, false].include?(array.first) + + type = array.first.class + [String, Integer, Float].include?(type) && array.all? { |element| element.instance_of?(type) } + end + + def clone_with_attributes(span, attributes) + copy = span.dup + copy.attributes = attributes + copy + end + end +end diff --git a/lib/langfuse/otel_setup.rb b/lib/langfuse/otel_setup.rb index e4bfaa1..c032950 100644 --- a/lib/langfuse/otel_setup.rb +++ b/lib/langfuse/otel_setup.rb @@ -16,6 +16,7 @@ module OtelSetup release sample_rate should_export_span + mask_otel_spans tracing_async batch_size flush_interval @@ -137,11 +138,14 @@ def build_tracer_provider(config) end def build_exporter(config) - OpenTelemetry::Exporter::OTLP::Exporter.new( + exporter = OpenTelemetry::Exporter::OTLP::Exporter.new( endpoint: "#{config.base_url}/api/public/otel/v1/traces", headers: build_headers(config.public_key, config.secret_key), compression: "gzip" ) + return exporter unless config.mask_otel_spans + + MaskingExporter.new(delegate: exporter, hook: config.mask_otel_spans, logger: config.logger) end def log_initialized(config) @@ -153,9 +157,15 @@ def validate_tracing_config!(config) raise ConfigurationError, "public_key is required" if blank?(config.public_key) raise ConfigurationError, "secret_key is required" if blank?(config.secret_key) raise ConfigurationError, "base_url cannot be empty" if blank?(config.base_url) - return if config.should_export_span.nil? || config.should_export_span.respond_to?(:call) - raise ConfigurationError, "should_export_span must respond to #call" + validate_callable!(config.should_export_span, "should_export_span") + validate_callable!(config.mask_otel_spans, "mask_otel_spans") + end + + def validate_callable!(value, name) + return if value.nil? || value.respond_to?(:call) + + raise ConfigurationError, "#{name} must respond to #call" end def tracing_config_snapshot(config) diff --git a/spec/langfuse/config_spec.rb b/spec/langfuse/config_spec.rb index 40b89cf..3276d64 100644 --- a/spec/langfuse/config_spec.rb +++ b/spec/langfuse/config_spec.rb @@ -592,6 +592,37 @@ def obj.call(data:) = data end end + describe "mask_otel_spans validation" do + let(:config) do + described_class.new do |c| + c.public_key = "pk_test" + c.secret_key = "sk_test" + end + end + + it "defaults mask_otel_spans to nil" do + expect(config.mask_otel_spans).to be_nil + end + + it "passes validation when mask_otel_spans is nil" do + config.mask_otel_spans = nil + expect { config.validate! }.not_to raise_error + end + + it "passes validation when mask_otel_spans responds to #call" do + config.mask_otel_spans = ->(spans:) { spans and nil } + expect { config.validate! }.not_to raise_error + end + + it "fails validation when mask_otel_spans does not respond to #call" do + config.mask_otel_spans = "not_callable" + expect { config.validate! }.to raise_error( + Langfuse::ConfigurationError, + "mask_otel_spans must respond to #call" + ) + end + end + describe "stale-while-revalidate integration" do it "works with all configuration options together" do config = described_class.new do |c| diff --git a/spec/langfuse/masking_exporter_spec.rb b/spec/langfuse/masking_exporter_spec.rb new file mode 100644 index 0000000..5ca5772 --- /dev/null +++ b/spec/langfuse/masking_exporter_spec.rb @@ -0,0 +1,293 @@ +# frozen_string_literal: true + +require "spec_helper" + +RSpec.describe Langfuse::MaskingExporter do + let(:logger) { instance_double(Logger, error: nil, warn: nil) } + let(:delegate) { OpenTelemetry::SDK::Trace::Export::InMemorySpanExporter.new } + let(:hook) { ->(**) {} } + let(:exporter) { described_class.new(delegate: delegate, hook: hook, logger: logger) } + + let(:langfuse_span) do + build_span_data(name: "langfuse-span", scope_name: "langfuse", + attributes: { "langfuse.observation.input" => "secret input" }) + end + let(:third_party_span) do + # Unfrozen attribute values so specs can prove snapshots copy rather than + # freeze the originals. + build_span_data(name: "chat gpt-4o", scope_name: "ruby-openai", + attributes: { "gen_ai.prompt" => +"ssn 123-45-6789", "gen_ai.system" => +"openai" }) + end + + def batch + [langfuse_span, third_party_span] + end + + def build_span_data(name:, scope_name:, attributes: {}) + OpenTelemetry::SDK::Trace::SpanData.new( + name, :internal, OpenTelemetry::Trace::Status.ok, OpenTelemetry::Trace::INVALID_SPAN_ID, + attributes.size, 0, 0, 1_000, 2_000, attributes, nil, nil, + OpenTelemetry::SDK::Resources::Resource.create({ "service.name" => "test-app" }), + OpenTelemetry::SDK::InstrumentationScope.new(scope_name, "1.0.0"), + OpenTelemetry::Trace.generate_span_id, OpenTelemetry::Trace.generate_trace_id, + OpenTelemetry::Trace::TraceFlags::DEFAULT, OpenTelemetry::Trace::Tracestate::DEFAULT, false + ) + end + + def identifier(span) + "#{span.hex_trace_id}:#{span.hex_span_id}" + end + + describe "#export" do + context "when the hook returns nil" do + it "exports the original batch unchanged" do + result = exporter.export(batch) + + expect(result).to eq(OpenTelemetry::SDK::Trace::Export::SUCCESS) + expect(delegate.finished_spans.map(&:name)).to eq(["langfuse-span", "chat gpt-4o"]) + end + + it "passes through the original span objects" do + exporter.export(batch) + + expect(delegate.finished_spans[0]).to equal(langfuse_span) + expect(delegate.finished_spans[1]).to equal(third_party_span) + end + end + + context "with snapshots given to the hook" do + let(:seen) { [] } + let(:hook) do + lambda do |spans:| + seen << spans + nil + end + end + + it "exposes both Langfuse-owned and third-party spans" do + exporter.export(batch) + + expect(seen.first.keys).to contain_exactly(identifier(langfuse_span), identifier(third_party_span)) + end + + it "exposes trace/span ids, name, scope, and frozen attributes" do + exporter.export(batch) + + snapshot = seen.first[identifier(third_party_span)] + expect(snapshot.trace_id).to eq(third_party_span.hex_trace_id) + expect(snapshot.span_id).to eq(third_party_span.hex_span_id) + expect(snapshot.name).to eq("chat gpt-4o") + expect(snapshot.scope_name).to eq("ruby-openai") + expect(snapshot.attributes).to be_frozen + expect(snapshot.attributes["gen_ai.prompt"]).to be_frozen + expect(snapshot.resource_attributes).to include("service.name" => "test-app") + end + + it "gives the hook a frozen mapping" do + exporter.export(batch) + + expect(seen.first).to be_frozen + end + + it "does not freeze the original span attribute values" do + exporter.export(batch) + + expect(third_party_span.attributes["gen_ai.prompt"]).not_to be_frozen + end + end + + context "with sparse patches" do + let(:hook) do + lambda do |spans:| + target = spans.values.find { |snapshot| snapshot.scope_name == "ruby-openai" } + { + "#{target.trace_id}:#{target.span_id}" => { + delete: ["gen_ai.prompt"], + set: { "gen_ai.prompt.masked" => true } + } + } + end + end + + it "deletes and sets attributes on the exported copy" do + exporter.export(batch) + + masked = delegate.finished_spans.find { |span| span.name == "chat gpt-4o" } + expect(masked.attributes).to eq("gen_ai.system" => "openai", "gen_ai.prompt.masked" => true) + end + + it "leaves unpatched spans as the original objects" do + exporter.export(batch) + + expect(delegate.finished_spans.find { |span| span.name == "langfuse-span" }).to equal(langfuse_span) + end + + it "never mutates the original span data" do + exporter.export(batch) + + expect(third_party_span.attributes).to eq( + "gen_ai.prompt" => "ssn 123-45-6789", "gen_ai.system" => "openai" + ) + end + end + + context "with delete-then-set precedence" do + let(:hook) do + lambda do |spans:| + spans.each_key.to_h do |id| + [id, { delete: ["gen_ai.prompt"], set: { "gen_ai.prompt" => "" } }] + end + end + end + + it "lets the replacement win when a key appears in both" do + exporter.export([third_party_span]) + + expect(delegate.finished_spans.first.attributes["gen_ai.prompt"]).to eq("") + end + end + + context "when the hook raises" do + let(:hook) { ->(**) { raise "boom" } } + + it "drops the whole batch and returns FAILURE" do + result = exporter.export(batch) + + expect(result).to eq(OpenTelemetry::SDK::Trace::Export::FAILURE) + expect(delegate.finished_spans).to be_empty + end + + it "logs a sanitized error" do + exporter.export(batch) + + expect(logger).to have_received(:error).with(/mask_otel_spans raised RuntimeError/) + end + end + + context "when the hook returns an invalid top-level result" do + let(:hook) { ->(**) { "not a hash" } } + + it "drops the whole batch and returns FAILURE" do + result = exporter.export(batch) + + expect(result).to eq(OpenTelemetry::SDK::Trace::Export::FAILURE) + expect(delegate.finished_spans).to be_empty + expect(logger).to have_received(:error).with(/returned String/) + end + end + + context "when one span patch is malformed" do + let(:hook) do + lambda do |spans:| + langfuse_id, third_party_id = spans.keys + { + langfuse_id => "not a patch", + third_party_id => { delete: ["gen_ai.prompt"] } + } + end + end + + it "drops only that span and keeps the rest of the batch" do + exporter.export(batch) + + expect(delegate.finished_spans.map(&:name)).to eq(["chat gpt-4o"]) + expect(logger).to have_received(:error).with(/malformed patch for span 'langfuse-span'/) + end + end + + context "when a patch has unknown keys" do + let(:hook) do + ->(spans:) { spans.each_key.to_h { |id| [id, { remove: ["gen_ai.prompt"] }] } } + end + + it "treats the patch as malformed and drops the span" do + exporter.export([third_party_span]) + + expect(delegate.finished_spans).to be_empty + expect(logger).to have_received(:error).with(/malformed patch/) + end + end + + context "when a replacement attribute value is invalid" do + let(:hook) do + ->(spans:) { spans.each_key.to_h { |id| [id, { set: { "gen_ai.prompt" => Object.new } }] } } + end + + it "omits the attribute instead of exporting the original value" do + exporter.export([third_party_span]) + + exported = delegate.finished_spans.first + expect(exported.attributes).not_to have_key("gen_ai.prompt") + expect(exported.attributes["gen_ai.system"]).to eq("openai") + end + + it "logs the key without the value" do + exporter.export([third_party_span]) + + expect(logger).to have_received(:warn) do |message| + expect(message).to include("gen_ai.prompt") + expect(message).not_to include("ssn 123-45-6789") + end + end + end + + context "with valid replacement value types" do + let(:hook) do + lambda do |spans:| + spans.each_key.to_h do |id| + [id, { set: { + "string" => "ok", "int" => 1, "float" => 1.5, "bool" => true, + "bools" => [true, false], "strings" => %w[a b], "empty" => [] + } }] + end + end + end + + it "accepts scalars and homogeneous scalar arrays" do + exporter.export([third_party_span]) + + attributes = delegate.finished_spans.first.attributes + expect(attributes).to include( + "string" => "ok", "int" => 1, "float" => 1.5, "bool" => true, + "bools" => [true, false], "strings" => %w[a b], "empty" => [] + ) + expect(logger).not_to have_received(:warn) + end + end + + context "with a heterogeneous array replacement" do + let(:hook) do + ->(spans:) { spans.each_key.to_h { |id| [id, { set: { "mixed" => ["a", 1] } }] } } + end + + it "rejects the value and omits the attribute" do + exporter.export([third_party_span]) + + expect(delegate.finished_spans.first.attributes).not_to have_key("mixed") + expect(logger).to have_received(:warn).with(/'mixed'/) + end + end + + context "with patches keyed by unknown identifiers" do + let(:hook) { ->(**) { { "deadbeef:cafebabe" => { delete: ["x"] } } } } + + it "ignores them and exports the batch" do + exporter.export(batch) + + expect(delegate.finished_spans.size).to eq(2) + end + end + end + + describe "#force_flush and #shutdown" do + it "delegates force_flush" do + expect(delegate).to receive(:force_flush).with(timeout: 5) + exporter.force_flush(timeout: 5) + end + + it "delegates shutdown" do + expect(delegate).to receive(:shutdown).with(timeout: 5) + exporter.shutdown(timeout: 5) + end + end +end diff --git a/spec/langfuse/otel_setup_spec.rb b/spec/langfuse/otel_setup_spec.rb index 7f76798..fd0793e 100644 --- a/spec/langfuse/otel_setup_spec.rb +++ b/spec/langfuse/otel_setup_spec.rb @@ -256,4 +256,65 @@ expect(exporter.finished_spans).to be_empty end end + + describe ".build_exporter" do + before do + allow(described_class).to receive(:build_exporter).and_call_original + end + + it "returns the plain OTLP exporter when mask_otel_spans is nil" do + expect(described_class.send(:build_exporter, config)).to be_a(OpenTelemetry::Exporter::OTLP::Exporter) + end + + it "wraps the OTLP exporter in a MaskingExporter when mask_otel_spans is configured" do + config.mask_otel_spans = ->(**) {} + expect(described_class.send(:build_exporter, config)).to be_a(Langfuse::MaskingExporter) + end + end + + describe "export-stage masking" do + let(:seen_spans) { [] } + let(:mask_hook) do + lambda do |spans:| + seen_spans.concat(spans.values) + spans.filter_map do |id, snapshot| + next unless snapshot.attributes.key?("gen_ai.prompt") + + [id, { set: { "gen_ai.prompt" => "" } }] + end.to_h + end + end + + before do + allow(described_class).to receive(:build_exporter).and_call_original + allow(OpenTelemetry::Exporter::OTLP::Exporter).to receive(:new).and_return(exporter) + Langfuse.reset! + Langfuse.configure do |c| + c.public_key = config.public_key + c.secret_key = config.secret_key + c.base_url = config.base_url + c.tracing_async = false + c.mask_otel_spans = mask_hook + c.logger = logger + end + end + + it "masks accepted third-party gen_ai spans on the Langfuse export copy" do + tracer = Langfuse.tracer_provider.tracer("langsmith.client") + tracer.start_span("llm-call", attributes: { "gen_ai.prompt" => "secret" }).finish + Langfuse.force_flush(timeout: 1) + + exported = exporter.finished_spans.find { |span| span.name == "llm-call" } + expect(exported.attributes["gen_ai.prompt"]).to eq("") + end + + it "runs should_export_span before masking so rejected spans never reach the hook" do + Langfuse.tracer_provider.tracer("dalli").start_span("cache-span").finish + tracer = Langfuse.tracer_provider.tracer("langsmith.client") + tracer.start_span("llm-call", attributes: { "gen_ai.prompt" => "x" }).finish + Langfuse.force_flush(timeout: 1) + + expect(seen_spans.map(&:name)).to eq(["llm-call"]) + end + end end From cb77bfec140d9fd12130fd72b71a1480d2b09894 Mon Sep 17 00:00:00 2001 From: kadekillary Date: Mon, 13 Jul 2026 07:49:47 -0700 Subject: [PATCH 2/3] [AAI-236] Sanitize hook error logs and deep-copy snapshot arrays --- lib/langfuse/masking_exporter.rb | 9 +++++++-- spec/langfuse/masking_exporter_spec.rb | 23 ++++++++++++++++++++--- 2 files changed, 27 insertions(+), 5 deletions(-) diff --git a/lib/langfuse/masking_exporter.rb b/lib/langfuse/masking_exporter.rb index 95e1613..571dd6e 100644 --- a/lib/langfuse/masking_exporter.rb +++ b/lib/langfuse/masking_exporter.rb @@ -98,7 +98,9 @@ def mask_batch(batch) apply_patches(batch, patches) rescue StandardError => e - @logger.error("Langfuse mask_otel_spans raised #{e.class}: #{e.message}; dropping the Langfuse export batch") + # Only the exception class is logged: hook exception messages can carry + # the sensitive attribute values the hook exists to mask. + @logger.error("Langfuse mask_otel_spans raised #{e.class}; dropping the Langfuse export batch") nil end @@ -128,9 +130,12 @@ def frozen_attributes(attributes) (attributes || {}).transform_values { |value| frozen_copy(value) }.freeze end + # Arrays are copied element-wise so snapshot values never alias mutable + # strings still referenced by the original span data. def frozen_copy(value) case value - when String, Array then value.dup.freeze + when String then value.dup.freeze + when Array then value.map { |element| frozen_copy(element) }.freeze else value end end diff --git a/spec/langfuse/masking_exporter_spec.rb b/spec/langfuse/masking_exporter_spec.rb index 5ca5772..67734b7 100644 --- a/spec/langfuse/masking_exporter_spec.rb +++ b/spec/langfuse/masking_exporter_spec.rb @@ -94,6 +94,20 @@ def identifier(span) expect(third_party_span.attributes["gen_ai.prompt"]).not_to be_frozen end + + it "copies array attribute elements instead of aliasing them" do + original = [+"tag-one", +"tag-two"] + span = build_span_data(name: "array-span", scope_name: "ruby-openai", + attributes: { "gen_ai.tags" => original }) + + exporter.export([span]) + + snapshot_value = seen.first[identifier(span)].attributes["gen_ai.tags"] + expect(snapshot_value).to be_frozen + expect(snapshot_value.first).to be_frozen + expect(snapshot_value.first).not_to equal(original.first) + expect(original.first).not_to be_frozen + end end context "with sparse patches" do @@ -148,7 +162,7 @@ def identifier(span) end context "when the hook raises" do - let(:hook) { ->(**) { raise "boom" } } + let(:hook) { ->(**) { raise "leaky ssn 123-45-6789" } } it "drops the whole batch and returns FAILURE" do result = exporter.export(batch) @@ -157,10 +171,13 @@ def identifier(span) expect(delegate.finished_spans).to be_empty end - it "logs a sanitized error" do + it "logs only the exception class, never the message" do exporter.export(batch) - expect(logger).to have_received(:error).with(/mask_otel_spans raised RuntimeError/) + expect(logger).to have_received(:error) do |message| + expect(message).to include("mask_otel_spans raised RuntimeError") + expect(message).not_to include("leaky ssn") + end end end From 3c444c6f2ef30795fef54a1a4d34624fb54d4218 Mon Sep 17 00:00:00 2001 From: kadekillary Date: Wed, 15 Jul 2026 06:03:07 -0700 Subject: [PATCH 3/3] fix(tracing): harden masking exporter --- lib/langfuse/masking_exporter.rb | 129 +++++++++++++++---------- lib/langfuse/otel_setup.rb | 1 + spec/langfuse/masking_exporter_spec.rb | 57 ++++++++++- 3 files changed, 133 insertions(+), 54 deletions(-) diff --git a/lib/langfuse/masking_exporter.rb b/lib/langfuse/masking_exporter.rb index 571dd6e..a272115 100644 --- a/lib/langfuse/masking_exporter.rb +++ b/lib/langfuse/masking_exporter.rb @@ -23,7 +23,39 @@ module Langfuse # @return [Hash] frozen resource attributes OtelSpanSnapshot = Data.define( :trace_id, :span_id, :name, :scope_name, :attributes, :resource_attributes - ) + ) do + class << self + # Build an immutable snapshot without freezing objects owned by the + # original span data. + # @api private + def from_span(span) + new( + trace_id: span.hex_trace_id, + span_id: span.hex_span_id, + name: frozen_copy(span.name), + scope_name: frozen_copy(span.instrumentation_scope&.name), + attributes: frozen_attributes(span.attributes), + resource_attributes: frozen_attributes(span.resource&.attribute_enumerator&.to_h) + ) + end + + private + + def frozen_attributes(attributes) + (attributes || {}).to_h do |key, value| + [frozen_copy(key), frozen_copy(value)] + end.freeze + end + + def frozen_copy(value) + case value + when String then value.dup.freeze + when Array then value.map { |element| frozen_copy(element) }.freeze + else value + end + end + end + end # Export-stage masking wrapper around the Langfuse OTLP exporter. # @@ -85,7 +117,10 @@ def shutdown(timeout: nil) # @return [Array, nil] the masked batch, or nil to drop it def mask_batch(batch) - patches = @hook.call(spans: snapshot_batch(batch)) + snapshots = snapshot_batch(batch) + return unless snapshots + + patches = @hook.call(spans: snapshots) return batch if patches.nil? unless patches.is_a?(Hash) @@ -96,6 +131,14 @@ def mask_batch(batch) return nil end + unless patches.each_key.all? { |identifier| snapshots.key?(identifier) } + @logger.error( + "Langfuse mask_otel_spans returned a patch for an unknown span identifier; " \ + "dropping the Langfuse export batch" + ) + return nil + end + apply_patches(batch, patches) rescue StandardError => e # Only the exception class is logged: hook exception messages can carry @@ -105,7 +148,18 @@ def mask_batch(batch) end def snapshot_batch(batch) - batch.to_h { |span| [span_identifier(span), snapshot(span)] }.freeze + batch.each_with_object({}) do |span, snapshots| + identifier = span_identifier(span) + if snapshots.key?(identifier) + @logger.error( + "Langfuse mask_otel_spans received duplicate span identifiers; " \ + "dropping the Langfuse export batch" + ) + return nil + end + + snapshots[identifier] = OtelSpanSnapshot.from_span(span) + end.freeze end # Stable identifier the hook uses to key sparse patches. @@ -113,41 +167,14 @@ def span_identifier(span) "#{span.hex_trace_id}:#{span.hex_span_id}" end - def snapshot(span) - OtelSpanSnapshot.new( - trace_id: span.hex_trace_id, - span_id: span.hex_span_id, - name: span.name, - scope_name: span.instrumentation_scope&.name, - attributes: frozen_attributes(span.attributes), - resource_attributes: frozen_attributes(span.resource&.attribute_enumerator&.to_h) - ) - end - - # Copies mutable values so freezing the snapshot never freezes objects - # still referenced by the original span data. - def frozen_attributes(attributes) - (attributes || {}).transform_values { |value| frozen_copy(value) }.freeze - end - - # Arrays are copied element-wise so snapshot values never alias mutable - # strings still referenced by the original span data. - def frozen_copy(value) - case value - when String then value.dup.freeze - when Array then value.map { |element| frozen_copy(element) }.freeze - else value - end - end - # Unpatched spans are passed through untouched; patched spans are cloned # so the originals stay pristine for any other exporter. def apply_patches(batch, patches) batch.filter_map do |span| - patch = patches[span_identifier(span)] - next span unless patch + identifier = span_identifier(span) + next span unless patches.key?(identifier) - apply_patch(span, patch) + apply_patch(span, patches.fetch(identifier)) end end @@ -167,16 +194,24 @@ def apply_patch(span, patch) def valid_patch?(patch) patch.is_a?(Hash) && (patch.keys - PATCH_KEYS).empty? && - (patch[:delete].nil? || patch[:delete].is_a?(Array)) && - (patch[:set].nil? || patch[:set].is_a?(Hash)) + valid_delete_keys?(patch[:delete]) && + valid_set_keys?(patch[:set]) + end + + def valid_delete_keys?(keys) + keys.nil? || (keys.is_a?(Array) && keys.all? { |key| valid_attribute_key?(key) }) + end + + def valid_set_keys?(attributes) + attributes.nil? || (attributes.is_a?(Hash) && attributes.each_key.all? { |key| valid_attribute_key?(key) }) end # Deletes run before sets so a replacement wins when a key appears in both. def patched_attributes(span, patch) attributes = (span.attributes || {}).dup - Array(patch[:delete]).each { |key| attributes.delete(key.to_s) } + Array(patch[:delete]).each { |key| attributes.delete(key) } (patch[:set] || {}).each do |key, value| - apply_replacement(attributes, key.to_s, value) + apply_replacement(attributes, key, value) end attributes.freeze end @@ -197,26 +232,22 @@ def apply_replacement(attributes, key, value) end def valid_attribute_value?(value) - case value - when String, Integer, Float, true, false then true - when Array then homogeneous_scalar_array?(value) - else false - end + OpenTelemetry::SDK::Internal.valid_value?(value) end - def homogeneous_scalar_array?(array) - return true if array.empty? - # Booleans mix TrueClass and FalseClass but form one OpenTelemetry type. - return array.all? { |element| [true, false].include?(element) } if [true, false].include?(array.first) - - type = array.first.class - [String, Integer, Float].include?(type) && array.all? { |element| element.instance_of?(type) } + def valid_attribute_key?(key) + OpenTelemetry::SDK::Internal.valid_key?(key) end def clone_with_attributes(span, attributes) copy = span.dup copy.attributes = attributes + copy.total_recorded_attributes = attributes.size + dropped_attribute_count(span) copy end + + def dropped_attribute_count(span) + [span.total_recorded_attributes.to_i - (span.attributes || {}).size, 0].max + end end end diff --git a/lib/langfuse/otel_setup.rb b/lib/langfuse/otel_setup.rb index c032950..eb8b21b 100644 --- a/lib/langfuse/otel_setup.rb +++ b/lib/langfuse/otel_setup.rb @@ -3,6 +3,7 @@ require "opentelemetry/sdk" require "opentelemetry/exporter/otlp" require "base64" +require_relative "masking_exporter" module Langfuse # OpenTelemetry initialization and setup for Langfuse tracing. diff --git a/spec/langfuse/masking_exporter_spec.rb b/spec/langfuse/masking_exporter_spec.rb index 67734b7..3adbd74 100644 --- a/spec/langfuse/masking_exporter_spec.rb +++ b/spec/langfuse/masking_exporter_spec.rb @@ -78,6 +78,8 @@ def identifier(span) expect(snapshot.span_id).to eq(third_party_span.hex_span_id) expect(snapshot.name).to eq("chat gpt-4o") expect(snapshot.scope_name).to eq("ruby-openai") + expect(snapshot.name).to be_frozen + expect(snapshot.scope_name).to be_frozen expect(snapshot.attributes).to be_frozen expect(snapshot.attributes["gen_ai.prompt"]).to be_frozen expect(snapshot.resource_attributes).to include("service.name" => "test-app") @@ -108,6 +110,16 @@ def identifier(span) expect(snapshot_value.first).not_to equal(original.first) expect(original.first).not_to be_frozen end + + it "copies span and scope names instead of exposing shared strings" do + span = build_span_data(name: +"mutable-name", scope_name: +"mutable-scope") + + exporter.export([span]) + + snapshot = seen.first[identifier(span)] + expect(snapshot.name).not_to equal(span.name) + expect(snapshot.scope_name).not_to equal(span.instrumentation_scope.name) + end end context "with sparse patches" do @@ -212,6 +224,17 @@ def identifier(span) end end + context "when a present patch is false" do + let(:hook) { ->(spans:) { { spans.keys.first => false } } } + + it "drops the span instead of exporting it unchanged" do + exporter.export([third_party_span]) + + expect(delegate.finished_spans).to be_empty + expect(logger).to have_received(:error).with(/malformed patch/) + end + end + context "when a patch has unknown keys" do let(:hook) do ->(spans:) { spans.each_key.to_h { |id| [id, { remove: ["gen_ai.prompt"] }] } } @@ -254,7 +277,8 @@ def identifier(span) spans.each_key.to_h do |id| [id, { set: { "string" => "ok", "int" => 1, "float" => 1.5, "bool" => true, - "bools" => [true, false], "strings" => %w[a b], "empty" => [] + "bools" => [true, false], "numbers" => [1, 2.5], + "strings" => %w[a b], "empty" => [] } }] end end @@ -266,10 +290,18 @@ def identifier(span) attributes = delegate.finished_spans.first.attributes expect(attributes).to include( "string" => "ok", "int" => 1, "float" => 1.5, "bool" => true, - "bools" => [true, false], "strings" => %w[a b], "empty" => [] + "bools" => [true, false], "numbers" => [1, 2.5], + "strings" => %w[a b], "empty" => [] ) expect(logger).not_to have_received(:warn) end + + it "keeps OpenTelemetry dropped-attribute accounting valid" do + exporter.export([third_party_span]) + + exported = delegate.finished_spans.first + expect(exported.total_recorded_attributes).to be >= exported.attributes.size + end end context "with a heterogeneous array replacement" do @@ -288,10 +320,25 @@ def identifier(span) context "with patches keyed by unknown identifiers" do let(:hook) { ->(**) { { "deadbeef:cafebabe" => { delete: ["x"] } } } } - it "ignores them and exports the batch" do - exporter.export(batch) + it "drops the batch instead of exporting unmatched sensitive spans" do + result = exporter.export(batch) + + expect(result).to eq(OpenTelemetry::SDK::Trace::Export::FAILURE) + expect(delegate.finished_spans).to be_empty + expect(logger).to have_received(:error).with(/unknown span identifier/) + end + end + + context "with duplicate span identifiers" do + it "drops the batch instead of silently collapsing snapshots" do + duplicate = third_party_span.dup + duplicate.name = "duplicate" - expect(delegate.finished_spans.size).to eq(2) + result = exporter.export([third_party_span, duplicate]) + + expect(result).to eq(OpenTelemetry::SDK::Trace::Export::FAILURE) + expect(delegate.finished_spans).to be_empty + expect(logger).to have_received(:error).with(/duplicate span identifiers/) end end end