Bump pyhpke to v0.4.0.

Bump pyhpke to v0.4.0.

Author: dajiaji

cwt/__init__.py

@@ -13,7 +13,6 @@
 from .encrypted_cose_key import EncryptedCOSEKey
 from .exceptions import CWTError, DecodeError, EncodeError, VerifyError
 from .helpers.hcert import load_pem_hcert_dsc
-from .hpke_cipher_suite import HPKECipherSuite
 from .recipient import Recipient
 from .signer import Signer
 

cwt/algs/ec2.py

@@ -1,7 +1,6 @@
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, List, Optional, Union
 
 import cryptography
-import pyhpke
 from cryptography.hazmat.primitives import hashes
 from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.hazmat.primitives.asymmetric.ec import (
@@ -24,8 +23,7 @@
     COSE_KEY_TYPES,
 )
 from ..cose_key_interface import COSEKeyInterface
-from ..exceptions import DecodeError, EncodeError, VerifyError
-from ..hpke_cipher_suite import HPKECipherSuite
+from ..exceptions import EncodeError, VerifyError
 from ..utils import i2osp, os2ip, to_cis
 from .asymmetric import AsymmetricKey
 from .symmetric import AESCCMKey, AESGCMKey, ChaCha20Key, HMACKey
@@ -275,28 +273,6 @@ def verify(self, msg: bytes, sig: bytes):
         except ValueError as err:
             raise VerifyError("Invalid signature.") from err
 
