add support for CipherContext.update_nonce

This only supports ChaCha20 and ciphers in CTR mode.

Author: reaperhulk

CHANGELOG.rst

@@ -26,6 +26,9 @@ Changelog
   and :class:`~cryptography.hazmat.primitives.ciphers.algorithms.ARC4` into
   :doc:`/hazmat/decrepit/index` and deprecated them in the ``cipher`` module.
   They will be removed from the ``cipher`` module in 48.0.0.
+* Added :meth:`~cryptography.hazmat.primitives.ciphers.CipherContext.update_nonce`
+  for altering the ``nonce`` of a cipher context without initializing a new
+  instance. See the docs for additional restrictions.
 
 .. _v42-0-3:
 

docs/hazmat/primitives/symmetric-encryption.rst

@@ -693,6 +693,27 @@ Interfaces
         :meth:`update` and :meth:`finalize` will raise an
         :class:`~cryptography.exceptions.AlreadyFinalized` exception.
 
+    .. method:: update_nonce(nonce)
+
+        .. versionadded:: 43.0.0
+
+        This method allows updating the nonce for an already existing context.
+        Normally the nonce is set when the context is created and internally
+        incremented as data as passed. However, in some scenarios the same key
+        is used repeatedly but the nonce changes non-sequentially (e.g. ``QUIC``),
+        which requires updating the context with the new nonce.
+
+        This method only works for contexts using
+        :class:`~cryptography.hazmat.primitives.ciphers.algorithms.ChaCha20` or
+        :class:`~cryptography.hazmat.primitives.ciphers.modes.CTR` mode.
+
+        :param nonce: The nonce to update the context with.
+        :type data: :term:`bytes-like`
+        :raises cryptography.exceptions.UnsupportedAlgorithm: If the
+            algorithm does not support updating the nonce.
+        :raises ValueError: If the nonce is not the correct length for the
+            algorithm.
+
 .. class:: AEADCipherContext
 
     When calling ``encryptor`` or ``decryptor`` on a ``Cipher`` object

src/cryptography/hazmat/primitives/ciphers/base.py

