Sanitize gRPC metadata headers built from resource attributes

grpc-go hard-fails the entire RPC when an outgoing metadata pair fails its
charset/printability validation, unlike the CE-extension path where an
invalid entry is just dropped. Add SanitizeMetadataHeaders (reusing
SanitizeExtensionName for keys, plus a printable-ASCII value sanitizer) and
apply it once when building the static header provider, so a malformed
resource-attribute key or value can no longer take down chip-ingress
emission for the node.

Author: pkcll

pkg/beholder/client.go

@@ -217,7 +217,7 @@ func NewGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 		opts = append(opts, chipingress.WithTracerProvider(tracerProvider))
 
 		if len(resourceAttrs) > 0 {
-			opts = append(opts, chipingress.WithHeaderProvider(chipingress.NewStaticHeaderProvider(resourceAttrs)))
+			opts = append(opts, chipingress.WithHeaderProvider(chipingress.NewStaticHeaderProvider(chipingress.SanitizeMetadataHeaders(resourceAttrs))))
 		}
 
 		chipIngressClient, err = chipingress.NewClient(cfg.ChipIngressEmitterGRPCEndpoint, opts...)

pkg/chipingress/header_provider.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"fmt"
 	"maps"
+	"sort"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -115,6 +116,58 @@ func NewStaticHeaderProvider(headers map[string]string) HeaderProvider {
 	return newStaticHeaderProvider(headers, false)
 }
 
