diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 1f9b339e2883..8e1479b5d800 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -10,6 +10,11 @@ Changelog * The :mod:`X.509 verification ` APIs are now considered stable and are subject to our API stability policy. +* Added the :doc:`/cobblestone` recipe, an implementation of the + Cobblestone-128 and Cobblestone-256 instantiations of the `C2SP + chunked-encryption specification + `_ for streaming authenticated + encryption of large messages. * Parsing a Signed Certificate Timestamp list now rejects encodings that carry trailing bytes after the list or after an individual SCT, instead of silently ignoring them. diff --git a/docs/cobblestone.rst b/docs/cobblestone.rst new file mode 100644 index 000000000000..35c98116b8eb --- /dev/null +++ b/docs/cobblestone.rst @@ -0,0 +1,185 @@ +Cobblestone (streaming symmetric encryption) +============================================= + +.. currentmodule:: cryptography.cobblestone + +Cobblestone provides authenticated symmetric encryption of large +messages — up to 4 PiB — as a stream, without ever holding the whole +message in memory. It is an implementation of the `C2SP +chunked-encryption specification`_'s two named instantiations: +**Cobblestone-128** (SHA-512 and AES-128-GCM, the recommended choice) +and **Cobblestone-256** (SHA-512 and AES-256-GCM, for environments +that mandate 256-bit keys). + +.. doctest:: + + >>> from cryptography.cobblestone import ( + ... Cobblestone128Decryptor, Cobblestone128Encryptor + ... ) + >>> key = Cobblestone128Encryptor.generate_key() + >>> encryptor = Cobblestone128Encryptor( + ... key, context=b"example-app file encryption" + ... ) + >>> ciphertext = encryptor.update(b"a secret message") + >>> ciphertext += encryptor.finalize() + >>> decryptor = Cobblestone128Decryptor( + ... key, context=b"example-app file encryption" + ... ) + >>> decryptor.update(ciphertext) + decryptor.finalize() + b'a secret message' + +.. class:: Cobblestone128Encryptor(key, context) + + .. versionadded:: 50.0.0 + + Encrypts a single message under ``key`` with Cobblestone-128. Each + instance must be used for exactly one message: call :meth:`update` + (or :meth:`update_into`) any number of times, then call + :meth:`finalize` exactly once. The concatenation of the returned + bytes is the ciphertext. + + :param key: A 16-byte key. This **must** be kept secret, and + **must** be uniformly random (e.g. the output of + :meth:`generate_key` — never a password). A single key may be + used to encrypt a practically unlimited number of messages. + :type key: :term:`bytes-like` + :param context: Application-provided context, bound to the + ciphertext. Decryption fails unless the same value is passed to + :class:`Cobblestone128Decryptor`. It is not secret, may be + empty, and is not part of the ciphertext, so it must be + available to the decrypting party independently. It can be used + for domain separation, e.g. ``b"myapp v2 backup encryption"``. + :type context: :term:`bytes-like` + :raises ValueError: If ``key`` is not 16 bytes. + + .. staticmethod:: generate_key() + + Generates a fresh 16-byte key. + + :return bytes: A new key. + + .. method:: update(data) + + Encrypts ``data``. Data is internally buffered into 16 KiB + chunks, so between 0 and ``len(data) + 16 KiB`` bytes of + ciphertext are returned. + + :param data: The data to encrypt. + :type data: :term:`bytes-like` + :return bytes: The next portion of the ciphertext. + + .. method:: update_into(data, buf) + + Encrypts ``data``, writing the resulting ciphertext into + ``buf``, and returns the number of bytes written. This avoids + allocating a new buffer for each call. + + :param data: The data to encrypt. + :type data: :term:`bytes-like` + :param buf: A writable buffer to write the ciphertext into. A + buffer of ``len(data) + len(data) // 1024 + 16456`` bytes + is always large enough. + :type buf: :term:`bytes-like` + :return int: The number of bytes written to ``buf``. + :raises ValueError: If ``buf`` is too small. + + .. method:: finalize() + + Encrypts the final chunk and returns the last portion of the + ciphertext. This must always be called, and the instance cannot + be used afterwards. + + :return bytes: The remainder of the ciphertext. + :raises cryptography.exceptions.AlreadyFinalized: If + ``finalize`` has already been called. + +.. class:: Cobblestone128Decryptor(key, context) + + .. versionadded:: 50.0.0 + + Decrypts a single message encrypted by + :class:`Cobblestone128Encryptor` with the same ``key`` and + ``context``. Call :meth:`update` (or :meth:`update_into`) with the + ciphertext any number of times, then call :meth:`finalize` exactly + once. The concatenation of the returned bytes is the plaintext. + + Any returned plaintext is authenticated, but until + :meth:`finalize` returns successfully the message could still turn + out to be truncated: an application acting on streamed plaintext + before that point must be prepared to discard its work if a later + call raises :class:`~cryptography.exceptions.InvalidTag`. + + Once any method raises + :class:`~cryptography.exceptions.InvalidTag`, the instance is + permanently unusable and all further calls raise + :class:`~cryptography.exceptions.AlreadyFinalized`. + + :param key: The 16-byte key the message was encrypted with. + :type key: :term:`bytes-like` + :param context: The context value the message was encrypted with. + :type context: :term:`bytes-like` + :raises ValueError: If ``key`` is not 16 bytes. + + .. method:: update(data) + + Processes ``data``, which need not be aligned to any boundary, + and returns the plaintext of all complete chunks that have been + authenticated so far. + + :param data: The next portion of the ciphertext. + :type data: :term:`bytes-like` + :return bytes: The next portion of the plaintext. + :raises cryptography.exceptions.InvalidTag: If the ciphertext + was encrypted with a different key or context, or has been + modified. + + .. method:: update_into(data, buf) + + Like ``update``, but writes the plaintext into ``buf`` and + returns the number of bytes written. + + :param data: The next portion of the ciphertext. + :type data: :term:`bytes-like` + :param buf: A writable buffer to write the plaintext into. A + buffer of ``len(data) + 16400`` bytes is always large + enough. + :type buf: :term:`bytes-like` + :return int: The number of bytes written to ``buf``. + :raises ValueError: If ``buf`` is too small. + :raises cryptography.exceptions.InvalidTag: If the ciphertext + was encrypted with a different key or context, or has been + modified. Note that in this case unauthenticated data may + have been written to ``buf`` and must not be used. + + .. method:: finalize() + + Decrypts and authenticates the final chunk, verifying that the + entire message has been processed, and returns the final + portion of the plaintext. This must always be called: a + successful return is what guarantees the complete message was + authentic and not truncated. + + :return bytes: The remainder of the plaintext. + :raises cryptography.exceptions.InvalidTag: If the ciphertext + was truncated or otherwise modified. + :raises cryptography.exceptions.AlreadyFinalized: If + ``finalize`` has already been called. + +.. class:: Cobblestone256Encryptor(key, context) + + .. versionadded:: 50.0.0 + + Exactly like :class:`Cobblestone128Encryptor`, but implements + Cobblestone-256: the ``key`` is 32 bytes and messages are encrypted + with AES-256-GCM. Use this when a 256-bit key is mandated; + otherwise Cobblestone-128 is recommended. + +.. class:: Cobblestone256Decryptor(key, context) + + .. versionadded:: 50.0.0 + + Exactly like :class:`Cobblestone128Decryptor`, but decrypts + messages produced by :class:`Cobblestone256Encryptor` with a + 32-byte key. + +.. _`C2SP chunked-encryption specification`: https://c2sp.org/chunked-encryption diff --git a/docs/index.rst b/docs/index.rst index a70a58886b6d..10c69b1545ab 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -68,6 +68,7 @@ hazmat layer only when necessary. :caption: The recipes layer fernet + cobblestone x509/index .. toctree:: diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt index 5e6d5433b919..df4c2e037e92 100644 --- a/docs/spelling_wordlist.txt +++ b/docs/spelling_wordlist.txt @@ -24,6 +24,7 @@ CentOS changelog Changelog ciphertext +Cobblestone codebook committer committers @@ -42,6 +43,7 @@ Decapsulate Decapsulation declaratively decrypt +Decryptor decrypts Decrypts decrypted @@ -61,6 +63,7 @@ duplicative El embeddability Encodings +Encryptor endian Euler extendable @@ -122,6 +125,7 @@ parsers Parsers PEM PHC +PiB pickleable plaintext Poly diff --git a/src/cryptography/cobblestone.py b/src/cryptography/cobblestone.py new file mode 100644 index 000000000000..09f67af9b1c0 --- /dev/null +++ b/src/cryptography/cobblestone.py @@ -0,0 +1,21 @@ +# 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 + +from cryptography.hazmat.bindings._rust import ( + cobblestone as _cobblestone, +) + +Cobblestone128Decryptor = _cobblestone.Cobblestone128Decryptor +Cobblestone128Encryptor = _cobblestone.Cobblestone128Encryptor +Cobblestone256Decryptor = _cobblestone.Cobblestone256Decryptor +Cobblestone256Encryptor = _cobblestone.Cobblestone256Encryptor + +__all__ = [ + "Cobblestone128Decryptor", + "Cobblestone128Encryptor", + "Cobblestone256Decryptor", + "Cobblestone256Encryptor", +] diff --git a/src/cryptography/hazmat/bindings/_rust/cobblestone.pyi b/src/cryptography/hazmat/bindings/_rust/cobblestone.pyi new file mode 100644 index 000000000000..95e958382904 --- /dev/null +++ b/src/cryptography/hazmat/bindings/_rust/cobblestone.pyi @@ -0,0 +1,33 @@ +# 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 cryptography.utils import Buffer + +class Cobblestone128Encryptor: + def __init__(self, key: Buffer, context: Buffer) -> None: ... + @staticmethod + def generate_key() -> bytes: ... + def update(self, data: Buffer) -> bytes: ... + def update_into(self, data: Buffer, buf: Buffer) -> int: ... + def finalize(self) -> bytes: ... + +class Cobblestone128Decryptor: + def __init__(self, key: Buffer, context: Buffer) -> None: ... + def update(self, data: Buffer) -> bytes: ... + def update_into(self, data: Buffer, buf: Buffer) -> int: ... + def finalize(self) -> bytes: ... + +class Cobblestone256Encryptor: + def __init__(self, key: Buffer, context: Buffer) -> None: ... + @staticmethod + def generate_key() -> bytes: ... + def update(self, data: Buffer) -> bytes: ... + def update_into(self, data: Buffer, buf: Buffer) -> int: ... + def finalize(self) -> bytes: ... + +class Cobblestone256Decryptor: + def __init__(self, key: Buffer, context: Buffer) -> None: ... + def update(self, data: Buffer) -> bytes: ... + def update_into(self, data: Buffer, buf: Buffer) -> int: ... + def finalize(self) -> bytes: ... diff --git a/src/rust/src/backend/aead.rs b/src/rust/src/backend/aead.rs index ec9f2bd32340..e98ab96fb06b 100644 --- a/src/rust/src/backend/aead.rs +++ b/src/rust/src/backend/aead.rs @@ -673,7 +673,7 @@ impl AesGcm { } #[pyo3(signature = (nonce, data, associated_data, buf))] - fn decrypt_into( + pub(crate) fn decrypt_into( &self, py: pyo3::Python<'_>, nonce: CffiBuf<'_>, diff --git a/src/rust/src/cobblestone.rs b/src/rust/src/cobblestone.rs new file mode 100644 index 000000000000..24fc6a66b1b7 --- /dev/null +++ b/src/rust/src/cobblestone.rs @@ -0,0 +1,763 @@ +// 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. + +// Implementation of the C2SP chunked-encryption specification +// (https://c2sp.org/chunked-encryption), providing its two named +// instantiations: Cobblestone-128 (SHA-512 and AES-128-GCM) and +// Cobblestone-256 (SHA-512 and AES-256-GCM). + +use pyo3::types::{PyAnyMethods, PyBytesMethods}; + +use crate::backend::aead::AesGcm; +use crate::backend::kdf::HkdfExpand; +use crate::buf::{CffiBuf, CffiMutBuf}; +use crate::error::{CryptographyError, CryptographyResult}; +use crate::{exceptions, types}; + +const CHUNK_SIZE: usize = 16 * 1024; +const SALT_LEN: usize = 24; +const COMMITMENT_LEN: usize = 32; +const HEADER_LEN: usize = SALT_LEN + COMMITMENT_LEN; +// The chunk counter is a 38-bit big-endian integer, limiting messages to +// 4 PiB - 1. +const MAX_CHUNK_COUNT: u64 = 1 << 38; +const INFO_PREFIX: &[u8] = b"c2sp.org/chunked-encryption@v1+"; + +// The scheme requires an AEAD nonce of at least 96 bits, and every AEAD we +// might instantiate it with (AES-GCM, ChaCha20-Poly1305) uses exactly 96. +const NONCE_LEN: usize = 12; + +struct AeadParams { + // Name from the IANA AEAD Algorithms registry, used in the HKDF info. + iana_name: &'static [u8], + key_len: usize, + tag_len: usize, +} + +impl AeadParams { + fn wire_chunk_size(&self) -> usize { + CHUNK_SIZE + self.tag_len + } +} + +static AES_128_GCM: AeadParams = AeadParams { + iana_name: b"AEAD_AES_128_GCM", + key_len: 16, + tag_len: 16, +}; + +static AES_256_GCM: AeadParams = AeadParams { + iana_name: b"AEAD_AES_256_GCM", + key_len: 32, + tag_len: 16, +}; + +fn message_too_large_error() -> CryptographyError { + CryptographyError::from(pyo3::exceptions::PyValueError::new_err( + "Message exceeds the maximum chunked encryption length (2**38 chunks).", + )) +} + +fn check_key_length(params: &AeadParams, key: &[u8]) -> CryptographyResult<()> { + if key.len() != params.key_len { + return Err(CryptographyError::from( + pyo3::exceptions::PyValueError::new_err(format!( + "key must be {} bytes.", + params.key_len + )), + )); + } + Ok(()) +} + +struct DerivedKeys<'p> { + key: pyo3::Bound<'p, pyo3::types::PyBytes>, + base_nonce: [u8; NONCE_LEN], + commitment: [u8; COMMITMENT_LEN], +} + +fn derive_keys<'p>( + py: pyo3::Python<'p>, + params: &AeadParams, + input_key: &[u8], + salt: &[u8], + context: &[u8], +) -> CryptographyResult> { + let mut info = Vec::with_capacity( + INFO_PREFIX.len() + params.iana_name.len() + 1 + salt.len() + context.len(), + ); + info.extend_from_slice(INFO_PREFIX); + info.extend_from_slice(params.iana_name); + info.push(0x00); + info.extend_from_slice(salt); + info.extend_from_slice(context); + + let algorithm = types::SHA512.get(py)?.call0()?; + let mut hkdf = HkdfExpand::new( + py, + algorithm.unbind(), + params.key_len + NONCE_LEN + COMMITMENT_LEN, + Some(pyo3::types::PyBytes::new(py, &info).unbind()), + None, + )?; + let okm = hkdf.derive(py, CffiBuf::from_bytes(py, input_key))?; + let okm_bytes = okm.as_bytes(); + + let key = pyo3::types::PyBytes::new(py, &okm_bytes[..params.key_len]); + let mut base_nonce = [0; NONCE_LEN]; + base_nonce.copy_from_slice(&okm_bytes[params.key_len..params.key_len + NONCE_LEN]); + let mut commitment = [0; COMMITMENT_LEN]; + commitment.copy_from_slice(&okm_bytes[params.key_len + NONCE_LEN..]); + Ok(DerivedKeys { + key, + base_nonce, + commitment, + }) +} + +// The per-chunk nonce sequence: the base nonce XOR'd with an incrementing +// chunk counter. +struct ChunkNonces { + base_nonce: [u8; NONCE_LEN], + counter: u64, +} + +impl ChunkNonces { + fn check_capacity(&self, n_chunks: u64) -> CryptographyResult<()> { + if n_chunks > MAX_CHUNK_COUNT - self.counter { + return Err(message_too_large_error()); + } + Ok(()) + } + + fn next(&mut self) -> CryptographyResult<[u8; NONCE_LEN]> { + self.check_capacity(1)?; + let mut nonce = self.base_nonce; + for (n, c) in nonce[NONCE_LEN - 8..] + .iter_mut() + .zip(self.counter.to_be_bytes()) + { + *n ^= c; + } + self.counter += 1; + Ok(nonce) + } +} + +// A single-message AEAD context that encrypts or decrypts successive chunks. +struct ChunkCipher { + aead: AesGcm, + nonces: ChunkNonces, + tag_len: usize, +} + +impl ChunkCipher { + fn new( + py: pyo3::Python<'_>, + params: &AeadParams, + keys: &DerivedKeys<'_>, + ) -> CryptographyResult { + let aead = AesGcm::new(py, keys.key.clone().unbind().into_any())?; + Ok(ChunkCipher { + aead, + nonces: ChunkNonces { + base_nonce: keys.base_nonce, + counter: 0, + }, + tag_len: params.tag_len, + }) + } + + // `out` must be exactly `plaintext.len() + self.tag_len` bytes. + fn encrypt_chunk( + &mut self, + py: pyo3::Python<'_>, + plaintext: &[u8], + out: &mut [u8], + ) -> CryptographyResult<()> { + let nonce = self.nonces.next()?; + self.aead.encrypt_into( + py, + CffiBuf::from_bytes(py, &nonce), + CffiBuf::from_bytes(py, plaintext), + None, + CffiMutBuf::from_bytes(py, out), + )?; + Ok(()) + } + + // `ciphertext` includes the trailing tag; `out` must be exactly + // `ciphertext.len() - self.tag_len` bytes. + fn decrypt_chunk( + &mut self, + py: pyo3::Python<'_>, + ciphertext: &[u8], + out: &mut [u8], + ) -> CryptographyResult<()> { + let nonce = self.nonces.next()?; + self.aead.decrypt_into( + py, + CffiBuf::from_bytes(py, &nonce), + CffiBuf::from_bytes(py, ciphertext), + None, + CffiMutBuf::from_bytes(py, out), + )?; + Ok(()) + } +} + +struct ChunkedEncryptor { + cipher: ChunkCipher, + buffer: Vec, + header: [u8; HEADER_LEN], + header_pending: bool, + finalized: bool, +} + +impl ChunkedEncryptor { + fn new( + py: pyo3::Python<'_>, + params: &AeadParams, + key: &[u8], + context: &[u8], + ) -> CryptographyResult { + check_key_length(params, key)?; + let mut salt = [0; SALT_LEN]; + cryptography_openssl::rand::rand_bytes(&mut salt)?; + let keys = derive_keys(py, params, key, &salt, context)?; + let cipher = ChunkCipher::new(py, params, &keys)?; + let mut header = [0; HEADER_LEN]; + header[..SALT_LEN].copy_from_slice(&salt); + header[SALT_LEN..].copy_from_slice(&keys.commitment); + Ok(ChunkedEncryptor { + cipher, + buffer: Vec::with_capacity(CHUNK_SIZE), + header, + header_pending: true, + finalized: false, + }) + } + + fn check_active(&self) -> CryptographyResult<()> { + if self.finalized { + return Err(exceptions::already_finalized_error()); + } + Ok(()) + } + + fn update_out_len(&self, data_len: usize) -> usize { + let n_chunks = (self.buffer.len() + data_len) / CHUNK_SIZE; + let header = if self.header_pending { HEADER_LEN } else { 0 }; + header + n_chunks * (CHUNK_SIZE + self.cipher.tag_len) + } + + // `out` must be at least `self.update_out_len(data.len())` bytes; both + // callers check this before any state is modified, so the header and + // every chunk written below are guaranteed to fit. + fn update_impl( + &mut self, + py: pyo3::Python<'_>, + mut data: &[u8], + out: &mut [u8], + ) -> CryptographyResult { + // Check the chunk counter limit up front so that the error doesn't + // leave a partially written output behind. + let n_chunks = ((self.buffer.len() + data.len()) / CHUNK_SIZE) as u64; + self.cipher.nonces.check_capacity(n_chunks)?; + + let wire_chunk = CHUNK_SIZE + self.cipher.tag_len; + let mut written = 0; + if self.header_pending { + out[..HEADER_LEN].copy_from_slice(&self.header); + written += HEADER_LEN; + self.header_pending = false; + } + if !self.buffer.is_empty() { + let take = std::cmp::min(CHUNK_SIZE - self.buffer.len(), data.len()); + self.buffer.extend_from_slice(&data[..take]); + data = &data[take..]; + if self.buffer.len() == CHUNK_SIZE { + self.cipher.encrypt_chunk( + py, + &self.buffer, + &mut out[written..written + wire_chunk], + )?; + written += wire_chunk; + self.buffer.clear(); + } + } + while data.len() >= CHUNK_SIZE { + let (chunk, rest) = data.split_at(CHUNK_SIZE); + self.cipher + .encrypt_chunk(py, chunk, &mut out[written..written + wire_chunk])?; + written += wire_chunk; + data = rest; + } + self.buffer.extend_from_slice(data); + Ok(written) + } + + fn update<'p>( + &mut self, + py: pyo3::Python<'p>, + data: &[u8], + ) -> CryptographyResult> { + self.check_active()?; + let out_len = self.update_out_len(data.len()); + Ok(pyo3::types::PyBytes::new_with(py, out_len, |b| { + let n = self.update_impl(py, data, b)?; + debug_assert_eq!(n, out_len); + Ok(()) + })?) + } + + fn update_into( + &mut self, + py: pyo3::Python<'_>, + data: &[u8], + out: &mut [u8], + ) -> CryptographyResult { + self.check_active()?; + let out_len = self.update_out_len(data.len()); + if out.len() < out_len { + return Err(CryptographyError::from( + pyo3::exceptions::PyValueError::new_err(format!( + "buffer must be at least {out_len} bytes" + )), + )); + } + self.update_impl(py, data, out) + } + + fn finalize<'p>( + &mut self, + py: pyo3::Python<'p>, + ) -> CryptographyResult> { + self.check_active()?; + self.finalized = true; + let header = if self.header_pending { HEADER_LEN } else { 0 }; + let out_len = header + self.buffer.len() + self.cipher.tag_len; + let result = pyo3::types::PyBytes::new_with(py, out_len, |b| { + if header != 0 { + b[..HEADER_LEN].copy_from_slice(&self.header); + } + self.cipher + .encrypt_chunk(py, &self.buffer, &mut b[header..])?; + Ok(()) + })?; + self.buffer.clear(); + Ok(result) + } +} + +enum DecrypterState { + // Buffering the salt and commitment; `key` and `context` are retained + // until they arrive and the keys can be derived. + Header { + key: Vec, + context: Vec, + buf: Vec, + }, + Body { + cipher: ChunkCipher, + buffer: Vec, + }, +} + +struct ChunkedDecryptor { + params: &'static AeadParams, + // `None` once finalized, or after any error: a failed context cannot + // process more data. + state: Option, +} + +impl ChunkedDecryptor { + fn new(params: &'static AeadParams, key: &[u8], context: &[u8]) -> CryptographyResult { + check_key_length(params, key)?; + Ok(ChunkedDecryptor { + params, + state: Some(DecrypterState::Header { + key: key.to_vec(), + context: context.to_vec(), + buf: Vec::with_capacity(HEADER_LEN), + }), + }) + } + + fn active_state(&mut self) -> CryptographyResult<&mut DecrypterState> { + match &mut self.state { + Some(state) => Ok(state), + None => Err(exceptions::already_finalized_error()), + } + } + + fn update_out_len(params: &AeadParams, state: &DecrypterState, data_len: usize) -> usize { + let body_len = match state { + DecrypterState::Header { buf, .. } => data_len.saturating_sub(HEADER_LEN - buf.len()), + DecrypterState::Body { buffer, .. } => buffer.len() + data_len, + }; + (body_len / params.wire_chunk_size()) * CHUNK_SIZE + } + + fn update_impl( + py: pyo3::Python<'_>, + params: &AeadParams, + state: &mut DecrypterState, + mut data: &[u8], + out: &mut [u8], + ) -> CryptographyResult { + match state { + DecrypterState::Header { key, context, buf } => { + let take = std::cmp::min(HEADER_LEN - buf.len(), data.len()); + buf.extend_from_slice(&data[..take]); + data = &data[take..]; + if buf.len() < HEADER_LEN { + debug_assert!(data.is_empty()); + return Ok(0); + } + let (salt, commitment) = buf.split_at(SALT_LEN); + let keys = derive_keys(py, params, key, salt, context)?; + if !cryptography_crypto::constant_time::bytes_eq(&keys.commitment, commitment) { + return Err(CryptographyError::from(exceptions::InvalidTag::new_err(()))); + } + let mut cipher = ChunkCipher::new(py, params, &keys)?; + let mut buffer = Vec::with_capacity(params.wire_chunk_size()); + let written = Self::decrypt_chunks(py, &mut cipher, &mut buffer, data, out)?; + *state = DecrypterState::Body { cipher, buffer }; + Ok(written) + } + DecrypterState::Body { cipher, buffer } => { + Self::decrypt_chunks(py, cipher, buffer, data, out) + } + } + } + + fn decrypt_chunks( + py: pyo3::Python<'_>, + cipher: &mut ChunkCipher, + buffer: &mut Vec, + mut data: &[u8], + out: &mut [u8], + ) -> CryptographyResult { + let wire_chunk = CHUNK_SIZE + cipher.tag_len; + // Check the chunk counter limit up front so that the error doesn't + // leave a partially written output behind. + let n_chunks = ((buffer.len() + data.len()) / wire_chunk) as u64; + cipher.nonces.check_capacity(n_chunks)?; + + // Any complete wire chunk is necessarily not the final chunk (the + // final chunk is always shorter), so it can be decrypted, and its + // plaintext released, as soon as it is available. If the ciphertext + // ends on a chunk boundary, the missing final chunk is detected in + // finalize(). + let mut written = 0; + if !buffer.is_empty() { + let take = std::cmp::min(wire_chunk - buffer.len(), data.len()); + buffer.extend_from_slice(&data[..take]); + data = &data[take..]; + if buffer.len() == wire_chunk { + cipher.decrypt_chunk(py, buffer, &mut out[written..written + CHUNK_SIZE])?; + written += CHUNK_SIZE; + buffer.clear(); + } + } + while data.len() >= wire_chunk { + let (chunk, rest) = data.split_at(wire_chunk); + cipher.decrypt_chunk(py, chunk, &mut out[written..written + CHUNK_SIZE])?; + written += CHUNK_SIZE; + data = rest; + } + buffer.extend_from_slice(data); + Ok(written) + } + + fn update<'p>( + &mut self, + py: pyo3::Python<'p>, + data: &[u8], + ) -> CryptographyResult> { + let params = self.params; + let state = self.active_state()?; + let out_len = Self::update_out_len(params, state, data.len()); + let result = pyo3::types::PyBytes::new_with(py, out_len, |b| { + let n = Self::update_impl(py, params, state, data, b)?; + debug_assert_eq!(n, out_len); + Ok(()) + }); + if result.is_err() { + self.state = None; + } + Ok(result?) + } + + fn update_into( + &mut self, + py: pyo3::Python<'_>, + data: &[u8], + out: &mut [u8], + ) -> CryptographyResult { + let params = self.params; + let state = self.active_state()?; + let out_len = Self::update_out_len(params, state, data.len()); + if out.len() < out_len { + return Err(CryptographyError::from( + pyo3::exceptions::PyValueError::new_err(format!( + "buffer must be at least {out_len} bytes" + )), + )); + } + let result = Self::update_impl(py, params, state, data, out); + if result.is_err() { + self.state = None; + } + result + } + + fn finalize<'p>( + &mut self, + py: pyo3::Python<'p>, + ) -> CryptographyResult> { + // Whether it succeeds or fails, finalization consumes the state: no + // further data may be processed. + match self.state.take() { + None => Err(exceptions::already_finalized_error()), + // The full salt and commitment never arrived: the message is + // truncated. + Some(DecrypterState::Header { .. }) => { + Err(CryptographyError::from(exceptions::InvalidTag::new_err(()))) + } + Some(DecrypterState::Body { mut cipher, buffer }) => { + if buffer.len() < cipher.tag_len { + // The final chunk is missing (the ciphertext ended + // exactly on a chunk boundary) or is too short to be + // valid: the message is truncated. + return Err(CryptographyError::from(exceptions::InvalidTag::new_err(()))); + } + // buffer is always shorter than a wire chunk, so the final + // chunk's plaintext is necessarily shorter than CHUNK_SIZE, + // as the specification requires. + let out_len = buffer.len() - cipher.tag_len; + Ok(pyo3::types::PyBytes::new_with(py, out_len, |b| { + cipher.decrypt_chunk(py, &buffer, b)?; + Ok(()) + })?) + } + } + } +} + +#[pyo3::pyclass(module = "cryptography.hazmat.bindings._rust.cobblestone")] +pub(crate) struct Cobblestone128Encryptor { + inner: ChunkedEncryptor, +} + +#[pyo3::pymethods] +impl Cobblestone128Encryptor { + #[new] + fn new( + py: pyo3::Python<'_>, + key: CffiBuf<'_>, + context: CffiBuf<'_>, + ) -> CryptographyResult { + Ok(Cobblestone128Encryptor { + inner: ChunkedEncryptor::new(py, &AES_128_GCM, key.as_bytes(), context.as_bytes())?, + }) + } + + #[staticmethod] + fn generate_key( + py: pyo3::Python<'_>, + ) -> CryptographyResult> { + crate::backend::rand::get_rand_bytes(py, AES_128_GCM.key_len) + } + + fn update<'p>( + &mut self, + py: pyo3::Python<'p>, + data: CffiBuf<'_>, + ) -> CryptographyResult> { + self.inner.update(py, data.as_bytes()) + } + + fn update_into( + &mut self, + py: pyo3::Python<'_>, + data: CffiBuf<'_>, + mut buf: CffiMutBuf<'_>, + ) -> CryptographyResult { + self.inner + .update_into(py, data.as_bytes(), buf.as_mut_bytes()) + } + + fn finalize<'p>( + &mut self, + py: pyo3::Python<'p>, + ) -> CryptographyResult> { + self.inner.finalize(py) + } +} + +#[pyo3::pyclass(module = "cryptography.hazmat.bindings._rust.cobblestone")] +pub(crate) struct Cobblestone128Decryptor { + inner: ChunkedDecryptor, +} + +#[pyo3::pymethods] +impl Cobblestone128Decryptor { + #[new] + fn new(key: CffiBuf<'_>, context: CffiBuf<'_>) -> CryptographyResult { + Ok(Cobblestone128Decryptor { + inner: ChunkedDecryptor::new(&AES_128_GCM, key.as_bytes(), context.as_bytes())?, + }) + } + + fn update<'p>( + &mut self, + py: pyo3::Python<'p>, + data: CffiBuf<'_>, + ) -> CryptographyResult> { + self.inner.update(py, data.as_bytes()) + } + + fn update_into( + &mut self, + py: pyo3::Python<'_>, + data: CffiBuf<'_>, + mut buf: CffiMutBuf<'_>, + ) -> CryptographyResult { + self.inner + .update_into(py, data.as_bytes(), buf.as_mut_bytes()) + } + + fn finalize<'p>( + &mut self, + py: pyo3::Python<'p>, + ) -> CryptographyResult> { + self.inner.finalize(py) + } +} + +#[pyo3::pyclass(module = "cryptography.hazmat.bindings._rust.cobblestone")] +pub(crate) struct Cobblestone256Encryptor { + inner: ChunkedEncryptor, +} + +#[pyo3::pymethods] +impl Cobblestone256Encryptor { + #[new] + fn new( + py: pyo3::Python<'_>, + key: CffiBuf<'_>, + context: CffiBuf<'_>, + ) -> CryptographyResult { + Ok(Cobblestone256Encryptor { + inner: ChunkedEncryptor::new(py, &AES_256_GCM, key.as_bytes(), context.as_bytes())?, + }) + } + + #[staticmethod] + fn generate_key( + py: pyo3::Python<'_>, + ) -> CryptographyResult> { + crate::backend::rand::get_rand_bytes(py, AES_256_GCM.key_len) + } + + fn update<'p>( + &mut self, + py: pyo3::Python<'p>, + data: CffiBuf<'_>, + ) -> CryptographyResult> { + self.inner.update(py, data.as_bytes()) + } + + fn update_into( + &mut self, + py: pyo3::Python<'_>, + data: CffiBuf<'_>, + mut buf: CffiMutBuf<'_>, + ) -> CryptographyResult { + self.inner + .update_into(py, data.as_bytes(), buf.as_mut_bytes()) + } + + fn finalize<'p>( + &mut self, + py: pyo3::Python<'p>, + ) -> CryptographyResult> { + self.inner.finalize(py) + } +} + +#[pyo3::pyclass(module = "cryptography.hazmat.bindings._rust.cobblestone")] +pub(crate) struct Cobblestone256Decryptor { + inner: ChunkedDecryptor, +} + +#[pyo3::pymethods] +impl Cobblestone256Decryptor { + #[new] + fn new(key: CffiBuf<'_>, context: CffiBuf<'_>) -> CryptographyResult { + Ok(Cobblestone256Decryptor { + inner: ChunkedDecryptor::new(&AES_256_GCM, key.as_bytes(), context.as_bytes())?, + }) + } + + fn update<'p>( + &mut self, + py: pyo3::Python<'p>, + data: CffiBuf<'_>, + ) -> CryptographyResult> { + self.inner.update(py, data.as_bytes()) + } + + fn update_into( + &mut self, + py: pyo3::Python<'_>, + data: CffiBuf<'_>, + mut buf: CffiMutBuf<'_>, + ) -> CryptographyResult { + self.inner + .update_into(py, data.as_bytes(), buf.as_mut_bytes()) + } + + fn finalize<'p>( + &mut self, + py: pyo3::Python<'p>, + ) -> CryptographyResult> { + self.inner.finalize(py) + } +} + +#[pyo3::pymodule(gil_used = false)] +#[pyo3(name = "cobblestone")] +pub(crate) mod cobblestone_mod { + #[pymodule_export] + use super::{ + Cobblestone128Decryptor, Cobblestone128Encryptor, Cobblestone256Decryptor, + Cobblestone256Encryptor, + }; +} + +#[cfg(test)] +mod tests { + use super::{ChunkNonces, MAX_CHUNK_COUNT, NONCE_LEN}; + + // Reaching the chunk counter limit requires processing a 4 PiB message, + // so this error path can't be exercised from the Python tests. + #[test] + fn test_chunk_nonces_counter_limit() { + let mut nonces = ChunkNonces { + base_nonce: [0; NONCE_LEN], + counter: MAX_CHUNK_COUNT - 1, + }; + assert!(nonces.check_capacity(1).is_ok()); + assert!(nonces.check_capacity(2).is_err()); + // The final counter value is usable... + assert!(nonces.next().is_ok()); + // ...but nothing past it. + assert!(nonces.check_capacity(0).is_ok()); + assert!(nonces.check_capacity(1).is_err()); + assert!(nonces.next().is_err()); + } +} diff --git a/src/rust/src/lib.rs b/src/rust/src/lib.rs index c72013d9c375..c7bcb3827201 100644 --- a/src/rust/src/lib.rs +++ b/src/rust/src/lib.rs @@ -34,6 +34,7 @@ use crate::error::CryptographyResult; mod asn1; mod backend; mod buf; +mod cobblestone; mod declarative_asn1; mod error; mod exceptions; @@ -132,6 +133,8 @@ mod _rust { #[pymodule_export] use crate::asn1::asn1_mod; #[pymodule_export] + use crate::cobblestone::cobblestone_mod; + #[pymodule_export] use crate::exceptions::exceptions; #[pymodule_export] use crate::oid::ObjectIdentifier; diff --git a/tests/test_cobblestone.py b/tests/test_cobblestone.py new file mode 100644 index 000000000000..e407509f6161 --- /dev/null +++ b/tests/test_cobblestone.py @@ -0,0 +1,407 @@ +# 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. + + +import os + +import pytest + +from cryptography.cobblestone import ( + Cobblestone128Decryptor, + Cobblestone128Encryptor, + Cobblestone256Decryptor, + Cobblestone256Encryptor, +) +from cryptography.exceptions import AlreadyFinalized, InvalidTag + +CHUNK_SIZE = 16 * 1024 +TAG_LEN = 16 +WIRE_CHUNK_SIZE = CHUNK_SIZE + TAG_LEN +SALT_LEN = 24 +COMMITMENT_LEN = 32 +HEADER_LEN = SALT_LEN + COMMITMENT_LEN + +VARIANTS = [ + pytest.param( + (Cobblestone128Encryptor, Cobblestone128Decryptor, 16), + id="cobblestone128", + ), + pytest.param( + (Cobblestone256Encryptor, Cobblestone256Decryptor, 32), + id="cobblestone256", + ), +] + + +def _encrypt_all(encryptor_cls, key: bytes, context: bytes, plaintext: bytes): + enc = encryptor_cls(key, context) + return enc.update(plaintext) + enc.finalize() + + +def _decrypt_all(decryptor_cls, key: bytes, context: bytes, ciphertext: bytes): + dec = decryptor_cls(key, context) + return dec.update(ciphertext) + dec.finalize() + + +MESSAGE_LENGTHS = [ + 0, + 1, + 57, + CHUNK_SIZE - 1, + CHUNK_SIZE, + CHUNK_SIZE + 1, + 20 * 1024, + 2 * CHUNK_SIZE - 1, + 2 * CHUNK_SIZE, + 2 * CHUNK_SIZE + 1, + 3 * CHUNK_SIZE + 5000, +] + + +@pytest.mark.parametrize("variant", VARIANTS) +class TestCobblestone: + @pytest.mark.parametrize("length", MESSAGE_LENGTHS) + def test_round_trip(self, variant, length): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + context = b"test context" + plaintext = os.urandom(length) + + ciphertext = _encrypt_all(encryptor_cls, key, context, plaintext) + n_chunks = length // CHUNK_SIZE + 1 + assert len(ciphertext) == HEADER_LEN + length + n_chunks * TAG_LEN + assert ( + _decrypt_all(decryptor_cls, key, context, ciphertext) == plaintext + ) + + @pytest.mark.parametrize("piece_size", [1, 57, 1024, 16384, 16400]) + def test_streaming(self, variant, piece_size): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + context = b"" + plaintext = os.urandom(2 * CHUNK_SIZE + 12345) + + enc = encryptor_cls(key, context) + ciphertext = b"" + for i in range(0, len(plaintext), piece_size): + ciphertext += enc.update(plaintext[i : i + piece_size]) + ciphertext += enc.finalize() + # The result matches a single-shot encryption's structure, and + # decrypts to the plaintext regardless of how the ciphertext is + # split up. + assert len(ciphertext) == HEADER_LEN + len(plaintext) + 3 * TAG_LEN + + dec = decryptor_cls(key, context) + decrypted = b"" + for i in range(0, len(ciphertext), piece_size): + decrypted += dec.update(ciphertext[i : i + piece_size]) + decrypted += dec.finalize() + assert decrypted == plaintext + + def test_empty_message(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + enc = encryptor_cls(key, b"ctx") + ciphertext = enc.finalize() + assert len(ciphertext) == HEADER_LEN + TAG_LEN + assert _decrypt_all(decryptor_cls, key, b"ctx", ciphertext) == b"" + + def test_update_with_empty_data_emits_header(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + enc = encryptor_cls(key, b"") + header = enc.update(b"") + assert len(header) == HEADER_LEN + assert enc.update(b"") == b"" + ciphertext = header + enc.finalize() + assert _decrypt_all(decryptor_cls, key, b"", ciphertext) == b"" + + def test_exact_chunk_boundary_has_empty_final_chunk(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + plaintext = os.urandom(CHUNK_SIZE) + ciphertext = _encrypt_all(encryptor_cls, key, b"", plaintext) + # One full chunk plus an empty final chunk. + assert len(ciphertext) == HEADER_LEN + WIRE_CHUNK_SIZE + TAG_LEN + assert _decrypt_all(decryptor_cls, key, b"", ciphertext) == plaintext + + def test_decrypter_streams_plaintext_incrementally(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + plaintext = os.urandom(3 * CHUNK_SIZE) + ciphertext = _encrypt_all(encryptor_cls, key, b"", plaintext) + + dec = decryptor_cls(key, b"") + out = dec.update(ciphertext[: HEADER_LEN + WIRE_CHUNK_SIZE]) + assert out == plaintext[:CHUNK_SIZE] + out = dec.update(ciphertext[HEADER_LEN + WIRE_CHUNK_SIZE :]) + assert out == plaintext[CHUNK_SIZE:] + assert dec.finalize() == b"" + + def test_wrong_key(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + ciphertext = _encrypt_all(encryptor_cls, key, b"", b"message") + dec = decryptor_cls(encryptor_cls.generate_key(), b"") + with pytest.raises(InvalidTag): + dec.update(ciphertext) + + def test_wrong_context(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + ciphertext = _encrypt_all(encryptor_cls, key, b"context a", b"msg") + dec = decryptor_cls(key, b"context b") + with pytest.raises(InvalidTag): + dec.update(ciphertext) + + @pytest.mark.parametrize( + "position", + [ + 0, # salt + SALT_LEN, # commitment + HEADER_LEN, # first chunk ciphertext + HEADER_LEN + WIRE_CHUNK_SIZE - 1, # first chunk tag + HEADER_LEN + WIRE_CHUNK_SIZE + 3, # final chunk + ], + ) + def test_tampering_detected(self, variant, position): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + plaintext = os.urandom(CHUNK_SIZE + 100) + ciphertext = bytearray( + _encrypt_all(encryptor_cls, key, b"", plaintext) + ) + ciphertext[position] ^= 1 + + dec = decryptor_cls(key, b"") + with pytest.raises(InvalidTag): + dec.update(bytes(ciphertext)) + dec.finalize() + + def test_swapped_chunks_detected(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + plaintext = os.urandom(2 * CHUNK_SIZE + 100) + ciphertext = _encrypt_all(encryptor_cls, key, b"", plaintext) + + chunk0_start = HEADER_LEN + chunk1_start = HEADER_LEN + WIRE_CHUNK_SIZE + chunk2_start = HEADER_LEN + 2 * WIRE_CHUNK_SIZE + swapped = ( + ciphertext[:HEADER_LEN] + + ciphertext[chunk1_start:chunk2_start] + + ciphertext[chunk0_start:chunk1_start] + + ciphertext[chunk2_start:] + ) + dec = decryptor_cls(key, b"") + with pytest.raises(InvalidTag): + dec.update(swapped) + + @pytest.mark.parametrize( + "length", + [ + 0, + 1, + HEADER_LEN - 1, + HEADER_LEN, # no final chunk at all + HEADER_LEN + TAG_LEN - 1, # final chunk shorter than its tag + HEADER_LEN + WIRE_CHUNK_SIZE, # ends on a chunk boundary + HEADER_LEN + WIRE_CHUNK_SIZE + TAG_LEN - 1, + ], + ) + def test_truncation_detected(self, variant, length): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + plaintext = os.urandom(CHUNK_SIZE + 100) + ciphertext = _encrypt_all(encryptor_cls, key, b"", plaintext) + + dec = decryptor_cls(key, b"") + with pytest.raises(InvalidTag): + dec.update(ciphertext[:length]) + dec.finalize() + + def test_extension_detected(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + ciphertext = _encrypt_all(encryptor_cls, key, b"", b"message") + + dec = decryptor_cls(key, b"") + with pytest.raises(InvalidTag): + dec.update(ciphertext + b"extra garbage bytes!") + dec.finalize() + + def test_ciphertexts_are_randomized(self, variant): + encryptor_cls, _, _ = variant + key = encryptor_cls.generate_key() + assert _encrypt_all(encryptor_cls, key, b"", b"data") != _encrypt_all( + encryptor_cls, key, b"", b"data" + ) + + def test_update_into(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + plaintext = os.urandom(CHUNK_SIZE + 100) + + enc = encryptor_cls(key, b"") + buf = bytearray(HEADER_LEN + 2 * WIRE_CHUNK_SIZE) + n = enc.update_into(plaintext, buf) + assert n == HEADER_LEN + WIRE_CHUNK_SIZE + ciphertext = bytes(buf[:n]) + enc.finalize() + assert _decrypt_all(decryptor_cls, key, b"", ciphertext) == plaintext + + dec = decryptor_cls(key, b"") + out = bytearray(2 * CHUNK_SIZE) + n = dec.update_into(ciphertext, out) + assert n == CHUNK_SIZE + assert bytes(out[:n]) + dec.finalize() == plaintext + + def test_update_into_accepts_larger_buffer(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + enc = encryptor_cls(key, b"") + buf = bytearray(10 * WIRE_CHUNK_SIZE) + n = enc.update_into(b"abc", buf) + assert n == HEADER_LEN + ciphertext = bytes(buf[:n]) + enc.finalize() + assert _decrypt_all(decryptor_cls, key, b"", ciphertext) == b"abc" + + def test_update_into_buffer_too_small(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + enc = encryptor_cls(key, b"") + with pytest.raises(ValueError, match="buffer must be at least"): + enc.update_into(b"abc", bytearray(HEADER_LEN - 1)) + # The context remains usable after the failed call. + ciphertext = enc.update(b"abc") + enc.finalize() + + dec = decryptor_cls(key, b"") + with pytest.raises(ValueError, match="buffer must be at least"): + dec.update_into( + ciphertext + bytes(WIRE_CHUNK_SIZE), bytearray(CHUNK_SIZE - 1) + ) + assert dec.update(ciphertext) == b"" + assert dec.finalize() == b"abc" + + def test_update_into_zero_output(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + dec = decryptor_cls(key, b"") + assert dec.update_into(b"", bytearray(0)) == 0 + + def test_use_after_finalize(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + enc = encryptor_cls(key, b"") + ciphertext = enc.update(b"data") + enc.finalize() + with pytest.raises(AlreadyFinalized): + enc.update(b"more") + with pytest.raises(AlreadyFinalized): + enc.update_into(b"more", bytearray(WIRE_CHUNK_SIZE)) + with pytest.raises(AlreadyFinalized): + enc.finalize() + + dec = decryptor_cls(key, b"") + dec.update(ciphertext) + dec.finalize() + with pytest.raises(AlreadyFinalized): + dec.update(b"more") + with pytest.raises(AlreadyFinalized): + dec.update_into(b"more", bytearray(WIRE_CHUNK_SIZE)) + with pytest.raises(AlreadyFinalized): + dec.finalize() + + def test_decryptor_unusable_after_invalid_tag(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + ciphertext = bytearray(_encrypt_all(encryptor_cls, key, b"", b"data")) + ciphertext[-1] ^= 1 + + dec = decryptor_cls(key, b"") + dec.update(bytes(ciphertext)) + with pytest.raises(InvalidTag): + dec.finalize() + # All subsequent operations fail. + with pytest.raises(AlreadyFinalized): + dec.update(b"") + with pytest.raises(AlreadyFinalized): + dec.finalize() + + def test_decryptor_update_into_unusable_after_invalid_tag(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + plaintext = os.urandom(CHUNK_SIZE + 100) + ciphertext = bytearray( + _encrypt_all(encryptor_cls, key, b"", plaintext) + ) + ciphertext[HEADER_LEN] ^= 1 # corrupt the first full chunk + + dec = decryptor_cls(key, b"") + buf = bytearray(2 * CHUNK_SIZE) + with pytest.raises(InvalidTag): + dec.update_into(bytes(ciphertext), buf) + # All subsequent operations fail. + with pytest.raises(AlreadyFinalized): + dec.update_into(b"", buf) + with pytest.raises(AlreadyFinalized): + dec.finalize() + + def test_finalize_only_decryptor_rejects_empty_stream(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + dec = decryptor_cls(key, b"") + with pytest.raises(InvalidTag): + dec.finalize() + + def test_generate_key(self, variant): + encryptor_cls, _, key_len = variant + key = encryptor_cls.generate_key() + assert isinstance(key, bytes) + assert len(key) == key_len + assert encryptor_cls.generate_key() != encryptor_cls.generate_key() + + def test_invalid_key_size(self, variant): + encryptor_cls, decryptor_cls, key_len = variant + for length in [0, key_len - 1, key_len + 1, 64]: + with pytest.raises(ValueError): + encryptor_cls(b"\x00" * length, b"") + with pytest.raises(ValueError): + decryptor_cls(b"\x00" * length, b"") + + def test_invalid_types(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = encryptor_cls.generate_key() + with pytest.raises(TypeError): + encryptor_cls("not bytes", b"") + with pytest.raises(TypeError): + encryptor_cls(key, "not bytes") + with pytest.raises(TypeError): + decryptor_cls("not bytes", b"") + enc = encryptor_cls(key, b"") + with pytest.raises(TypeError): + enc.update("not bytes") + with pytest.raises(TypeError): + enc.update_into(b"", b"immutable") + + def test_accepts_buffers(self, variant): + encryptor_cls, decryptor_cls, _ = variant + key = bytearray(encryptor_cls.generate_key()) + plaintext = os.urandom(1000) + enc = encryptor_cls(key, memoryview(b"ctx")) + ciphertext = enc.update(memoryview(plaintext)) + enc.finalize() + dec = decryptor_cls(memoryview(bytes(key)), bytearray(b"ctx")) + assert dec.update(bytearray(ciphertext)) + dec.finalize() == plaintext + + +class TestVariantsAreDistinct: + def test_key_sizes_differ(self): + assert len(Cobblestone128Encryptor.generate_key()) == 16 + assert len(Cobblestone256Encryptor.generate_key()) == 32 + + def test_cross_variant_key_sizes_rejected(self): + key128 = Cobblestone128Encryptor.generate_key() + key256 = Cobblestone256Encryptor.generate_key() + with pytest.raises(ValueError, match="key must be 16 bytes"): + Cobblestone128Encryptor(key256, b"") + with pytest.raises(ValueError, match="key must be 32 bytes"): + Cobblestone256Decryptor(key128, b"") diff --git a/tests/wycheproof/test_cobblestone.py b/tests/wycheproof/test_cobblestone.py new file mode 100644 index 000000000000..5787491c78e8 --- /dev/null +++ b/tests/wycheproof/test_cobblestone.py @@ -0,0 +1,56 @@ +# 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. + + +import binascii +import hashlib +import typing +import zlib + +import pytest + +from cryptography.cobblestone import ( + Cobblestone128Decryptor, + Cobblestone256Decryptor, +) +from cryptography.exceptions import AlreadyFinalized, InvalidTag + +from .utils import wycheproof_tests + +_DECRYPTORS: dict[str, typing.Any] = { + "AEAD_AES_128_GCM": Cobblestone128Decryptor, + "AEAD_AES_256_GCM": Cobblestone256Decryptor, +} + + +@wycheproof_tests( + "c2sp_chunked_encryption_aes_128_gcm_test.json", + "c2sp_chunked_encryption_aes_256_gcm_test.json", +) +def test_cobblestone(backend, wycheproof): + assert wycheproof.testgroup["sha"] == "SHA-512" + decryptor_cls = _DECRYPTORS[wycheproof.testgroup["aead"]] + key = binascii.unhexlify(wycheproof.testcase["key"]) + ctx = binascii.unhexlify(wycheproof.testcase["ctx"]) + ct = zlib.decompress(binascii.unhexlify(wycheproof.testcase["ct"])) + + if wycheproof.valid: + dec = decryptor_cls(key, ctx) + msg = dec.update(ct) + dec.finalize() + assert len(msg) == wycheproof.testcase["msgLength"] + assert ( + hashlib.sha512(msg).hexdigest() == wycheproof.testcase["msgSha512"] + ) + elif wycheproof.has_flag("InvalidKeySize"): + with pytest.raises(ValueError): + decryptor_cls(key, ctx) + else: + dec = decryptor_cls(key, ctx) + with pytest.raises(InvalidTag): + dec.update(ct) + dec.finalize() + # The failed context must keep failing rather than report a clean + # end of message. + with pytest.raises((InvalidTag, AlreadyFinalized)): + dec.finalize()