Merge pull request #444 from kentakayama/add-detached-content-converter

Add detached content converter

Author: dajiaji

cwt/cose_message.py

@@ -1,4 +1,6 @@
-from typing import Any, Dict, List, Optional, TypeVar
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional, Tuple, TypeVar
 
 from cbor2 import CBORTag, loads
 
@@ -98,6 +100,20 @@ def __init__(self, type: COSETypes, msg: List[Any]):
             raise ValueError(f"Invalid COSETypes({type}) for COSE message.")
         return
 
+    def __eq__(self: COSEMessage, other: object) -> bool:
+        if not isinstance(other, COSEMessage):
+            return NotImplemented
+        return (
+            self._type == other._type
+            and self._protected == other._protected
+            and self._unprotected == other._unprotected
+            and self._payload == other._payload
+            and self._other_fields == other._other_fields
+        )
+
+    def __ne__(self: COSEMessage, other: object) -> bool:
+        return not self.__eq__(other)
+
     @classmethod
     def loads(cls, msg: bytes):
         tagged = loads(msg)
@@ -266,8 +282,8 @@ def _validate_cose_message(self, msg: List[Any]):
             raise ValueError("The protected headers should be bytes.")
         if not isinstance(msg[1], dict):
             raise ValueError("The unprotected headers should be Dict[int, Any].")
-        if not isinstance(msg[2], bytes):
-            raise ValueError("The payload should be bytes.")
+        if not isinstance(msg[2], bytes) and msg[2] is not None:
+            raise ValueError("The payload should be bytes or null.")
 
         countersignatures = msg[1].get(11, None)
         if countersignatures is None:
@@ -287,3 +303,36 @@ def _validate_cose_message(self, msg: List[Any]):
     def _get_kid(self, sig: list) -> Optional[bytes]:
         kid = sig[1].get(4, None)
         return kid if kid else self._loads(sig[0]).get(4, None)
+
+    def detach_payload(self: Self) -> Tuple[COSEMessage, bytes]:
+        """
+        Detach a payload from the COSE message
+
+        Returns:
+            Tuple[COSEMessage, bytes]: A byte string of the encoded COSE or a
+                cbor2.CBORTag object, and a byte string of the detached payload.
+        Raises:
+            ValueError: The payload does not exist.
+        """
+
+        if not isinstance(self._payload, bytes):
+            raise ValueError("The payload does not exist.")
+
+        return COSEMessage(self._type, [self._msg[0], self._msg[1], None, *self._msg[3:]]), self._payload
+
+    def attach_payload(self: Self, payload: bytes) -> COSEMessage:
+        """
+        Attach a detached content to the COSE message
+
+        Args:
+            payload (bytes): A byte string of detached payload.
+        Returns:
+            COSEMessage: The COSE message (self).
+        Raises:
+            ValueError: The payload already exist.
+        """
+
+        if self._payload is not None:
+            raise ValueError("The payload already exist.")
+
+        return COSEMessage(self._type, [self._msg[0], self._msg[1], payload, *self._msg[3:]])

tests/test_cose_message.py

@@ -306,7 +306,6 @@ def test_cose_usage_examples_cose_mac_countersignature(self):
             (COSETypes.MAC0, [], "Invalid COSE message."),
             (COSETypes.MAC0, [{}, {}, b""], "The protected headers should be bytes."),
             (COSETypes.MAC0, [b"", b"", b""], "The unprotected headers should be Dict[int, Any]."),
-            (COSETypes.MAC0, [b"", {}, {}], "The payload should be bytes."),
             (COSETypes.MAC0, [b"", {11: {}}, b""], "The countersignature should be array."),
             (COSETypes.MAC0, [b"", {11: []}, b""], "Invalid countersignature."),
             (COSETypes.MAC0, [b"", {11: [b""]}, b""], "Invalid COSE message."),
@@ -461,3 +460,49 @@ def test_cose_message_counterverify_with_different_abbreviated_countersignature(
             COSEMessage.loads(countersigned).counterverify(pub_key)
             pytest.fail("counterverify() should not fail.")
         assert "Failed to verify." in str(err.value)
+
+    def test_cose_message_detach_and_attach(self):
+        """
+        Detach the payload from a COSE message.
+        For example, [an example message](https://github.com/cose-wg/Examples/blob/master/ecdsa-examples/ecdsa-sig-01.json)
+        ```
+        18([
+          / protected: / h'A201260300',
+          / unprotected: / {4: h'3131'},
+          / payload: / h'546869732069732074686520636F6E74656E742E',
+          / signature: / h'6520BBAF2081D7E0ED0F95F76EB0733D667005F7467CEC4B87B9381A6BA1EDE8E00DF29F32A37230F39A842A54821FDD223092819D7728EFB9D3A0080B75380B'
+        ])
+        ```
+        would be separated into
+        ```
+        18([
+          / protected: / h'A201260300',
+          / unprotected: / {4: h'3131'},
+          / payload: / null / detached /,
+          / signature: / h'6520BBAF2081D7E0ED0F95F76EB0733D667005F7467CEC4B87B9381A6BA1EDE8E00DF29F32A37230F39A842A54821FDD223092819D7728EFB9D3A0080B75380B'
+        ])
+        ```
+        and
+        ```
+        546869732069732074686520636F6E74656E742E
+        ```
+
+        """
+        ecdsa_cose_sign1_example = COSEMessage.loads(
+            bytes.fromhex(
+                "D28445A201260300A10442313154546869732069732074686520636F6E74656E742E58406520BBAF2081D7E0ED0F95F76EB0733D667005F7467CEC4B87B9381A6BA1EDE8E00DF29F32A37230F39A842A54821FDD223092819D7728EFB9D3A0080B75380B"
+            )
+        )
+        expected_detached_cose_message = COSEMessage.loads(
+            bytes.fromhex(
+                "D28445A201260300A104423131F658406520BBAF2081D7E0ED0F95F76EB0733D667005F7467CEC4B87B9381A6BA1EDE8E00DF29F32A37230F39A842A54821FDD223092819D7728EFB9D3A0080B75380B"
+            )
+        )
+        expected_payload = bytes.fromhex("546869732069732074686520636F6E74656E742E")
+
+        detached_content_cose_message, detached_payload = ecdsa_cose_sign1_example.detach_payload()
+        assert expected_detached_cose_message == detached_content_cose_message
+        assert expected_payload == detached_payload
+
+        reverted_cose_message = detached_content_cose_message.attach_payload(detached_payload)
+        assert ecdsa_cose_sign1_example == reverted_cose_message