Merge branch 'main' into fix-generate-zero-length-protected-header-for-non-aead

Author: kentakayama

cwt/cose.py

@@ -396,12 +396,17 @@ def decode_with_headers(
         p, u = self._decode_headers(data.value[0], data.value[1])
         alg = p[1] if 1 in p else u.get(1, 0)
 
+        # Local variable `protected` is byte encoded protected header
+        # Sender is allowed to encode empty protected header into a bstr-wrapped zero-length map << {} >> (0x40A0)
+        # but Recipient MUST treat it as a zero-length byte string h'' (0x40) while decoding
+        protected = data.value[0] if len(p) > 0 else b""
+
         err: Exception = ValueError("key is not found.")
 
         # Encrypt0
         if data.tag == 16:
             kid = self._get_kid(p, u)
-            aad = self._dumps(["Encrypt0", data.value[0], external_aad])
+            aad = self._dumps(["Encrypt0", protected, external_aad])
             nonce = u.get(5, None)
             if kid:
                 for _, k in enumerate(keys):
@@ -436,7 +441,7 @@ def decode_with_headers(
         # MAC0
         if data.tag == 17:
             kid = self._get_kid(p, u)
-            msg = self._dumps(["MAC0", data.value[0], external_aad, payload])
+            msg = self._dumps(["MAC0", protected, external_aad, payload])
             if kid:
                 for _, k in enumerate(keys):
                     if k.kid != kid:
@@ -457,7 +462,7 @@ def decode_with_headers(
 
         # MAC
         if data.tag == 97:
-            to_be_maced = self._dumps(["MAC", data.value[0], external_aad, payload])
+            to_be_maced = self._dumps(["MAC", protected, external_aad, payload])
             rs = Recipients.from_list(data.value[4], self._verify_kid, context)
             mac_auth_key = rs.derive_key(keys, alg, external_aad, "Mac_Recipient")
             mac_auth_key.verify(to_be_maced, data.value[3])
@@ -466,7 +471,7 @@ def decode_with_headers(
         # Signature1
         if data.tag == 18:
             kid = self._get_kid(p, u)
-            to_be_signed = self._dumps(["Signature1", data.value[0], external_aad, payload])
+            to_be_signed = self._dumps(["Signature1", protected, external_aad, payload])
             if kid:
                 for _, k in enumerate(keys):
                     if k.kid != kid:
@@ -511,7 +516,7 @@ def decode_with_headers(
                         to_be_signed = self._dumps(
                             [
                                 "Signature",
-                                data.value[0],
+                                protected,
                                 sig[0],
                                 external_aad,
                                 payload,
@@ -527,7 +532,7 @@ def decode_with_headers(
                     to_be_signed = self._dumps(
                         [
                             "Signature",
-                            data.value[0],
+                            protected,
                             sig[0],
                             external_aad,
                             payload,

tests/test_algs_symmetric.py

@@ -877,7 +877,7 @@ def test_aesctr_key_decrypt_with_invalid_nonce(self):
         nonce = token_bytes(16)
         encrypted = key.encrypt(b"Hello world!", nonce=nonce)
         # alternate the nonce by incrementing the last byte
-        invalid_nonce = nonce[0:-1] + (nonce[-1] + 1 % 256).to_bytes(1, "big")
+        invalid_nonce = nonce[0:-1] + ((nonce[-1] + 1) % 256).to_bytes(1, "big")
         assert nonce != invalid_nonce
         decrypted = key.decrypt(encrypted, nonce=invalid_nonce)
         assert encrypted != decrypted
@@ -1035,7 +1035,7 @@ def test_aescbc_key_decrypt_with_invalid_nonce(self):
         nonce = token_bytes(16)
         encrypted = key.encrypt(b"Hello world!", nonce=nonce)
         # alternate the nonce by incrementing the last byte
-        invalid_nonce = nonce[0:-1] + (nonce[-1] + 1 % 256).to_bytes(1, "big")
+        invalid_nonce = nonce[0:-1] + ((nonce[-1] + 1) % 256).to_bytes(1, "big")
         assert nonce != invalid_nonce
         decrypted = key.decrypt(encrypted, nonce=invalid_nonce)
         assert encrypted != decrypted

tests/test_cose_sample_with_encode.py

@@ -1,6 +1,7 @@
 from secrets import token_bytes
 
 import pytest
+from cbor2 import dumps, loads
 
 from cwt import COSE, COSEAlgs, COSEHeaders, COSEKey, COSEMessage, Recipient, Signer
 
@@ -968,6 +969,17 @@ def test_cose_usage_examples_cose_signature1(self):
         )
         assert b"Hello world!" == recipient.decode(encoded3, pub_key)
 
+        # zero-length map protected header
+        encoded4 = sender.encode(
+            b"Hello world!",
+            priv_key,
+            unprotected={COSEHeaders.ALG: COSEAlgs.ES256, COSEHeaders.KID: b"01"},
+        )
+        loaded_encoded4 = loads(encoded4)
+        loaded_encoded4.value[0] = bytes.fromhex("a0")  # << {} >>
+        encoded4 = dumps(loaded_encoded4)
+        assert b"Hello world!" == recipient.decode(encoded4, pub_key)
+
     def test_cose_usage_examples_cose_signature1_countersignature(self):
         # The sender side:
         priv_key = COSEKey.from_jwk(