implement decrypt_into on aessiv (#13782)

Author: reaperhulk

CHANGELOG.rst

@@ -47,6 +47,9 @@ Changelog
   :class:`~cryptography.hazmat.primitives.ciphers.aead.AESSIV`, and
   :class:`~cryptography.hazmat.primitives.ciphers.aead.ChaCha20Poly1305` to
   allow encrypting directly into a pre-allocated buffer.
+* Added a ``decrypt_into`` method to
+  :class:`~cryptography.hazmat.primitives.ciphers.aead.AESSIV` to allow
+  decrypting directly into a pre-allocated buffer.
 
 .. _v46-0-3:
 

docs/hazmat/primitives/aead.rst

@@ -559,6 +559,31 @@ also support providing integrity for associated data which is not encrypted.
             when the ciphertext has been changed, but will also occur when the
             key or associated data are wrong.
 
+    .. method:: decrypt_into(data, associated_data, buf)
+
+        .. versionadded:: 47.0.0
+
+        Decrypts the ``data`` and authenticates the ``associated_data``. If you
+        called encrypt with ``associated_data`` you must pass the same
+        ``associated_data`` in decrypt or the integrity check will fail. The
+        output is written into the ``buf`` parameter.
+
+        :param bytes data: The data to decrypt (with tag **prepended**).
+        :param list associated_data: An optional ``list`` of ``bytes-like objects``. This
+            is additional data that should be authenticated with the key, but
+            is not encrypted. Can be ``None`` if none was used during
+            encryption.
+        :param buf: A writable :term:`bytes-like` object that must be exactly
+            ``len(data) - 16`` bytes. The plaintext will be written to this
+            buffer.
+        :returns int: The number of bytes written to the buffer (always
+            ``len(data) - 16``).
+        :raises ValueError: If the buffer is not the correct size.
+        :raises cryptography.exceptions.InvalidTag: If the authentication tag
+            doesn't validate this exception will be raised. This will occur
+            when the ciphertext has been changed, but will also occur when the
+            key or associated data are wrong.
+
 .. class:: AESCCM(key, tag_length=16)
 
     .. versionadded:: 2.0

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

@@ -98,6 +98,12 @@ class AESSIV:
         data: Buffer,
         associated_data: Sequence[Buffer] | None,
     ) -> bytes: ...
+    def decrypt_into(
+        self,
+        data: Buffer,
+        associated_data: Sequence[Buffer] | None,
+        buf: Buffer,
+    ) -> int: ...
 
 class AESOCB3:
     def __init__(self, key: Buffer) -> None: ...

src/rust/src/backend/aead.rs

@@ -199,35 +199,68 @@ impl EvpCipherAead {
         aad: Option<Aad<'_>>,
         nonce: Option<&[u8]>,
     ) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
+        // Temporary while we remove this function
+        if ciphertext.len() < self.tag_len {
+            return Err(CryptographyError::from(exceptions::InvalidTag::new_err(())));
+        }
+
         let mut ctx = openssl::cipher_ctx::CipherCtx::new()?;
         ctx.copy(&self.base_decryption_ctx)?;
-        Self::decrypt_with_context(
+        Ok(pyo3::types::PyBytes::new_with(
             py,
+            ciphertext.len() - self.tag_len,
+            |b| {
+                EvpCipherAead::decrypt_with_context(
+                    ctx,
+                    ciphertext,
+                    aad,
+                    nonce,
+                    self.tag_len,
+                    self.tag_first,
+                    false,
+                    b,
+                )?;
+                Ok(())
+            },
+        )?)
+    }
+
+    fn decrypt_into(
+        &self,
+        // We have this arg so we have consistent arguments with decrypt_into in
+        // LazyEvpCipherAead. We can remove it when we remove LazyEvpCipherAead.
+        _py: pyo3::Python<'_>,
+        ciphertext: &[u8],
+        aad: Option<Aad<'_>>,
+        nonce: Option<&[u8]>,
+        buf: &mut [u8],
+    ) -> CryptographyResult<()> {
+        let mut ctx = openssl::cipher_ctx::CipherCtx::new()?;
+        ctx.copy(&self.base_decryption_ctx)?;
+
+        Self::decrypt_with_context(
             ctx,
             ciphertext,
             aad,
             nonce,
             self.tag_len,
             self.tag_first,
             false,
+            buf,
         )
     }
 
     #[allow(clippy::too_many_arguments)]
-    fn decrypt_with_context<'p>(
-        py: pyo3::Python<'p>,
+    fn decrypt_with_context(
         mut ctx: openssl::cipher_ctx::CipherCtx,
         ciphertext: &[u8],
         aad: Option<Aad<'_>>,
         nonce: Option<&[u8]>,
         tag_len: usize,
         tag_first: bool,
         is_ccm: bool,
-    ) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
-        if ciphertext.len() < tag_len {
-            return Err(CryptographyError::from(exceptions::InvalidTag::new_err(())));
-        }
-
+        buf: &mut [u8],
+    ) -> CryptographyResult<()> {
         let tag;
         let ciphertext_data;
         if tag_first {
@@ -253,16 +286,10 @@ impl EvpCipherAead {
 
         Self::process_aad(&mut ctx, aad)?;
 
-        Ok(pyo3::types::PyBytes::new_with(
-            py,
-            ciphertext_data.len(),
-            |b| {
-                Self::process_data(&mut ctx, ciphertext_data, b, is_ccm)
-                    .map_err(|_| exceptions::InvalidTag::new_err(()))?;
+        Self::process_data(&mut ctx, ciphertext_data, buf, is_ccm)
+            .map_err(|_| exceptions::InvalidTag::new_err(()))?;
 
-                Ok(())
-            },
-        )?)
+        Ok(())
     }
 }
 
@@ -350,16 +377,27 @@ impl LazyEvpCipherAead {
             decryption_ctx.decrypt_init(Some(self.cipher), Some(key_buf.as_bytes()), None)?;
         }
 
-        EvpCipherAead::decrypt_with_context(
+        if ciphertext.len() < self.tag_len {
+            return Err(CryptographyError::from(exceptions::InvalidTag::new_err(())));
+        }
+
+        Ok(pyo3::types::PyBytes::new_with(
             py,
-            decryption_ctx,
-            ciphertext,
-            aad,
-            nonce,
-            self.tag_len,
-            self.tag_first,
-            self.is_ccm,
-        )
+            ciphertext.len() - self.tag_len,
+            |b| {
+                EvpCipherAead::decrypt_with_context(
+                    decryption_ctx,
+                    ciphertext,
+                    aad,
+                    nonce,
+                    self.tag_len,
+                    self.tag_first,
+                    self.is_ccm,
+                    b,
+                )?;
+                Ok(())
+            },
+        )?)
     }
 }
 
