implement decrypt_into for aesgcmsiv (#13788)

Author: reaperhulk

CHANGELOG.rst

@@ -50,6 +50,7 @@ Changelog
 * Added ``decrypt_into`` methods to
   :class:`~cryptography.hazmat.primitives.ciphers.aead.AESCCM`,
   :class:`~cryptography.hazmat.primitives.ciphers.aead.AESGCM`,
+  :class:`~cryptography.hazmat.primitives.ciphers.aead.AESGCMSIV`,
   :class:`~cryptography.hazmat.primitives.ciphers.aead.AESOCB3`,
   :class:`~cryptography.hazmat.primitives.ciphers.aead.AESSIV`, and
   :class:`~cryptography.hazmat.primitives.ciphers.aead.ChaCha20Poly1305` to

docs/hazmat/primitives/aead.rst

@@ -386,6 +386,33 @@ also support providing integrity for associated data which is not encrypted.
             when the ciphertext has been changed, but will also occur when the
             key, nonce, or associated data are wrong.
 
+    .. method:: decrypt_into(nonce, 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 nonce: A 12-byte value.
+        :type nonce: :term:`bytes-like`
+        :param data: The data to decrypt (with tag appended).
+        :type data: :term:`bytes-like`
+        :param associated_data: Additional data to authenticate. Can be
+            ``None`` if none was passed during encryption.
+        :type associated_data: :term:`bytes-like`
+        :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, nonce, or associated data are wrong.
+
 .. class:: AESOCB3(key)
 
     .. versionadded:: 36.0.0

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

@@ -180,3 +180,10 @@ class AESGCMSIV:
         data: Buffer,
         associated_data: Buffer | None,
     ) -> bytes: ...
+    def decrypt_into(
+        self,
+        nonce: Buffer,
+        data: Buffer,
+        associated_data: Buffer | None,
+        buf: Buffer,
+    ) -> int: ...

src/rust/src/backend/aead.rs

@@ -192,40 +192,6 @@ impl EvpCipherAead {
         Ok(())
     }
 
-    #[cfg(not(any(CRYPTOGRAPHY_IS_BORINGSSL, CRYPTOGRAPHY_IS_AWSLC)))]
-    fn decrypt<'p>(
-        &self,
-        py: pyo3::Python<'p>,
-        ciphertext: &[u8],
-        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)?;
-        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
@@ -428,38 +394,6 @@ impl EvpAead {
         Ok(())
     }
 
-    fn decrypt<'p>(
-        &self,
-        py: pyo3::Python<'p>,
-        ciphertext: &[u8],
-        aad: Option<Aad<'_>>,
-        nonce: Option<&[u8]>,
-    ) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
-        if ciphertext.len() < self.tag_len {
-            return Err(CryptographyError::from(exceptions::InvalidTag::new_err(())));
-        }
-
-        let ad = if let Some(Aad::Single(ad)) = &aad {
-            check_length(ad.as_bytes())?;
-            ad.as_bytes()
-        } else {
-            assert!(aad.is_none());
-            b""
-        };
-
-        Ok(pyo3::types::PyBytes::new_with(
-            py,
-            ciphertext.len() - self.tag_len,
-            |b| {
-                self.ctx
-                    .decrypt(ciphertext, nonce.unwrap_or(b""), ad, b)
-                    .map_err(|_| exceptions::InvalidTag::new_err(()))?;
-
-                Ok(())
-            },
-        )?)
-    }
-
     fn decrypt_into(
         &self,
         _py: pyo3::Python<'_>,
@@ -1659,14 +1593,66 @@ impl AesGcmSiv {
         associated_data: Option<CffiBuf<'_>>,
     ) -> CryptographyResult<pyo3::Bound<'p, pyo3::types::PyBytes>> {
         let nonce_bytes = nonce.as_bytes();
+        let data_bytes = data.as_bytes();
+
+        if nonce_bytes.len() != 12 {
+            return Err(CryptographyError::from(
+                pyo3::exceptions::PyValueError::new_err("Nonce must be 12 bytes long"),
+            ));
+        }
+
+        if data_bytes.len() < self.ctx.tag_len {
+            return Err(CryptographyError::from(exceptions::InvalidTag::new_err(())));
+        }
+
+        Ok(pyo3::types::PyBytes::new_with(
+            py,
+            data_bytes.len() - 16,
+            |b| {
+                let buf = CffiMutBuf::from_bytes(py, b);
+                self.decrypt_into(py, nonce, data, associated_data, buf)?;
+                Ok(())
+            },
+        )?)
+    }
+
+    #[pyo3(signature = (nonce, data, associated_data, buf))]
+    fn decrypt_into(
+        &self,
+        py: pyo3::Python<'_>,
+        nonce: CffiBuf<'_>,
+        data: CffiBuf<'_>,
+        associated_data: Option<CffiBuf<'_>>,
+        mut buf: CffiMutBuf<'_>,
+    ) -> CryptographyResult<usize> {
+        let nonce_bytes = nonce.as_bytes();
+        let data_bytes = data.as_bytes();
         let aad = associated_data.map(Aad::Single);
+
         if nonce_bytes.len() != 12 {
             return Err(CryptographyError::from(
                 pyo3::exceptions::PyValueError::new_err("Nonce must be 12 bytes long"),
             ));
         }
+
+        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(py, data.as_bytes(), aad, Some(nonce_bytes))
+            .decrypt_into(py, data_bytes, aad, Some(nonce_bytes), buf.as_mut_bytes())?;
+
+        Ok(expected_len)
     }
 }
 

