From a58f7783b72c187705c837e18e14a087e8ed2044 Mon Sep 17 00:00:00 2001 From: Vishwak Thatikonda Date: Fri, 23 Oct 2026 16:54:54 -0700 Subject: [PATCH] fix(parser): decode signed Kafka header bytes as UTF-8 KafkaRecordSchema decoded each header value with String.fromCodePoint(...value), which throws a RangeError that escapes safeParse. Lambda sends header bytes as signed Java bytes, so any byte >= 128 arrives negative and every real event with a non-ASCII or binary header failed to parse. Spreading the array also exceeded the call stack for large headers. Validate header values as integers from -128 to 255, so out-of-range values become Zod issues, and decode with Buffer.from(value).toString('utf8'), which masks each number to a byte, has no argument limit, and matches how @aws-lambda-powertools/kafka decodes headers. Add a record with a signed UTF-8 header to the Kafka test event, matching the one Powertools for Python added, and cover out-of-range values and a large header. Closes #5830 --- packages/parser/src/schemas/kafka.ts | 6 +- packages/parser/tests/events/kafka/base.json | 15 ++++ .../parser/tests/unit/envelopes/kafka.test.ts | 8 +-- .../parser/tests/unit/schema/kafka.test.ts | 68 +++++++++++++++++++ 4 files changed, 91 insertions(+), 6 deletions(-) diff --git a/packages/parser/src/schemas/kafka.ts b/packages/parser/src/schemas/kafka.ts index d69f37e718..5b922d7791 100644 --- a/packages/parser/src/schemas/kafka.ts +++ b/packages/parser/src/schemas/kafka.ts @@ -22,8 +22,10 @@ const KafkaRecordSchema = z.object({ headers: z.array( z.record( z.string(), - z.array(z.number()).transform((value) => { - return String.fromCodePoint(...value); + // Lambda sends header bytes as signed Java bytes, so bytes >= 128 arrive negative. + // Buffer.from masks each number to a byte, and decoding matches @aws-lambda-powertools/kafka. + z.array(z.number().int().min(-128).max(255)).transform((value) => { + return Buffer.from(value).toString('utf8'); }) ) ), diff --git a/packages/parser/tests/events/kafka/base.json b/packages/parser/tests/events/kafka/base.json index 4c17ea2490..51b03ac832 100644 --- a/packages/parser/tests/events/kafka/base.json +++ b/packages/parser/tests/events/kafka/base.json @@ -17,6 +17,21 @@ "headerKey": [104, 101, 97, 100, 101, 114, 86, 97, 108, 117, 101] } ] + }, + { + "topic": "mymessage-with-unsigned", + "partition": 0, + "offset": 15, + "timestamp": 1545084650987, + "timestampType": "CREATE_TIME", + "value": "eyJrZXkiOiJ2YWx1ZSJ9", + "headers": [ + { + "headerKey": [ + 104, 101, 108, 108, 111, 45, 119, 111, 114, 108, 100, 45, -61, -85 + ] + } + ] } ] } diff --git a/packages/parser/tests/unit/envelopes/kafka.test.ts b/packages/parser/tests/unit/envelopes/kafka.test.ts index 6eb741f767..c621ffff21 100644 --- a/packages/parser/tests/unit/envelopes/kafka.test.ts +++ b/packages/parser/tests/unit/envelopes/kafka.test.ts @@ -96,7 +96,7 @@ describe('Envelope: Kafka', () => { const result = KafkaEnvelope.parse(event, z.string()); // Assess - expect(result).toEqual(['{"key":"value"}']); + expect(result).toEqual(['{"key":"value"}', '{"key":"value"}']); }); it('parses a Kafka event and applies the schema transformation', () => { @@ -110,7 +110,7 @@ describe('Envelope: Kafka', () => { ); // Assess - expect(result).toEqual([{ key: 'value' }]); + expect(result).toEqual([{ key: 'value' }, { key: 'value' }]); }); it('parses a self managed Kafka event', () => { @@ -122,7 +122,7 @@ describe('Envelope: Kafka', () => { const result = KafkaEnvelope.parse(event, z.string()); // Assess - expect(result).toEqual(['{"key":"value"}']); + expect(result).toEqual(['{"key":"value"}', '{"key":"value"}']); }); }); @@ -137,7 +137,7 @@ describe('Envelope: Kafka', () => { // Assess expect(result).toEqual({ success: true, - data: ['{"key":"value"}'], + data: ['{"key":"value"}', '{"key":"value"}'], }); }); diff --git a/packages/parser/tests/unit/schema/kafka.test.ts b/packages/parser/tests/unit/schema/kafka.test.ts index 845d20fcdf..464cf29e09 100644 --- a/packages/parser/tests/unit/schema/kafka.test.ts +++ b/packages/parser/tests/unit/schema/kafka.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from 'vitest'; +import type { z } from 'zod'; import { KafkaMskEventSchema, KafkaSelfManagedEventSchema, @@ -14,6 +15,11 @@ describe('Schema: Kafka', () => { eventsPath: 'kafka', filename: 'base', }); + // Typed as the schema input, so header values are the raw byte arrays + const rawEvent = getTestEvent>({ + eventsPath: 'kafka', + filename: 'base', + }); it('parses a Kafka MSK event', () => { // Prepare @@ -47,6 +53,19 @@ describe('Schema: Kafka', () => { }, ], }, + { + topic: 'mymessage-with-unsigned', + partition: 0, + offset: 15, + timestamp: 1545084650987, + timestampType: 'CREATE_TIME', + value: '{"key":"value"}', + headers: [ + { + headerKey: 'hello-world-ë', + }, + ], + }, ], }, }); @@ -61,6 +80,42 @@ describe('Schema: Kafka', () => { expect(() => KafkaMskEventSchema.parse(event)).toThrow(); }); + it.each([-129, 256, 1.5])( + 'returns an issue for a header byte of %s instead of throwing', + (byte) => { + // Prepare + const event = structuredClone(rawEvent); + event.records['mytopic-0'][0].headers = [{ headerKey: [byte] }]; + + // Act + const result = KafkaMskEventSchema.safeParse(event); + + // Assess + expect(result.success).toBe(false); + expect(result.error?.issues).toEqual([ + expect.objectContaining({ + path: ['records', 'mytopic-0', 0, 'headers', 0, 'headerKey', 0], + }), + ]); + } + ); + + it('decodes a large header without exceeding the call stack', () => { + // Prepare + const event = structuredClone(rawEvent); + event.records['mytopic-0'][0].headers = [ + { headerKey: new Array(200_000).fill(65) }, + ]; + + // Act + const result = KafkaMskEventSchema.parse(event); + + // Assess + expect(result.records['mytopic-0'][0].headers).toEqual([ + { headerKey: 'A'.repeat(200_000) }, + ]); + }); + it('parses a Kafka self-managed event', () => { // Prepare const event = omit( @@ -92,6 +147,19 @@ describe('Schema: Kafka', () => { }, ], }, + { + topic: 'mymessage-with-unsigned', + partition: 0, + offset: 15, + timestamp: 1545084650987, + timestampType: 'CREATE_TIME', + value: '{"key":"value"}', + headers: [ + { + headerKey: 'hello-world-ë', + }, + ], + }, ], }, });