fix: validate v1 compat attribute names

Signed-off-by: MickeyWzt <289528356+MickeyWzt@users.noreply.github.com>

Author: MickeyWzt

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))

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,

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,

src/cloudevents/v1/exceptions.py

@@ -25,6 +25,10 @@ class InvalidRequiredFields(GenericException):
     pass
 
 
+class InvalidAttributeName(GenericException):
+    pass
+
+
 class InvalidStructuredJSON(GenericException):
     pass
 

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,14 @@
 }
 
 
+def _validate_attribute_name(name: str) -> None:
+    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 +68,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 +99,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:

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:

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:

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)