@@ -1060,8 +1098,50 @@ impl AesSiv {
         data: CffiBuf<'_>,
         associated_data: Option<pyo3::Bound<'_, pyo3::types::PyList>>,
     ) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
+        if data.as_bytes().len() < self.ctx.tag_len {
+            return Err(CryptographyError::from(exceptions::InvalidTag::new_err(())));
+        }
+        Ok(pyo3::types::PyBytes::new_with(
+            py,
+            data.as_bytes().len() - self.ctx.tag_len,
+            |b| {
+                let buf = CffiMutBuf::from_bytes(py, b);
+                self.decrypt_into(py, data, associated_data, buf)?;
+                Ok(())
+            },
+        )?)
+    }
+
+    #[pyo3(signature = (data, associated_data, buf))]
+    fn decrypt_into(
+        &self,
+        py: pyo3::Python<'_>,
+        data: CffiBuf<'_>,
+        associated_data: Option<pyo3::Bound<'_, pyo3::types::PyList>>,
+        mut buf: CffiMutBuf<'_>,
+    ) -> CryptographyResult<usize> {
+        let data_bytes = data.as_bytes();
         let aad = associated_data.map(Aad::List);
-        self.ctx.decrypt(py, data.as_bytes(), aad, None)
+
+        // We need to do this check early to prevent underflow when computing expected_len
+        if data_bytes.len() < self.ctx.tag_len {
+            return Err(CryptographyError::from(exceptions::InvalidTag::new_err(())));
+        }
+
+        let expected_len = data_bytes.len() - self.ctx.tag_len;
+        if buf.as_mut_bytes().len() != expected_len {
+            return Err(CryptographyError::from(
+                pyo3::exceptions::PyValueError::new_err(format!(
+                    "buffer must be {} bytes",
+                    expected_len
+                )),
+            ));
+        }
+
+        self.ctx
+            .decrypt_into(py, data_bytes, aad, None, buf.as_mut_bytes())?;
+
+        Ok(expected_len)
     }
 }
 

tests/hazmat/primitives/test_aead.py

@@ -953,6 +953,15 @@ def test_bad_generate_key(self, backend):
         with pytest.raises(ValueError):
             AESSIV.generate_key(128)
 
+    def test_data_too_short(self, backend):
+        key = AESSIV.generate_key(256)
+        aessiv = AESSIV(key)
+        with pytest.raises(InvalidTag):
+            aessiv.decrypt(b"tooshort", None)
+        with pytest.raises(InvalidTag):
+            buf = bytearray(16)
+            aessiv.decrypt_into(b"tooshort", None, buf)
+
     def test_associated_data_none_equal_to_empty_list(self, backend):
         key = AESSIV.generate_key(256)
         aessiv = AESSIV(key)
@@ -999,6 +1008,41 @@ def test_encrypt_into_buffer_incorrect_size(self, ptlen, buflen, backend):
         with pytest.raises(ValueError, match="buffer must be"):
             aessiv.encrypt_into(pt, None, buf)
 
+    def test_decrypt_into(self, backend):
+        key = AESSIV.generate_key(256)
+        aessiv = AESSIV(key)
+        pt = b"decrypt me"
+        ad = [b"additional"]
+        ct = aessiv.encrypt(pt, ad)
+        buf = bytearray(len(pt))
+        n = aessiv.decrypt_into(ct, ad, buf)
+        assert n == len(pt)
+        assert buf == pt
+
+    @pytest.mark.parametrize(
+        ("ctlen", "buflen"), [(26, 9), (26, 11), (31, 14), (36, 21)]
+    )
+    def test_decrypt_into_buffer_incorrect_size(self, ctlen, buflen, backend):
+        key = AESSIV.generate_key(256)
+        aessiv = AESSIV(key)
+        ct = b"x" * ctlen
+        buf = bytearray(buflen)
+        with pytest.raises(ValueError, match="buffer must be"):
+            aessiv.decrypt_into(ct, None, buf)
+
+    def test_decrypt_into_invalid_tag(self, backend):
+        key = AESSIV.generate_key(256)
+        aessiv = AESSIV(key)
+        pt = b"some data"
+        ad = [b"additional"]
+        ct = aessiv.encrypt(pt, ad)
+        # Corrupt the ciphertext
+        corrupted_ct = bytearray(ct)
+        corrupted_ct[0] ^= 1
+        buf = bytearray(len(pt))
+        with pytest.raises(InvalidTag):
+            aessiv.decrypt_into(bytes(corrupted_ct), ad, buf)
+
 
 @pytest.mark.skipif(
     not _aead_supported(AESGCMSIV),