[codex] validate v1 compat attribute names

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

@@ -13,6 +13,7 @@
 #    under the License.
 
 import datetime
+import re
 import typing
 import uuid
 
@@ -24,6 +25,15 @@
     "1.0": v1.Event._ce_required_fields,
     "0.3": v03.Event._ce_required_fields,
 }
+_attribute_name_pattern = re.compile(r"^[a-z0-9]+$")
+
+
+def _validate_attribute_name(name: str) -> None:
+    if not _attribute_name_pattern.fullmatch(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):
@@ -59,6 +69,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 +100,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_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)