@@ -33,6 +33,14 @@ def finalize(self) -> bytes:
         Returns the results of processing the final block as bytes.
         """
 
+    @abc.abstractmethod
+    def update_nonce(self, nonce: bytes) -> None:
+        """
+        Updates the nonce for the cipher context to the provided value.
+        Raises an exception if it does not support reset or if the
+        provided nonce does not have a valid length.
+        """
+
 
 class AEADCipherContext(CipherContext, metaclass=abc.ABCMeta):
     @abc.abstractmethod

src/rust/src/backend/ciphers.rs

@@ -12,6 +12,8 @@ use pyo3::IntoPy;
 struct CipherContext {
     ctx: openssl::cipher_ctx::CipherCtx,
     py_mode: pyo3::PyObject,
+    side: openssl::symm::Mode,
+    update_nonce_allowed: bool,
 }
 
 impl CipherContext {
@@ -41,6 +43,7 @@ impl CipherContext {
             }
         };
 
+        let mut update_nonce_allowed = false;
         let iv_nonce = if mode.is_instance(types::MODE_WITH_INITIALIZATION_VECTOR.get(py)?)? {
             Some(
                 mode.getattr(pyo3::intern!(py, "initialization_vector"))?
@@ -52,11 +55,13 @@ impl CipherContext {
                     .extract::<CffiBuf<'_>>()?,
             )
         } else if mode.is_instance(types::MODE_WITH_NONCE.get(py)?)? {
+            update_nonce_allowed = true;
             Some(
                 mode.getattr(pyo3::intern!(py, "nonce"))?
                     .extract::<CffiBuf<'_>>()?,
             )
         } else if algorithm.is_instance(types::CHACHA20.get(py)?)? {
+            update_nonce_allowed = true;
             Some(
                 algorithm
                     .getattr(pyo3::intern!(py, "nonce"))?
@@ -111,9 +116,36 @@ impl CipherContext {
         Ok(CipherContext {
             ctx,
             py_mode: mode.into(),
+            side,
+            update_nonce_allowed,
         })
     }
 
+    fn update_nonce(&mut self, nonce: CffiBuf<'_>) -> CryptographyResult<()> {
+        if !self.update_nonce_allowed {
+            return Err(CryptographyError::from(
+                exceptions::UnsupportedAlgorithm::new_err((
+                    "This algorithm or mode does not support resetting the nonce.",
+                    exceptions::Reasons::UNSUPPORTED_CIPHER,
+                )),
+            ));
+        }
+        if nonce.as_bytes().len() != self.ctx.iv_length() {
+            return Err(CryptographyError::from(
+                pyo3::exceptions::PyValueError::new_err(format!(
+                    "Nonce must be {} bytes long",
+                    self.ctx.iv_length()
+                )),
+            ));
+        }
+        let init_op = match self.side {
+            openssl::symm::Mode::Encrypt => openssl::cipher_ctx::CipherCtxRef::encrypt_init,
+            openssl::symm::Mode::Decrypt => openssl::cipher_ctx::CipherCtxRef::decrypt_init,
+        };
+        init_op(&mut self.ctx, None, None, Some(nonce.as_bytes()))?;
+        Ok(())
+    }
+
     fn update<'p>(
         &mut self,
         py: pyo3::Python<'p>,
@@ -234,6 +266,10 @@ impl PyCipherContext {
         get_mut_ctx(self.ctx.as_mut())?.update(py, buf.as_bytes())
     }
 
+    fn update_nonce(&mut self, nonce: CffiBuf<'_>) -> CryptographyResult<()> {
+        get_mut_ctx(self.ctx.as_mut())?.update_nonce(nonce)
+    }
+
     fn update_into(
         &mut self,
         py: pyo3::Python<'_>,
@@ -338,6 +374,10 @@ impl PyAEADEncryptionContext {
             })?
             .clone_ref(py))
     }
+
+    fn update_nonce(&mut self, nonce: CffiBuf<'_>) -> CryptographyResult<()> {
+        get_mut_ctx(self.ctx.as_mut())?.update_nonce(nonce)
+    }
 }
 
 #[pyo3::prelude::pymethods]
@@ -466,6 +506,10 @@ impl PyAEADDecryptionContext {
         self.ctx = None;
         Ok(result)
     }
+
+    fn update_nonce(&mut self, nonce: CffiBuf<'_>) -> CryptographyResult<()> {
+        get_mut_ctx(self.ctx.as_mut())?.update_nonce(nonce)
+    }
 }
 
 #[pyo3::prelude::pyfunction]

tests/hazmat/primitives/test_aes.py

@@ -8,11 +8,12 @@
 
 import pytest
 
+from cryptography.exceptions import AlreadyFinalized, _Reasons
 from cryptography.hazmat.bindings._rust import openssl as rust_openssl
 from cryptography.hazmat.primitives.ciphers import algorithms, base, modes
 
 from ...doubles import DummyMode
-from ...utils import load_nist_vectors
+from ...utils import load_nist_vectors, raises_unsupported_algorithm
 from .utils import _load_all_params, generate_encrypt_test
 
 
@@ -305,3 +306,43 @@ def test_alternate_aes_classes(mode, alg_cls, backend):
     dec = cipher.decryptor()
     pt = dec.update(ct) + dec.finalize()
     assert pt == data
+
+
+def test_update_nonce(backend):
+    data = b"helloworld" * 10
+    nonce = b"\x00" * 16
+    c = base.Cipher(
+        algorithms.AES(b"\x00" * 16),
+        modes.CTR(nonce),
+    )
+    enc = c.encryptor()
+    ct1 = enc.update(data)
+    assert len(ct1) == len(data)
+    for _ in range(2):
+        enc.update_nonce(nonce)
+        assert enc.update(data) == ct1
+    enc.finalize()
+    with pytest.raises(AlreadyFinalized):
+        enc.update_nonce(nonce)
+    dec = c.decryptor()
+    assert dec.update(ct1) == data
+    for _ in range(2):
+        dec.update_nonce(nonce)
+        assert dec.update(ct1) == data
+    dec.finalize()
+    with pytest.raises(AlreadyFinalized):
+        dec.update_nonce(nonce)
+
+
+def test_update_nonce_invalid_mode(backend):
+    iv = b"\x00" * 16
+    c = base.Cipher(
+        algorithms.AES(b"\x00" * 16),
+        modes.CBC(iv),
+    )
+    enc = c.encryptor()
+    with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_CIPHER):
+        enc.update_nonce(iv)
+    dec = c.decryptor()
+    with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_CIPHER):
+        dec.update_nonce(iv)

tests/hazmat/primitives/test_aes_gcm.py

@@ -8,10 +8,11 @@
 
 import pytest
 
+from cryptography.exceptions import _Reasons
 from cryptography.hazmat.bindings._rust import openssl as rust_openssl
 from cryptography.hazmat.primitives.ciphers import algorithms, base, modes
 
-from ...utils import load_nist_vectors
+from ...utils import load_nist_vectors, raises_unsupported_algorithm
 from .utils import generate_aead_test
 
 
@@ -230,3 +231,16 @@ def test_alternate_aes_classes(self, alg, backend):
         dec = cipher.decryptor()
         pt = dec.update(ct) + dec.finalize_with_tag(enc.tag)
         assert pt == data
+
+    def test_update_nonce_invalid_mode(self, backend):
+        nonce = b"\x00" * 12
+        c = base.Cipher(
+            algorithms.AES(b"\x00" * 16),
+            modes.GCM(nonce),
+        )
+        enc = c.encryptor()
+        with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_CIPHER):
+            enc.update_nonce(nonce)
+        dec = c.decryptor()
+        with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_CIPHER):
+            dec.update_nonce(nonce)

tests/hazmat/primitives/test_chacha20.py

@@ -9,6 +9,7 @@
 
 import pytest
 
+from cryptography.exceptions import AlreadyFinalized
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms
 
 from ...utils import load_nist_vectors
@@ -90,3 +91,36 @@ def test_partial_blocks(self, backend):
         ct_partial_3 = enc_partial.update(pt[len_partial * 2 :])
 
         assert ct_full == ct_partial_1 + ct_partial_2 + ct_partial_3
+
+    def test_update_nonce(self, backend):
+        data = b"helloworld" * 10
+        key = b"\x00" * 32
+        nonce = b"\x00" * 16
+        cipher = Cipher(algorithms.ChaCha20(key, nonce), None)
+        enc = cipher.encryptor()
+        ct1 = enc.update(data)
+        assert len(ct1) == len(data)
+        for _ in range(2):
+            enc.update_nonce(nonce)
+            assert enc.update(data) == ct1
+        enc.finalize()
+        with pytest.raises(AlreadyFinalized):
+            enc.update_nonce(nonce)
+        dec = cipher.decryptor()
+        assert dec.update(ct1) == data
+        for _ in range(2):
+            dec.update_nonce(nonce)
+            assert dec.update(ct1) == data
+        dec.finalize()
+        with pytest.raises(AlreadyFinalized):
+            dec.update_nonce(nonce)
+
+    def test_nonce_reset_invalid_length(self, backend):
+        key = b"\x00" * 32
+        nonce = b"\x00" * 16
+        cipher = Cipher(algorithms.ChaCha20(key, nonce), None)
+        enc = cipher.encryptor()
+        with pytest.raises(ValueError):
+            enc.update_nonce(nonce[:-1])
+        with pytest.raises(ValueError):
+            enc.update_nonce(nonce + b"\x00")