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_" 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_" (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