tests/hazmat/primitives/test_aead.py

@@ -1287,6 +1287,10 @@ def test_invalid_nonce_length(self, backend):
         with pytest.raises(ValueError):
             aesgcmsiv.decrypt(nonce, pt, None)
 
+        with pytest.raises(ValueError):
+            buf = bytearray(16)
+            aesgcmsiv.decrypt_into(nonce, b"x" * 20, None, buf)
+
     def test_empty(self):
         key = AESGCMSIV.generate_key(256)
         aesgcmsiv = AESGCMSIV(key)
@@ -1468,3 +1472,49 @@ def test_encrypt_into_buffer_incorrect_size(self, ptlen, buflen, backend):
         buf = bytearray(buflen)
         with pytest.raises(ValueError, match="buffer must be"):
             aesgcmsiv.encrypt_into(nonce, pt, None, buf)
+
+    def test_decrypt_into(self, backend):
+        key = AESGCMSIV.generate_key(256)
+        aesgcmsiv = AESGCMSIV(key)
+        nonce = os.urandom(12)
+        pt = b"decrypt me"
+        ad = b"additional"
+        ct = aesgcmsiv.encrypt(nonce, pt, ad)
+        buf = bytearray(len(pt))
+        n = aesgcmsiv.decrypt_into(nonce, 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 = AESGCMSIV.generate_key(256)
+        aesgcmsiv = AESGCMSIV(key)
+        nonce = os.urandom(12)
+        ct = b"x" * ctlen
+        buf = bytearray(buflen)
+        with pytest.raises(ValueError, match="buffer must be"):
+            aesgcmsiv.decrypt_into(nonce, ct, None, buf)
+
+    def test_decrypt_into_invalid_tag(self, backend):
+        key = AESGCMSIV.generate_key(256)
+        aesgcmsiv = AESGCMSIV(key)
+        nonce = os.urandom(12)
+        pt = b"some data"
+        ad = b"additional"
+        ct = aesgcmsiv.encrypt(nonce, pt, ad)
+        # Corrupt the ciphertext
+        corrupted_ct = bytearray(ct)
+        corrupted_ct[0] ^= 1
+        buf = bytearray(len(pt))
+        with pytest.raises(InvalidTag):
+            aesgcmsiv.decrypt_into(nonce, bytes(corrupted_ct), ad, buf)
+
+    def test_decrypt_into_data_too_short(self, backend):
+        key = AESGCMSIV.generate_key(256)
+        aesgcmsiv = AESGCMSIV(key)
+        nonce = os.urandom(12)
+        buf = bytearray(16)
+        with pytest.raises(InvalidTag):
+            aesgcmsiv.decrypt_into(nonce, b"short", None, buf)