diff options
-rw-r--r-- | CHANGELOG.rst | 2 | ||||
-rw-r--r-- | docs/hazmat/primitives/aead.rst | 88 | ||||
-rw-r--r-- | src/cryptography/hazmat/backends/openssl/aead.py | 56 | ||||
-rw-r--r-- | src/cryptography/hazmat/primitives/ciphers/aead.py | 63 | ||||
-rw-r--r-- | tests/hazmat/primitives/test_aead.py | 157 |
5 files changed, 351 insertions, 15 deletions
diff --git a/CHANGELOG.rst b/CHANGELOG.rst index cc993beb..278d977a 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -27,6 +27,8 @@ Changelog objects from X.509 certificate extensions. * Added support for :class:`~cryptography.hazmat.primitives.ciphers.aead.ChaCha20Poly1305`. +* Added support for + :class:`~cryptography.hazmat.primitives.ciphers.aead.AESCCM`. * Added support for :doc:`/hazmat/primitives/asymmetric/x25519`. * Added support for serializing and deserializing Diffie-Hellman parameters with diff --git a/docs/hazmat/primitives/aead.rst b/docs/hazmat/primitives/aead.rst index 54343b86..94b08f0a 100644 --- a/docs/hazmat/primitives/aead.rst +++ b/docs/hazmat/primitives/aead.rst @@ -77,3 +77,91 @@ also support providing integrity for associated data which is not encrypted. 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:: AESCCM(key) + + .. versionadded:: 2.0 + + .. note: + + AES-CCM is provided largely for compatibility with existing protocols. + Due to its construction it is not as computationally efficient as + other AEAD ciphers. + + The AES-CCM construction is composed of the + :class:`~cryptography.hazmat.primitives.ciphers.algorithms.AES` block + cipher utilizing Counter with CBC-MAC (CCM) (specified in :rfc:`3610`). + + :param bytes key: A 128, 192, or 256-bit key. This **must** be kept secret. + + :raises cryptography.exceptions.UnsupportedAlgorithm: If the version of + OpenSSL does not support AES-CCM. + + .. doctest:: + + >>> import os + >>> from cryptography.hazmat.primitives.ciphers.aead import AESCCM + >>> data = b"a secret message" + >>> aad = b"authenticated but unencrypted data" + >>> key = AESCCM.generate_key(bit_length=128) + >>> aesccm = AESCCM(key) + >>> nonce = os.urandom(13) + >>> ct = aesccm.encrypt(nonce, data, aad) + >>> aesccm.decrypt(nonce, ct, aad) + 'a secret message' + + .. classmethod:: generate_key(bit_length) + + Securely generates a random AES-CCM key. + + :param bit_length: The bit length of the key to generate. Must be + 128, 192, or 256. + + :returns bytes: The generated key. + + .. method:: encrypt(nonce, data, associated_data, tag_length=16) + + .. warning:: + + Reuse of a ``nonce`` with a given ``key`` compromises the security + of any message with that ``nonce`` and ``key`` pair. + + Encrypts and authenticates the ``data`` provided as well as + authenticating the ``associated_data``. The output of this can be + passed directly to the ``decrypt`` method. + + :param bytes nonce: A value of between 7 and 13 bytes. The maximum + length is determined by the length of the ciphertext you are + encrypting and must satisfy the condition: + ``len(data) < 2 ** (8 * (15 - len(nonce)))`` + **NEVER REUSE A NONCE** with a key. + :param bytes data: The data to encrypt. + :param bytes associated_data: Additional data that should be + authenticated with the key, but is not encrypted. Can be ``None``. + :param int tag_length: The length of the authentication tag. This + defaults to 16 bytes and it is **strongly** recommended that you + do not make it shorter unless absolutely necessary. Valid tag + lengths are 4, 6, 8, 12, 14, and 16. + :returns bytes: The ciphertext bytes with the tag appended. + + .. method:: decrypt(nonce, data, associated_data, tag_length=16) + + 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. + + :param bytes nonce: A value of between 7 and 13 bytes. This + is the same value used when you originally called encrypt. + **NEVER REUSE A NONCE** with a key. + :param bytes data: The data to decrypt (with tag appended). + :param bytes associated_data: Additional data to authenticate. Can be + ``None`` if none was passed during encryption. + :param int tag_length: The length of the authentication tag. This + defaults to 16 bytes. You only need to change this if your existing + ciphertext has a shorter tag. Valid tag lengths are 4, 6, 8, 12, + 14, and 16. + :returns bytes: The original plaintext. + :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. diff --git a/src/cryptography/hazmat/backends/openssl/aead.py b/src/cryptography/hazmat/backends/openssl/aead.py index 4fde6eae..5402acb3 100644 --- a/src/cryptography/hazmat/backends/openssl/aead.py +++ b/src/cryptography/hazmat/backends/openssl/aead.py @@ -13,10 +13,13 @@ _DECRYPT = 0 def _aead_cipher_name(cipher): from cryptography.hazmat.primitives.ciphers.aead import ( - ChaCha20Poly1305 + AESCCM, ChaCha20Poly1305 ) - assert isinstance(cipher, ChaCha20Poly1305) - return b"chacha20-poly1305" + if isinstance(cipher, ChaCha20Poly1305): + return b"chacha20-poly1305" + else: + assert isinstance(cipher, AESCCM) + return "aes-{0}-ccm".format(len(cipher._key) * 8).encode("ascii") def _aead_setup(backend, cipher_name, key, nonce, tag, tag_len, operation): @@ -61,6 +64,18 @@ def _aead_setup(backend, cipher_name, key, nonce, tag, tag_len, operation): return ctx +def _set_length(backend, ctx, data_len): + intptr = backend._ffi.new("int *") + res = backend._lib.EVP_CipherUpdate( + ctx, + backend._ffi.NULL, + intptr, + backend._ffi.NULL, + data_len + ) + backend.openssl_assert(res != 0) + + def _process_aad(backend, ctx, associated_data): outlen = backend._ffi.new("int *") res = backend._lib.EVP_CipherUpdate( @@ -78,10 +93,15 @@ def _process_data(backend, ctx, data): def _encrypt(backend, cipher, nonce, data, associated_data, tag_length): + from cryptography.hazmat.primitives.ciphers.aead import AESCCM cipher_name = _aead_cipher_name(cipher) ctx = _aead_setup( backend, cipher_name, cipher._key, nonce, None, tag_length, _ENCRYPT ) + # CCM requires us to pass the length of the data before processing anything + # However calling this with any other AEAD results in an error + if isinstance(cipher, AESCCM): + _set_length(backend, ctx, len(data)) _process_aad(backend, ctx, associated_data) processed_data = _process_data(backend, ctx, data) @@ -100,6 +120,7 @@ def _encrypt(backend, cipher, nonce, data, associated_data, tag_length): def _decrypt(backend, cipher, nonce, data, associated_data, tag_length): + from cryptography.hazmat.primitives.ciphers.aead import AESCCM if len(data) < tag_length: raise InvalidTag tag = data[-tag_length:] @@ -108,12 +129,29 @@ def _decrypt(backend, cipher, nonce, data, associated_data, tag_length): ctx = _aead_setup( backend, cipher_name, cipher._key, nonce, tag, tag_length, _DECRYPT ) + # CCM requires us to pass the length of the data before processing anything + # However calling this with any other AEAD results in an error + if isinstance(cipher, AESCCM): + _set_length(backend, ctx, len(data)) + _process_aad(backend, ctx, associated_data) - processed_data = _process_data(backend, ctx, data) - outlen = backend._ffi.new("int *") - res = backend._lib.EVP_CipherFinal_ex(ctx, backend._ffi.NULL, outlen) - if res == 0: - backend._consume_errors() - raise InvalidTag + # CCM has a different error path if the tag doesn't match. Errors are + # raised in Update and Final is irrelevant. + if isinstance(cipher, AESCCM): + outlen = backend._ffi.new("int *") + buf = backend._ffi.new("unsigned char[]", len(data)) + res = backend._lib.EVP_CipherUpdate(ctx, buf, outlen, data, len(data)) + if res != 1: + backend._consume_errors() + raise InvalidTag + + processed_data = backend._ffi.buffer(buf, outlen[0])[:] + else: + processed_data = _process_data(backend, ctx, data) + outlen = backend._ffi.new("int *") + res = backend._lib.EVP_CipherFinal_ex(ctx, backend._ffi.NULL, outlen) + if res == 0: + backend._consume_errors() + raise InvalidTag return processed_data diff --git a/src/cryptography/hazmat/primitives/ciphers/aead.py b/src/cryptography/hazmat/primitives/ciphers/aead.py index 8df55fab..189cb5b1 100644 --- a/src/cryptography/hazmat/primitives/ciphers/aead.py +++ b/src/cryptography/hazmat/primitives/ciphers/aead.py @@ -53,3 +53,66 @@ class ChaCha20Poly1305(object): utils._check_bytes("associated_data", associated_data) if len(nonce) != 12: raise ValueError("Nonce must be 12 bytes") + + +class AESCCM(object): + def __init__(self, key): + utils._check_bytes("key", key) + if len(key) not in (16, 24, 32): + raise ValueError("AESCCM key must be 128, 192, or 256 bits.") + + self._key = key + if not backend.aead_cipher_supported(self): + raise exceptions.UnsupportedAlgorithm( + "AESCCM is not supported by this version of OpenSSL", + exceptions._Reasons.UNSUPPORTED_CIPHER + ) + + @classmethod + def generate_key(cls, bit_length): + if not isinstance(bit_length, int): + raise TypeError("bit_length must be an integer") + + if bit_length not in (128, 192, 256): + raise ValueError("bit_length must be 128, 192, or 256") + + return os.urandom(bit_length // 8) + + def encrypt(self, nonce, data, associated_data, tag_length=16): + if associated_data is None: + associated_data = b"" + + self._check_params(nonce, data, associated_data, tag_length) + self._validate_lengths(nonce, len(data)) + return aead._encrypt( + backend, self, nonce, data, associated_data, tag_length + ) + + def decrypt(self, nonce, data, associated_data, tag_length=16): + if associated_data is None: + associated_data = b"" + + self._check_params(nonce, data, associated_data, tag_length) + return aead._decrypt( + backend, self, nonce, data, associated_data, tag_length + ) + + def _validate_lengths(self, nonce, data_len): + # For information about computing this, see + # https://tools.ietf.org/html/rfc3610#section-2.1 + l = 15 - len(nonce) + if 2 ** (8 * l) < data_len: + raise ValueError("Nonce too long for data") + + def _check_params(self, nonce, data, associated_data, tag_length): + if not isinstance(tag_length, int): + raise TypeError("tag_length must be an integer") + + if tag_length not in (4, 6, 8, 12, 14, 16): + raise ValueError("Invalid tag_length") + + utils._check_bytes("nonce", nonce) + utils._check_bytes("data", data) + utils._check_bytes("associated_data", associated_data) + if not 7 <= len(nonce) <= 13: + raise ValueError("Nonce must be between 7 and 13 bytes") diff --git a/tests/hazmat/primitives/test_aead.py b/tests/hazmat/primitives/test_aead.py index a1ca5ae7..9700a1ab 100644 --- a/tests/hazmat/primitives/test_aead.py +++ b/tests/hazmat/primitives/test_aead.py @@ -11,23 +11,27 @@ import pytest from cryptography.exceptions import InvalidTag, UnsupportedAlgorithm, _Reasons from cryptography.hazmat.backends.interfaces import CipherBackend -from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305 +from cryptography.hazmat.primitives.ciphers.aead import ( + AESCCM, ChaCha20Poly1305 +) +from .utils import _load_all_params from ...utils import ( - load_nist_vectors, load_vectors_from_file, raises_unsupported_algorithm + load_nist_ccm_vectors, load_nist_vectors, load_vectors_from_file, + raises_unsupported_algorithm ) -def _chacha20poly1305_supported(): +def _aead_supported(cls): try: - ChaCha20Poly1305(b"0" * 32) + cls(b"0" * 32) return True except UnsupportedAlgorithm: return False @pytest.mark.skipif( - _chacha20poly1305_supported(), + _aead_supported(ChaCha20Poly1305), reason="Requires OpenSSL without ChaCha20Poly1305 support" ) @pytest.mark.requires_backend_interface(interface=CipherBackend) @@ -37,7 +41,7 @@ def test_chacha20poly1305_unsupported_on_older_openssl(backend): @pytest.mark.skipif( - not _chacha20poly1305_supported(), + not _aead_supported(ChaCha20Poly1305), reason="Does not support ChaCha20Poly1305" ) @pytest.mark.requires_backend_interface(interface=CipherBackend) @@ -146,3 +150,144 @@ class TestChaCha20Poly1305(object): assert computed_pt == pt computed_ct = chacha.encrypt(nonce, pt, aad) assert computed_ct == ct + tag + + +@pytest.mark.skipif( + _aead_supported(AESCCM), + reason="Requires OpenSSL without AES-CCM support" +) +@pytest.mark.requires_backend_interface(interface=CipherBackend) +def test_aesccm_unsupported_on_older_openssl(backend): + with raises_unsupported_algorithm(_Reasons.UNSUPPORTED_CIPHER): + AESCCM(AESCCM.generate_key(128)) + + +@pytest.mark.skipif( + not _aead_supported(AESCCM), + reason="Does not support AESCCM" +) +@pytest.mark.requires_backend_interface(interface=CipherBackend) +class TestAESCCM(object): + def test_default_tag_length(self, backend): + key = AESCCM.generate_key(128) + aesccm = AESCCM(key) + nonce = os.urandom(12) + pt = b"hello" + ct = aesccm.encrypt(nonce, pt, None) + assert len(ct) == len(pt) + 16 + + def test_invalid_tag_length(self, backend): + key = AESCCM.generate_key(128) + aesccm = AESCCM(key) + pt = b"hello" + nonce = os.urandom(12) + with pytest.raises(ValueError): + aesccm.encrypt(nonce, pt, None, tag_length=7) + + with pytest.raises(ValueError): + aesccm.encrypt(nonce, pt, None, tag_length=2) + + def test_invalid_nonce_length(self, backend): + key = AESCCM.generate_key(128) + aesccm = AESCCM(key) + pt = b"hello" + nonce = os.urandom(14) + with pytest.raises(ValueError): + aesccm.encrypt(nonce, pt, None) + + with pytest.raises(ValueError): + aesccm.encrypt(nonce[:6], pt, None) + + @pytest.mark.parametrize( + "vector", + _load_all_params( + os.path.join("ciphers", "AES", "CCM"), + [ + "DVPT128.rsp", "DVPT192.rsp", "DVPT256.rsp", + "VADT128.rsp", "VADT192.rsp", "VADT256.rsp", + "VNT128.rsp", "VNT192.rsp", "VNT256.rsp", + "VPT128.rsp", "VPT192.rsp", "VPT256.rsp", + ], + load_nist_ccm_vectors + ) + ) + def test_vectors(self, vector, backend): + key = binascii.unhexlify(vector["key"]) + nonce = binascii.unhexlify(vector["nonce"]) + adata = binascii.unhexlify(vector["adata"])[:vector["alen"]] + ct = binascii.unhexlify(vector["ct"]) + pt = binascii.unhexlify(vector["payload"])[:vector["plen"]] + aesccm = AESCCM(key) + if vector.get('fail'): + with pytest.raises(InvalidTag): + aesccm.decrypt(nonce, ct, adata, vector["tlen"]) + else: + computed_pt = aesccm.decrypt(nonce, ct, adata, vector["tlen"]) + assert computed_pt == pt + assert aesccm.encrypt(nonce, pt, adata, vector["tlen"]) == ct + + def test_roundtrip(self, backend): + key = AESCCM.generate_key(128) + aesccm = AESCCM(key) + pt = b"encrypt me" + ad = b"additional" + nonce = os.urandom(12) + ct = aesccm.encrypt(nonce, pt, ad, 16) + computed_pt = aesccm.decrypt(nonce, ct, ad, 16) + assert computed_pt == pt + + def test_nonce_too_long(self, backend): + key = AESCCM.generate_key(128) + aesccm = AESCCM(key) + pt = b"encrypt me" * 6600 + # pt can be no more than 65536 bytes when nonce is 13 bytes + nonce = os.urandom(13) + with pytest.raises(ValueError): + aesccm.encrypt(nonce, pt, None, 16) + + @pytest.mark.parametrize( + ("nonce", "data", "associated_data", "tag_length"), + [ + [object(), b"data", b"", 16], + [b"0" * 12, object(), b"", 16], + [b"0" * 12, b"data", object(), 16], + [b"0" * 12, b"data", b"", object()] + ] + ) + def test_params_not_bytes(self, nonce, data, associated_data, tag_length, + backend): + key = AESCCM.generate_key(128) + aesccm = AESCCM(key) + with pytest.raises(TypeError): + aesccm.encrypt(nonce, data, associated_data, tag_length) + + def test_bad_key(self, backend): + with pytest.raises(TypeError): + AESCCM(object()) + + with pytest.raises(ValueError): + AESCCM(b"0" * 31) + + def test_bad_generate_key(self, backend): + with pytest.raises(TypeError): + AESCCM.generate_key(object()) + + with pytest.raises(ValueError): + AESCCM.generate_key(129) + + def test_associated_data_none_equal_to_empty_bytestring(self, backend): + key = AESCCM.generate_key(128) + aesccm = AESCCM(key) + nonce = os.urandom(12) + ct1 = aesccm.encrypt(nonce, b"some_data", None) + ct2 = aesccm.encrypt(nonce, b"some_data", b"") + assert ct1 == ct2 + pt1 = aesccm.decrypt(nonce, ct1, None) + pt2 = aesccm.decrypt(nonce, ct2, b"") + assert pt1 == pt2 + + def test_decrypt_data_too_short(self, backend): + key = AESCCM.generate_key(128) + aesccm = AESCCM(key) + with pytest.raises(InvalidTag): + aesccm.decrypt(b"0" * 12, b"0", None) |