AESCCM support (#3700)

2 files changed, 110 insertions, 9 deletions

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")