fix!: reject duplicate CBOR map keys

CHANGES.rst

@@ -6,6 +6,8 @@ Unreleased
 
 - **Breaking change**: Drop support for Python 3.10.
 - Add support for Python 3.15.
+- **Breaking change**: Reject CBOR maps with duplicate keys on decoding (e.g. COSE header maps, COSE_Key and CWT claims) instead of silently keeping the last value, as RFC 9052 Section 3 requires for COSE header parameters.
+    - Bump the minimum supported version of cbor2 to 6.1.0, which provides ``allow_duplicate_keys``.
 - Correct COSE signing header annotations to accept dictionaries and ResolvedHeader in strict type checkers.
 - Annotate factory class methods (``COSE.new``, ``CWT.new``, ``Claims.new``, ``Claims.from_json``, ``Signer.new``, ``Signer.from_jwk``, ``Signer.from_pem``, ``COSEMessage.loads``, etc.) to return ``Self`` for strict type checkers.
 - Modernize type annotations to PEP 604/585 syntax (e.g. ``X | None``, ``dict[str, Any]``) and add pyupgrade (``--py311-plus``) to pre-commit.

cwt/cbor_processor.py

@@ -14,6 +14,6 @@ def _dumps(self, obj: Any) -> bytes:
 
     def _loads(self, s: bytes) -> dict[str | int, Any]:
         try:
-            return loads(s)
+            return loads(s, allow_duplicate_keys=False)
         except Exception as err:
             raise DecodeError("Failed to decode.") from err

cwt/cose_key.py

@@ -188,7 +188,7 @@ def from_bytes(cls, key_data: bytes) -> COSEKeyInterface:
             ValueError: Invalid arguments.
             DecodeError: Failed to decode the key data.
         """
-        params = cbor2.loads(key_data)
+        params = cbor2.loads(key_data, allow_duplicate_keys=False)
         return cls.new(params)
 
     @classmethod

cwt/cose_message.py

@@ -116,7 +116,7 @@ def __ne__(self: COSEMessage, other: object) -> bool:
 
     @classmethod
     def loads(cls, msg: bytes) -> Self:
-        tagged = loads(msg)
+        tagged = loads(msg, allow_duplicate_keys=False)
         if not isinstance(tagged, CBORTag):
             raise ValueError("Invalid COSE format.")
         type = COSE_TAG_TO_TYPE.get(tagged.tag, None)

cwt/encrypted_cose_key.py

@@ -70,5 +70,5 @@ def to_cose_key(key: list[Any], encryption_key: COSEKeyInterface) -> COSEKeyInte
             DecodeError: Failed to decode the COSE key.
             VerifyError: Failed to verify the COSE key.
         """
-        res = cbor2.loads(COSE().decode(CBORTag(16, key), encryption_key))
+        res = cbor2.loads(COSE().decode(CBORTag(16, key), encryption_key), allow_duplicate_keys=False)
         return COSEKey.new(res)

cwt/recipient.py