-    def seal(self, suite: HPKECipherSuite, msg: bytes, aad: bytes = b"") -> Tuple[bytes, bytes]:
-        # if self._alg != -1:
-        #     raise ValueError("alg should be HPKE(-1).");
-        if self._private_key is not None:
-            raise ValueError("Private key cannot be used for HPKE encryption.")
-
-        ctx = self._create_hpke_context(suite)
-        enc, sender = ctx.setup_send(self._public_key, b"")  # TODO: Add support for info
-        return enc, sender.aead.seal(aad, msg)
-
-    def open(self, suite: HPKECipherSuite, enc: bytes, msg: bytes, aad: bytes = b"") -> bytes:
-        # if self._alg != -1:
-        #     raise ValueError("alg should be HPKE(-1).");
-        if self._private_key is None:
-            raise ValueError("Public key cannot be used for HPKE decryption.")
-        ctx = self._create_hpke_context(suite)
-        try:
-            recipient = ctx.setup_recv(enc, self._private_key, b"")  # TODO: Add support for info
-            return recipient.aead.open(aad, msg)
-        except Exception as err:
-            raise DecodeError("Failed to decrypt.") from err
-
     def derive_key(
         self,
         context: Union[List[Any], Dict[str, Any]],
@@ -355,26 +331,3 @@ def _os_to_der(self, key_size: int, sig: bytes) -> bytes:
         r = os2ip(sig[:num_bytes])
         s = os2ip(sig[num_bytes:])
         return encode_dss_signature(r, s)
-
-    def _create_hpke_context(self, suite: HPKECipherSuite) -> Any:
-        if self._crv == 1:
-            if suite.kem != 0x0010:
-                raise ValueError("KEM id should be 0x0010.")
-            if suite.kdf == 0x0001 and suite.aead == 0x0001:
-                return pyhpke.Suite__DHKEM_P256_HKDF_SHA256__HKDF_SHA256__AES_128_GCM
-            if suite.kdf == 0x0001 and suite.aead == 0x0003:
-                return pyhpke.Suite__DHKEM_P256_HKDF_SHA256__HKDF_SHA256__ChaCha20Poly1305
-            if suite.kdf == 0x0003 and suite.aead == 0x0001:
-                return pyhpke.Suite__DHKEM_P256_HKDF_SHA256__HKDF_SHA512__AES_128_GCM
-            raise ValueError("Unsupported kdf/aead combination.")
-        elif self._crv == 2:
-            if suite.kem != 0x0011:
-                raise ValueError("KEM id should be 0x0011.")
-            raise ValueError("Unsupported kdf/aead combination.")
-        elif self._crv == 3:
-            if suite.kem != 0x0012:
-                raise ValueError("KEM id should be 0x0012.")
-            if suite.kdf == 0x0001 and suite.aead == 0x0001:
-                return pyhpke.Suite__DHKEM_P521_HKDF_SHA512__HKDF_SHA512__AES_256_GCM
-            raise ValueError("Unsupported kdf/aead combination.")
-        raise ValueError("Invalid crv for HPKE.")

cwt/cose_key_interface.py

@@ -1,12 +1,11 @@
-from typing import Any, Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, List, Optional, Union
 
 from .cbor_processor import CBORProcessor
 from .const import (
     COSE_KEY_OPERATION_VALUES,
     COSE_KEY_TYPES,
     COSE_NAMED_ALGORITHMS_SUPPORTED,
 )
-from .hpke_cipher_suite import HPKECipherSuite
 
 
 class COSEKeyInterface(CBORProcessor):
@@ -227,39 +226,6 @@ def decrypt(self, msg: bytes, nonce: bytes, aad: bytes) -> bytes:
         """
         raise NotImplementedError
 
-    def seal(self, suite: HPKECipherSuite, msg: bytes, aad: bytes = b"") -> Tuple[bytes, bytes]:
-        """
-        Encrypts the specified message with HPKE.
-
-        Args:
-            msg (bytes): A message to be encrypted.
-            aad (bytes): Additional authenticated data.
-        Returns:
-            Tuple[bytes, bytes]: The encapsulation key and the ciphertext respectively.
-        Raises:
-            NotImplementedError: Not implemented.
-            ValueError: Invalid arguments.
-            EncodeError: Failed to encrypt the message.
-        """
-        raise NotImplementedError
-
-    def open(self, suite: HPKECipherSuite, enc: bytes, msg: bytes, aad: bytes = b"") -> bytes:
-        """
-        Decrypts the specified message with HPKE.
-
-        Args:
-            enc (bytes): An encapsulated key.
-            msg (bytes): An encrypted message.
-            aad (bytes): Additional authenticated data.
-        Returns:
-            bytes: The byte string of the decrypted data.
-        Raises:
-            NotImplementedError: Not implemented.
-            ValueError: Invalid arguments.
-            DecodeError: Failed to decrypt the message.
-        """
-        raise NotImplementedError
-
     def wrap_key(self, key_to_wrap: bytes) -> bytes:
         """
         Wraps a key.

cwt/hpke_cipher_suite.py

@@ -1,7 +1,9 @@
 from typing import Any, Dict, List, Optional, Union
 
+from pyhpke import AEADId, CipherSuite, KDFId, KEMId, KEMKey, KEMKeyInterface
+
 from ..cose_key_interface import COSEKeyInterface
-from ..hpke_cipher_suite import HPKECipherSuite
+from ..exceptions import DecodeError, EncodeError
 from ..recipient_interface import RecipientInterface
 
 
@@ -25,7 +27,7 @@ def __init__(
             raise ValueError("kdf id(2) not found in HPKE sender information(-4).")
         if 3 not in unprotected[-4]:
             raise ValueError("aead id(3) not found in HPKE sender information(-4).")
-        self._suite = HPKECipherSuite(unprotected[-4][1], unprotected[-4][2], unprotected[-4][3])
+        self._suite = CipherSuite.new(KEMId(unprotected[-4][1]), KDFId(unprotected[-4][2]), AEADId(unprotected[-4][3]))
         return
 
     def apply(
@@ -39,14 +41,19 @@ def apply(
             raise ValueError("recipient_key should be set.")
 
         self._recipient_key = recipient_key
+        self._kem_key = self._to_kem_key(recipient_key)
         return self._recipient_key
 
     def to_list(self, payload: bytes = b"", external_aad: bytes = b"", aad_context: str = "Enc_Recipient") -> List[Any]:
         enc_structure = [aad_context, self._dumps(self._protected), external_aad]
         aad = self._dumps(enc_structure)
-        enc, self._ciphertext = self._recipient_key.seal(self._suite, payload, aad)
+        enc, sender = self._suite.create_sender_context(self._kem_key)
         self._unprotected[-4][4] = enc
-        return super().to_list(payload, external_aad, aad_context)
+        try:
+            self._ciphertext = sender.seal(payload, aad=aad)
+            return super().to_list(payload, external_aad, aad_context)
+        except Exception as err:
+            raise EncodeError("Failed to seal.") from err
 
     def decrypt(
         self,
@@ -61,4 +68,11 @@ def decrypt(
     ) -> bytes:
         enc_structure = [aad_context, self._dumps(self._protected), external_aad]
         aad = self._dumps(enc_structure)
-        return key.open(self._suite, self._unprotected[-4][4], self._ciphertext, aad)
+        recipient = self._suite.create_recipient_context(self._unprotected[-4][4], self._to_kem_key(key))
+        try:
+            return recipient.open(self._ciphertext, aad=aad)
+        except Exception as err:
+            raise DecodeError("Failed to open.") from err
+
+    def _to_kem_key(self, src: COSEKeyInterface) -> KEMKeyInterface:
+        return KEMKey.from_pyca_cryptography_key(src.key)