Merge pull request #451 from dajiaji/refine-handling-detached-payload

Add detached_payload parameter to sign/verify funcs.

Author: dajiaji

cwt/cose.py

@@ -290,6 +290,7 @@ def decode(
         keys: Union[COSEKeyInterface, List[COSEKeyInterface]],
         context: Optional[Union[Dict[str, Any], List[Any]]] = None,
         external_aad: bytes = b"",
+        detached_payload: Optional[bytes] = None,
     ) -> bytes:
         """
         Verifies and decodes COSE data, and returns only payload.
@@ -303,14 +304,15 @@ def decode(
                 structure for key deriviation functions.
             external_aad(bytes): External additional authenticated data supplied by
                 application.
+            detached_payload (Optional[bytes]): The detached payload that should be verified with data.
         Returns:
             bytes: A byte string of decoded payload.
         Raises:
             ValueError: Invalid arguments.
             DecodeError: Failed to decode data.
             VerifyError: Failed to verify data.
         """
-        _, _, res = self.decode_with_headers(data, keys, context, external_aad)
+        _, _, res = self.decode_with_headers(data, keys, context, external_aad, detached_payload)
         return res
 
     def decode_with_headers(
@@ -319,6 +321,7 @@ def decode_with_headers(
         keys: Union[COSEKeyInterface, List[COSEKeyInterface]],
         context: Optional[Union[Dict[str, Any], List[Any]]] = None,
         external_aad: bytes = b"",
+        detached_payload: Optional[bytes] = None,
     ) -> Tuple[Dict[int, Any], Dict[int, Any], bytes]:
         """
         Verifies and decodes COSE data, and returns protected headers, unprotected headers and payload.
@@ -332,6 +335,7 @@ def decode_with_headers(
                 structure for key deriviation functions.
             external_aad(bytes): External additional authenticated data supplied by
                 application.
+            detached_payload (Optional[bytes]): The detached payload that should be verified with data.
         Returns:
             Tuple[Dict[int, Any], Dict[int, Any], bytes]: A dictionary data of decoded protected headers, and a dictionary data of unprotected headers, and a byte string of decoded payload.
         Raises:
@@ -376,6 +380,14 @@ def decode_with_headers(
         else:
             raise ValueError(f"Unsupported or unknown CBOR tag({data.tag}).")
 
+        payload = data.value[2]
+        if detached_payload is not None:
+            if data.value[2] is not None:
+                raise ValueError("The payload already exists.")
+            payload = detached_payload
+        if payload is None:
+            raise ValueError("detached_payload should be set.")
+
         # protected: Union[Dict[int, Any], bytes] = self._loads(data.value[0]) if data.value[0] else b""
         # unprotected = data.value[1]
         # if not isinstance(unprotected, dict):
@@ -396,18 +408,18 @@ def decode_with_headers(
                         continue
                     try:
                         if not isinstance(p, bytes) and alg in COSE_ALGORITHMS_HPKE.values():  # HPKE
-                            hpke = HPKE(p, u, data.value[2])
+                            hpke = HPKE(p, u, payload)
                             res = hpke.decode(k, aad)
                             if not isinstance(res, bytes):
                                 raise TypeError("Internal type error.")
                             return p, u, res
-                        return p, u, k.decrypt(data.value[2], nonce, aad)
+                        return p, u, k.decrypt(payload, nonce, aad)
                     except Exception as e:
                         err = e
                 raise err
             for _, k in enumerate(keys):
                 try:
-                    return p, u, k.decrypt(data.value[2], nonce, aad)
+                    return p, u, k.decrypt(payload, nonce, aad)
                 except Exception as e:
                     err = e
             raise err
@@ -418,42 +430,42 @@ def decode_with_headers(
             nonce = u.get(5, b"")
             enc_key = rs.derive_key(keys, alg, external_aad, "Enc_Recipient")
             aad = self._dumps(["Encrypt", data.value[0], external_aad])
-            return p, u, enc_key.decrypt(data.value[2], nonce, aad)
+            return p, u, enc_key.decrypt(payload, nonce, aad)
 
         # MAC0
         if data.tag == 17:
             kid = self._get_kid(p, u)
-            msg = self._dumps(["MAC0", data.value[0], external_aad, data.value[2]])
+            msg = self._dumps(["MAC0", data.value[0], external_aad, payload])
             if kid:
                 for _, k in enumerate(keys):
                     if k.kid != kid:
                         continue
                     try:
                         k.verify(msg, data.value[3])
-                        return p, u, data.value[2]
+                        return p, u, payload
                     except Exception as e:
                         err = e
                 raise err
             for _, k in enumerate(keys):
                 try:
                     k.verify(msg, data.value[3])
-                    return p, u, data.value[2]
+                    return p, u, payload
                 except Exception as e:
                     err = e
             raise err
 
         # MAC
         if data.tag == 97:
-            to_be_maced = self._dumps(["MAC", data.value[0], external_aad, data.value[2]])
+            to_be_maced = self._dumps(["MAC", data.value[0], 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])
-            return p, u, data.value[2]
+            return p, u, payload
 
         # Signature1
         if data.tag == 18:
             kid = self._get_kid(p, u)
-            to_be_signed = self._dumps(["Signature1", data.value[0], external_aad, data.value[2]])
+            to_be_signed = self._dumps(["Signature1", data.value[0], external_aad, payload])
             if kid:
                 for _, k in enumerate(keys):
                     if k.kid != kid:
@@ -462,7 +474,7 @@ def decode_with_headers(
                         if self._ca_certs:
                             k.validate_certificate(self._ca_certs)
                         k.verify(to_be_signed, data.value[3])
-                        return p, u, data.value[2]
+                        return p, u, payload
                     except Exception as e:
                         err = e
                 raise err
@@ -471,7 +483,7 @@ def decode_with_headers(
                     if self._ca_certs:
                         k.validate_certificate(self._ca_certs)
                     k.verify(to_be_signed, data.value[3])
-                    return p, u, data.value[2]
+                    return p, u, payload
                 except Exception as e:
                     err = e
             raise err
@@ -501,11 +513,11 @@ def decode_with_headers(
                                 data.value[0],
                                 sig[0],
                                 external_aad,
-                                data.value[2],
+                                payload,
                             ]
                         )
                         k.verify(to_be_signed, sig[2])
-                        return p, u, data.value[2]
+                        return p, u, payload
                     except Exception as e:
                         err = e
                 continue
@@ -517,11 +529,11 @@ def decode_with_headers(
                             data.value[0],
                             sig[0],
                             external_aad,
-                            data.value[2],
+                            payload,
                         ]
                     )
                     k.verify(to_be_signed, sig[2])
-                    return p, u, data.value[2]
+                    return p, u, payload
                 except Exception as e:
                     err = e
         raise err

cwt/cose_message.py

@@ -29,7 +29,7 @@ def __init__(self, type: COSETypes, msg: List[Any]):
         self._type = type
         self._protected = msg[0]
         self._unprotected = msg[1]
-        self._payload = msg[2]
+        # self._payload = msg[2]  # msg[2] is mutable and has no readable alias to avoid complexity.
         self._other_fields: List[bytes] = []
         self._recipients: List[List[Any]] = []
         self._signatures: List[List[Any]] = []
@@ -150,7 +150,7 @@ def payload(self) -> bytes:
         """
         The payload of the COSE message.
         """
-        return self._payload
+        return self._msg[2]
 
     @property
     def other_fields(self) -> List[bytes]:
@@ -180,7 +180,14 @@ def dumps(self) -> bytes:
         tag = COSE_TYPE_TO_TAG.get(self._type, -1)
         return self._dumps(CBORTag(tag, self._msg)) if tag > 0 else self._dumps(self._msg)
 
-    def countersign(self, signer: Signer, aad: bytes = b"", abbreviated: bool = False, tagged: bool = False) -> COSEMessage:
+    def countersign(
+        self,
+        signer: Signer,
+        aad: bytes = b"",
+        abbreviated: bool = False,
+        tagged: bool = False,
+        detached_payload: Optional[bytes] = None,
+    ) -> COSEMessage:
         """
         Countersigns to the COSE message with the signer specified.
 
@@ -189,21 +196,29 @@ def countersign(self, signer: Signer, aad: bytes = b"", abbreviated: bool = Fals
             aad (bytes): The application supplied additional authenticated data.
             abbreviated(bool): The type of the countersignature (abbreviated or not).
             tagged(bool): The indicator whether the countersignature is tagged or not.
+            detached_payload (Optional[bytes]): The detached payload that should be
+                countersigned with the COSEMessage.
         Returns:
             COSEMessage: The COSE message (self).
         Raises:
             ValueError: Invalid arguments.
             EncodeError: Failed to countersign.
         """
+        payload = self._msg[2]
+        if detached_payload is not None:
+            if self._msg[2] is not None:
+                raise ValueError("The payload already exists.")
+            payload = detached_payload
+
         if abbreviated:
-            to_be_signed = ["CounterSignature0V2", self._protected, aad, self._payload]
+            to_be_signed = ["CounterSignature0V2", self._protected, aad, payload]
             for other_field in self._other_fields:
                 to_be_signed.append(other_field)
             signer.sign(self._dumps(to_be_signed))
             self._unprotected[12] = signer.signature
             return self
 
-        to_be_signed = ["CounterSignatureV2", self._protected, signer.protected, aad, self._payload]
+        to_be_signed = ["CounterSignatureV2", self._protected, signer.protected, aad, payload]
         for other_field in self._other_fields:
             to_be_signed.append(other_field)
         signer.sign(self._dumps(to_be_signed))
@@ -217,23 +232,36 @@ def countersign(self, signer: Signer, aad: bytes = b"", abbreviated: bool = Fals
         self._unprotected[11].append([signer.protected, signer.unprotected, signer.signature])
         return self
 
-    def counterverify(self, key: COSEKeyInterface, aad: bytes = b"") -> Optional[List[Any]]:
+    def counterverify(
+        self,
+        key: COSEKeyInterface,
+        aad: bytes = b"",
+        detached_payload: Optional[bytes] = None,
+    ) -> Optional[List[Any]]:
         """
         Verifies a countersignature in the COSE message with the verification key specified.
 
         Args:
             key(COSEKeyInterface): A COSEKey that is used to verify a signature in the COSE message.
             aad (bytes): The application supplied additional authenticated data.
+            detached_payload (Optional[bytes]): The detached payload that should be
+                counterverified with the COSEMessage.
         Returns:
             Optional[List[Any]]: The COSE signature verified.
         Raises:
             ValueError: Invalid arguments.
             VerifyError: Failed to verify.
         """
+        payload = self._msg[2]
+        if detached_payload is not None:
+            if self._msg[2] is not None:
+                raise ValueError("The payload already exists.")
+            payload = detached_payload
+
         err: Exception = ValueError("Countersignature not found.")
         acs = self._unprotected.get(12, None)
         if acs:
-            to_be_signed = ["CounterSignature0V2", self._protected, aad, self._payload]
+            to_be_signed = ["CounterSignature0V2", self._protected, aad, payload]
             for other_field in self._other_fields:
                 to_be_signed.append(other_field)
             try:
@@ -245,7 +273,7 @@ def counterverify(self, key: COSEKeyInterface, aad: bytes = b"") -> Optional[Lis
         cs = self._unprotected.get(11, None)
         if not cs:
             raise err
-        to_be_signed = ["CounterSignatureV2", self._protected, b"", aad, self._payload]
+        to_be_signed = ["CounterSignatureV2", self._protected, b"", aad, payload]
         for other_field in self._other_fields:
             to_be_signed.append(other_field)
         if isinstance(cs[0], bytes):
@@ -310,8 +338,9 @@ def detach_payload(self) -> Tuple[COSEMessage, bytes]:
         if self._msg[2] is None:
             raise ValueError("The payload does not exist.")
 
+        payload = self._msg[2]
         self._msg[2] = None
-        return self, self._payload
+        return self, payload
 
     def attach_payload(self, payload: bytes) -> COSEMessage:
         """
@@ -326,7 +355,6 @@ def attach_payload(self, payload: bytes) -> COSEMessage:
         """
 
         if self._msg[2] is not None:
-            raise ValueError("The payload already exist.")
-        self._payload = payload
-
+            raise ValueError("The payload already exists.")
+        self._msg[2] = payload
         return self