diff --git a/CHANGES.rst b/CHANGES.rst index bc5bb61f..0d5261a9 100644 --- a/CHANGES.rst +++ b/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. diff --git a/cwt/cbor_processor.py b/cwt/cbor_processor.py index db58d3c0..2613ac29 100644 --- a/cwt/cbor_processor.py +++ b/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 diff --git a/cwt/cose_key.py b/cwt/cose_key.py index 80278739..853c5978 100644 --- a/cwt/cose_key.py +++ b/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 diff --git a/cwt/cose_message.py b/cwt/cose_message.py index 969b912f..b29e65fd 100644 --- a/cwt/cose_message.py +++ b/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) diff --git a/cwt/encrypted_cose_key.py b/cwt/encrypted_cose_key.py index 9b35c212..c534b79e 100644 --- a/cwt/encrypted_cose_key.py +++ b/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) diff --git a/cwt/recipient.py b/cwt/recipient.py index 630b6207..759088a6 100644 --- a/cwt/recipient.py +++ b/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]) diff --git a/poetry.lock b/poetry.lock index c3b7d95f..b0f01bf7 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1327,4 +1327,4 @@ python-discovery = ">=1.6.1" [metadata] lock-version = "2.1" python-versions = "^3.11,<4.0" -content-hash = "5594589275105923500589b551081547f19691e515f0396b2973f51811cdbf16" +content-hash = "efa333d7076bdfca0729b804876225807d5c570126a27a6cfcdba89b21a8bfee" diff --git a/pyproject.toml b/pyproject.toml index 4945f792..a3137525 100644 --- a/pyproject.toml +++ b/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" diff --git a/tests/test_cose.py b/tests/test_cose.py index 7c6bb789..08362161 100644 --- a/tests/test_cose.py +++ b/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.", diff --git a/tests/test_duplicate_map_keys.py b/tests/test_duplicate_map_keys.py new file mode 100644 index 00000000..0804fd78 --- /dev/null +++ b/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.")