Generate zero-length protected header for non AEAD

cwt/const.py

@@ -118,14 +118,9 @@
     "verifyMAC": 10,  # JWK-like lowerCamelCase
 }
 
-# COSE Algorithms for Content Encryption Key (CEK).
-COSE_ALGORITHMS_CEK = {
-    "A128CTR": -65534,  # AES-CTR mode w/ 128-bit key (Deprecated)
-    "A192CTR": -65533,  # AES-CTR mode w/ 192-bit key (Deprecated)
-    "A256CTR": -65532,  # AES-CTR mode w/ 256-bit key (Deprecated)
-    "A128CBC": -65531,  # AES-CBC mode w/ 128-bit key (Deprecated)
-    "A192CBC": -65530,  # AES-CBC mode w/ 192-bit key (Deprecated)
-    "A256CBC": -65529,  # AES-CBC mode w/ 256-bit key (Deprecated)
+
+# COSE AEAD Algorithms
+COSE_ALGORITHMS_CEK_AEAD = {
     "A128GCM": 1,  # AES-GCM mode w/ 128-bit key, 128-bit tag
     "A192GCM": 2,  # AES-GCM mode w/ 192-bit key, 128-bit tag
     "A256GCM": 3,  # AES-GCM mode w/ 256-bit key, 128-bit tag
@@ -141,6 +136,22 @@
     # etc.
 }
 
+# COSE non AEAD Algorithms defined in RFC9459
+COSE_ALGORITHMS_CEK_NON_AEAD = {
+    "A128CTR": -65534,  # AES-CTR mode w/ 128-bit key (Deprecated)
+    "A192CTR": -65533,  # AES-CTR mode w/ 192-bit key (Deprecated)
+    "A256CTR": -65532,  # AES-CTR mode w/ 256-bit key (Deprecated)
+    "A128CBC": -65531,  # AES-CBC mode w/ 128-bit key (Deprecated)
+    "A192CBC": -65530,  # AES-CBC mode w/ 192-bit key (Deprecated)
+    "A256CBC": -65529,  # AES-CBC mode w/ 256-bit key (Deprecated)
+}
+
+# COSE Algorithms for Content Encryption Key (CEK).
+COSE_ALGORITHMS_CEK = {
+    **COSE_ALGORITHMS_CEK_AEAD,
+    **COSE_ALGORITHMS_CEK_NON_AEAD,
+}
+
 COSE_KEY_LEN = {
     -65534: 128,  # AES-CTR w/ 128-bit key (Deprecated)
     -65533: 192,  # AES-CTR w/ 192-bit key (Deprecated)

cwt/cose.py

@@ -6,6 +6,7 @@
 from .cbor_processor import CBORProcessor
 from .const import (
     COSE_ALGORITHMS_CEK,
+    COSE_ALGORITHMS_CEK_NON_AEAD,
     COSE_ALGORITHMS_CKDM,
     COSE_ALGORITHMS_CKDM_KEY_AGREEMENT,
     COSE_ALGORITHMS_CKDM_KEY_AGREEMENT_DIRECT,
@@ -393,7 +394,7 @@ def decode_with_headers(
         # if not isinstance(unprotected, dict):
         #     raise ValueError("unprotected header should be dict.")
         p, u = self._decode_headers(data.value[0], data.value[1])
-        alg = self._get_alg(p)
+        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)
@@ -553,9 +554,18 @@ def _encode_headers(
         u = to_cose_header(unprotected)
         if key is not None:
             if self._alg_auto_inclusion:
-                p[1] = key.alg
+                if key.alg in COSE_ALGORITHMS_CEK_NON_AEAD.values():
+                    u[1] = key.alg
+                else:
+                    p[1] = key.alg
             if self._kid_auto_inclusion and key.kid:
                 u[4] = key.kid
+
+        # Check the protected header is empty if the algorithm is non AEAD (AES-CBC or AES-CTR)
+        # because section 4 of RFC9459 says "The 'protected' header MUST be a zero-length byte string."
+        alg = p[1] if 1 in p else u.get(1, 0)
+        if alg in COSE_ALGORITHMS_CEK_NON_AEAD.values() and len(p) > 0:
+            raise ValueError("protected header MUST be zero-length")
         return p, u
 
     def _decode_headers(self, protected: Any, unprotected: Any) -> Tuple[Dict[int, Any], Dict[int, Any]]:

tests/test_recipient.py

@@ -797,6 +797,25 @@ def test_recipients_aes(self, kw_alg, enc_alg):
         kw_key = COSEKey.from_symmetric_key(alg=kw_alg)
         enc_key = COSEKey.from_symmetric_key(alg=enc_alg)
 
+        # The sender side (must fail):
+        with pytest.raises(ValueError) as err:
+            r = Recipient.new(protected={"alg": kw_alg}, sender_key=kw_key)
+            pytest.fail("encode_and_encrypt() should fail.")
+        assert "The protected header must be a zero-length string in key wrap mode with an AE algorithm." in str(err.value)
+
+        # The sender side (must fail):
+        r = Recipient.new(unprotected={"alg": kw_alg}, sender_key=kw_key)
+        sender = COSE.new(alg_auto_inclusion=True)
+        with pytest.raises(ValueError) as err:
+            encoded = sender.encode_and_encrypt(
+                b"Hello world!",
+                enc_key,
+                protected={"kid": "actually-not-protected"},
+                recipients=[r],
+            )
+            pytest.fail("encode_and_encrypt() should fail.")
+        assert "protected header MUST be zero-length" in str(err.value)
+
         # The sender side:
         r = Recipient.new(unprotected={"alg": kw_alg}, sender_key=kw_key)
         sender = COSE.new(alg_auto_inclusion=True)
@@ -832,13 +851,13 @@ def test_recipients_hpke(self, rsk1, rsk2, enc_alg):
                 "y": "BGU5soLgsu_y7GN2I3EPUXS9EZ7Sw0qif-V70JtInFI",
             }
         )
-        r = Recipient.new(protected={1: 35}, recipient_key=rpk)
+        r = Recipient.new(unprotected={1: 35}, recipient_key=rpk)
         r.encode(enc_key.key)
         sender = COSE.new()
         encoded = sender.encode_and_encrypt(
             b"This is the content.",
             enc_key,
-            protected={"alg": enc_alg},
+            unprotected={"alg": enc_alg},
             recipients=[r],
         )
         recipient = COSE.new()
@@ -861,7 +880,7 @@ def test_recipients_ecdh_es(self, key_agreement_alg, key_agreement_alg_id, kw_al
             "alg": kw_alg,
             "supp_pub": {
                 "key_data_length": len(enc_key.key) * 8,
-                "protected": {1: key_agreement_alg_id},
+                "protected": {},
             },
         }
 
@@ -886,15 +905,15 @@ def test_recipients_ecdh_es(self, key_agreement_alg, key_agreement_alg_id, kw_al
                 "y": "BGU5soLgsu_y7GN2I3EPUXS9EZ7Sw0qif-V70JtInFI",
             }
         )
-        r = Recipient.new(protected={"alg": key_agreement_alg}, sender_key=rsk1, recipient_key=rpk2, context=context)
+        r = Recipient.new(unprotected={"alg": key_agreement_alg}, sender_key=rsk1, recipient_key=rpk2, context=context)
 
         nonce = enc_key.generate_nonce()
         sender = COSE.new()
         encoded = sender.encode(
             b"Hello world!",
             enc_key,
-            protected={"alg": enc_alg},
-            unprotected={"iv": nonce},
+            protected={},
+            unprotected={"alg": enc_alg, "iv": nonce},
             recipients=[r],
         )