From 6e4d686316dceb8484c2298f275b33445a3a6bac Mon Sep 17 00:00:00 2001
From: zhanghao <112082360+zhanghao0806@users.noreply.github.com>
Date: Fri, 30 Oct 2026 16:41:03 +0800
Subject: [PATCH] fix(http): reject invalid UTF-8 in encoded attributes

Signed-off-by: zhanghao <112082360+zhanghao0806@users.noreply.github.com>
---
 src/cloudevents/core/bindings/http.py      |  2 +-
 tests/test_core/test_bindings/test_http.py | 35 ++++++++++++++++++++++
 2 files changed, 36 insertions(+), 1 deletion(-)

diff --git a/src/cloudevents/core/bindings/http.py b/src/cloudevents/core/bindings/http.py
index fbdcc3b0..6f1f8a57 100644
--- a/src/cloudevents/core/bindings/http.py
+++ b/src/cloudevents/core/bindings/http.py
@@ -65,7 +65,7 @@ def _decode_header_value(attr_name: str, value: str) -> Any:
     :param value: The percent-encoded header value
     :return: Decoded value (datetime for 'time' attribute, string otherwise)
     """
-    decoded = unquote(value)
+    decoded = unquote(value, encoding="utf-8", errors="strict")
     if attr_name == TIME_ATTR:
         return isoparse(decoded)
     return decoded
diff --git a/tests/test_core/test_bindings/test_http.py b/tests/test_core/test_bindings/test_http.py
index cf46ba4d..c7399fbb 100644
--- a/tests/test_core/test_bindings/test_http.py
+++ b/tests/test_core/test_bindings/test_http.py
@@ -1139,3 +1139,38 @@ def test_convenience_with_explicit_format_override() -> None:
 
     assert recovered.get_type() == event.get_type()
     assert recovered.get_data() == event.get_data()
+
+
+@pytest.mark.parametrize("reader", [from_binary, from_http])
+@pytest.mark.parametrize("value", ["%FF", "%C3%28", "%ED%A0%80", "%F0%9F%98"])
+def test_binary_rejects_invalid_percent_encoded_utf8(reader: Any, value: str) -> None:
+    message = HTTPMessage(
+        headers={
+            "ce-specversion": "1.0",
+            "ce-id": "123",
+            "ce-type": "example.created",
+            "ce-source": "/example",
+            "ce-subject": value,
+        },
+        body=b"",
+    )
+    with pytest.raises(UnicodeDecodeError):
+        reader(message, JSONFormat())
+
+
+@pytest.mark.parametrize(
+    "encoded,decoded",
+    [("%E2%82%AC", "€"), ("%F0%9F%98%80", "😀"), ("%41", "A"), ("a+b", "a+b")],
+)
+def test_binary_accepts_valid_percent_encoded_utf8(encoded: str, decoded: str) -> None:
+    message = HTTPMessage(
+        headers={
+            "ce-specversion": "1.0",
+            "ce-id": "123",
+            "ce-type": "example.created",
+            "ce-source": "/example",
+            "ce-subject": encoded,
+        },
+        body=b"",
+    )
+    assert from_binary(message, JSONFormat()).get_attributes()["subject"] == decoded