Remove aad parameter from HPKE one-shot public API (#14203)

Per RFC 9180 Section 8.1, applications using single-shot APIs should
use the info parameter for auxiliary authenticated information rather
than aad. This change:

- Removes aad parameter from public Suite.encrypt/decrypt in Rust
- Adds module-level _encrypt_with_aad/_decrypt_with_aad functions
  in the private Rust hpke module for test vector validation
- Updates tests to use the module-level functions for aad
- Updates type stubs, documentation, and CHANGELOG

Fixes #14073

Co-authored-by: Claude <noreply@anthropic.com>

Author: alex

CHANGELOG.rst

@@ -84,6 +84,8 @@ Changelog
   :class:`~cryptography.hazmat.primitives.asymmetric.utils.NoDigestInfo`.
 * Added :meth:`~cryptography.hazmat.primitives.hashes.Hash.hash`, a one-shot
   method for computing hashes.
+* Added :doc:`/hazmat/primitives/hpke` support implementing :rfc:`9180` for
+  hybrid authenticated encryption.
 
 .. _v46-0-5:
 

docs/hazmat/primitives/hpke.rst

@@ -19,9 +19,9 @@ 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.
+context (e.g., "MyApp-v1-UserMessages"). Per :rfc:`9180#section-8.1`,
+applications using single-shot APIs should use the ``info`` parameter for
+specifying auxiliary authenticated information.
 
 .. code-block:: python
 
@@ -51,27 +51,27 @@ along with the ciphertext.
     :param aead: The authenticated encryption algorithm.
     :type aead: :class:`AEAD`
 
-    .. method:: encrypt(plaintext, public_key, info=b"", aad=b"")
+    .. method:: encrypt(plaintext, public_key, info=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.
+        :param bytes info: Application-specific context string for binding the
+            encryption to a specific application or protocol.
         :returns: The encapsulated key concatenated with ciphertext (enc || ct).
         :rtype: bytes
 
-    .. method:: decrypt(ciphertext, private_key, info=b"", aad=b"")
+    .. method:: decrypt(ciphertext, private_key, info=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.
+        :param bytes info: Application-specific context string (must match the
+            value used during encryption).
         :returns: The decrypted plaintext.
         :rtype: bytes
         :raises cryptography.exceptions.InvalidTag: If decryption fails.

src/cryptography/hazmat/bindings/_rust/openssl/hpke.pyi

@@ -21,12 +21,25 @@ class Suite:
         plaintext: Buffer,
         public_key: x25519.X25519PublicKey,
         info: Buffer | None = None,
-        aad: Buffer | None = None,
     ) -> bytes: ...
     def decrypt(
         self,
         ciphertext: Buffer,
         private_key: x25519.X25519PrivateKey,
         info: Buffer | None = None,
-        aad: Buffer | None = None,
     ) -> bytes: ...
+
+def _encrypt_with_aad(
+    suite: Suite,
+    plaintext: Buffer,
+    public_key: x25519.X25519PublicKey,
+    info: Buffer | None = None,
+    aad: Buffer | None = None,
+) -> bytes: ...
+def _decrypt_with_aad(
+    suite: Suite,
+    ciphertext: Buffer,
+    private_key: x25519.X25519PrivateKey,
+    info: Buffer | None = None,
+    aad: Buffer | None = None,
+) -> bytes: ...

src/rust/src/backend/hpke.rs

@@ -2,14 +2,14 @@
 // 2.0, and the BSD License. See the LICENSE file in the root of this repository
 // for complete details.
 
+use pyo3::types::{PyAnyMethods, PyBytesMethods};
+
 use crate::backend::aead::AesGcm;
 use crate::backend::kdf::{hkdf_extract, HkdfExpand};
 use crate::backend::x25519;
 use crate::buf::CffiBuf;
 use crate::error::{CryptographyError, CryptographyResult};
-use crate::exceptions;
-use crate::types;
-use pyo3::types::{PyAnyMethods, PyBytesMethods};
+use crate::{exceptions, types};
 
 const HPKE_VERSION: &[u8] = b"HPKE-v1";
 const HPKE_MODE_BASE: u8 = 0x00;
@@ -235,6 +235,66 @@ impl Suite {
         self.hkdf_expand(py, prk, &labeled_info, length)
     }
 
+    fn encrypt_inner<'p>(
+        &self,
+        py: pyo3::Python<'p>,
+        plaintext: CffiBuf<'_>,
+        public_key: &pyo3::Bound<'_, pyo3::PyAny>,
+        info: Option<CffiBuf<'_>>,
+        aad: Option<CffiBuf<'_>>,
+    ) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
+        let info_bytes: &[u8] = info.as_ref().map(|b| b.as_bytes()).unwrap_or(b"");
+
+        let (shared_secret, enc) = self.encap(py, public_key)?;
+        let (key, base_nonce) = self.key_schedule(py, shared_secret.as_bytes(), info_bytes)?;
+
+        let aesgcm = AesGcm::new(py, pyo3::types::PyBytes::new(py, &key).unbind().into_any())?;
+        let ct = aesgcm.encrypt(py, CffiBuf::from_bytes(py, &base_nonce), plaintext, aad)?;
+
+        let enc_bytes = enc.as_bytes();
+        let ct_bytes = ct.as_bytes();
+        Ok(pyo3::types::PyBytes::new_with(
+            py,
+            enc_bytes.len() + ct_bytes.len(),
+            |buf| {
+                buf[..enc_bytes.len()].copy_from_slice(enc_bytes);
+                buf[enc_bytes.len()..].copy_from_slice(ct_bytes);
+                Ok(())
+            },
+        )?)
+    }
+
+    fn decrypt_inner<'p>(
+        &self,
+        py: pyo3::Python<'p>,
+        ciphertext: CffiBuf<'_>,
+        private_key: &pyo3::Bound<'_, pyo3::PyAny>,
+        info: Option<CffiBuf<'_>>,
+        aad: Option<CffiBuf<'_>>,
+    ) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
+        let ct_bytes = ciphertext.as_bytes();
+        if ct_bytes.len() < kem_params::X25519_NENC + aead_params::AES_128_GCM_NT {
+            return Err(CryptographyError::from(exceptions::InvalidTag::new_err(())));
+        }
+
+        let info_bytes: &[u8] = info.as_ref().map(|b| b.as_bytes()).unwrap_or(b"");
+
+        let (enc, ct) = ct_bytes.split_at(kem_params::X25519_NENC);
+
+        let shared_secret = self
+            .decap(py, enc, private_key)
+            .map_err(|_| CryptographyError::from(exceptions::InvalidTag::new_err(())))?;
+        let (key, base_nonce) = self.key_schedule(py, shared_secret.as_bytes(), info_bytes)?;
+
+        let aesgcm = AesGcm::new(py, pyo3::types::PyBytes::new(py, &key).unbind().into_any())?;
+        aesgcm.decrypt(
+            py,
+            CffiBuf::from_bytes(py, &base_nonce),
+            CffiBuf::from_bytes(py, ct),
+            aad,
+        )
+    }
+
     fn key_schedule(
         &self,
         py: pyo3::Python<'_>,
@@ -297,71 +357,59 @@ impl Suite {
         })
     }
 
