SIV support

.github/workflows/ci.yml

@@ -26,7 +26,7 @@ jobs:
         PYTHON:
           - {VERSION: "3.9", TOXENV: "flake", COVERAGE: "false"}
           - {VERSION: "3.9", TOXENV: "rust", COVERAGE: "false"}
-          - {VERSION: "3.9", TOXENV: "docs", COVERAGE: "false"}
+          - {VERSION: "3.9", TOXENV: "docs", COVERAGE: "false", OPENSSL: {TYPE: "openssl", VERSION: "3.0.2"}}
           - {VERSION: "pypy-3.7", TOXENV: "pypy3-nocoverage", COVERAGE: "false"}
           - {VERSION: "pypy-3.8", TOXENV: "pypy3-nocoverage", COVERAGE: "false"}
           - {VERSION: "3.9", TOXENV: "py39", OPENSSL: {TYPE: "openssl", VERSION: "1.1.0l"}}

CHANGELOG.rst

@@ -48,6 +48,9 @@ Changelog
 * Added support for 12-15 byte (96 to 120 bit) nonces to
   :class:`~cryptography.hazmat.primitives.ciphers.aead.AESOCB3`. This class
   previously supported only 12 byte (96 bit).
+* Added support for
+  :class:`~cryptography.hazmat.primitives.ciphers.aead.AESSIV` when using
+  OpenSSL 3.0.0+.
 
 .. _v36-0-2:
 

docs/hazmat/primitives/aead.rst

@@ -228,6 +228,92 @@ 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.
 
+.. class:: AESSIV(key)
+
+    .. versionadded:: 37.0
+
+    The SIV (synthetic initialization vector) construction is defined in
+    :rfc:`5297`. Depending on how it is used, SIV allows either
+    deterministic authenticated encryption or nonce-based,
+    misuse-resistant authenticated encryption.
+
+    :param key: A 256, 384, or 512-bit key (double sized from typical AES).
+        This **must** be kept secret.
+    :type key: :term:`bytes-like`
+
+    :raises cryptography.exceptions.UnsupportedAlgorithm: If the version of
+        OpenSSL does not support AES-SIV.
+
+    .. doctest::
+
+        >>> import os
+        >>> from cryptography.hazmat.primitives.ciphers.aead import AESSIV
+        >>> data = b"a secret message"
+        >>> nonce = os.urandom(16)
+        >>> aad = [b"authenticated but unencrypted data", nonce]
+        >>> key = AESSIV.generate_key(bit_length=512)  # AES256 requires 512-bit keys for SIV
+        >>> aessiv = AESSIV(key)
+        >>> ct = aessiv.encrypt(data, aad)
+        >>> aessiv.decrypt(ct, aad)
+        b'a secret message'
+
+    .. classmethod:: generate_key(bit_length)
+
+        Securely generates a random AES-SIV key.
+
+        :param bit_length: The bit length of the key to generate. Must be
+            256, 384, or 512. AES-SIV splits the key into an encryption and
+            MAC key, so these lengths correspond to AES 128, 192, and 256.
+
+        :returns bytes: The generated key.
+
+    .. method:: encrypt(data, associated_data)
+
+        .. note::
+
+            SIV performs nonce-based authenticated encryption when a component of
+            the associated data is a nonce. The final associated data in the
+            list is used for the nonce.
+
+            Random nonces should have at least 128-bits of entropy. If a nonce is
+            reused with SIV authenticity is retained and confidentiality is only
+            compromised to the extent that an attacker can determine that the
+            same plaintext (and same associated data) was protected with the same
+            nonce and key.
+
+            If you do not supply a nonce encryption is deterministic and the same
+            (plaintext, key) pair will always produce the same ciphertext.
+
+        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 data: The data to encrypt.
+        :param list associated_data: An optional ``list`` of ``bytes``. This
+            is additional data that should be authenticated with the key, but
+            is not encrypted. Can be ``None``.  In SIV mode the final element
+            of this list is treated as a ``nonce``.
+        :returns bytes: The ciphertext bytes with the 16 byte tag **prepended**.
+        :raises OverflowError: If ``data`` or an ``associated_data`` element
+            is larger than 2\ :sup:`31` - 1 bytes.
+
+    .. method:: decrypt(data, associated_data)
+
+        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 data: The data to decrypt (with tag **prepended**).
+        :param list associated_data: An optional ``list`` of ``bytes``. This
+            is additional data that should be authenticated with the key, but
+            is not encrypted. Can be ``None`` if none was used during
+            encryption.
+        :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 or associated data are wrong.
+
 .. class:: AESCCM(key, tag_length=16)
 
     .. versionadded:: 2.0

