Merge branch 'main' into follow-core-hpke-15

Author: dajiaji

CHANGES.rst

@@ -4,6 +4,40 @@ Changes
 Unreleased
 ----------
 
+Version 3.2.0
+-------------
+
+Released 2025-09-17
+
+- Fix a DeprecationWarning in a test util. `#637 <https://github.com/dajiaji/python-cwt/pull/637>`__ by @achamayou
+- Add a cwt.utils.ResolvedHeaders to allow tstr header labels and values. `#636 <https://github.com/dajiaji/python-cwt/pull/636>`__ by @achamayou
+- Typo fix. `#635 <https://github.com/dajiaji/python-cwt/pull/635>`__ by @achamayou
+- Prevent mutation of supp_pub.protected in to_cis. `#630 <https://github.com/dajiaji/python-cwt/pull/630>`__
+- Use enums on tests. `#625 <https://github.com/dajiaji/python-cwt/pull/625>`__
+- Update dependencies.
+    - Bump cryptography to 46.0.2. `#646 <https://github.com/dajiaji/python-cwt/pull/646>`__
+    - Bump cbor2 to 5.7.0. `#632 <https://github.com/dajiaji/python-cwt/pull/632>`__
+- Update dev dependencies.
+    - Bump pytest-cov to 7.0.0. `#644 <https://github.com/dajiaji/python-cwt/pull/644>`__
+    - Bump tox to 4.30.2. `#642 <https://github.com/dajiaji/python-cwt/pull/642>`__
+    - Bump pytest to 8.4.2. `#641 <https://github.com/dajiaji/python-cwt/pull/641>`__
+
+
+    - Bump pre-commit/black to 25.1.0. `#642 <https://github.com/dajiaji/python-cwt/pull/642>`__
+    - Bump pre-commit/blacken-docs to 1.19.1. `#642 <https://github.com/dajiaji/python-cwt/pull/642>`__
+    - Bump pre-commit/flake8 to 7.2.0. `#642 <https://github.com/dajiaji/python-cwt/pull/642>`__
+    - Bump pre-commit/isort to 6.0.1. `#642 <https://github.com/dajiaji/python-cwt/pull/642>`__
+    - Bump pre-commit/mirrors-mypy to 1.15.0. `#642 <https://github.com/dajiaji/python-cwt/pull/642>`__
+    - Bump pre-commit to 4.2.0. `#642 <https://github.com/dajiaji/python-cwt/pull/642>`__
+
+    - Bump pre-commit/black to 25.1.0. `#646 <https://github.com/dajiaji/python-cwt/pull/646>`__
+    - Bump pre-commit/blacken-docs to 1.19.1. `#646 <https://github.com/dajiaji/python-cwt/pull/646>`__
+    - Bump pre-commit/flake8 to 7.2.0. `#646 <https://github.com/dajiaji/python-cwt/pull/646>`__
+    - Bump pre-commit/isort to 6.0.1. `#646 <https://github.com/dajiaji/python-cwt/pull/646>`__
+    - Bump pre-commit/mirrors-mypy to 1.15.0. `#646 <https://github.com/dajiaji/python-cwt/pull/646>`__
+    - Bump pre-commit to 4.2.0. `#646 <https://github.com/dajiaji/python-cwt/pull/646>`__
+    - Bump requests to 2.32.4. `#646 <https://github.com/dajiaji/python-cwt/pull/646>`__
+
 Version 3.1.0
 -------------
 

README.md

@@ -54,6 +54,15 @@ assert b"Hello world!" == recipient.decode(encoded, mac_key)
 ## You can get decoded protected/unprotected headers with the payload as follows:
 # protected, unprotected, payload = recipient.decode_with_headers(encoded, mac_key)
 # assert b"Hello world!" == payload
