feat: sanitize utf8 strings with binary in spans

router-tests/telemetry/attribute_processor_test.go

@@ -0,0 +1,222 @@
+package telemetry
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/wundergraph/cosmo/router-tests/testenv"
+	"github.com/wundergraph/cosmo/router/core"
+	"github.com/wundergraph/cosmo/router/pkg/config"
+	"github.com/wundergraph/cosmo/router/pkg/trace/tracetest"
+	"go.opentelemetry.io/otel/attribute"
+	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
+	"go.uber.org/zap/zapcore"
+)
+
+func TestAttributeProcessorIntegration(t *testing.T) {
+	t.Parallel()
+
+	t.Run("SanitizeUTF8 logs warning when invalid UTF-8 is detected", func(t *testing.T) {
+		t.Parallel()
+
+		exporter := tracetest.NewInMemoryExporter(t)
+
+		// Create a string with invalid UTF-8 bytes
+		invalidUTF8Value := string([]byte{0x80, 0x81, 0x82})
+		sanitizedValue := strings.ToValidUTF8(invalidUTF8Value, "\ufffd")
+		attrKey := "custom.invalid_utf8_attr"
+
+		testenv.Run(t, &testenv.Config{
+			TraceExporter: exporter,
+			TracingSanitizeUTF8: &config.SanitizeUTF8Config{
+				Enabled:          true,
+				LogSanitizations: true,
+			},
+			// Add a custom tracing attribute with invalid UTF-8 as default value
+			CustomTracingAttributes: []config.CustomAttribute{
+				{
+					Key:     attrKey,
+					Default: invalidUTF8Value,
+				},
+			},
+			LogObservation: testenv.LogObservationConfig{
+				Enabled:  true,
+				LogLevel: zapcore.WarnLevel,
+			},
+		}, func(t *testing.T, xEnv *testenv.Environment) {
+			res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{
+				Query: `query { employees { id } }`,
+			})
+			require.Contains(t, res.Body, `"employees"`)
+
+			// Verify that spans were created
+			sn := exporter.GetSpans().Snapshots()
+			require.NotEmpty(t, sn)
+
+			// Verify that the invalid UTF-8 attribute was sanitized (replaced with U+FFFD)
+			sanitizedAttr := attribute.String(attrKey, sanitizedValue)
+			require.Contains(t, sn[0].Attributes(), sanitizedAttr)
+
+			// Verify that the warning log was emitted
+			logEntries := xEnv.Observer().FilterMessageSnippet("Invalid UTF-8 in span attribute").All()
+			require.GreaterOrEqual(t, len(logEntries), 1)
+
+			// Verify the log contains the attribute key
+			logEntry := logEntries[0]
+			contextMap := logEntry.ContextMap()
+			require.Equal(t, attrKey, contextMap["key"])
+		})
+	})
+
+	t.Run("SanitizeUTF8 does not log when logging is disabled", func(t *testing.T) {
+		t.Parallel()
+
+		exporter := tracetest.NewInMemoryExporter(t)
+
+		// Create a string with invalid UTF-8 bytes
+		invalidUTF8Value := string([]byte{0x80, 0x81, 0x82})
+		sanitizedValue := strings.ToValidUTF8(invalidUTF8Value, "\ufffd")
+		attrKey := "custom.invalid_utf8_attr_no_log"
+
+		testenv.Run(t, &testenv.Config{
+			TraceExporter: exporter,
+			TracingSanitizeUTF8: &config.SanitizeUTF8Config{
+				Enabled:          true,
+				LogSanitizations: false, // Logging disabled
+			},
+			CustomTracingAttributes: []config.CustomAttribute{
+				{
+					Key:     attrKey,
+					Default: invalidUTF8Value,
+				},
+			},
+			LogObservation: testenv.LogObservationConfig{
+				Enabled:  true,
+				LogLevel: zapcore.WarnLevel,
+			},
+		}, func(t *testing.T, xEnv *testenv.Environment) {
+			res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{
+				Query: `query { employees { id } }`,
+			})
+			require.Contains(t, res.Body, `"employees"`)
+
+			// Verify that spans were created
+			sn := exporter.GetSpans().Snapshots()
+			require.NotEmpty(t, sn)
+
+			// Verify that the invalid UTF-8 attribute was still sanitized
+			sanitizedAttr := attribute.String(attrKey, sanitizedValue)
+			require.Contains(t, sn[0].Attributes(), sanitizedAttr)
+
+			// Verify that NO warning log was emitted for the sanitization
+			logEntries := xEnv.Observer().FilterMessageSnippet("Invalid UTF-8 in span attribute").All()
+			require.Empty(t, logEntries)
+		})
+	})
+
+	t.Run("SanitizeUTF8 disabled leaves invalid UTF-8 unchanged", func(t *testing.T) {
+		t.Parallel()
+
+		exporter := tracetest.NewInMemoryExporter(t)
+
+		// Create a string with invalid UTF-8 bytes
+		invalidUTF8Value := string([]byte{0x80, 0x81, 0x82})
+		attrKey := "custom.invalid_utf8_unchanged"
+
+		testenv.Run(t, &testenv.Config{
+			TraceExporter: exporter,
+			TracingSanitizeUTF8: &config.SanitizeUTF8Config{
+				Enabled: false, // Disabled
+			},
+			CustomTracingAttributes: []config.CustomAttribute{
+				{
+					Key:     attrKey,
+					Default: invalidUTF8Value,
+				},
+			},
+		}, func(t *testing.T, xEnv *testenv.Environment) {
+			res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{
+				Query: `query { employees { id } }`,
+			})
+			require.Contains(t, res.Body, `"employees"`)
+
+			// Verify that spans were created
+			sn := exporter.GetSpans().Snapshots()
+			require.NotEmpty(t, sn)
+
+			// Verify that the invalid UTF-8 attribute was NOT sanitized
+			require.Contains(t, sn[0].Attributes(), attribute.String(attrKey, invalidUTF8Value))
+		})
+	})
+
+	t.Run("IPAnonymization redacts IP attributes", func(t *testing.T) {
+		t.Parallel()
+
+		exporter := tracetest.NewInMemoryExporter(t)
+
+		testenv.Run(t, &testenv.Config{
+			TraceExporter: exporter,
+			IPAnonymization: &core.IPAnonymizationConfig{
+				Enabled: true,
+				Method:  core.Redact,
+			},
+		}, func(t *testing.T, xEnv *testenv.Environment) {
+			res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{
+				Query: `query { employees { id } }`,
+			})
+			require.Contains(t, res.Body, `"employees"`)
+
+			sn := exporter.GetSpans().Snapshots()
+			require.NotEmpty(t, sn)
+
+			// Check that IP addresses are redacted (checks both http.client_ip and net.sock.peer.addr)
+			redactedIPCount := 0
+			for _, span := range sn {
+				for _, attr := range span.Attributes() {
+					if attr.Key == semconv.HTTPClientIPKey || attr.Key == semconv.NetSockPeerAddrKey {
+						redactedIPCount++
+						require.Equal(t, "[REDACTED]", attr.Value.AsString())
+					}
+				}
+			}
+			require.Positive(t, redactedIPCount)
+		})
+	})
+
+	t.Run("IPAnonymization hashes IP attributes", func(t *testing.T) {
+		t.Parallel()
+
+		exporter := tracetest.NewInMemoryExporter(t)
+
+		testenv.Run(t, &testenv.Config{
+			TraceExporter: exporter,
+			IPAnonymization: &core.IPAnonymizationConfig{
+				Enabled: true,
+				Method:  core.Hash,
+			},
+		}, func(t *testing.T, xEnv *testenv.Environment) {
+			res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{
+				Query: `query { employees { id } }`,
+			})
+			require.Contains(t, res.Body, `"employees"`)
+
+			sn := exporter.GetSpans().Snapshots()
+			require.NotEmpty(t, sn)
+
+			// Check that IP addresses are hashed (64 char hex) in spans that have them
+			hashedIPCount := 0
+			for _, span := range sn {
+				for _, attr := range span.Attributes() {
+					if attr.Key == semconv.HTTPClientIPKey || attr.Key == semconv.NetSockPeerAddrKey {
+						hashedIPCount++
+						value := attr.Value.AsString()
+						require.Len(t, value, 64)
+						require.NotEqual(t, "[REDACTED]", value)
+					}
+				}
+			}
+			require.Positive(t, hashedIPCount)
+		})
+	})
+}

