Merge pull request #400 from dajiaji/add-support-for-cwt-claims-in-he…

…aders

Add experimental support for CWT claims in headers.

README.md

@@ -14,6 +14,7 @@ implementation compliant with:
 - [RFC9338: CBOR Object Signing and Encryption (COSE): Countersignatures](https://www.rfc-editor.org/rfc/rfc9338.html) - experimental
 - [RFC8392: CWT (CBOR Web Token)](https://tools.ietf.org/html/rfc8392)
 - [draft-04: Use of HPKE with COSE](https://www.ietf.org/archive/id/draft-ietf-cose-hpke-04.html) - experimental
+- [draft-05: CWT Claims in COSE Headers](https://www.ietf.org/archive/id/draft-ietf-cose-cwt-claims-in-headers-05.html) - experimental
 - and related various specifications. See [Referenced Specifications](#referenced-specifications).
 
 It is designed to make users who already know about [JWS](https://tools.ietf.org/html/rfc7515)/[JWE](https://tools.ietf.org/html/rfc7516)/[JWT](https://tools.ietf.org/html/rfc7519)
@@ -1674,6 +1675,7 @@ Python CWT is (partially) compliant with following specifications:
 - [RFC8230: Using RSA Algorithms with COSE Messages](https://tools.ietf.org/html/rfc8230)
 - [RFC8152: CBOR Object Signing and Encryption (COSE)](https://tools.ietf.org/html/rfc8152)
 - [draft-04: Use of HPKE with COSE](https://www.ietf.org/archive/id/draft-ietf-cose-hpke-04.html) - experimental
+- [draft-05: CWT Claims in COSE Headers](https://www.ietf.org/archive/id/draft-ietf-cose-cwt-claims-in-headers-05.html) - experimental
 - [Electronic Health Certificate Specification](https://github.com/ehn-dcc-development/hcert-spec/blob/main/hcert_spec.md)
 - [Technical Specifications for Digital Green Certificates Volume 1](https://ec.europa.eu/health/sites/default/files/ehealth/docs/digital-green-certificates_v1_en.pdf)
 

cwt/cwt.py

@@ -337,11 +337,12 @@ def decode(
         if isinstance(cwt, CBORTag) and cwt.tag == CWT.CBOR_TAG:
             cwt = cwt.value
         keys = [keys] if isinstance(keys, COSEKeyInterface) else keys
+        p: Dict[int, Any] = {}
         while isinstance(cwt, CBORTag):
-            cwt = self._cose.decode(cwt, keys)
+            p, u, cwt = self._cose.decode_with_headers(cwt, keys)
             cwt = self._loads(cwt)
         if not no_verify:
-            self._verify(cwt)
+            self._verify(cwt, p)
         return cwt
 
     def set_private_claim_names(self, claim_names: Dict[str, int]):
@@ -398,7 +399,7 @@ def _validate(self, claims: Union[Dict[int, Any], bytes]):
         Claims.validate(claims)
         return
 
-    def _verify(self, claims: Union[Dict[int, Any], bytes]):
+    def _verify(self, claims: Union[Dict[int, Any], bytes], protected: Dict[int, Any] = {}):
         if not isinstance(claims, dict):
             raise DecodeError("Failed to decode.")
 
@@ -416,6 +417,11 @@ def _verify(self, claims: Union[Dict[int, Any], bytes]):
                     raise VerifyError("The token is not yet valid.")
             else:
                 raise ValueError("nbf should be int or float.")
+
+        if 13 in protected:  # CWT claims in protected headers
+            for k, v in protected[13].items():
+                if k in claims and claims[k] != v:
+                    raise VerifyError(f"The CWT claim({k}) value in protected header does not match the values in the payload.")
         return
 
     def _set_default_value(self, claims: Union[Dict[int, Any], bytes]):

tests/test_cwt.py

@@ -10,9 +10,9 @@
 
 import cbor2
 import pytest
-from cbor2 import CBORTag
+from cbor2 import CBORTag, dumps
 
-from cwt import CWT, Claims, COSEKey, DecodeError, Recipient, VerifyError
+from cwt import COSE, CWT, Claims, COSEKey, DecodeError, Recipient, VerifyError
 from cwt.cose_key_interface import COSEKeyInterface
 from cwt.signer import Signer
 
@@ -605,3 +605,37 @@ def test_cwt__verify_with_invalid_args(self, ctx, invalid, msg):
             ctx._verify(invalid)
             pytest.fail("_verify should fail.")
         assert msg in str(err.value)
+
+    def test_cwt_decode_with_cwt_claims_in_headers(self, ctx):
+        cose = COSE.new(alg_auto_inclusion=True, kid_auto_inclusion=True)
+        payload = dumps({1: "https://as.example", 2: "someone"})
+        protected = {13: {1: "https://as.example"}}
+        mac_key = COSEKey.from_symmetric_key(alg="HS256", kid="01")
+        cwt = cose.encode_and_mac(payload, mac_key, protected)
+        decoded = ctx.decode(cwt, mac_key)
+        assert 1 in decoded and decoded[1] == "https://as.example"
+        assert 2 in decoded and decoded[2] == "someone"
+
+    @pytest.mark.parametrize(
+        "invalid, msg",
+        [
+            (
+                {1: "https://asx.example", 2: "someone"},
+                "The CWT claim(1) value in protected header does not match the values in the payload.",
+            ),
+            (
+                {1: "https://as.example", 2: "someonex"},
+                "The CWT claim(2) value in protected header does not match the values in the payload.",
+            ),
+        ],
+    )
+    def test_cwt_decode_with_invalid_cwt_claims_in_headers(self, ctx, invalid, msg):
+        cose = COSE.new(alg_auto_inclusion=True, kid_auto_inclusion=True)
+        payload = dumps({1: "https://as.example", 2: "someone"})
+        protected = {13: invalid}
+        mac_key = COSEKey.from_symmetric_key(alg="HS256", kid="01")
+        cwt = cose.encode_and_mac(payload, mac_key, protected)
+        with pytest.raises(VerifyError) as err:
+            ctx.decode(cwt, mac_key)
+            pytest.fail("verify should fail.")
+        assert msg in str(err.value)