+// SanitizeMetadataValue replaces any byte outside the printable ASCII range [0x20-0x7E]
+// with '?'. grpc-go hard-fails the entire RPC when an outgoing metadata value fails this
+// check (unlike the CE-extension path, where an invalid entry is simply dropped), so
+// values headed for gRPC metadata must be normalized before being sent.
+func SanitizeMetadataValue(val string) string {
+	b := []byte(val)
+	out := make([]byte, len(b))
+	for i, c := range b {
+		if c >= 0x20 && c <= 0x7E {
+			out[i] = c
+		} else {
+			out[i] = '?'
+		}
+	}
+	return string(out)
+}
+
+// SanitizeMetadataHeaders sanitizes a map of resource-attribute headers for use as outgoing
+// gRPC metadata (e.g. via NewStaticHeaderProvider). Keys are sanitized with
+// SanitizeExtensionName — the same strict [a-z0-9] charset used for CloudEvent extensions —
+// which is a subset of grpc's allowed metadata-key charset, so a sanitized key can never trip
+// grpc's key validation or the reserved "-bin" suffix, and produces the same key stem as the
+// corresponding CE extension (differing only by the CloudEvents Kafka binding's "ce_" prefix
+// once on the wire). Values are sanitized via SanitizeMetadataValue, since grpc-go fails the
+// whole RPC on a non-printable value. Entries that sanitize to an empty key, or that collide
+// with a reserved extension name (see reservedExtensionNames), are skipped. Keys are applied
+// in sorted order so duplicate sanitized keys resolve deterministically (first in sorted order
+// wins), matching WithResourceAttributeExtensions' collision handling.
+func SanitizeMetadataHeaders(in map[string]string) map[string]string {
+	keys := make([]string, 0, len(in))
+	for k := range in {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+
+	out := make(map[string]string, len(in))
+	for _, k := range keys {
+		name := SanitizeExtensionName(k)
+		if name == "" {
+			continue
+		}
+		if _, reserved := reservedExtensionNames[name]; reserved {
+			continue
+		}
+		if _, exists := out[name]; exists {
+			continue
+		}
+		out[name] = SanitizeMetadataValue(in[k])
+	}
+	return out
+}
+
 // newRotatingHeaderProvider returns a HeaderProvider that refreshes its
 // headers every ttl using signer. initialHeaders, if non-empty, are served
 // until the first rotation occurs.

pkg/chipingress/header_provider_test.go

@@ -4,13 +4,18 @@ import (
 	"context"
 	"crypto/ed25519"
 	"encoding/hex"
+	"net"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress"
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb"
 )
 
 // fakeSigner is a minimal chipingress.Signer used by tests that need to
@@ -277,3 +282,110 @@ func TestNewStaticHeaderProvider(t *testing.T) {
 	require.True(t, ok)
 	assert.False(t, tlsReq.RequireTransportSecurity())
 }
+
+func TestSanitizeMetadataValue(t *testing.T) {
+	tests := []struct {
+		name string
+		in   string
+		want string
+	}{
+		{name: "printable ASCII is unchanged", in: "chain-1_prod.v2", want: "chain-1_prod.v2"},
+		{name: "empty", in: "", want: ""},
+		{name: "control character replaced", in: "value\nwith\tcontrol", want: "value?with?control"},
+		{name: "non-ASCII UTF-8 replaced byte-wise", in: "café", want: "caf??"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, chipingress.SanitizeMetadataValue(tt.in))
+		})
+	}
+}
+
+func TestSanitizeMetadataHeaders(t *testing.T) {
+	t.Run("standard OTel-style keys are sanitized to the same stem as CE extensions", func(t *testing.T) {
+		in := map[string]string{
+			"service.name":  "beholder",
+			"chain_id":      "1",
+			"node-operator": "acme",
+		}
+		got := chipingress.SanitizeMetadataHeaders(in)
+		assert.Equal(t, map[string]string{
+			"servicename":  "beholder",
+			"chainid":      "1",
+			"nodeoperator": "acme",
+		}, got)
+	})
+
+	t.Run("empty-after-sanitize keys are dropped", func(t *testing.T) {
+		got := chipingress.SanitizeMetadataHeaders(map[string]string{"---": "value"})
+		assert.Empty(t, got)
+	})
+
+	t.Run("reserved names are dropped", func(t *testing.T) {
+		got := chipingress.SanitizeMetadataHeaders(map[string]string{chipingress.IdempotencyKeyAttr: "should-not-appear", "subject": "should-not-appear"})
+		assert.Empty(t, got)
+	})
+
+	t.Run("non-printable values are sanitized", func(t *testing.T) {
+		got := chipingress.SanitizeMetadataHeaders(map[string]string{"chain_id": "1\n2"})
+		assert.Equal(t, "1?2", got["chainid"])
+	})
+
+	t.Run("duplicate sanitized keys resolve deterministically to sorted-first key", func(t *testing.T) {
+		got := chipingress.SanitizeMetadataHeaders(map[string]string{"service.name": "from-dotted", "service_name": "from-snake"})
+		// sorted order: "service.name" < "service_name" ('.' < '_' in ASCII), so the dotted key wins.
+		assert.Equal(t, "from-dotted", got["servicename"])
+	})
+}
+
+// pingServer is a minimal ChipIngressServer that always answers Ping successfully.
+type pingServer struct {
+	pb.UnimplementedChipIngressServer
+}
+
+func (pingServer) Ping(context.Context, *pb.EmptyRequest) (*pb.PingResponse, error) {
+	return &pb.PingResponse{}, nil
+}
+
+// TestSanitizeMetadataHeaders_AvoidsRPCFailure is a regression/guard test for the core reason
+// SanitizeMetadataHeaders exists: grpc-go hard-fails an entire RPC (codes.Internal) when an
+// outgoing metadata pair fails its charset validation. An unsanitized resource-attribute key or
+// value (dots, non-printable characters) reproduces that failure; running it through
+// SanitizeMetadataHeaders first must not.
+func TestSanitizeMetadataHeaders_AvoidsRPCFailure(t *testing.T) {
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	defer lis.Close()
+
+	srv := grpc.NewServer()
+	pb.RegisterChipIngressServer(srv, pingServer{})
+	go func() { _ = srv.Serve(lis) }()
+	defer srv.Stop()
+
+	dirty := map[string]string{"k8s.pod.name": "pod-\x01abc"}
+
+	t.Run("unsanitized headers fail the RPC", func(t *testing.T) {
+		client, err := chipingress.NewClient(lis.Addr().String(),
+			chipingress.WithInsecureConnection(),
+			chipingress.WithHeaderProvider(chipingress.NewStaticHeaderProvider(dirty)),
+		)
+		require.NoError(t, err)
+		defer client.Close() //nolint:errcheck
+
+		_, err = client.Ping(t.Context(), &chipingress.EmptyRequest{})
+		require.Error(t, err)
+		assert.Equal(t, codes.Internal, status.Code(err))
+	})
+
+	t.Run("sanitized headers succeed", func(t *testing.T) {
+		client, err := chipingress.NewClient(lis.Addr().String(),
+			chipingress.WithInsecureConnection(),
+			chipingress.WithHeaderProvider(chipingress.NewStaticHeaderProvider(chipingress.SanitizeMetadataHeaders(dirty))),
+		)
+		require.NoError(t, err)
+		defer client.Close() //nolint:errcheck
+
+		_, err = client.Ping(t.Context(), &chipingress.EmptyRequest{})
+		require.NoError(t, err)
+	})
+}