router-tests/testenv/testenv.go

@@ -308,6 +308,8 @@ type Config struct {
 	DisableParentBasedSampler          bool
 	TLSConfig                          *core.TlsConfig
 	TraceExporter                      trace.SpanExporter
+	TracingSanitizeUTF8                *config.SanitizeUTF8Config
+	IPAnonymization                    *core.IPAnonymizationConfig
 	CustomMetricAttributes             []config.CustomAttribute
 	CustomTelemetryAttributes          []config.CustomAttribute
 	CustomTracingAttributes            []config.CustomAttribute
@@ -1483,19 +1485,25 @@ func configureRouter(listenerAddr string, testConfig *Config, routerConfig *node
 	if testConfig.TraceExporter != nil {
 		testConfig.PropagationConfig.TraceContext = true
 
+		tracingConfig := config.Tracing{
+			Enabled:                    true,
+			SamplingRate:               1,
+			ParentBasedSampler:         !testConfig.DisableParentBasedSampler,
+			OperationContentAttributes: testConfig.OperationContentAttributes,
+			Exporters:                  []config.TracingExporter{},
+			Propagation:                testConfig.PropagationConfig,
+			TracingGlobalFeatures:      config.TracingGlobalFeatures{},
+			ResponseTraceHeader:        testConfig.ResponseTraceHeader,
+		}
+
+		if testConfig.TracingSanitizeUTF8 != nil {
+			tracingConfig.SanitizeUTF8 = *testConfig.TracingSanitizeUTF8
+		}
+
 		c := core.TraceConfigFromTelemetry(&config.Telemetry{
 			ServiceName:        "cosmo-router",
 			ResourceAttributes: testConfig.CustomResourceAttributes,
-			Tracing: config.Tracing{
-				Enabled:                    true,
-				SamplingRate:               1,
-				ParentBasedSampler:         !testConfig.DisableParentBasedSampler,
-				OperationContentAttributes: testConfig.OperationContentAttributes,
-				Exporters:                  []config.TracingExporter{},
-				Propagation:                testConfig.PropagationConfig,
-				TracingGlobalFeatures:      config.TracingGlobalFeatures{},
-				ResponseTraceHeader:        testConfig.ResponseTraceHeader,
-			},
+			Tracing:            tracingConfig,
 		})
 
 		c.TestMemoryExporter = testConfig.TraceExporter
@@ -1505,6 +1513,10 @@ func configureRouter(listenerAddr string, testConfig *Config, routerConfig *node
 		)
 	}
 
+	if testConfig.IPAnonymization != nil {
+		routerOpts = append(routerOpts, core.WithAnonymization(testConfig.IPAnonymization))
+	}
+
 	if testConfig.CustomTelemetryAttributes != nil {
 		routerOpts = append(routerOpts, core.WithTelemetryAttributes(testConfig.CustomTelemetryAttributes))
 	}

router/core/engine_loader_hooks.go

@@ -16,6 +16,7 @@ import (
 	"github.com/wundergraph/cosmo/router/internal/unique"
 	"github.com/wundergraph/cosmo/router/pkg/metric"
 	rotel "github.com/wundergraph/cosmo/router/pkg/otel"
+	rtrace "github.com/wundergraph/cosmo/router/pkg/trace"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/engine/resolve"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
@@ -247,7 +248,7 @@ func (f *engineLoaderHooks) OnFinished(ctx context.Context, ds resolve.DataSourc
 	if responseInfo.Err != nil {
 		// Set error status. This is the fetch error from the engine
 		// Downstream errors are extracted from the subgraph response
-		span.SetStatus(codes.Error, responseInfo.Err.Error())
+		rtrace.SetSanitizedSpanStatus(span, codes.Error, responseInfo.Err.Error())
 		span.RecordError(responseInfo.Err)
 
 		var errorCodesAttr []string

router/core/graphql_prehandler.go

@@ -598,7 +598,7 @@ func (h *PreHandler) handleOperation(req *http.Request, httpOperation *httpOpera
 		span.SetAttributes(otel.WgEnginePersistedOperationCacheHit.Bool(operationKit.parsedOperation.PersistedOperationCacheHit))
 		if err != nil {
 			span.RecordError(err)
-			span.SetStatus(codes.Error, err.Error())
+			rtrace.SetSanitizedSpanStatus(span, codes.Error, err.Error())
 
 			var poNotFoundErr *persistedoperation.PersistentOperationNotFoundError
 			if h.operationBlocker.logUnknownOperationsEnabled && errors.As(err, &poNotFoundErr) {

router/core/router.go

@@ -52,6 +52,7 @@ import (
 	"github.com/wundergraph/cosmo/router/pkg/otel/otelconfig"
 	"github.com/wundergraph/cosmo/router/pkg/statistics"
 	rtrace "github.com/wundergraph/cosmo/router/pkg/trace"
+	"github.com/wundergraph/cosmo/router/pkg/trace/attributeprocessor"
 	"github.com/wundergraph/cosmo/router/pkg/watcher"
 	"github.com/wundergraph/graphql-go-tools/v2/pkg/netpoll"
 )
@@ -825,10 +826,11 @@ func (r *Router) bootstrap(ctx context.Context) error {
 			Logger:            r.logger,
 			Config:            r.traceConfig,
 			ServiceInstanceID: r.instanceID,
-			IPAnonymization: &rtrace.IPAnonymizationConfig{
+			IPAnonymization: &attributeprocessor.IPAnonymizationConfig{
 				Enabled: r.ipAnonymization.Enabled,
-				Method:  rtrace.IPAnonymizationMethod(r.ipAnonymization.Method),
+				Method:  attributeprocessor.IPAnonymizationMethod(r.ipAnonymization.Method),
 			},
+			SanitizeUTF8:   r.traceConfig.SanitizeUTF8,
 			MemoryExporter: r.traceConfig.TestMemoryExporter,
 		})
 		if err != nil {
@@ -2339,6 +2341,10 @@ func TraceConfigFromTelemetry(cfg *config.Telemetry) *rtrace.Config {
 		Propagators:                propagators,
 		ResponseTraceHeader:        cfg.Tracing.ResponseTraceHeader,
 		OperationContentAttributes: cfg.Tracing.OperationContentAttributes,
+		SanitizeUTF8: &attributeprocessor.SanitizeUTF8Config{
+			Enabled:          cfg.Tracing.SanitizeUTF8.Enabled,
+			LogSanitizations: cfg.Tracing.SanitizeUTF8.LogSanitizations,
+		},
 	}
 }
 

router/pkg/config/config.go

@@ -70,6 +70,11 @@ type ResponseTraceHeader struct {
 	HeaderName string `yaml:"header_name" envDefault:"x-wg-trace-id"`
 }
 
+type SanitizeUTF8Config struct {
+	Enabled          bool `yaml:"enabled" envDefault:"false" env:"ENABLED"`
+	LogSanitizations bool `yaml:"log_sanitizations" envDefault:"false" env:"LOG_SANITIZATIONS"`
+}
+
 type Tracing struct {
 	Enabled             bool                `yaml:"enabled" envDefault:"true" env:"TRACING_ENABLED"`
 	SamplingRate        float64             `yaml:"sampling_rate" envDefault:"1" env:"TRACING_SAMPLING_RATE"`
@@ -82,6 +87,9 @@ type Tracing struct {
 	OperationContentAttributes bool `yaml:"operation_content_attributes" envDefault:"false" env:"TRACING_OPERATION_CONTENT_ATTRIBUTES"`
 
 	TracingGlobalFeatures `yaml:",inline"`
+
+	// SanitizeUTF8 configures sanitization of invalid UTF-8 sequences in span attribute values
+	SanitizeUTF8 SanitizeUTF8Config `yaml:"sanitize_utf8" envPrefix:"TRACING_SANITIZE_UTF8_"`
 }
 
 type PropagationConfig struct {

router/pkg/config/config.schema.json

@@ -1068,6 +1068,23 @@
                   }
                 }
               }
+            },
+            "sanitize_utf8": {
+              "type": "object",
+              "description": "Configuration for sanitizing invalid UTF-8 sequences in span attribute values.",
+              "additionalProperties": false,
+              "properties": {
+                "enabled": {
+                  "type": "boolean",
+                  "default": false,
+                  "description": "Enable the sanitization of invalid UTF-8 sequences in span attribute values. Invalid sequences are replaced with the Unicode replacement character (U+FFFD)."
+                },
+                "log_sanitizations": {
+                  "type": "boolean",
+                  "default": false,
+                  "description": "Log a warning when invalid UTF-8 sequences are sanitized. The log includes the attribute key and original value."
+                }
+              }
             }
           }
         },

router/pkg/config/testdata/config_defaults.json

@@ -28,7 +28,11 @@
       "Attributes": null,
       "OperationContentAttributes": false,
       "ExportGraphQLVariables": false,
-      "WithNewRoot": false
+      "WithNewRoot": false,
+      "SanitizeUTF8": {
+        "Enabled": false,
+        "LogSanitizations": false
+      }
     },
     "Metrics": {
       "Attributes": null,