From e8b34bc478933118378e35a21a2bde000db09b1e Mon Sep 17 00:00:00 2001
From: Pavel <177363085+pkcll@users.noreply.github.com>
Date: Tue, 28 Jul 2026 16:45:04 +0100
Subject: [PATCH 1/3] chipingress: send resource attributes as prefixed gRPC
 metadata
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Resource attributes describe the producer, not any individual event. Carrying
them as per-event CloudEvent extensions repeated identical bytes for every
event in a batch — for a thousand events with ten attributes, roughly 300 KB
that counts against maxGRPCRequestSize and so reduces how many events fit per
batch. OTLP factors resource out of the payload for the same reason. Send them
once per request as gRPC metadata instead.

Every emitted key is ResourceHeaderPrefix ("resource_") followed by a key
normalized to grpc's charset, which grpc-go gives as [0-9a-z-_.]
(internal/metadata.ValidateKey). Structure therefore survives —
csa_public_key becomes resource_csa_public_key rather than collapsing to
csapublickey — which is what lets chip-ingress emit the forwarded header
verbatim. Values still go through SanitizeMetadataValue, because grpc-go fails
an entire RPC, auth header included, on one non-printable value. A trailing
"-bin" is rewritten so grpc does not try to base64-decode a plain-text
attribute.

The prefix is a wire contract with chip-ingress, which forwards metadata
carrying it onto every Kafka record a request produces. Requiring it inbound
and preserving it outbound keeps the namespace closed, and that is what
removes the need for a reserved-key set on either side. The header interceptor
appends to outgoing metadata rather than replacing, so an attribute named
X-Beholder-Node-Auth-Token would have sent a second value under the key
carrying the CSA node auth token and broken authentication; prefixed, it
becomes resource_x-beholder-node-auth-token and collides with nothing. The
same holds for authorization, te, content-type, the grpc- prefix and
pseudo-headers, so reservedMetadataHeaderNames, reservedMetadataKeys and
isReservedMetadataKey are all deleted rather than extended. A test asserts the
property directly, in place of the set it replaces.

Removes EventOpt, NewEventWithOpts and WithResourceAttributeExtensions, which
existed only for the extension path and have no callers in either repository;
NewEvent returns to being the single event constructor. With
SanitizeMetadataHeaders the sole consumer of the shared key helper, fold it in
and delete resource_attributes.go and the resourceAttrKey pair type — the
sanitized output map doubles as the dedupe set. SanitizeMetadataKey becomes
unexported, since nothing outside the package used it.

Adds ResourceHeaderPrefix. The same constant exists in chip-ingress as
constants.ResourceHeaderPrefix; duplicating a wire contract across
repositories matches how authHeaderKey is already spelled in both
pkg/beholder and pkg/chipingress, and the two must stay byte-identical or
forwarding silently stops.
---
 pkg/chipingress/client.go               |  59 +++-------
 pkg/chipingress/client_test.go          | 149 ++++++++++--------------
 pkg/chipingress/header_provider.go      |  96 ++++++++++++---
 pkg/chipingress/header_provider_test.go | 105 +++++++++++++----
 pkg/chipingress/resource_attributes.go  |  46 --------
 pkg/chipingress/types.go                |  42 +++----
 6 files changed, 251 insertions(+), 246 deletions(-)
 delete mode 100644 pkg/chipingress/resource_attributes.go