@@ -130,7 +130,7 @@ def from_list(
             raise ValueError("Invalid recipient format.")
         if not isinstance(recipient[0], bytes):
             raise ValueError("protected header should be bytes.")
-        protected = {} if not recipient[0] else cbor2.loads(recipient[0])
+        protected = {} if not recipient[0] else cbor2.loads(recipient[0], allow_duplicate_keys=False)
         if not isinstance(recipient[1], Mapping):
             raise ValueError("unprotected header should be dict.")
         unprotected = to_mutable(recipient[1])

pyproject.toml

@@ -26,7 +26,7 @@ exclude = [
 
 [tool.poetry.dependencies]
 python = "^3.11,<4.0"
-cbor2 = ">=5.4.2,<7,!=6.0.0"
+cbor2 = ">=6.1.0,<7"
 cryptography = ">=42.0.1,<52"
 pyhpke = ">=0.5.3,<1.0.0"
 

tests/test_cose.py

@@ -1143,11 +1143,6 @@ def test_cose_encode_for_mac_with_valid_args(self, p, u):
                 "The alg(4) in the headers does not match the alg(5) in the populated key.",
             ),
             # (
-            #     {COSEHeaders.ALG: COSEAlgs.HS256_64, COSEHeaders.ALG: COSEAlgs.HS256},
-            #     {},
-            #     "The protected header has duplicated keys.",
-            # ),
-            # (
             #     {},
             #     {COSEHeaders.ALG: COSEAlgs.HS256_64, COSEHeaders.ALG: COSEAlgs.HS256},
             #     "Unsupported or unknown alg: xxx.",

tests/test_duplicate_map_keys.py

@@ -0,0 +1,97 @@
+"""
+Tests for rejecting CBOR maps with duplicate keys (RFC 9052 Section 3).
+"""
+
+import cbor2
+import pytest
+from cbor2 import CBORTag, dumps
+
+import cwt
+from cwt import COSE, COSEKey, COSEMessage, DecodeError, EncryptedCOSEKey, Recipient
+
+# {1: 4, 1: 5}: a map with the same label (alg) twice.
+DUP_ALG_MAP = b"\xa2" + dumps(1) + dumps(4) + dumps(1) + dumps(5)
+# {4: b"a", 4: b"poc"}: a map with the same label (kid) twice.
+DUP_KID_MAP = b"\xa2" + dumps(4) + dumps(b"a") + dumps(4) + dumps(b"poc")
+
+
+@pytest.fixture
+def mac_key():
+    return COSEKey.from_symmetric_key(alg="HS256", kid="poc")
+
+
+class TestDuplicateMapKeys:
+    def test_decode_mac0_with_duplicated_protected_header_label(self, mac_key):
+        tag = mac_key.sign(dumps(["MAC0", DUP_ALG_MAP, b"", b"payload"]))
+        wire = dumps(CBORTag(17, [DUP_ALG_MAP, {}, b"payload", tag]))
+        with pytest.raises(DecodeError) as err:
+            COSE.new().decode_with_headers(wire, mac_key)
+            pytest.fail("decode_with_headers() should fail.")
+        assert "Failed to decode." in str(err.value)
+        assert isinstance(err.value.__cause__, cbor2.CBORDecodeError)
+
+    def test_decode_mac0_with_duplicated_unprotected_header_label(self, mac_key):
+        protected = dumps({1: 5})
+        tag = mac_key.sign(dumps(["MAC0", protected, b"", b"payload"]))
+        wire = b"\xd1\x84" + dumps(protected) + DUP_KID_MAP + dumps(b"payload") + dumps(tag)
+        with pytest.raises(DecodeError) as err:
+            COSE.new().decode_with_headers(wire, mac_key)
+            pytest.fail("decode_with_headers() should fail.")
+        assert "Failed to decode." in str(err.value)
+
+    def test_decode_sign1_with_duplicated_protected_header_label(self):
+        # {1: -7, 1: -7}: ES256 twice.
+        protected = b"\xa2" + dumps(1) + dumps(-7) + dumps(1) + dumps(-7)
+        key = COSEKey.generate_symmetric_key(alg="HS256")  # never reached; decoding fails first.
+        wire = dumps(CBORTag(18, [protected, {}, b"payload", b"\x00" * 64]))
+        with pytest.raises(DecodeError) as err:
+            COSE.new().decode(wire, key)
+            pytest.fail("decode() should fail.")
+        assert "Failed to decode." in str(err.value)
+
+    def test_decode_cwt_with_duplicated_claim_keys(self, mac_key):
+        # {1: "a", 1: "b"}: iss twice.
+        claims = b"\xa2" + dumps(1) + dumps("a") + dumps(1) + dumps("b")
+        token = COSE.new(alg_auto_inclusion=True, kid_auto_inclusion=True).encode_and_mac(claims, mac_key)
+        with pytest.raises(DecodeError) as err:
+            cwt.decode(token, mac_key)
+            pytest.fail("decode() should fail.")
+        assert "Failed to decode." in str(err.value)
+
+    def test_cose_key_from_bytes_with_duplicated_labels(self):
+        # {1: 4, 1: 4, -1: k}: kty twice.
+        key_data = b"\xa3" + dumps(1) + dumps(4) + dumps(1) + dumps(4) + dumps(-1) + dumps(b"x" * 32)
+        with pytest.raises(cbor2.CBORDecodeError):
+            COSEKey.from_bytes(key_data)
+            pytest.fail("from_bytes() should fail.")
+
+    def test_encrypted_cose_key_to_cose_key_with_duplicated_labels(self):
+        enc_key = COSEKey.generate_symmetric_key(alg="ChaCha20/Poly1305", kid="enc")
+        cose = COSE()
+
+        def encrypt(key_data: bytes) -> list:
+            p = {1: enc_key.alg}
+            u = {4: enc_key.kid, 5: enc_key.generate_nonce()}
+            return cose.encode_and_encrypt(key_data, enc_key, p, u, out="cbor2/CBORTag").value
+
+        # alg: HS256, k: b"x" * 32
+        rest = dumps(3) + dumps(5) + dumps(-1) + dumps(b"x" * 32)
+
+        # {1: 4, 3: 5, -1: k}: without duplicates, the key can be decrypted and decoded.
+        assert EncryptedCOSEKey.to_cose_key(encrypt(b"\xa3" + dumps(1) + dumps(4) + rest), enc_key).kty == 4
+
+        # {1: 4, 1: 4, 3: 5, -1: k}: kty twice.
+        with pytest.raises(cbor2.CBORDecodeError):
+            EncryptedCOSEKey.to_cose_key(encrypt(b"\xa4" + dumps(1) + dumps(4) + dumps(1) + dumps(4) + rest), enc_key)
+            pytest.fail("to_cose_key() should fail.")
+
+    def test_recipient_from_list_with_duplicated_protected_header_label(self):
+        with pytest.raises(cbor2.CBORDecodeError):
+            Recipient.from_list([DUP_ALG_MAP, {}, b""])
+            pytest.fail("from_list() should fail.")
+
+    def test_cose_message_loads_with_duplicated_unprotected_header_label(self):
+        wire = b"\xd1\x84" + dumps(b"") + DUP_KID_MAP + dumps(b"payload") + dumps(b"tag")
+        with pytest.raises(cbor2.CBORDecodeError):
+            COSEMessage.loads(wire)
+            pytest.fail("loads() should fail.")