+
+## Note that to pass header parameters with tstr labels, or tstr values, and avoid
+# clashes with short-string names such as "alg" or value encoding to bstr, you can
+# resolve the headers yourself, and pass a cwt.utils.ResolvedHeader({...}).
+#
+# For example:
+#   protected=cwt.utils.ResolvedHeader({
+#     "string label": "value"
+#   })
 ```
 
 **CWT API**

cwt/__init__.py

@@ -27,7 +27,7 @@
 from .recipient import Recipient
 from .signer import Signer
 
-__version__ = "3.1.0"
+__version__ = "3.2.0"
 __title__ = "cwt"
 __description__ = "A Python implementation of CWT/COSE"
 __url__ = "https://python-cwt.readthedocs.io"

cwt/cbor_processor.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict
+from typing import Any, Dict, Union
 
 from cbor2 import dumps, loads
 
@@ -12,7 +12,7 @@ def _dumps(self, obj: Any) -> bytes:
         except Exception as err:
             raise EncodeError("Failed to encode.") from err
 
-    def _loads(self, s: bytes) -> Dict[int, Any]:
+    def _loads(self, s: bytes) -> Dict[Union[str, int], Any]:
         try:
             return loads(s)
         except Exception as err:

cwt/cose.py

@@ -23,7 +23,7 @@
 from .recipient_interface import RecipientInterface
 from .recipients import Recipients
 from .signer import Signer
-from .utils import sort_keys_for_deterministic_encoding, to_cose_header
+from .utils import ResolvedHeader, sort_keys_for_deterministic_encoding, to_cose_header
 
 
 class COSE(CBORProcessor):
@@ -134,8 +134,8 @@ def encode(
         self,
         payload: bytes,
         key: Optional[COSEKeyInterface] = None,
-        protected: Optional[dict] = None,
-        unprotected: Optional[dict] = None,
+        protected: Optional[Union[dict, ResolvedHeader]] = None,
+        unprotected: Optional[Union[dict, ResolvedHeader]] = None,
         recipients: List[RecipientInterface] = [],
         signers: List[Signer] = [],
         external_aad: bytes = b"",
@@ -150,8 +150,8 @@ def encode(
         Args:
             payload (bytes): A content to be MACed, signed or encrypted.
             key (Optional[COSEKeyInterface]): A content encryption key as COSEKey.
-            protected (Optional[dict]): Parameters that are to be cryptographically protected.
-            unprotected (Optional[dict]): Parameters that are not cryptographically protected.
+            protected (Optional[Union[dict, ResolvedHeader]]): Parameters that are to be cryptographically protected.
+            unprotected (Optional[Union[dict, ResolvedHeader]]): Parameters that are not cryptographically protected.
             recipients (List[RecipientInterface]): A list of recipient information structures.
             signers (List[Signer]): A list of signer information objects for
                 multiple signer cases.
@@ -385,7 +385,7 @@ def decode_with_headers(
                 Since non-AEAD ciphers DO NOT provide neither authentication nor integrity
                 of decrypted message, make sure to validate them outside of this library.
         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.
+            Tuple[Dict[Union[str, int], Any], Dict[Union[str, 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:
             ValueError: Invalid arguments.
             DecodeError: Failed to decode data.
@@ -605,10 +605,10 @@ def decode_with_headers(
     def _encode_headers(
         self,
         key: Optional[COSEKeyInterface],
-        protected: Optional[dict],
-        unprotected: Optional[dict],
+        protected: Optional[Union[dict, ResolvedHeader]],
+        unprotected: Optional[Union[dict, ResolvedHeader]],
         enable_non_aead: bool,
-    ) -> Tuple[Dict[int, Any], Dict[int, Any]]:
+    ) -> Tuple[Dict[Union[str, int], Any], Dict[Union[str, int], Any]]:
         p = to_cose_header(protected)
         u = to_cose_header(unprotected)
         if key is not None:
@@ -635,30 +635,32 @@ def _encode_headers(
                 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]]:
-        p: Union[Dict[int, Any], bytes]
+    def _decode_headers(
+        self, protected: Any, unprotected: Any
+    ) -> Tuple[Dict[Union[str, int], Any], Dict[Union[str, int], Any]]:
+        p: Union[Dict[Union[str, int], Any], bytes]
         p = self._loads(protected) if protected else {}
         if isinstance(p, bytes):
             if len(p) > 0:
                 raise ValueError("Invalid protected header.")
             p = {}
-        u: Dict[int, Any] = unprotected
+        u: Dict[Union[str, int], Any] = unprotected
         if not isinstance(u, dict):
             raise ValueError("unprotected header should be dict.")
         return p, u
 
     def _validate_cose_message(
         self,
         key: Optional[COSEKeyInterface],
-        p: Dict[int, Any],
-        u: Dict[int, Any],
+        p: Dict[Union[str, int], Any],
+        u: Dict[Union[str, int], Any],
         recipients: List[RecipientInterface],
         signers: List[Signer],
     ) -> int:
         if len(recipients) > 0 and len(signers) > 0:
             raise ValueError("Both recipients and signers are specified.")
 
-        h: Dict[int, Any] = {}
+        h: Dict[Union[str, int], Any] = {}
         iv_count: int = 0
         for k, v in p.items():
             if k == 2:  # crit
@@ -780,8 +782,8 @@ def _encode_and_encrypt(
         self,
         payload: bytes,
         key: Optional[COSEKeyInterface],
-        p: Dict[int, Any],
-        u: Dict[int, Any],
+        p: Dict[Union[str, int], Any],
+        u: Dict[Union[str, int], Any],
         recipients: List[RecipientInterface],
         external_aad: bytes,
         out: str,
@@ -849,8 +851,8 @@ def _encode_and_mac(
         self,
         payload: bytes,
         key: Optional[COSEKeyInterface],
-        p: Dict[int, Any],
-        u: Dict[int, Any],
+        p: Dict[Union[str, int], Any],
+        u: Dict[Union[str, int], Any],
         recipients: List[RecipientInterface],
         external_aad: bytes,
         out: str,
@@ -895,8 +897,8 @@ def _encode_and_sign(
         self,
         payload: bytes,
         key: Optional[COSEKeyInterface],
-        p: Dict[int, Any],
-        u: Dict[int, Any],
+        p: Dict[Union[str, int], Any],
+        u: Dict[Union[str, int], Any],
         signers: List[Signer],
         external_aad: bytes,
         out: str,

cwt/cose_message.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 from cbor2 import CBORTag, loads
 
@@ -132,14 +132,14 @@ def type(self) -> COSETypes:
         return self._type
 
     @property
-    def protected(self) -> Dict[int, Any]:
+    def protected(self) -> Dict[Union[str, int], Any]:
         """
         The protected headers as a CBOR object.
         """
         return self._loads(self._protected)
 
     @property
-    def unprotected(self) -> Dict[int, Any]:
+    def unprotected(self) -> Dict[Union[str, int], Any]:
         """
         The unprotected headers as a CBOR object.
         """

cwt/cwt.py

@@ -316,7 +316,7 @@ def decode(
         data: bytes,
         keys: Union[COSEKeyInterface, List[COSEKeyInterface]],
         no_verify: bool = False,
-    ) -> Union[Dict[int, Any], bytes]:
+    ) -> Union[Dict[Union[str, int], Any], bytes]:
         """
         Verifies and decodes CWT.
 
@@ -333,11 +333,11 @@ def decode(
             DecodeError: Failed to decode the CWT.
             VerifyError: Failed to verify the CWT.
         """
-        cwt: Union[bytes, CBORTag, Dict[int, Any]] = self._loads(data)
+        cwt: Union[bytes, CBORTag, Dict[Union[str, int], Any]] = self._loads(data)
         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] = {}
+        p: Dict[Union[str, int], Any] = {}
         while isinstance(cwt, CBORTag):
             p, u, cwt = self._cose.decode_with_headers(cwt, keys)
             cwt = self._loads(cwt)
@@ -399,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], protected: Dict[int, Any] = {}):
+    def _verify(self, claims: Union[Dict[Union[str, int], Any], bytes], protected: Dict[Union[str, int], Any] = {}):
         if not isinstance(claims, dict):
             raise DecodeError("Failed to decode.")
 
@@ -484,7 +484,7 @@ def decode(
     data: bytes,
     keys: Union[COSEKeyInterface, List[COSEKeyInterface]],
     no_verify: bool = False,
-) -> Union[Dict[int, Any], bytes]:
+) -> Union[Dict[Union[str, int], Any], bytes]:
     return _cwt.decode(data, keys, no_verify)
 
 

cwt/recipient.py

@@ -20,7 +20,7 @@
 from .recipient_algs.ecdh_direct_hkdf import ECDH_DirectHKDF
 from .recipient_algs.hpke import HPKE
 from .recipient_interface import RecipientInterface
-from .utils import to_cose_header, to_recipient_context
+from .utils import ResolvedHeader, to_cose_header, to_recipient_context
 
 
 class Recipient:
@@ -31,8 +31,8 @@ class Recipient:
     @classmethod
     def new(
         cls,
-        protected: dict = {},
-        unprotected: dict = {},
+        protected: Union[dict, ResolvedHeader] = {},
+        unprotected: Union[dict, ResolvedHeader] = {},
         ciphertext: bytes = b"",
         recipients: List[Any] = [],
         sender_key: Optional[COSEKeyInterface] = None,
@@ -46,8 +46,8 @@ def new(
         Creates a recipient from a CBOR-like dictionary with numeric keys.
 
         Args:
-            protected (dict): Parameters that are to be cryptographically protected.
-            unprotected (dict): Parameters that are not cryptographically protected.
+            protected (Union[dict, ResolvedHeader]): Parameters that are to be cryptographically protected.
+            unprotected (Union[dict, ResolvedHeader]): Parameters that are not cryptographically protected.
             ciphertext (List[Any]): A cipher text.
             sender_key (Optional[COSEKeyInterface]): A sender private key as COSEKey.
             recipient_key (Optional[COSEKeyInterface]): A recipient public key as COSEKey.
@@ -80,7 +80,7 @@ def new(
         if alg == -6:
             return DirectKey(p, u)
         if alg in COSE_ALGORITHMS_KEY_WRAP.values():
-            if len(protected) > 0:
+            if len(p) > 0:
                 raise ValueError("The protected header must be a zero-length string in key wrap mode with an AE algorithm.")
             if not sender_key:
                 sender_key = COSEKey.from_symmetric_key(alg=alg)

cwt/recipient_algs/aes_key_wrap.py

@@ -15,7 +15,7 @@ class AESKeyWrap(RecipientInterface):
 
     def __init__(
         self,
-        unprotected: Dict[int, Any],
+        unprotected: Dict[Union[str, int], Any],
         ciphertext: bytes = b"",
         recipients: List[Any] = [],
         sender_key: Optional[COSEKeyInterface] = None,

cwt/recipient_algs/direct.py

@@ -1,13 +1,13 @@
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Union
 
 from ..recipient_interface import RecipientInterface
 
 
 class Direct(RecipientInterface):
     def __init__(
         self,
-        protected: Dict[int, Any],
-        unprotected: Dict[int, Any],
+        protected: Dict[Union[str, int], Any],
+        unprotected: Dict[Union[str, int], Any],
         ciphertext: bytes = b"",
         recipients: List[Any] = [],
     ):