-    #[pyo3(signature = (plaintext, public_key, info=None, aad=None))]
+    #[pyo3(signature = (plaintext, public_key, info=None))]
     fn encrypt<'p>(
         &self,
         py: pyo3::Python<'p>,
         plaintext: CffiBuf<'_>,
         public_key: &pyo3::Bound<'_, pyo3::PyAny>,
         info: Option<CffiBuf<'_>>,
-        aad: Option<CffiBuf<'_>>,
     ) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
-        let info_bytes: &[u8] = info.as_ref().map(|b| b.as_bytes()).unwrap_or(b"");
-
-        let (shared_secret, enc) = self.encap(py, public_key)?;
-        let (key, base_nonce) = self.key_schedule(py, shared_secret.as_bytes(), info_bytes)?;
-
-        let aesgcm = AesGcm::new(py, pyo3::types::PyBytes::new(py, &key).unbind().into_any())?;
-        let ct = aesgcm.encrypt(py, CffiBuf::from_bytes(py, &base_nonce), plaintext, aad)?;
-
-        let enc_bytes = enc.as_bytes();
-        let ct_bytes = ct.as_bytes();
-        Ok(pyo3::types::PyBytes::new_with(
-            py,
-            enc_bytes.len() + ct_bytes.len(),
-            |buf| {
-                buf[..enc_bytes.len()].copy_from_slice(enc_bytes);
-                buf[enc_bytes.len()..].copy_from_slice(ct_bytes);
-                Ok(())
-            },
-        )?)
+        self.encrypt_inner(py, plaintext, public_key, info, None)
     }
 
-    #[pyo3(signature = (ciphertext, private_key, info=None, aad=None))]
+    #[pyo3(signature = (ciphertext, private_key, info=None))]
     fn decrypt<'p>(
         &self,
         py: pyo3::Python<'p>,
         ciphertext: CffiBuf<'_>,
         private_key: &pyo3::Bound<'_, pyo3::PyAny>,
         info: Option<CffiBuf<'_>>,
-        aad: Option<CffiBuf<'_>>,
     ) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
-        let ct_bytes = ciphertext.as_bytes();
-        if ct_bytes.len() < kem_params::X25519_NENC + aead_params::AES_128_GCM_NT {
-            return Err(CryptographyError::from(exceptions::InvalidTag::new_err(())));
-        }
-
-        let info_bytes: &[u8] = info.as_ref().map(|b| b.as_bytes()).unwrap_or(b"");
-
-        let (enc, ct) = ct_bytes.split_at(kem_params::X25519_NENC);
+        self.decrypt_inner(py, ciphertext, private_key, info, None)
+    }
+}
 