diff --git a/pkg/chipingress/client.go b/pkg/chipingress/client.go
index f51ce93bb7..d1224b75e8 100644
--- a/pkg/chipingress/client.go
+++ b/pkg/chipingress/client.go
@@ -5,7 +5,6 @@ import (
 	"crypto/tls"
 	"fmt"
 	"net"
-	"strings"
 	"time"
 
 	"github.com/google/uuid"
@@ -215,9 +214,13 @@ func WithHeaderProvider(provider HeaderProvider) Opt {
 	return func(c *clientConfig) { c.headerProvider = provider }
 }
 
-// WithResourceAttributeHeaders returns an Opt that attaches the provided resource attributes
-// as sanitized gRPC metadata headers. It combines SanitizeMetadataHeaders with
-// NewStaticHeaderProvider so the safe, validated path is used by default.
+// WithResourceAttributeHeaders returns an Opt that attaches the provided resource attributes as
+// gRPC metadata on every request, under ResourceHeaderPrefix. It combines SanitizeMetadataHeaders
+// with NewStaticHeaderProvider so the safe, validated path is used by default.
+//
+// Attributes are attached once per request rather than to individual events because they describe the
+// producer, not any one event. Chip-ingress fans them out onto every Kafka record the request
+// produces.
 func WithResourceAttributeHeaders(attrs map[string]string) Opt {
 	return WithHeaderProvider(NewStaticHeaderProvider(SanitizeMetadataHeaders(attrs)))
 }
@@ -254,11 +257,15 @@ func WithTracerProvider(provider trace.TracerProvider) Opt {
 	return func(c *clientConfig) { c.tracerProvider = provider }
 }
 
+// nopInfoHeaderKey is the metadata key WithNOPLookup sets, asking chip-ingress to look up NOP info
+// for the authenticated CSA key.
+const nopInfoHeaderKey = "x-include-nop-info"
+
 func WithNOPLookup() Opt {
 	return func(c *clientConfig) {
 		c.nopInfoHeaderProvider = headerProviderFunc(func(ctx context.Context) (map[string]string, error) {
 			return map[string]string{
-				"x-include-nop-info": "true",
+				nopInfoHeaderKey: "true",
 			}, nil
 		})
 	}
@@ -283,42 +290,12 @@ func newHeaderInterceptor(provider HeaderProvider) grpc.UnaryClientInterceptor {
 	}
 }
 
-// EventOpt configures a CloudEvent after its well-known attributes have been set by NewEvent.
-type EventOpt func(*ce.Event)
-
-// sanitizeExtensionName lower-cases name and strips every rune outside [a-z0-9], the character
-// set the CloudEvents spec requires for extension attribute names.
-func sanitizeExtensionName(name string) string {
-	var b strings.Builder
-	for _, r := range strings.ToLower(name) {
-		if (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') {
-			b.WriteRune(r)
-		}
-	}
-	return b.String()
-}
-
-// WithResourceAttributeExtensions returns an EventOpt that sets a CloudEvent extension for each
-// entry in attrs, sanitizing keys via sanitizeExtensionName so they satisfy the CloudEvents
-// extension-name character set. Entries that sanitize to an empty string, or that collide with a
-// reserved extension name (see reservedExtensionNames), are skipped. Keys are applied in sorted
-// order so that if two distinct keys sanitize to the same name, the result is deterministic.
-func WithResourceAttributeExtensions(attrs map[string]string) EventOpt {
-	return func(event *ce.Event) {
-		for _, pair := range sanitizeResourceAttributeKeys(attrs, nil) {
-			event.SetExtension(pair.name, attrs[pair.key])
-		}
-	}
-}
-
 // NewEvent creates a new CloudEvent with the specified domain, entity, payload, and optional attributes.
+//
+// Resource attributes are deliberately not stamped here. They describe the producer rather than any
+// individual event, so they travel once per request as gRPC metadata (see
+// WithResourceAttributeHeaders) instead of being repeated on every event in a batch.
 func NewEvent(domain, entity string, payload []byte, attributes map[string]any) (CloudEvent, error) {
-	return NewEventWithOpts(domain, entity, payload, attributes)
-}
-
-// NewEventWithOpts creates a new CloudEvent like NewEvent, additionally applying opts (e.g.
-// WithResourceAttributeExtensions) to the event before its data is set.
-func NewEventWithOpts(domain, entity string, payload []byte, attributes map[string]any, opts ...EventOpt) (CloudEvent, error) {
 	event := ce.NewEvent()
 	event.SetSource(domain)
 	event.SetType(entity)
@@ -352,10 +329,6 @@ func NewEventWithOpts(domain, entity string, payload []byte, attributes map[stri
 		event.SetExtension(IdempotencyKeyAttr, val)
 	}
 
-	for _, opt := range opts {
-		opt(&event)
-	}
-
 	err := event.SetData(ceformat.ContentTypeProtobuf, payload)
 	if err != nil {
 		return ce.Event{}, fmt.Errorf("could not set data on event: %w", err)
diff --git a/pkg/chipingress/client_test.go b/pkg/chipingress/client_test.go
index c224e1bb42..16cd5e9657 100644
--- a/pkg/chipingress/client_test.go
+++ b/pkg/chipingress/client_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"strings"
 	"testing"
 	"time"
 
@@ -168,89 +169,6 @@ func TestNewEvent_IdempotencyKey(t *testing.T) {
 	})
 }
 
-func Test_sanitizeExtensionName(t *testing.T) {
-	tests := []struct {
-		name string
-		in   string
-		want string
-	}{
-		{name: "snake_case", in: "chain_id", want: "chainid"},
-		{name: "dotted", in: "k8s.pod.name", want: "k8spodname"},
-		{name: "already valid", in: "chainid", want: "chainid"},
-		{name: "upper case is lowered", in: "ChainID", want: "chainid"},
-		{name: "empty", in: "", want: ""},
-		{name: "all invalid characters", in: "---...", want: ""},
-		{name: "mixed valid and invalid", in: "Service-Name.1", want: "servicename1"},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			assert.Equal(t, tt.want, sanitizeExtensionName(tt.in))
-		})
-	}
-}
-
-func TestNewEventWithOpts_WithResourceAttributeExtensions(t *testing.T) {
-	payload := []byte("body")
-
-	t.Run("sanitized keys/values land on the event", func(t *testing.T) {
-		attrs := map[string]string{"chain_id": "1", "k8s.pod.name": "pod-abc"}
-		event, err := NewEventWithOpts("domain", "entity", payload, nil, WithResourceAttributeExtensions(attrs))
-		require.NoError(t, err)
-		ext := event.Extensions()
-		assert.Equal(t, "1", ext["chainid"])
-		assert.Equal(t, "pod-abc", ext["k8spodname"])
-	})
-
-	t.Run("empty sanitized name is dropped", func(t *testing.T) {
-		attrs := map[string]string{"---": "value"}
-		event, err := NewEventWithOpts("domain", "entity", payload, nil, WithResourceAttributeExtensions(attrs))
-		require.NoError(t, err)
-		assert.Len(t, event.Extensions(), 1) // only the always-set recordedtime extension
-	})
-
-	t.Run("reserved name is skipped", func(t *testing.T) {
-		attrs := map[string]string{IdempotencyKeyAttr: "should-not-override", "subject": "should-not-override"}
-		event, err := NewEventWithOpts("domain", "entity", payload, map[string]any{IdempotencyKeyAttr: "real-key"}, WithResourceAttributeExtensions(attrs))
-		require.NoError(t, err)
-		ext := event.Extensions()
-		assert.Equal(t, "real-key", ext[IdempotencyKeyAttr])
-		assert.Empty(t, event.Subject())
-	})
-
-	t.Run("duplicate sanitized names resolve deterministically to sorted-first key", func(t *testing.T) {
-		attrs := map[string]string{"service.name": "from-dotted", "service_name": "from-snake"}
-		event, err := NewEventWithOpts("domain", "entity", payload, nil, WithResourceAttributeExtensions(attrs))
-		require.NoError(t, err)
-		// sorted order: "service.name" < "service_name" ('.' < '_' in ASCII), so the dotted key wins.
-		assert.Equal(t, "from-dotted", event.Extensions()["servicename"])
-	})
-
-	t.Run("omitting all opts is a no-op", func(t *testing.T) {
-		event, err := NewEventWithOpts("domain", "entity", payload, nil)
-		require.NoError(t, err)
-		assert.Len(t, event.Extensions(), 1) // only the always-set recordedtime extension
-	})
-}
-
-// TestNewEvent_UnchangedSignature is a backward-compatibility guard: NewEvent's exported
-// signature must stay exactly as it was before EventOpt/NewEventWithOpts were introduced, and
-// must remain equivalent to calling NewEventWithOpts with no opts.
-func TestNewEvent_UnchangedSignature(t *testing.T) {
-	payload := []byte("body")
-	attributes := map[string]any{"subject": "example-subject"}
-
-	viaNewEvent, err := NewEvent("domain", "entity", payload, attributes)
-	require.NoError(t, err)
-
-	viaNewEventWithOpts, err := NewEventWithOpts("domain", "entity", payload, attributes)
-	require.NoError(t, err)
-
-	assert.Equal(t, viaNewEventWithOpts.Subject(), viaNewEvent.Subject())
-	assert.Equal(t, viaNewEventWithOpts.Extensions()["recordedtime"].(ce.Timestamp).Truncate(time.Second),
-		viaNewEvent.Extensions()["recordedtime"].(ce.Timestamp).Truncate(time.Second))
-	assert.Equal(t, viaNewEventWithOpts.Data(), viaNewEvent.Data())
-}
-
 func TestEventToProto(t *testing.T) {
 	// Create a test protobuf message
 	testProto := pb.PingResponse{Message: "test message"}
@@ -684,14 +602,21 @@ func TestOptions(t *testing.T) {
 	t.Run("WithResourceAttributeHeaders", func(t *testing.T) {
 		config := defaultCfg
 		WithResourceAttributeHeaders(map[string]string{
-			"Chain-ID": "1",
-			"id":       "skipped", // reserved extension name
-			"chain_id": "2",       // duplicate sanitized key, first wins
+			"Chain-ID":       "1",   // lower-cased, separator preserved
+			"csa_public_key": "abc", // preserved verbatim
+			// Namespaced rather than dropped: prefixing puts them out of reach of the real keys.
+			"te":          "harmless",
+			authHeaderKey: "harmless",
 		})(&config)
 		assert.NotNil(t, config.headerProvider)
 		headers, err := config.headerProvider.Headers(t.Context())
 		require.NoError(t, err)
-		assert.Equal(t, map[string]string{"chainid": "1"}, headers)
+		assert.Equal(t, map[string]string{
+			ResourceHeaderPrefix + "chain-id":                     "1",
+			ResourceHeaderPrefix + "csa_public_key":               "abc",
+			ResourceHeaderPrefix + "te":                           "harmless",
+			ResourceHeaderPrefix + strings.ToLower(authHeaderKey): "harmless",
+		}, headers)
 	})
 
 	t.Run("WithBasicAuth", func(t *testing.T) {
@@ -808,6 +733,56 @@ func TestClient_ChainedHeaderProviders(t *testing.T) {
 	assert.Equal(t, []string{"true"}, capture.lastMD.Get("x-include-nop-info"))
 }
 
+// TestClient_AuthHeaderCoexistsWithResourceAttributes pins down the property the resource-attribute
+// work must never break: the CSA node auth token and the resource-attribute headers travel by two
+// different mechanisms — per-RPC credentials (WithTokenAuth) and a unary interceptor
+// (WithResourceAttributeHeaders) — and both must arrive intact, exactly once, on the same request.
+//
+// It also pins the property that lets the client carry attributes without a reserved-key deny-list:
+// an attribute named after the auth header is namespaced under ResourceHeaderPrefix, so it cannot
+// append a second value under the auth header's own key.
+func TestClient_AuthHeaderCoexistsWithResourceAttributes(t *testing.T) {
+	lis, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	defer lis.Close()
+
+	srv := gp.NewServer()
+	capture := &capturingServer{}
+	pb.RegisterChipIngressServer(srv, capture)
+	go func() { _ = srv.Serve(lis) }()
+	defer srv.Stop()
+
+	const authToken = "2:deadbeef:1:cafe"
+
+	client, err := NewClient(lis.Addr().String(),
+		WithInsecureConnection(),
+		WithTokenAuth(&mockHeaderProvider{headers: map[string]string{authHeaderKey: authToken}}),
+		WithResourceAttributeHeaders(map[string]string{
+			"csa_public_key": "abc123",
+			"service.name":   "chainlink",
+			// Namespaced away from the auth key rather than appended to it.
+			authHeaderKey: "forged",
+		}),
+		WithNOPLookup(),
+	)
+	require.NoError(t, err)
+	defer client.Close() //nolint:errcheck
+
+	_, err = client.Ping(t.Context(), &EmptyRequest{})
+	require.NoError(t, err)
+
+	require.NotNil(t, capture.lastMD)
+	// grpc lower-cases metadata keys on the wire.
+	assert.Equal(t, []string{authToken}, capture.lastMD.Get(authHeaderKey),
+		"the auth token must arrive exactly once, unmodified")
+	assert.Equal(t, []string{"abc123"}, capture.lastMD.Get(ResourceHeaderPrefix+"csa_public_key"))
+	assert.Equal(t, []string{"chainlink"}, capture.lastMD.Get(ResourceHeaderPrefix+"service.name"))
+	assert.Equal(t, []string{"true"}, capture.lastMD.Get("x-include-nop-info"))
+	// The forged attribute landed in the resource namespace, harmlessly.
+	assert.Equal(t, []string{"forged"},
+		capture.lastMD.Get(ResourceHeaderPrefix+strings.ToLower(authHeaderKey)))
+}
+
 func TestWithTLS(t *testing.T) {
 	serverName := "example.com"
 	config := defaultCfg
diff --git a/pkg/chipingress/header_provider.go b/pkg/chipingress/header_provider.go
index 47e2dfcd9b..1f9799cab5 100644
--- a/pkg/chipingress/header_provider.go
+++ b/pkg/chipingress/header_provider.go
@@ -8,6 +8,8 @@ import (
 	"errors"
 	"fmt"
 	"maps"
+	"sort"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -111,6 +113,13 @@ func newStaticHeaderProvider(headers map[string]string, requireTLS bool) HeaderP
 // NewStaticHeaderProvider returns a HeaderProvider that always returns the given headers,
 // for use with WithHeaderProvider to attach fixed, non-auth gRPC metadata (e.g. resource
 // attributes) to every request.
+//
+// This is for the non-auth interceptor path only. It reports RequireTransportSecurity() == false,
+// which WithHeaderProvider never consults — the HeaderProvider interface declares only Headers,
+// and grpc asks only credentials.PerRPCCredentials about transport security. Do not pass the
+// result to WithTokenAuth: that path takes its TLS requirement from the client config
+// (!c.insecureConnection), not from the provider, so the false here would be silently ignored
+// rather than honoured. Use NewHeaderProvider for auth headers.
 func NewStaticHeaderProvider(headers map[string]string) HeaderProvider {
 	return newStaticHeaderProvider(headers, false)
 }
@@ -132,27 +141,78 @@ func SanitizeMetadataValue(val string) string {
 	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) or a gRPC-reserved header name
-// (see reservedMetadataKeys), are skipped. Keys are applied in sorted order so duplicate
-// sanitized keys resolve deterministically (first in sorted order wins), matching
-// WithResourceAttributeExtensions' collision handling.
+// sanitizeMetadataKey normalizes a resource-attribute key into a valid outgoing gRPC metadata
+// key, without the ResourceHeaderPrefix that SanitizeMetadataHeaders adds. grpc-go accepts keys
+// matching [0-9a-z-_.] (see internal/metadata.ValidateKey), so the key's structure is preserved:
+// "csa_public_key" stays "csa_public_key" and "service.name" stays "service.name", which is what
+// lets chip-ingress emit the forwarded header verbatim.
 //
-// Note: unlike the CloudEvents Kafka binding, gRPC metadata keys are NOT prefixed with "ce_" —
-// that prefix is a CloudEvents-binding concept, not a metadata one, and reusing it here would
-// collide with the CE binding's own "ce_<name>" Kafka header if the server ever forwards gRPC
-// metadata verbatim onto Kafka.
+// The rules are: lower-case (grpc requires it), keep '.', '-' and '_', replace every other
+// character with '_', and return "" for a key with no [a-z0-9] character left, since a key of
+// pure separators carries no information. A trailing "-bin" is rewritten to "_bin" because grpc
+// treats a "-bin" suffix as declaring a base64-encoded binary value and would try to decode it.
+func sanitizeMetadataKey(key string) string {
+	var b strings.Builder
+	b.Grow(len(key))
+
+	hasAlnum := false
+	for _, r := range strings.ToLower(key) {
+		switch {
+		case (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9'):
+			hasAlnum = true
+			b.WriteRune(r)
+		case r == '.' || r == '-' || r == '_':
+			b.WriteRune(r)
+		default:
+			b.WriteByte('_')
+		}
+	}
+	if !hasAlnum {
+		return ""
+	}
+
+	out := b.String()
+	if suffix := "-bin"; strings.HasSuffix(out, suffix) {
+		out = strings.TrimSuffix(out, suffix) + "_bin"
+	}
+	return out
+}
+
+// SanitizeMetadataHeaders sanitizes a map of resource attributes for use as outgoing gRPC metadata
+// (e.g. via NewStaticHeaderProvider). Every emitted key is ResourceHeaderPrefix followed by a key
+// normalized to grpc's charset, so service.name becomes resource_service.name and csa_public_key
+// becomes resource_csa_public_key. Chip-ingress forwards keys carrying that prefix onto every Kafka
+// record a request produces, emitting them unchanged. Values go through SanitizeMetadataValue,
+// because grpc-go fails the whole RPC — auth header included — on a single non-printable value.
+//
+// The prefix is what makes this safe without a deny-list. The header interceptor appends to outgoing
+// metadata rather than replacing it, so an attribute landing on an existing header name would send
+// two values under one key — an attribute named X-Beholder-Node-Auth-Token would have broken
+// authentication that way. Because every emitted key is prefixed, no attribute can reach a reserved
+// gRPC key: that one becomes resource_x-beholder-node-auth-token, which collides with nothing, and
+// the same holds for authorization, te, content-type, the grpc- prefix and pseudo-headers.
+//
+// Entries whose key normalizes to "" are skipped, since a bare prefix carries no information. If two
+// keys normalize to the same name the first in lexicographic order of the original keys wins, so the
+// result is deterministic.
 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) // deterministic: first in sorted order wins a normalized-name collision
+
 	out := make(map[string]string, len(in))
-	for _, pair := range sanitizeResourceAttributeKeys(in, reservedMetadataKeys) {
-		out[pair.name] = SanitizeMetadataValue(in[pair.key])
+	for _, k := range keys {
+		name := sanitizeMetadataKey(k)
+		if name == "" {
+			continue
+		}
+		name = ResourceHeaderPrefix + name
+		if _, dup := out[name]; dup {
+			continue
+		}
+		out[name] = SanitizeMetadataValue(in[k])
 	}
 	return out
 }
diff --git a/pkg/chipingress/header_provider_test.go b/pkg/chipingress/header_provider_test.go
index 8069420fd2..18b39c24ac 100644
--- a/pkg/chipingress/header_provider_test.go
+++ b/pkg/chipingress/header_provider_test.go
@@ -5,6 +5,7 @@ import (
 	"crypto/ed25519"
 	"encoding/hex"
 	"net"
+	"strings"
 	"testing"
 	"time"
 
@@ -301,45 +302,101 @@ func TestSanitizeMetadataValue(t *testing.T) {
 	}
 }
 
+const rp = chipingress.ResourceHeaderPrefix
+
 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)
+	t.Run("keys are prefixed and keep their structure", func(t *testing.T) {
+		got := chipingress.SanitizeMetadataHeaders(map[string]string{
+			"service.name":   "beholder",
+			"csa_public_key": "abc123",
+			"node-operator":  "acme",
+			"DonID":          "don-1",
+		})
 		assert.Equal(t, map[string]string{
-			"servicename":  "beholder",
-			"chainid":      "1",
-			"nodeoperator": "acme",
+			rp + "service.name":   "beholder",
+			rp + "csa_public_key": "abc123",
+			rp + "node-operator":  "acme",
+			rp + "donid":          "don-1",
 		}, 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("structure-preserving normalization", func(t *testing.T) {
+		tests := []struct {
+			name string
+			in   string
+			want string
+		}{
+			// grpc accepts [0-9a-z-_.], so structure survives and chip-ingress can emit the
+			// forwarded header verbatim.
+			{"snake case preserved", "csa_public_key", rp + "csa_public_key"},
+			{"dotted preserved", "service.name", rp + "service.name"},
+			{"upper-cased is lowered", "DonID", rp + "donid"},
+			{"mixed separators preserved", "k8s.pod-name_1", rp + "k8s.pod-name_1"},
+			{"illegal characters become underscores", "chain id/2:x", rp + "chain_id_2_x"},
+			{"non-ascii becomes underscores", "héllo", rp + "h_llo"},
+			// A "-bin" suffix tells grpc the value is base64-encoded binary; rewrite it so grpc
+			// does not try to decode a plain-text resource attribute.
+			{"bin suffix is rewritten", "payload-bin", rp + "payload_bin"},
+			{"bin substring is untouched", "payload-binary", rp + "payload-binary"},
+		}
+		for _, tt := range tests {
+			t.Run(tt.name, func(t *testing.T) {
+				assert.Equal(t, map[string]string{tt.want: "v"},
+					chipingress.SanitizeMetadataHeaders(map[string]string{tt.in: "v"}))
+			})
+		}
 	})
 
-	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("keys with nothing left after normalization are dropped", func(t *testing.T) {
+		// A bare prefix carries no information.
+		for _, key := range []string{"", "---", "__"} {
+			assert.Empty(t, chipingress.SanitizeMetadataHeaders(map[string]string{key: "value"}),
+				"key %q must be dropped", key)
+		}
 	})
 
-	t.Run("gRPC-reserved header 'te' is dropped", func(t *testing.T) {
-		got := chipingress.SanitizeMetadataHeaders(map[string]string{"te": "trailers"})
-		assert.Empty(t, got)
+	// This is the property that replaces the reserved-key set the prefix made redundant. The header
+	// interceptor appends to outgoing metadata rather than replacing, so an attribute landing on an
+	// existing header name would send two values under one key — for the CSA auth token that breaks
+	// authentication. Prefixing puts every attribute out of reach of every reserved gRPC key.
+	t.Run("no attribute can collide with a reserved gRPC metadata key", func(t *testing.T) {
+		for _, key := range []string{
+			"X-Beholder-Node-Auth-Token", // CSA auth token, via WithTokenAuth
+			"x-include-nop-info",         // WithNOPLookup
+			"authorization",              // WithBasicAuth
+			"te", "content-type", "cookie", "host", "user-agent",
+			"grpc-timeout", "grpc-encoding",
+		} {
+			got := chipingress.SanitizeMetadataHeaders(map[string]string{key: "forged"})
+			require.Len(t, got, 1, "key %q should still be sent, just namespaced", key)
+			for name := range got {
+				assert.True(t, strings.HasPrefix(name, rp), "key %q must be prefixed, got %q", key, name)
+				assert.NotEqual(t, strings.ToLower(key), name, "key %q must not reach the reserved name", key)
+			}
+		}
+	})
+
+	t.Run("CloudEvents context attribute names are kept, they mean nothing as gRPC metadata", func(t *testing.T) {
+		got := chipingress.SanitizeMetadataHeaders(map[string]string{"subject": "keep-me", "source": "keep-me-too"})
+		assert.Equal(t, map[string]string{rp + "subject": "keep-me", rp + "source": "keep-me-too"}, 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"])
+		assert.Equal(t, "1?2", got[rp+"chain_id"])
+	})
+
+	t.Run("duplicate normalized keys resolve deterministically to sorted-first key", func(t *testing.T) {
+		// Both normalize to chain_id; sorted order is "chain id" < "chain_id" (' ' < '_'), so the
+		// space-separated key wins.
+		got := chipingress.SanitizeMetadataHeaders(map[string]string{"chain id": "from-space", "chain_id": "from-snake"})
+		assert.Equal(t, map[string]string{rp + "chain_id": "from-space"}, got)
 	})
 
-	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"])
+	t.Run("keys that differ only in case collapse deterministically", func(t *testing.T) {
+		got := chipingress.SanitizeMetadataHeaders(map[string]string{"DonID": "upper", "donid": "lower"})
+		// sorted order: "DonID" < "donid" (upper-case sorts first in ASCII).
+		assert.Equal(t, map[string]string{rp + "donid": "upper"}, got)
 	})
 }
 
diff --git a/pkg/chipingress/resource_attributes.go b/pkg/chipingress/resource_attributes.go
deleted file mode 100644
index 2d5f974e8f..0000000000
--- a/pkg/chipingress/resource_attributes.go
+++ /dev/null
@@ -1,46 +0,0 @@
-package chipingress
-
-import "sort"
-
-// resourceAttrKey pairs a sanitized extension/metadata key name with the original
-// resource-attribute key it was derived from.
-type resourceAttrKey struct {
-	name string
-	key  string
-}
-
-// sanitizeResourceAttributeKeys returns the deduplicated, sorted list of resource-attribute
-// keys that survive sanitization and reservation checks. The returned pairs contain the
-// sanitized name and the original map key, so callers can apply their own value handling.
-//
-// Ordering is deterministic: original keys are sorted lexicographically, and if two keys
-// sanitize to the same name the first one in sorted order wins. extraReserved, if non-nil,
-// is consulted in addition to reservedExtensionNames.
-func sanitizeResourceAttributeKeys(attrs map[string]string, extraReserved map[string]struct{}) []resourceAttrKey {
-	keys := make([]string, 0, len(attrs))
-	for k := range attrs {
-		keys = append(keys, k)
-	}
-	sort.Strings(keys)
-
-	seen := make(map[string]struct{}, len(attrs))
-	result := make([]resourceAttrKey, 0, len(attrs))
-	for _, k := range keys {
-		name := sanitizeExtensionName(k)
-		if name == "" {
-			continue
-		}
-		if _, reserved := reservedExtensionNames[name]; reserved {
-			continue
-		}
-		if _, reserved := extraReserved[name]; reserved {
-			continue
-		}
-		if _, already := seen[name]; already {
-			continue
-		}
-		seen[name] = struct{}{}
-		result = append(result, resourceAttrKey{name: name, key: k})
-	}
-	return result
-}
diff --git a/pkg/chipingress/types.go b/pkg/chipingress/types.go
index 25eede3edb..34551bc93d 100644
--- a/pkg/chipingress/types.go
+++ b/pkg/chipingress/types.go
@@ -13,34 +13,20 @@ import (
 // Kafka headers named "ce_<name>" (e.g., ce_idempotencykey), enabling downstream deduplication.
 const IdempotencyKeyAttr = "idempotencykey"
 
-// reservedExtensionNames holds every CloudEvent extension name that NewEvent sets internally,
-// plus the CloudEvents core context attribute names (id, source, type, specversion, time,
-// subject, dataschema, datacontenttype) and the spec-forbidden "data" name. WithResourceAttributeExtensions
-// consults this set so that a resource attribute can never silently overwrite event-lifecycle
-// metadata or collide with a CloudEvents core attribute.
-var reservedExtensionNames = map[string]struct{}{
-	IdempotencyKeyAttr: {},
-	"recordedtime":     {},
-	"id":               {},
-	"source":           {},
-	"type":             {},
-	"specversion":      {},
-	"time":             {},
-	"subject":          {},
-	"dataschema":       {},
-	"datacontenttype":  {},
-	"data":             {},
-}
-
-// reservedMetadataKeys holds gRPC-reserved header names that could otherwise be reached by
-// sanitizeExtensionName's [a-z0-9] sanitization. Verified against grpc-go v1.79.1's
-// isReservedHeader: every other reserved header (pseudo-headers, "content-type", "grpc-*")
-// contains a ':' or '-' that sanitization strips, so "te" is the only one actually reachable.
-// SanitizeMetadataHeaders consults this set so that edge case is handled deterministically
-// rather than relying on grpc's own (silent) handling of a reserved header.
-var reservedMetadataKeys = map[string]struct{}{
-	"te": {},
-}
+// ResourceHeaderPrefix namespaces producer resource attributes sent as outgoing gRPC metadata.
+// SanitizeMetadataHeaders applies it to every key it emits.
+//
+// It is the wire contract with chip-ingress, which forwards metadata carrying this prefix onto every
+// Kafka record a request produces and emits the key unchanged. Requiring the prefix inbound and
+// preserving it outbound keeps the namespace closed, which is what makes the forwarding safe: a
+// client can only cause a header beginning with this prefix to be written, so a resource attribute
+// cannot shadow a "ce_" header, an identity header the server derives from the verified auth token,
+// or — on this side of the wire — a reserved gRPC metadata key such as the CSA auth token's.
+//
+// The same constant exists in chip-ingress as constants.ResourceHeaderPrefix. Duplicating it across
+// repositories is deliberate, matching how authHeaderKey is already spelled in both pkg/beholder and
+// pkg/chipingress; the two must stay byte-identical or forwarding silently stops.
+const ResourceHeaderPrefix = "resource_"
 
 type (
 	// Cloudevents types

From c131714d802fca10e0c28622836b02b58f089965 Mon Sep 17 00:00:00 2001
From: Pavel <177363085+pkcll@users.noreply.github.com>
Date: Fri, 28 Aug 2026 21:15:47 -0400
Subject: [PATCH 2/3] chipingress: validate resource attributes instead of
 rewriting, add caps

SanitizeMetadataHeaders now omits an invalid key, non-printable value,
duplicate, or over-limit attribute rather than rewriting it, so two
distinct configured keys can never collapse into one gRPC metadata
key and a non-printable value can never be silently byte-mangled.
Adds the 32-attribute / 128B-key / 512B-value / 4096B-total caps, and
warns plus meters every dropped attribute via WithResourceAttributeHeaders.
---
 pkg/chipingress/client.go               |  45 +++++-
 pkg/chipingress/header_provider.go      | 157 +++++++++++++--------
 pkg/chipingress/header_provider_test.go | 177 +++++++++++++++---------
 3 files changed, 252 insertions(+), 127 deletions(-)

diff --git a/pkg/chipingress/client.go b/pkg/chipingress/client.go
index d1224b75e8..33aadeba7c 100644
--- a/pkg/chipingress/client.go
+++ b/pkg/chipingress/client.go
@@ -5,12 +5,17 @@ import (
 	"crypto/tls"
 	"fmt"
 	"net"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/google/uuid"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
+	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
@@ -214,15 +219,51 @@ func WithHeaderProvider(provider HeaderProvider) Opt {
 	return func(c *clientConfig) { c.headerProvider = provider }
 }
 
+// resourceAttributeLogger receives one warning per dropped resource attribute. It defaults to a
+// no-op so library consumers who don't configure a logger see no output; tests substitute their own
+// via setResourceAttributeLogger.
+var resourceAttributeLogger atomic.Pointer[zap.Logger]
+
+func init() {
+	resourceAttributeLogger.Store(zap.NewNop())
+}
+
+// SetResourceAttributeLogger overrides the logger WithResourceAttributeHeaders warns through when it
+// drops a resource attribute. Intended for host applications that want dropped attributes surfaced
+// in their own logs; the default is a no-op logger.
+func SetResourceAttributeLogger(logger *zap.Logger) {
+	resourceAttributeLogger.Store(logger)
+}
+
+// resourceAttributeDropsCounter counts resource attributes SanitizeMetadataHeaders omitted, by
+// reason, against the global otel MeterProvider. It is created lazily against the global provider
+// (rather than a per-client one from WithMeterProvider) because WithResourceAttributeHeaders runs at
+// Opt-construction time, before any client-level configuration is wired.
+var resourceAttributeDropsCounter = sync.OnceValue(func() metric.Int64Counter {
+	c, _ := otel.Meter("github.com/smartcontractkit/chainlink-common/pkg/chipingress").
+		Int64Counter("chipingress.resource_attribute.dropped",
+			metric.WithDescription("Resource attributes omitted by SanitizeMetadataHeaders, by reason."))
+	return c
+})
+
 // WithResourceAttributeHeaders returns an Opt that attaches the provided resource attributes as
 // gRPC metadata on every request, under ResourceHeaderPrefix. It combines SanitizeMetadataHeaders
-// with NewStaticHeaderProvider so the safe, validated path is used by default.
+// with NewStaticHeaderProvider so the safe, validated path is used by default, and warns + meters
+// every attribute SanitizeMetadataHeaders had to omit.
 //
 // Attributes are attached once per request rather than to individual events because they describe the
 // producer, not any one event. Chip-ingress fans them out onto every Kafka record the request
 // produces.
 func WithResourceAttributeHeaders(attrs map[string]string) Opt {
-	return WithHeaderProvider(NewStaticHeaderProvider(SanitizeMetadataHeaders(attrs)))
+	sanitized, dropped := SanitizeMetadataHeaders(attrs)
+	for _, d := range dropped {
+		resourceAttributeLogger.Load().Warn("dropping invalid resource attribute",
+			zap.String("key", d.Key), zap.String("reason", d.Reason))
+		if counter := resourceAttributeDropsCounter(); counter != nil {
+			counter.Add(context.Background(), 1, metric.WithAttributes(attribute.String("reason", d.Reason)))
+		}
+	}
+	return WithHeaderProvider(NewStaticHeaderProvider(sanitized))
 }
 
 // WithInsecureConnection configures the client to use an insecure connection (no TLS).
diff --git a/pkg/chipingress/header_provider.go b/pkg/chipingress/header_provider.go
index 1f9799cab5..3a146528b2 100644
--- a/pkg/chipingress/header_provider.go
+++ b/pkg/chipingress/header_provider.go
@@ -124,66 +124,75 @@ 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] = '?'
+// Limits on resource attributes accepted by SanitizeMetadataHeaders. They reserve headroom in the
+// gRPC HEADERS frame for authentication and normal gRPC metadata, and bound how much of every Kafka
+// record's header space a producer's resource attributes can consume.
+const (
+	maxResourceAttributes          = 32
+	maxResourceAttributeKeyBytes   = 128
+	maxResourceAttributeValueBytes = 512
+	maxResourceAttributeTotalBytes = 4096 // sum of accepted key + value bytes, prefix excluded
+)
+
+// isPrintableASCII reports whether every byte of val is in the printable ASCII range [0x20, 0x7E].
+// grpc-go hard-fails the entire RPC — auth header included — when an outgoing metadata value fails
+// this check, so a value that does not pass is omitted rather than rewritten: a byte-mangled value
+// is a worse outcome than a dropped attribute for an operator-facing routing/observability field.
+func isPrintableASCII(val string) bool {
+	for i := 0; i < len(val); i++ {
+		if c := val[i]; c < 0x20 || c > 0x7E {
+			return false
 		}
 	}
-	return string(out)
+	return true
 }
 
-// sanitizeMetadataKey normalizes a resource-attribute key into a valid outgoing gRPC metadata
-// key, without the ResourceHeaderPrefix that SanitizeMetadataHeaders adds. grpc-go accepts keys
-// matching [0-9a-z-_.] (see internal/metadata.ValidateKey), so the key's structure is preserved:
-// "csa_public_key" stays "csa_public_key" and "service.name" stays "service.name", which is what
-// lets chip-ingress emit the forwarded header verbatim.
+// DroppedAttribute records a resource attribute SanitizeMetadataHeaders omitted, and why.
+type DroppedAttribute struct {
+	Key    string
+	Reason string
+}
+
+// Reasons a resource attribute can be omitted by SanitizeMetadataHeaders. Exposed as strings (not
+// an enum type) so callers can attach them to a log field or a metric attribute directly.
+const (
+	reasonInvalidKey    = "invalid_key"
+	reasonInvalidValue  = "invalid_value"
+	reasonDuplicateKey  = "duplicate_key"
+	reasonLimitExceeded = "limit_exceeded"
+)
+
+// sanitizeMetadataKey validates a resource-attribute key as a valid outgoing gRPC metadata key,
+// without the ResourceHeaderPrefix that SanitizeMetadataHeaders adds, and reports whether it is
+// valid. It never rewrites: a key that fails validation is omitted by the caller rather than
+// mutated, so two distinct configured keys can never collapse into one gRPC metadata key.
 //
-// The rules are: lower-case (grpc requires it), keep '.', '-' and '_', replace every other
-// character with '_', and return "" for a key with no [a-z0-9] character left, since a key of
-// pure separators carries no information. A trailing "-bin" is rewritten to "_bin" because grpc
-// treats a "-bin" suffix as declaring a base64-encoded binary value and would try to decode it.
-func sanitizeMetadataKey(key string) string {
-	var b strings.Builder
-	b.Grow(len(key))
-
-	hasAlnum := false
-	for _, r := range strings.ToLower(key) {
-		switch {
-		case (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9'):
-			hasAlnum = true
-			b.WriteRune(r)
-		case r == '.' || r == '-' || r == '_':
-			b.WriteRune(r)
-		default:
-			b.WriteByte('_')
-		}
+// Valid keys, once lower-cased, match [0-9a-z-_.]+ (grpc's own key charset — see
+// internal/metadata.ValidateKey) and do not end in "-bin", which grpc treats as declaring a
+// base64-encoded binary value. A valid key's structure survives untouched: "csa_public_key" stays
+// "csa_public_key" and "service.name" stays "service.name", which is what lets chip-ingress emit the
+// forwarded header verbatim.
+func sanitizeMetadataKey(key string) (string, bool) {
+	if key == "" {
+		return "", false
 	}
-	if !hasAlnum {
-		return ""
+	lower := strings.ToLower(key)
+	for _, r := range lower {
+		if !((r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') || r == '.' || r == '-' || r == '_') {
+			return "", false
+		}
 	}
-
-	out := b.String()
-	if suffix := "-bin"; strings.HasSuffix(out, suffix) {
-		out = strings.TrimSuffix(out, suffix) + "_bin"
+	if strings.HasSuffix(lower, "-bin") {
+		return "", false
 	}
-	return out
+	return lower, true
 }
 
-// SanitizeMetadataHeaders sanitizes a map of resource attributes for use as outgoing gRPC metadata
-// (e.g. via NewStaticHeaderProvider). Every emitted key is ResourceHeaderPrefix followed by a key
-// normalized to grpc's charset, so service.name becomes resource_service.name and csa_public_key
-// becomes resource_csa_public_key. Chip-ingress forwards keys carrying that prefix onto every Kafka
-// record a request produces, emitting them unchanged. Values go through SanitizeMetadataValue,
-// because grpc-go fails the whole RPC — auth header included — on a single non-printable value.
+// SanitizeMetadataHeaders validates a map of resource attributes for use as outgoing gRPC metadata
+// (e.g. via NewStaticHeaderProvider). Every emitted key is ResourceHeaderPrefix followed by the
+// validated key, unchanged, so service.name becomes resource_service.name and csa_public_key becomes
+// resource_csa_public_key. Chip-ingress forwards keys carrying that prefix onto every Kafka record a
+// request produces, emitting the key unchanged.
 //
 // The prefix is what makes this safe without a deny-list. The header interceptor appends to outgoing
 // metadata rather than replacing it, so an attribute landing on an existing header name would send
@@ -192,29 +201,59 @@ func sanitizeMetadataKey(key string) string {
 // gRPC key: that one becomes resource_x-beholder-node-auth-token, which collides with nothing, and
 // the same holds for authorization, te, content-type, the grpc- prefix and pseudo-headers.
 //
-// Entries whose key normalizes to "" are skipped, since a bare prefix carries no information. If two
-// keys normalize to the same name the first in lexicographic order of the original keys wins, so the
-// result is deterministic.
-func SanitizeMetadataHeaders(in map[string]string) map[string]string {
+// An attribute is omitted, rather than rewritten, when: its key is empty, exceeds
+// maxResourceAttributeKeyBytes, fails sanitizeMetadataKey's charset/[-bin] validation, or duplicates
+// an already-accepted key (first in sorted order of the original keys wins); its value exceeds
+// maxResourceAttributeValueBytes or is not printable ASCII (isPrintableASCII); or accepting it would
+// push the accepted count past maxResourceAttributes or the accepted key+value byte total past
+// maxResourceAttributeTotalBytes. Keys are processed in sorted order so every omission is
+// deterministic. dropped records each omission and why, for the caller to warn and meter.
+func SanitizeMetadataHeaders(in map[string]string) (map[string]string, []DroppedAttribute) {
 	keys := make([]string, 0, len(in))
 	for k := range in {
 		keys = append(keys, k)
 	}
-	sort.Strings(keys) // deterministic: first in sorted order wins a normalized-name collision
+	sort.Strings(keys) // deterministic: first in sorted order wins, and excess entries drop from the tail
 
 	out := make(map[string]string, len(in))
+	var dropped []DroppedAttribute
+	totalBytes := 0
 	for _, k := range keys {
-		name := sanitizeMetadataKey(k)
-		if name == "" {
+		if len(out) >= maxResourceAttributes {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonLimitExceeded})
+			continue
+		}
+		if len(k) > maxResourceAttributeKeyBytes {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonInvalidKey})
+			continue
+		}
+		name, ok := sanitizeMetadataKey(k)
+		if !ok {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonInvalidKey})
 			continue
 		}
 		name = ResourceHeaderPrefix + name
 		if _, dup := out[name]; dup {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonDuplicateKey})
+			continue
+		}
+		val := in[k]
+		if len(val) > maxResourceAttributeValueBytes {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonInvalidValue})
+			continue
+		}
+		if !isPrintableASCII(val) {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonInvalidValue})
+			continue
+		}
+		if totalBytes+len(name)-len(ResourceHeaderPrefix)+len(val) > maxResourceAttributeTotalBytes {
+			dropped = append(dropped, DroppedAttribute{Key: k, Reason: reasonLimitExceeded})
 			continue
 		}
-		out[name] = SanitizeMetadataValue(in[k])
+		totalBytes += len(name) - len(ResourceHeaderPrefix) + len(val)
+		out[name] = val
 	}
-	return out
+	return out, dropped
 }
 
 // newRotatingHeaderProvider returns a HeaderProvider that refreshes its
diff --git a/pkg/chipingress/header_provider_test.go b/pkg/chipingress/header_provider_test.go
index 18b39c24ac..ea705d7b75 100644
--- a/pkg/chipingress/header_provider_test.go
+++ b/pkg/chipingress/header_provider_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/ed25519"
 	"encoding/hex"
+	"fmt"
 	"net"
 	"strings"
 	"testing"
@@ -284,34 +285,17 @@ func TestNewStaticHeaderProvider(t *testing.T) {
 	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))
-		})
-	}
-}
-
 const rp = chipingress.ResourceHeaderPrefix
 
 func TestSanitizeMetadataHeaders(t *testing.T) {
-	t.Run("keys are prefixed and keep their structure", func(t *testing.T) {
-		got := chipingress.SanitizeMetadataHeaders(map[string]string{
+	t.Run("valid keys are prefixed and kept verbatim", func(t *testing.T) {
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{
 			"service.name":   "beholder",
 			"csa_public_key": "abc123",
 			"node-operator":  "acme",
-			"DonID":          "don-1",
+			"donid":          "don-1",
 		})
+		assert.Empty(t, dropped)
 		assert.Equal(t, map[string]string{
 			rp + "service.name":   "beholder",
 			rp + "csa_public_key": "abc123",
@@ -320,83 +304,142 @@ func TestSanitizeMetadataHeaders(t *testing.T) {
 		}, got)
 	})
 
-	t.Run("structure-preserving normalization", func(t *testing.T) {
+	t.Run("validate, don't rewrite", func(t *testing.T) {
 		tests := []struct {
-			name string
-			in   string
-			want string
+			name    string
+			in      string
+			want    string // "" if the key must be omitted
+			omitted bool
 		}{
-			// grpc accepts [0-9a-z-_.], so structure survives and chip-ingress can emit the
-			// forwarded header verbatim.
-			{"snake case preserved", "csa_public_key", rp + "csa_public_key"},
-			{"dotted preserved", "service.name", rp + "service.name"},
-			{"upper-cased is lowered", "DonID", rp + "donid"},
-			{"mixed separators preserved", "k8s.pod-name_1", rp + "k8s.pod-name_1"},
-			{"illegal characters become underscores", "chain id/2:x", rp + "chain_id_2_x"},
-			{"non-ascii becomes underscores", "héllo", rp + "h_llo"},
-			// A "-bin" suffix tells grpc the value is base64-encoded binary; rewrite it so grpc
-			// does not try to decode a plain-text resource attribute.
-			{"bin suffix is rewritten", "payload-bin", rp + "payload_bin"},
-			{"bin substring is untouched", "payload-binary", rp + "payload-binary"},
+			// grpc accepts [0-9a-z-_.], so a valid key's structure survives untouched and
+			// chip-ingress can emit the forwarded header verbatim.
+			{name: "snake case preserved", in: "csa_public_key", want: rp + "csa_public_key"},
+			{name: "dotted preserved", in: "service.name", want: rp + "service.name"},
+			{name: "upper-cased is lowered", in: "DonID", want: rp + "donid"},
+			{name: "mixed separators preserved", in: "k8s.pod-name_1", want: rp + "k8s.pod-name_1"},
+			// Invalid keys are OMITTED, never rewritten: silently collapsing two distinct
+			// configured keys into one gRPC metadata key is worse than dropping one.
+			{name: "illegal characters omit the attribute", in: "chain id/2:x", omitted: true},
+			{name: "non-ascii omits the attribute", in: "héllo", omitted: true},
+			{name: "empty key omits the attribute", in: "", omitted: true},
+			// A "-bin" suffix tells grpc the value is base64-encoded binary; omit rather than
+			// rewrite so a plain-text resource attribute can't silently start being decoded.
+			{name: "bin suffix omits the attribute", in: "payload-bin", omitted: true},
+			{name: "bin substring is untouched", in: "payload-binary", want: rp + "payload-binary"},
 		}
 		for _, tt := range tests {
 			t.Run(tt.name, func(t *testing.T) {
-				assert.Equal(t, map[string]string{tt.want: "v"},
-					chipingress.SanitizeMetadataHeaders(map[string]string{tt.in: "v"}))
+				got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{tt.in: "v"})
+				if tt.omitted {
+					assert.Empty(t, got)
+					require.Len(t, dropped, 1)
+					assert.Equal(t, tt.in, dropped[0].Key)
+					return
+				}
+				assert.Empty(t, dropped)
+				assert.Equal(t, map[string]string{tt.want: "v"}, got)
 			})
 		}
 	})
 
-	t.Run("keys with nothing left after normalization are dropped", func(t *testing.T) {
-		// A bare prefix carries no information.
-		for _, key := range []string{"", "---", "__"} {
-			assert.Empty(t, chipingress.SanitizeMetadataHeaders(map[string]string{key: "value"}),
-				"key %q must be dropped", key)
-		}
-	})
-
 	// This is the property that replaces the reserved-key set the prefix made redundant. The header
 	// interceptor appends to outgoing metadata rather than replacing, so an attribute landing on an
 	// existing header name would send two values under one key — for the CSA auth token that breaks
-	// authentication. Prefixing puts every attribute out of reach of every reserved gRPC key.
+	// authentication. Prefixing puts every valid attribute out of reach of every reserved gRPC key.
 	t.Run("no attribute can collide with a reserved gRPC metadata key", func(t *testing.T) {
 		for _, key := range []string{
-			"X-Beholder-Node-Auth-Token", // CSA auth token, via WithTokenAuth
-			"x-include-nop-info",         // WithNOPLookup
-			"authorization",              // WithBasicAuth
+			"x-include-nop-info", // WithNOPLookup
+			"authorization",      // WithBasicAuth
 			"te", "content-type", "cookie", "host", "user-agent",
 			"grpc-timeout", "grpc-encoding",
 		} {
-			got := chipingress.SanitizeMetadataHeaders(map[string]string{key: "forged"})
+			got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{key: "forged"})
+			assert.Empty(t, dropped, "key %q should be valid and namespaced, not dropped", key)
 			require.Len(t, got, 1, "key %q should still be sent, just namespaced", key)
 			for name := range got {
 				assert.True(t, strings.HasPrefix(name, rp), "key %q must be prefixed, got %q", key, name)
 				assert.NotEqual(t, strings.ToLower(key), name, "key %q must not reach the reserved name", key)
 			}
 		}
+		// The CSA auth token header contains uppercase letters and hyphens; hyphens are a valid
+		// gRPC metadata char, so this key is namespaced rather than omitted, same as the others.
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{"X-Beholder-Node-Auth-Token": "forged"})
+		assert.Empty(t, dropped)
+		assert.Equal(t, map[string]string{rp + "x-beholder-node-auth-token": "forged"}, got)
 	})
 
 	t.Run("CloudEvents context attribute names are kept, they mean nothing as gRPC metadata", func(t *testing.T) {
-		got := chipingress.SanitizeMetadataHeaders(map[string]string{"subject": "keep-me", "source": "keep-me-too"})
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{"subject": "keep-me", "source": "keep-me-too"})
+		assert.Empty(t, dropped)
 		assert.Equal(t, map[string]string{rp + "subject": "keep-me", rp + "source": "keep-me-too"}, 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[rp+"chain_id"])
+	t.Run("non-printable values omit the whole attribute", func(t *testing.T) {
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{"chain_id": "1\n2"})
+		assert.Empty(t, got)
+		require.Len(t, dropped, 1)
+		assert.Equal(t, "chain_id", dropped[0].Key)
+		assert.Equal(t, "invalid_value", dropped[0].Reason)
 	})
 
-	t.Run("duplicate normalized keys resolve deterministically to sorted-first key", func(t *testing.T) {
-		// Both normalize to chain_id; sorted order is "chain id" < "chain_id" (' ' < '_'), so the
-		// space-separated key wins.
-		got := chipingress.SanitizeMetadataHeaders(map[string]string{"chain id": "from-space", "chain_id": "from-snake"})
-		assert.Equal(t, map[string]string{rp + "chain_id": "from-space"}, got)
+	t.Run("duplicate keys resolve deterministically to sorted-first key", func(t *testing.T) {
+		// "DonID" and "donid" both validate to "donid"; sorted order of the ORIGINAL keys is
+		// "DonID" < "donid" (upper-case sorts first in ASCII), so "DonID" wins.
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{"DonID": "upper", "donid": "lower"})
+		assert.Equal(t, map[string]string{rp + "donid": "upper"}, got)
+		require.Len(t, dropped, 1)
+		assert.Equal(t, "donid", dropped[0].Key)
+		assert.Equal(t, "duplicate_key", dropped[0].Reason)
 	})
 
-	t.Run("keys that differ only in case collapse deterministically", func(t *testing.T) {
-		got := chipingress.SanitizeMetadataHeaders(map[string]string{"DonID": "upper", "donid": "lower"})
-		// sorted order: "DonID" < "donid" (upper-case sorts first in ASCII).
-		assert.Equal(t, map[string]string{rp + "donid": "upper"}, got)
+	t.Run("oversized key is omitted, not truncated", func(t *testing.T) {
+		longKey := strings.Repeat("a", 129)
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{longKey: "v"})
+		assert.Empty(t, got)
+		require.Len(t, dropped, 1)
+		assert.Equal(t, "invalid_key", dropped[0].Reason)
+	})
+
+	t.Run("oversized value is omitted, not truncated", func(t *testing.T) {
+		longVal := strings.Repeat("v", 513)
+		got, dropped := chipingress.SanitizeMetadataHeaders(map[string]string{"chain_id": longVal})
+		assert.Empty(t, got)
+		require.Len(t, dropped, 1)
+		assert.Equal(t, "invalid_value", dropped[0].Reason)
+	})
+
+	t.Run("attribute count is capped at 32, excess dropped deterministically", func(t *testing.T) {
+		in := make(map[string]string, 33)
+		for i := 0; i < 33; i++ {
+			in[fmt.Sprintf("attr_%02d", i)] = "v"
+		}
+		got, dropped := chipingress.SanitizeMetadataHeaders(in)
+		assert.Len(t, got, 32)
+		require.Len(t, dropped, 1)
+		// Sorted order: "attr_32" sorts last among "attr_00".."attr_32".
+		assert.Equal(t, "attr_32", dropped[0].Key)
+		assert.Equal(t, "limit_exceeded", dropped[0].Reason)
+	})
+
+	t.Run("total key+value bytes are capped at 4096, tail dropped deterministically", func(t *testing.T) {
+		// Each accepted attribute contributes len(key)+len(value) bytes (prefix excluded). 9
+		// attributes of 500 bytes each would total 4500, over the 4096 cap, so the last one or
+		// two (in sorted order) must be dropped.
+		in := make(map[string]string, 9)
+		val := strings.Repeat("v", 490)
+		for i := 0; i < 9; i++ {
+			in[fmt.Sprintf("attr_%d", i)] = val // key is 6 bytes, so each entry is 496 bytes
+		}
+		got, dropped := chipingress.SanitizeMetadataHeaders(in)
+		assert.NotEmpty(t, dropped)
+		for _, d := range dropped {
+			assert.Equal(t, "limit_exceeded", d.Reason)
+		}
+		total := 0
+		for name, v := range got {
+			total += len(name) - len(rp) + len(v)
+		}
+		assert.LessOrEqual(t, total, 4096)
 	})
 }
 
@@ -440,9 +483,11 @@ func TestSanitizeMetadataHeaders_AvoidsRPCFailure(t *testing.T) {
 	})
 
 	t.Run("sanitized headers succeed", func(t *testing.T) {
+		sanitized, dropped := chipingress.SanitizeMetadataHeaders(dirty)
+		require.Len(t, dropped, 1, "the non-printable value must be omitted, not rewritten")
 		client, err := chipingress.NewClient(lis.Addr().String(),
 			chipingress.WithInsecureConnection(),
-			chipingress.WithHeaderProvider(chipingress.NewStaticHeaderProvider(chipingress.SanitizeMetadataHeaders(dirty))),
+			chipingress.WithHeaderProvider(chipingress.NewStaticHeaderProvider(sanitized)),
 		)
 		require.NoError(t, err)
 		defer client.Close() //nolint:errcheck

From 30365376e5882b3c901ba021afd72cdefaa54974 Mon Sep 17 00:00:00 2001
From: Pavel <177363085+pkcll@users.noreply.github.com>
Date: Thu, 3 Sep 2026 16:53:43 -0400
Subject: [PATCH 3/3] chipingress: extract metadata key charset check into a
 predicate

staticcheck QF1001 flagged the negated conjunction in sanitizeMetadataKey.
A positive isValidMetadataKeyChar predicate reads better than a hand-applied
De Morgan inversion and keeps behaviour identical.
---
 pkg/chipingress/header_provider.go | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/pkg/chipingress/header_provider.go b/pkg/chipingress/header_provider.go
index 3a146528b2..f69f721d3d 100644
--- a/pkg/chipingress/header_provider.go
+++ b/pkg/chipingress/header_provider.go
@@ -162,6 +162,13 @@ const (
 	reasonLimitExceeded = "limit_exceeded"
 )
 
+// isValidMetadataKeyChar reports whether r is allowed in an outgoing gRPC metadata key. grpc-go
+// accepts [0-9a-z-_.] (see internal/metadata.ValidateKey); upper-case is handled by lower-casing
+// before this is called.
+func isValidMetadataKeyChar(r rune) bool {
+	return (r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') || r == '.' || r == '-' || r == '_'
+}
+
 // sanitizeMetadataKey validates a resource-attribute key as a valid outgoing gRPC metadata key,
 // without the ResourceHeaderPrefix that SanitizeMetadataHeaders adds, and reports whether it is
 // valid. It never rewrites: a key that fails validation is omitted by the caller rather than
@@ -178,7 +185,7 @@ func sanitizeMetadataKey(key string) (string, bool) {
 	}
 	lower := strings.ToLower(key)
 	for _, r := range lower {
-		if !((r >= 'a' && r <= 'z') || (r >= '0' && r <= '9') || r == '.' || r == '-' || r == '_') {
+		if !isValidMetadataKeyChar(r) {
 			return "", false
 		}
 	}