src/cryptography/hazmat/backends/openssl/aead.py

@@ -13,10 +13,13 @@
         AESCCM,
         AESGCM,
         AESOCB3,
+        AESSIV,
         ChaCha20Poly1305,
     )
 
-    _AEAD_TYPES = typing.Union[AESCCM, AESGCM, AESOCB3, ChaCha20Poly1305]
+    _AEAD_TYPES = typing.Union[
+        AESCCM, AESGCM, AESOCB3, AESSIV, ChaCha20Poly1305
+    ]
 
 _ENCRYPT = 1
 _DECRYPT = 0
@@ -27,6 +30,7 @@ def _aead_cipher_name(cipher: "_AEAD_TYPES") -> bytes:
         AESCCM,
         AESGCM,
         AESOCB3,
+        AESSIV,
         ChaCha20Poly1305,
     )
 
@@ -36,11 +40,29 @@ def _aead_cipher_name(cipher: "_AEAD_TYPES") -> bytes:
         return f"aes-{len(cipher._key) * 8}-ccm".encode("ascii")
     elif isinstance(cipher, AESOCB3):
         return f"aes-{len(cipher._key) * 8}-ocb".encode("ascii")
+    elif isinstance(cipher, AESSIV):
+        return f"aes-{len(cipher._key) * 8 // 2}-siv".encode("ascii")
     else:
         assert isinstance(cipher, AESGCM)
         return f"aes-{len(cipher._key) * 8}-gcm".encode("ascii")
 
 