-        let shared_secret = self
-            .decap(py, enc, private_key)
-            .map_err(|_| CryptographyError::from(exceptions::InvalidTag::new_err(())))?;
-        let (key, base_nonce) = self.key_schedule(py, shared_secret.as_bytes(), info_bytes)?;
+#[pyo3::pyfunction]
+#[pyo3(signature = (suite, plaintext, public_key, info=None, aad=None))]
+fn _encrypt_with_aad<'p>(
+    py: pyo3::Python<'p>,
+    suite: &Suite,
+    plaintext: CffiBuf<'_>,
+    public_key: &pyo3::Bound<'_, pyo3::PyAny>,
+    info: Option<CffiBuf<'_>>,
+    aad: Option<CffiBuf<'_>>,
+) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
+    suite.encrypt_inner(py, plaintext, public_key, info, aad)
+}
 
-        let aesgcm = AesGcm::new(py, pyo3::types::PyBytes::new(py, &key).unbind().into_any())?;
-        aesgcm.decrypt(
-            py,
-            CffiBuf::from_bytes(py, &base_nonce),
-            CffiBuf::from_bytes(py, ct),
-            aad,
-        )
-    }
+#[pyo3::pyfunction]
+#[pyo3(signature = (suite, ciphertext, private_key, info=None, aad=None))]
+fn _decrypt_with_aad<'p>(
+    py: pyo3::Python<'p>,
+    suite: &Suite,
+    ciphertext: CffiBuf<'_>,
+    private_key: &pyo3::Bound<'_, pyo3::PyAny>,
+    info: Option<CffiBuf<'_>>,
+    aad: Option<CffiBuf<'_>>,
+) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
+    suite.decrypt_inner(py, ciphertext, private_key, info, aad)
 }
 
 #[pyo3::pymodule(gil_used = false)]
 pub(crate) mod hpke {
+    // stable and nightly rustfmt disagree on import ordering
+    #[rustfmt::skip]
     #[pymodule_export]
-    use super::{Suite, AEAD, KDF, KEM};
+    use super::{_decrypt_with_aad, _encrypt_with_aad, Suite, AEAD, KDF, KEM};
 }

tests/hazmat/primitives/test_hpke.py

@@ -8,6 +8,7 @@
 import pytest
 
 from cryptography.exceptions import InvalidTag
+from cryptography.hazmat.bindings._rust import openssl as rust_openssl
 from cryptography.hazmat.primitives.asymmetric import x25519
 from cryptography.hazmat.primitives.hpke import (
     AEAD,
@@ -49,17 +50,13 @@ def test_roundtrip(self, kem, kdf, aead):
         sk_r = x25519.X25519PrivateKey.generate()
         pk_r = sk_r.public_key()
 
-        ciphertext = suite.encrypt(
-            b"Hello, HPKE!", pk_r, info=b"test", aad=b"additional data"
-        )
-        plaintext = suite.decrypt(
-            ciphertext, sk_r, info=b"test", aad=b"additional data"
-        )
+        ciphertext = suite.encrypt(b"Hello, HPKE!", pk_r, info=b"test")
+        plaintext = suite.decrypt(ciphertext, sk_r, info=b"test")
 
         assert plaintext == b"Hello, HPKE!"
 
     @pytest.mark.parametrize("kem,kdf,aead", SUPPORTED_SUITES)
-    def test_roundtrip_no_aad(self, kem, kdf, aead):
+    def test_roundtrip_no_info(self, kem, kdf, aead):
         suite = Suite(kem, kdf, aead)
 
         sk_r = x25519.X25519PrivateKey.generate()
@@ -88,10 +85,14 @@ def test_wrong_aad_fails(self):
         sk_r = x25519.X25519PrivateKey.generate()
         pk_r = sk_r.public_key()
 
-        ciphertext = suite.encrypt(b"Secret message", pk_r, aad=b"correct aad")
+        ciphertext = rust_openssl.hpke._encrypt_with_aad(
+            suite, b"Secret message", pk_r, aad=b"correct aad"
+        )
 
         with pytest.raises(InvalidTag):
-            suite.decrypt(ciphertext, sk_r, aad=b"wrong aad")
+            rust_openssl.hpke._decrypt_with_aad(
+                suite, ciphertext, sk_r, aad=b"wrong aad"
+            )
 
     def test_info_mismatch_fails(self):
         suite = Suite(KEM.X25519, KDF.HKDF_SHA256, AEAD.AES_128_GCM)
@@ -248,6 +249,10 @@ def test_vector_decryption(self, subtests):
                 pt_expected = bytes.fromhex(encryption["pt"])
 
                 # Combine enc || ct for single-shot decrypt
+                # Use internal function with AAD for test vector
+                # validation
                 ciphertext = enc + ct
-                pt = suite.decrypt(ciphertext, sk_r, info=info, aad=aad)
+                pt = rust_openssl.hpke._decrypt_with_aad(
+                    suite, ciphertext, sk_r, info=info, aad=aad
+                )
                 assert pt == pt_expected