From b5f5c40ae14c717a79a40846fc13bf9dae9f9656 Mon Sep 17 00:00:00 2001 From: zhanghao <112082360+zhanghao0806@users.noreply.github.com> Date: Fri, 30 Oct 2026 16:39:38 +0800 Subject: [PATCH] fix(kafka): ignore unrelated binary header values Signed-off-by: zhanghao <112082360+zhanghao0806@users.noreply.github.com> --- src/cloudevents/core/bindings/kafka.py | 5 ++- tests/test_core/test_bindings/test_kafka.py | 37 +++++++++++++++++++++ 2 files changed, 39 insertions(+), 3 deletions(-) diff --git a/src/cloudevents/core/bindings/kafka.py b/src/cloudevents/core/bindings/kafka.py index dae9e496..920b1472 100644 --- a/src/cloudevents/core/bindings/kafka.py +++ b/src/cloudevents/core/bindings/kafka.py @@ -165,18 +165,17 @@ def from_binary( attributes: dict[str, Any] = {} for header_name, header_value_bytes in message.headers.items(): - header_value = header_value_bytes.decode("utf-8") - normalized_name = header_name.lower() if normalized_name.startswith(CE_PREFIX): + header_value = header_value_bytes.decode("utf-8") attr_name = normalized_name[len(CE_PREFIX) :] if attr_name == TIME_ATTR: attributes[attr_name] = isoparse(header_value) else: attributes[attr_name] = header_value elif normalized_name == CONTENT_TYPE_HEADER: - attributes[DATACONTENTTYPE_ATTR] = header_value + attributes[DATACONTENTTYPE_ATTR] = header_value_bytes.decode("utf-8") # If message has a key, add it as partitionkey extension attribute if message.key is not None: diff --git a/tests/test_core/test_bindings/test_kafka.py b/tests/test_core/test_bindings/test_kafka.py index 54df8221..35f2115b 100644 --- a/tests/test_core/test_bindings/test_kafka.py +++ b/tests/test_core/test_bindings/test_kafka.py @@ -811,3 +811,40 @@ def test_from_structured_with_key_auto_detect_v03() -> None: assert event.get_type() == "com.example.test" assert event.get_extension("partitionkey") == "partition-key-456" assert event.get_attributes()["specversion"] == "0.3" + + +@pytest.mark.parametrize("reader", [from_binary, from_kafka]) +def test_binary_ignores_non_cloudevent_binary_headers(reader: Any) -> None: + message = KafkaMessage( + headers={ + "ce_specversion": b"1.0", + "ce_id": b"123", + "ce_type": b"example.created", + "ce_source": b"/example", + "content-type": b"application/json", + "correlation-bin": b"\xff\x00\xfe", + }, + key=None, + value=b'{"message":"hello"}', + ) + event = reader(message, JSONFormat()) + assert event.get_attributes()["id"] == "123" + assert event.get_data() == {"message": "hello"} + assert "correlation-bin" not in event.get_attributes() + + +@pytest.mark.parametrize("header", ["ce_subject", "content-type"]) +def test_binary_rejects_invalid_utf8_in_cloudevent_headers(header: str) -> None: + message = KafkaMessage( + headers={ + "ce_specversion": b"1.0", + "ce_id": b"123", + "ce_type": b"example.created", + "ce_source": b"/example", + header: b"\xff", + }, + key=None, + value=b"", + ) + with pytest.raises(UnicodeDecodeError): + from_binary(message, JSONFormat())