+def _evp_cipher(cipher_name: bytes, backend: "Backend"):
+    if cipher_name.endswith(b"-siv"):
+        evp_cipher = backend._lib.EVP_CIPHER_fetch(
+            backend._ffi.NULL,
+            cipher_name,
+            backend._ffi.NULL,
+        )
+        backend.openssl_assert(evp_cipher != backend._ffi.NULL)
+        evp_cipher = backend._ffi.gc(evp_cipher, backend._lib.EVP_CIPHER_free)
+    else:
+        evp_cipher = backend._lib.EVP_get_cipherbyname(cipher_name)
+        backend.openssl_assert(evp_cipher != backend._ffi.NULL)
+
+    return evp_cipher
+
+
 def _aead_setup(
     backend: "Backend",
     cipher_name: bytes,
@@ -50,8 +72,7 @@ def _aead_setup(
     tag_len: int,
     operation: int,
 ):
-    evp_cipher = backend._lib.EVP_get_cipherbyname(cipher_name)
-    backend.openssl_assert(evp_cipher != backend._ffi.NULL)
+    evp_cipher = _evp_cipher(cipher_name, backend)
     ctx = backend._lib.EVP_CIPHER_CTX_new()
     ctx = backend._ffi.gc(ctx, backend._lib.EVP_CIPHER_CTX_free)
     res = backend._lib.EVP_CipherInit_ex(
@@ -118,7 +139,10 @@ def _process_data(backend: "Backend", ctx, data: bytes) -> bytes:
     outlen = backend._ffi.new("int *")
     buf = backend._ffi.new("unsigned char[]", len(data))
     res = backend._lib.EVP_CipherUpdate(ctx, buf, outlen, data, len(data))
-    backend.openssl_assert(res != 0)
+    if res == 0:
+        # AES SIV can error here if the data is invalid on decrypt
+        backend._consume_errors()
+        raise InvalidTag
     return backend._ffi.buffer(buf, outlen[0])[:]
 
 
@@ -130,7 +154,7 @@ def _encrypt(
     associated_data: typing.List[bytes],
     tag_length: int,
 ) -> bytes:
-    from cryptography.hazmat.primitives.ciphers.aead import AESCCM
+    from cryptography.hazmat.primitives.ciphers.aead import AESCCM, AESSIV
 
     cipher_name = _aead_cipher_name(cipher)
     ctx = _aead_setup(
@@ -159,7 +183,14 @@ def _encrypt(
     backend.openssl_assert(res != 0)
     tag = backend._ffi.buffer(tag_buf)[:]
 
-    return processed_data + tag
+    if isinstance(cipher, AESSIV):
+        # RFC 5297 defines the output as IV || C, where the tag we generate is
+        # the "IV" and C is the ciphertext. This is the opposite of our
+        # other AEADs, which are Ciphertext || Tag
+        backend.openssl_assert(len(tag) == 16)
+        return tag + processed_data
+    else:
+        return processed_data + tag
 
 
 def _decrypt(
@@ -170,12 +201,20 @@ def _decrypt(
     associated_data: typing.List[bytes],
     tag_length: int,
 ) -> bytes:
-    from cryptography.hazmat.primitives.ciphers.aead import AESCCM
+    from cryptography.hazmat.primitives.ciphers.aead import AESCCM, AESSIV
 
     if len(data) < tag_length:
         raise InvalidTag
-    tag = data[-tag_length:]
-    data = data[:-tag_length]
+
+    if isinstance(cipher, AESSIV):
+        # RFC 5297 defines the output as IV || C, where the tag we generate is
+        # the "IV" and C is the ciphertext. This is the opposite of our
+        # other AEADs, which are Ciphertext || Tag
+        tag = data[:tag_length]
+        data = data[tag_length:]
+    else:
+        tag = data[-tag_length:]
+        data = data[:-tag_length]
     cipher_name = _aead_cipher_name(cipher)
     ctx = _aead_setup(
         backend, cipher_name, cipher._key, nonce, tag, tag_length, _DECRYPT

src/cryptography/hazmat/backends/openssl/backend.py

@@ -2035,7 +2035,15 @@ def aead_cipher_supported(self, cipher) -> bool:
         cipher_name = aead._aead_cipher_name(cipher)
         if self._fips_enabled and cipher_name not in self._fips_aead:
             return False
-        return self._lib.EVP_get_cipherbyname(cipher_name) != self._ffi.NULL
+        # SIV isn't loaded through get_cipherbyname but instead a new fetch API
+        # only available in 3.0+. But if we know we're on 3.0+ then we know
+        # it's supported.
+        if cipher_name.endswith(b"-siv"):
+            return self._lib.CRYPTOGRAPHY_OPENSSL_300_OR_GREATER == 1
+        else:
+            return (
+                self._lib.EVP_get_cipherbyname(cipher_name) != self._ffi.NULL
+            )
 
     @contextlib.contextmanager
     def _zeroed_bytearray(self, length: int) -> typing.Iterator[bytearray]:

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

@@ -289,3 +289,73 @@ def _check_params(
         utils._check_bytes("associated_data", associated_data)
         if len(nonce) < 12 or len(nonce) > 15:
             raise ValueError("Nonce must be between 12 and 15 bytes")
+
+
+class AESSIV(object):
+    _MAX_SIZE = 2**31 - 1
+
+    def __init__(self, key: bytes):
+        utils._check_byteslike("key", key)
+        if len(key) not in (32, 48, 64):
+            raise ValueError("AESSIV key must be 256, 384, or 512 bits.")
+
+        self._key = key
+
+        if not backend.aead_cipher_supported(self):
+            raise exceptions.UnsupportedAlgorithm(
+                "AES-SIV is not supported by this version of OpenSSL",
+                exceptions._Reasons.UNSUPPORTED_CIPHER,
+            )
+
+    @classmethod
+    def generate_key(cls, bit_length: int) -> bytes:
+        if not isinstance(bit_length, int):
+            raise TypeError("bit_length must be an integer")
+
+        if bit_length not in (256, 384, 512):
+            raise ValueError("bit_length must be 256, 384, or 512")
+
+        return os.urandom(bit_length // 8)
+
+    def encrypt(
+        self,
+        data: bytes,
+        associated_data: typing.Optional[typing.List[bytes]],
+    ) -> bytes:
+        if associated_data is None:
+            associated_data = []
+
+        self._check_params(data, associated_data)
+
+        if len(data) > self._MAX_SIZE or any(
+            len(ad) > self._MAX_SIZE for ad in associated_data
+        ):
+            # This is OverflowError to match what cffi would raise
+            raise OverflowError(
+                "Data or associated data too long. Max 2**31 - 1 bytes"
+            )
+
+        return aead._encrypt(backend, self, b"", data, associated_data, 16)
+
+    def decrypt(
+        self,
+        data: bytes,
+        associated_data: typing.Optional[typing.List[bytes]],
+    ) -> bytes:
+        if associated_data is None:
+            associated_data = []
+
+        self._check_params(data, associated_data)
+
+        return aead._decrypt(backend, self, b"", data, associated_data, 16)
+
+    def _check_params(
+        self,
+        data: bytes,
+        associated_data: typing.List,
+    ) -> None:
+        utils._check_bytes("data", data)
+        if not isinstance(associated_data, list) or not all(
+            isinstance(x, bytes) for x in associated_data
+        ):
+            raise TypeError("associated_data must be a list of bytes or None")