From d17ab70c37b1da9c65a51d8a9c1267bbd1b4c73f Mon Sep 17 00:00:00 2001 From: MickeyWzt <289528356+MickeyWzt@users.noreply.github.com> Date: Fri, 3 Jul 2026 23:37:22 +0800 Subject: [PATCH] fix: validate v1 compat attribute names Signed-off-by: MickeyWzt <289528356+MickeyWzt@users.noreply.github.com> --- src/cloudevents/core/spec.py | 14 ++++++++++- src/cloudevents/core/v03/event.py | 5 ++-- src/cloudevents/core/v1/event.py | 5 ++-- src/cloudevents/v1/exceptions.py | 4 ++++ src/cloudevents/v1/http/event.py | 17 +++++++++++++ tests/test_core/test_v03/test_event.py | 13 ++++++++++ tests/test_core/test_v1/test_event.py | 13 ++++++++++ tests/test_v1_compat/test_event_extensions.py | 24 +++++++++++++++++++ 8 files changed, 88 insertions(+), 7 deletions(-) diff --git a/src/cloudevents/core/spec.py b/src/cloudevents/core/spec.py index e3858189..37230a61 100644 --- a/src/cloudevents/core/spec.py +++ b/src/cloudevents/core/spec.py @@ -11,8 +11,20 @@ # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. -from typing import Literal +import re +from typing import Final, Literal SpecVersion = Literal["1.0", "0.3"] SPECVERSION_V1_0 = "1.0" SPECVERSION_V0_3 = "0.3" + +_ATTRIBUTE_NAME_PATTERN: Final[re.Pattern[str]] = re.compile(r"^[a-z0-9]+$") + + +def is_valid_attribute_name(name: str) -> bool: + """ + Return whether a name follows the CloudEvents attribute naming convention. + + See https://github.com/cloudevents/spec/blob/v1.0.2/cloudevents/spec.md#attribute-naming-convention + """ + return bool(_ATTRIBUTE_NAME_PATTERN.fullmatch(name)) diff --git a/src/cloudevents/core/v03/event.py b/src/cloudevents/core/v03/event.py index 4d51c3ed..9e126ed6 100644 --- a/src/cloudevents/core/v03/event.py +++ b/src/cloudevents/core/v03/event.py @@ -12,7 +12,6 @@ # License for the specific language governing permissions and limitations # under the License. -import re import uuid from collections import defaultdict from datetime import datetime, timezone @@ -27,7 +26,7 @@ InvalidAttributeValueError, MissingRequiredAttributeError, ) -from cloudevents.core.spec import SPECVERSION_V0_3 +from cloudevents.core.spec import SPECVERSION_V0_3, is_valid_attribute_name REQUIRED_ATTRIBUTES: Final[list[str]] = ["id", "source", "type", "specversion"] OPTIONAL_ATTRIBUTES: Final[list[str]] = [ @@ -274,7 +273,7 @@ def _validate_extension_attributes( msg=f"Extension attribute name must be at least 1 character long but was '{extension_attribute}'", ) ) - if not re.match(r"^[a-z0-9]+$", extension_attribute): + if not is_valid_attribute_name(extension_attribute): errors[extension_attribute].append( CustomExtensionAttributeError( attribute_name=extension_attribute, diff --git a/src/cloudevents/core/v1/event.py b/src/cloudevents/core/v1/event.py index a71a3d58..97e729c0 100644 --- a/src/cloudevents/core/v1/event.py +++ b/src/cloudevents/core/v1/event.py @@ -12,7 +12,6 @@ # License for the specific language governing permissions and limitations # under the License. -import re import uuid from collections import defaultdict from datetime import datetime, timezone @@ -27,7 +26,7 @@ InvalidAttributeValueError, MissingRequiredAttributeError, ) -from cloudevents.core.spec import SPECVERSION_V1_0 +from cloudevents.core.spec import SPECVERSION_V1_0, is_valid_attribute_name REQUIRED_ATTRIBUTES: Final[list[str]] = ["id", "source", "type", "specversion"] OPTIONAL_ATTRIBUTES: Final[list[str]] = [ @@ -259,7 +258,7 @@ def _validate_extension_attributes( msg=f"Extension attribute name must be at least 1 character long but was '{extension_attribute}'", ) ) - if not re.match(r"^[a-z0-9]+$", extension_attribute): + if not is_valid_attribute_name(extension_attribute): errors[extension_attribute].append( CustomExtensionAttributeError( attribute_name=extension_attribute, diff --git a/src/cloudevents/v1/exceptions.py b/src/cloudevents/v1/exceptions.py index 29294130..4b119fca 100644 --- a/src/cloudevents/v1/exceptions.py +++ b/src/cloudevents/v1/exceptions.py @@ -25,6 +25,10 @@ class InvalidRequiredFields(GenericException): pass +class InvalidAttributeName(GenericException): + pass + + class InvalidStructuredJSON(GenericException): pass diff --git a/src/cloudevents/v1/http/event.py b/src/cloudevents/v1/http/event.py index 377262d5..b37575b9 100644 --- a/src/cloudevents/v1/http/event.py +++ b/src/cloudevents/v1/http/event.py @@ -17,6 +17,7 @@ import uuid import cloudevents.v1.exceptions as cloud_exceptions +from cloudevents.core.spec import is_valid_attribute_name from cloudevents.v1 import abstract from cloudevents.v1.sdk.event import v03, v1 @@ -26,6 +27,19 @@ } +def _validate_attribute_name(name: str) -> None: + """ + Validate names against the CloudEvents attribute naming convention. + + See https://github.com/cloudevents/spec/blob/v1.0.2/cloudevents/spec.md#attribute-naming-convention + """ + if not is_valid_attribute_name(name): + raise cloud_exceptions.InvalidAttributeName( + f"Invalid CloudEvent attribute name '{name}': " + "attribute names must only contain lowercase ASCII letters and digits" + ) + + class CloudEvent(abstract.CloudEvent): """ Python-friendly cloudevent class supporting v1 events @@ -59,6 +73,8 @@ def __init__(self, attributes: typing.Mapping[str, str], data: typing.Any = None :type data: typing.Any """ self._attributes = {k.lower(): v for k, v in attributes.items()} + for attribute_name in self._attributes: + _validate_attribute_name(attribute_name) self.data = data if "specversion" not in self._attributes: self._attributes["specversion"] = "1.0" @@ -88,6 +104,7 @@ def get_data(self) -> typing.Optional[typing.Any]: return self.data def __setitem__(self, key: str, value: typing.Any) -> None: + _validate_attribute_name(key) self._attributes[key] = value def __delitem__(self, key: str) -> None: diff --git a/tests/test_core/test_v03/test_event.py b/tests/test_core/test_v03/test_event.py index 950308ed..efe8e64d 100644 --- a/tests/test_core/test_v03/test_event.py +++ b/tests/test_core/test_v03/test_event.py @@ -435,6 +435,19 @@ def test_required_attributes_null_or_empty( ] }, ), + ( + "example-extension", + { + "example-extension": [ + str( + CustomExtensionAttributeError( + "example-extension", + "Extension attribute 'example-extension' should only contain lowercase letters and numbers", + ) + ) + ] + }, + ), ], ) def test_custom_extension(extension_name: str, expected_error: dict) -> None: diff --git a/tests/test_core/test_v1/test_event.py b/tests/test_core/test_v1/test_event.py index 7d283216..5cf41f06 100644 --- a/tests/test_core/test_v1/test_event.py +++ b/tests/test_core/test_v1/test_event.py @@ -371,6 +371,19 @@ def test_required_attributes_null_or_empty( ] }, ), + ( + "example-extension", + { + "example-extension": [ + str( + CustomExtensionAttributeError( + "example-extension", + "Extension attribute 'example-extension' should only contain lowercase letters and numbers", + ) + ) + ] + }, + ), ], ) def test_custom_extension(extension_name: str, expected_error: dict) -> None: diff --git a/tests/test_v1_compat/test_event_extensions.py b/tests/test_v1_compat/test_event_extensions.py index 52059acd..fa33dbca 100644 --- a/tests/test_v1_compat/test_event_extensions.py +++ b/tests/test_v1_compat/test_event_extensions.py @@ -16,6 +16,7 @@ import pytest +import cloudevents.v1.exceptions as cloud_exceptions from cloudevents.v1.http import CloudEvent, from_http, to_binary, to_structured test_data = json.dumps({"data-key": "val"}) @@ -32,6 +33,29 @@ def test_cloudevent_access_extensions(specversion): assert event["ext1"] == "testval" +def test_cloudevent_rejects_invalid_extension_attribute_name(): + with pytest.raises(cloud_exceptions.InvalidAttributeName) as exc: + CloudEvent( + { + "type": "com.example.string", + "source": "https://example.com/event-producer", + "example-extension": "testval", + }, + test_data, + ) + + assert "example-extension" in str(exc.value) + + +def test_cloudevent_rejects_invalid_extension_attribute_setitem(): + event = CloudEvent(test_attributes, test_data) + + with pytest.raises(cloud_exceptions.InvalidAttributeName) as exc: + event["example-extension"] = "testval" + + assert "example-extension" in str(exc.value) + + @pytest.mark.parametrize("specversion", ["0.3", "1.0"]) def test_to_binary_extensions(specversion): event = CloudEvent(test_attributes, test_data)