Add HPKE (Hybrid Public Key Encryption) support per RFC 9180 (#14126)

MkDev11 · web-flow · commit d59470e5d521 · 2026-01-21T12:49:54.000-05:00
* Add HPKE support per RFC 9180 * Trigger CI re-run * Refactor HPKE API to match agreed design from issue discussion API changes per maintainer feedback: - Add Suite(KEM, KDF, AEAD) class with RFC parameter order - Add KEM, KDF, AEAD enum types for algorithm selection - suite.sender(public_key, info) returns SenderContext - suite.recipient(enc, private_key, info) returns RecipientContext - First encrypt() returns enc || ciphertext concatenated - Subsequent encrypt() calls return just ciphertext - Use encrypt/decrypt method names - Simpler KEM names (X25519 instead of DHKEM_X25519_HKDF_SHA256) This matches the API discussed in issue #14073 and addresses reviewer feedback about the API mismatch. * Fix ruff lint errors: unused import, line length, sorting * Fix ruff format * Fix mypy type errors with TypedDict for parameter dicts * Address reviewer feedback: dataclass, HKDF.extract, remove HPKEError - Replace TypedDict with frozen dataclasses for parameter types - Use HKDF.extract() static method instead of hmac.digest - Remove base HPKEError class, keep only MessageLimitReachedError - Use if/elif instead of match for Python 3.8 compatibility * Fix coverage: use assert instead of unreachable raise statements The type system guarantees only valid enum values can be passed to these internal functions, so use assert which is excluded from coverage. * Address reviewer feedback: remove first_message behavior, add test vectors - Remove first_message special behavior from SenderContext and RecipientContext - encrypt() now always returns just ciphertext, enc accessed via property - Remove try/except Exception - AESGCM.decrypt raises InvalidTag directly - Update tests to use new API (enc via sender.enc property) - Add test vector loading from RFC 9180 vectors (when available) - Update docstring example to show correct usage * Add test for load_vectors missing file branch * Simplify to single-shot API per maintainer feedback - Replace sender()/recipient() context methods with encrypt()/decrypt() - encrypt() returns enc || ciphertext (like Go's single-shot API) - decrypt() takes enc || ciphertext and returns plaintext - Make enum values opaque strings instead of RFC numeric values - Remove SenderContext, RecipientContext, MessageLimitReachedError - Update tests for new single-shot API * Format test file with ruff * Fix coverage: remove conditional branch in vector test * Address Alex's review feedback - Remove docstrings (prose docs only) - Use load_vectors_from_file helper - Use [] instead of .get() for vector keys - Parameterize roundtrip tests by KEM/KDF/AEAD - Merge error test class into main test class - Use subtest fixture instead of parametrize for vectors - Remove test_load_vectors_missing_file test - Load vectors inside test function, not at module level * Fix Alex's review comments - Remove comment from _key_schedule (mode = 0x00) - Remove parameterization from error case tests (only roundtrip tests) - Remove FileNotFoundError handling (vectors always exist) - Update docs to match actual Suite API implementation - Add HPKE test vectors from RFC 9180 * - Add _HPKE_MODE_BASE module-level constant (third request) - Add type checking in Suite.__init__ for kem/kdf/aead parameters - Rename NENC to X25519_ENC_LENGTH for clarity - Expand docs with security guidance on authenticated encryption, ephemeral keys, info/aad usage - Remove unnecessary 'Example' and 'Classes' section headers from docs * Fix spelling error in docs: ciphertexts -> ciphertext * Add tests for TypeError paths in Suite.__init__ for coverage * Fix line length in test_invalid_aead_type * Add type: ignore comments for intentional TypeError tests --------- Co-authored-by: mkdev11 <mkdev11@users.noreply.github.com>
diff --git a/docs/hazmat/primitives/hpke.rst b/docs/hazmat/primitives/hpke.rst
@@ -0,0 +1,101 @@
+.. hazmat::
+
+HPKE (Hybrid Public Key Encryption)
+===================================
+
+.. module:: cryptography.hazmat.primitives.hpke
+
+HPKE is a standard for public key encryption that combines a Key Encapsulation
+Mechanism (KEM), a Key Derivation Function (KDF), and an Authenticated
+Encryption with Associated Data (AEAD) scheme. It is defined in :rfc:`9180`.
+
+This implementation supports Base mode with DHKEM(X25519, HKDF-SHA256),
+HKDF-SHA256, and AES-128-GCM.
+
+HPKE provides authenticated encryption: the recipient can be certain that the
+message was encrypted by someone who knows the recipient's public key, but
+the sender is anonymous. Each call to :meth:`Suite.encrypt` generates a fresh
+ephemeral key pair, so encrypting the same plaintext twice will produce
+different ciphertext.
+
+The ``info`` parameter should be used to bind the encryption to a specific
+context (e.g., "MyApp-v1-UserMessages"). The ``aad`` parameter provides
+additional authenticated data that is not encrypted but is authenticated
+along with the ciphertext.
+
+.. code-block:: python
+
+    from cryptography.hazmat.primitives.hpke import Suite, KEM, KDF, AEAD
+    from cryptography.hazmat.primitives.asymmetric import x25519
+
+    suite = Suite(KEM.X25519, KDF.HKDF_SHA256, AEAD.AES_128_GCM)
+
+    # Generate recipient key pair
+    private_key = x25519.X25519PrivateKey.generate()
+    public_key = private_key.public_key()
+
+    # Encrypt
+    ciphertext = suite.encrypt(b"secret message", public_key, info=b"app info")
+
+    # Decrypt
+    plaintext = suite.decrypt(ciphertext, private_key, info=b"app info")
+
+.. class:: Suite(kem, kdf, aead)
+
+    An HPKE cipher suite combining a KEM, KDF, and AEAD.
+
+    :param kem: The key encapsulation mechanism.
+    :type kem: :class:`KEM`
+    :param kdf: The key derivation function.
+    :type kdf: :class:`KDF`
+    :param aead: The authenticated encryption algorithm.
+    :type aead: :class:`AEAD`
+
+    .. method:: encrypt(plaintext, public_key, info=b"", aad=b"")
+
+        Encrypt a message using HPKE.
+
+        :param bytes plaintext: The message to encrypt.
+        :param public_key: The recipient's public key.
+        :type public_key: :class:`~cryptography.hazmat.primitives.asymmetric.x25519.X25519PublicKey`
+        :param bytes info: Application-specific info string.
+        :param bytes aad: Additional authenticated data.
+        :returns: The encapsulated key concatenated with ciphertext (enc || ct).
+        :rtype: bytes
+
+    .. method:: decrypt(ciphertext, private_key, info=b"", aad=b"")
+
+        Decrypt a message using HPKE.
+
+        :param bytes ciphertext: The enc || ct value from :meth:`encrypt`.
+        :param private_key: The recipient's private key.
+        :type private_key: :class:`~cryptography.hazmat.primitives.asymmetric.x25519.X25519PrivateKey`
+        :param bytes info: Application-specific info string.
+        :param bytes aad: Additional authenticated data.
+        :returns: The decrypted plaintext.
+        :rtype: bytes
+        :raises cryptography.exceptions.InvalidTag: If decryption fails.
+
+.. class:: KEM
+
+    An enumeration of key encapsulation mechanisms.
+
+    .. attribute:: X25519
+
+        DHKEM(X25519, HKDF-SHA256)
+
+.. class:: KDF
+
+    An enumeration of key derivation functions.
+
+    .. attribute:: HKDF_SHA256
+
+        HKDF-SHA256
+
+.. class:: AEAD
+
+    An enumeration of authenticated encryption algorithms.
+
+    .. attribute:: AES_128_GCM
+
+        AES-128-GCM
diff --git a/docs/hazmat/primitives/index.rst b/docs/hazmat/primitives/index.rst
@@ -8,6 +8,7 @@ Primitives
 
     aead
     asymmetric/index
+    hpke
     constant-time
     key-derivation-functions
     keywrap
diff --git a/src/cryptography/hazmat/primitives/hpke.py b/src/cryptography/hazmat/primitives/hpke.py
@@ -0,0 +1,253 @@
+# This file is dual licensed under the terms of the Apache License, Version
+# 2.0, and the BSD License. See the LICENSE file in the root of this repository
+# for complete details.
+
+from __future__ import annotations
+
+import dataclasses
+import enum
+
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import x25519
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from cryptography.hazmat.primitives.kdf.hkdf import HKDF, HKDFExpand
+from cryptography.utils import int_to_bytes
+
+_HPKE_VERSION = b"HPKE-v1"
+_HPKE_MODE_BASE = 0x00
+
+
+class KEM(enum.Enum):
+    X25519 = "X25519"
+
+
+class KDF(enum.Enum):
+    HKDF_SHA256 = "HKDF_SHA256"
+
+
+class AEAD(enum.Enum):
+    AES_128_GCM = "AES_128_GCM"
+
+
+@dataclasses.dataclass(frozen=True)
+class _KEMParams:
+    id: int
+    nsecret: int
+    nenc: int
+    npk: int
+    nsk: int
+    hash: hashes.HashAlgorithm
+
+
+@dataclasses.dataclass(frozen=True)
+class _KDFParams:
+    id: int
+    nh: int
+    hash: hashes.HashAlgorithm
+
+
+@dataclasses.dataclass(frozen=True)
+class _AEADParams:
+    id: int
+    nk: int
+    nn: int
+    nt: int
+
+
+def _get_kem_params(kem: KEM) -> _KEMParams:
+    assert kem == KEM.X25519
+    return _KEMParams(
+        id=0x0020,
+        nsecret=32,
+        nenc=32,
+        npk=32,
+        nsk=32,
+        hash=hashes.SHA256(),
+    )
+
+
+def _get_kdf_params(kdf: KDF) -> _KDFParams:
+    assert kdf == KDF.HKDF_SHA256
+    return _KDFParams(
+        id=0x0001,
+        nh=32,
+        hash=hashes.SHA256(),
+    )
+
+
+def _get_aead_params(aead: AEAD) -> _AEADParams:
+    assert aead == AEAD.AES_128_GCM
+    return _AEADParams(
+        id=0x0001,
+        nk=16,
+        nn=12,
+        nt=16,
+    )
+
+
+class Suite:
+    def __init__(self, kem: KEM, kdf: KDF, aead: AEAD) -> None:
+        if not isinstance(kem, KEM):
+            raise TypeError("kem must be an instance of KEM")
+        if not isinstance(kdf, KDF):
+            raise TypeError("kdf must be an instance of KDF")
+        if not isinstance(aead, AEAD):
+            raise TypeError("aead must be an instance of AEAD")
+
+        self._kem = kem
+        self._kdf = kdf
+        self._aead = aead
+
+        self._kem_params = _get_kem_params(kem)
+        self._kdf_params = _get_kdf_params(kdf)
+        self._aead_params = _get_aead_params(aead)
+
+        # Build suite IDs
+        self._kem_suite_id = b"KEM" + int_to_bytes(self._kem_params.id, 2)
+        self._hpke_suite_id = (
+            b"HPKE"
+            + int_to_bytes(self._kem_params.id, 2)
+            + int_to_bytes(self._kdf_params.id, 2)
+            + int_to_bytes(self._aead_params.id, 2)
+        )
+
+    def _kem_labeled_extract(
+        self, salt: bytes, label: bytes, ikm: bytes
+    ) -> bytes:
+        labeled_ikm = _HPKE_VERSION + self._kem_suite_id + label + ikm
+        return HKDF.extract(
+            self._kdf_params.hash,
+            salt if salt else None,
+            labeled_ikm,
+        )
+
+    def _kem_labeled_expand(
+        self, prk: bytes, label: bytes, info: bytes, length: int
+    ) -> bytes:
+        labeled_info = (
+            int_to_bytes(length, 2)
+            + _HPKE_VERSION
+            + self._kem_suite_id
+            + label
+            + info
+        )
+        hkdf_expand = HKDFExpand(
+            algorithm=self._kdf_params.hash,
+            length=length,
+            info=labeled_info,
+        )
+        return hkdf_expand.derive(prk)
+
+    def _extract_and_expand(self, dh: bytes, kem_context: bytes) -> bytes:
+        eae_prk = self._kem_labeled_extract(b"", b"eae_prk", dh)
+        shared_secret = self._kem_labeled_expand(
+            eae_prk,
+            b"shared_secret",
+            kem_context,
+            self._kem_params.nsecret,
+        )
+        return shared_secret
+
+    def _encap(self, pk_r: x25519.X25519PublicKey) -> tuple[bytes, bytes]:
+        sk_e = x25519.X25519PrivateKey.generate()
+        pk_e = sk_e.public_key()
+        dh = sk_e.exchange(pk_r)
+        enc = pk_e.public_bytes_raw()
+        pk_rm = pk_r.public_bytes_raw()
+        kem_context = enc + pk_rm
+        shared_secret = self._extract_and_expand(dh, kem_context)
+        return shared_secret, enc
+
+    def _decap(self, enc: bytes, sk_r: x25519.X25519PrivateKey) -> bytes:
+        pk_e = x25519.X25519PublicKey.from_public_bytes(enc)
+        dh = sk_r.exchange(pk_e)
+        pk_rm = sk_r.public_key().public_bytes_raw()
+        kem_context = enc + pk_rm
+        shared_secret = self._extract_and_expand(dh, kem_context)
+        return shared_secret
+
+    def _hpke_labeled_extract(
+        self, salt: bytes, label: bytes, ikm: bytes
+    ) -> bytes:
+        labeled_ikm = _HPKE_VERSION + self._hpke_suite_id + label + ikm
+        return HKDF.extract(
+            self._kdf_params.hash,
+            salt if salt else None,
+            labeled_ikm,
+        )
+
+    def _hpke_labeled_expand(
+        self, prk: bytes, label: bytes, info: bytes, length: int
+    ) -> bytes:
+        labeled_info = (
+            int_to_bytes(length, 2)
+            + _HPKE_VERSION
+            + self._hpke_suite_id
+            + label
+            + info
+        )
+        hkdf_expand = HKDFExpand(
+            algorithm=self._kdf_params.hash,
+            length=length,
+            info=labeled_info,
+        )
+        return hkdf_expand.derive(prk)
+
+    def _key_schedule(
+        self, shared_secret: bytes, info: bytes
+    ) -> tuple[bytes, bytes]:
+        mode = _HPKE_MODE_BASE
+
+        psk_id_hash = self._hpke_labeled_extract(b"", b"psk_id_hash", b"")
+        info_hash = self._hpke_labeled_extract(b"", b"info_hash", info)
+        key_schedule_context = bytes([mode]) + psk_id_hash + info_hash
+
+        secret = self._hpke_labeled_extract(shared_secret, b"secret", b"")
+
+        key = self._hpke_labeled_expand(
+            secret, b"key", key_schedule_context, self._aead_params.nk
+        )
+        base_nonce = self._hpke_labeled_expand(
+            secret,
+            b"base_nonce",
+            key_schedule_context,
+            self._aead_params.nn,
+        )
+
+        return key, base_nonce
+
+    def encrypt(
+        self,
+        plaintext: bytes,
+        public_key: x25519.X25519PublicKey,
+        info: bytes = b"",
+        aad: bytes = b"",
+    ) -> bytes:
+        shared_secret, enc = self._encap(public_key)
+        key, base_nonce = self._key_schedule(shared_secret, info)
+        aead_impl = AESGCM(key)
+        ct = aead_impl.encrypt(base_nonce, plaintext, aad)
+        return enc + ct
+
+    def decrypt(
+        self,
+        ciphertext: bytes,
+        private_key: x25519.X25519PrivateKey,
+        info: bytes = b"",
+        aad: bytes = b"",
+    ) -> bytes:
+        nenc = self._kem_params.nenc
+        enc = ciphertext[:nenc]
+        ct = ciphertext[nenc:]
+        shared_secret = self._decap(enc, private_key)
+        key, base_nonce = self._key_schedule(shared_secret, info)
+        aead_impl = AESGCM(key)
+        return aead_impl.decrypt(base_nonce, ct, aad)
+
+
+__all__ = [
+    "AEAD",
+    "KDF",
+    "KEM",
+    "Suite",
+]
diff --git a/tests/hazmat/primitives/test_hpke.py b/tests/hazmat/primitives/test_hpke.py
