This commit is contained in:
Iliyan Angelov
2025-11-19 12:27:01 +02:00
parent 2043ac897c
commit 34b4c969d4
469 changed files with 26870 additions and 8329 deletions

View File

@@ -17,11 +17,11 @@ from cryptography.hazmat.primitives.ciphers.base import (
)
__all__ = [
"Cipher",
"CipherAlgorithm",
"BlockCipherAlgorithm",
"CipherContext",
"AEADCipherContext",
"AEADDecryptionContext",
"AEADEncryptionContext",
"BlockCipherAlgorithm",
"Cipher",
"CipherAlgorithm",
"CipherContext",
]

View File

@@ -4,375 +4,20 @@
from __future__ import annotations
import os
import typing
from cryptography import exceptions, utils
from cryptography.hazmat.backends.openssl import aead
from cryptography.hazmat.backends.openssl.backend import backend
from cryptography.hazmat.bindings._rust import FixedPool
class ChaCha20Poly1305:
_MAX_SIZE = 2**31 - 1
def __init__(self, key: bytes):
if not backend.aead_cipher_supported(self):
raise exceptions.UnsupportedAlgorithm(
"ChaCha20Poly1305 is not supported by this version of OpenSSL",
exceptions._Reasons.UNSUPPORTED_CIPHER,
)
utils._check_byteslike("key", key)
if len(key) != 32:
raise ValueError("ChaCha20Poly1305 key must be 32 bytes.")
self._key = key
self._pool = FixedPool(self._create_fn)
@classmethod
def generate_key(cls) -> bytes:
return os.urandom(32)
def _create_fn(self):
return aead._aead_create_ctx(backend, self, self._key)
def encrypt(
self,
nonce: bytes,
data: bytes,
associated_data: typing.Optional[bytes],
) -> bytes:
if associated_data is None:
associated_data = b""
if len(data) > self._MAX_SIZE or len(associated_data) > self._MAX_SIZE:
# This is OverflowError to match what cffi would raise
raise OverflowError(
"Data or associated data too long. Max 2**31 - 1 bytes"
)
self._check_params(nonce, data, associated_data)
with self._pool.acquire() as ctx:
return aead._encrypt(
backend, self, nonce, data, [associated_data], 16, ctx
)
def decrypt(
self,
nonce: bytes,
data: bytes,
associated_data: typing.Optional[bytes],
) -> bytes:
if associated_data is None:
associated_data = b""
self._check_params(nonce, data, associated_data)
with self._pool.acquire() as ctx:
return aead._decrypt(
backend, self, nonce, data, [associated_data], 16, ctx
)
def _check_params(
self,
nonce: bytes,
data: bytes,
associated_data: bytes,
) -> None:
utils._check_byteslike("nonce", nonce)
utils._check_byteslike("data", data)
utils._check_byteslike("associated_data", associated_data)
if len(nonce) != 12:
raise ValueError("Nonce must be 12 bytes")
class AESCCM:
_MAX_SIZE = 2**31 - 1
def __init__(self, key: bytes, tag_length: int = 16):
utils._check_byteslike("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 isinstance(tag_length, int):
raise TypeError("tag_length must be an integer")
if tag_length not in (4, 6, 8, 10, 12, 14, 16):
raise ValueError("Invalid tag_length")
self._tag_length = tag_length
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: int) -> bytes:
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: bytes,
data: bytes,
associated_data: typing.Optional[bytes],
) -> bytes:
if associated_data is None:
associated_data = b""
if len(data) > self._MAX_SIZE or len(associated_data) > self._MAX_SIZE:
# This is OverflowError to match what cffi would raise
raise OverflowError(
"Data or associated data too long. Max 2**31 - 1 bytes"
)
self._check_params(nonce, data, associated_data)
self._validate_lengths(nonce, len(data))
return aead._encrypt(
backend, self, nonce, data, [associated_data], self._tag_length
)
def decrypt(
self,
nonce: bytes,
data: bytes,
associated_data: typing.Optional[bytes],
) -> bytes:
if associated_data is None:
associated_data = b""
self._check_params(nonce, data, associated_data)
return aead._decrypt(
backend, self, nonce, data, [associated_data], self._tag_length
)
def _validate_lengths(self, nonce: bytes, data_len: int) -> None:
# For information about computing this, see
# https://tools.ietf.org/html/rfc3610#section-2.1
l_val = 15 - len(nonce)
if 2 ** (8 * l_val) < data_len:
raise ValueError("Data too long for nonce")
def _check_params(
self, nonce: bytes, data: bytes, associated_data: bytes
) -> None:
utils._check_byteslike("nonce", nonce)
utils._check_byteslike("data", data)
utils._check_byteslike("associated_data", associated_data)
if not 7 <= len(nonce) <= 13:
raise ValueError("Nonce must be between 7 and 13 bytes")
class AESGCM:
_MAX_SIZE = 2**31 - 1
def __init__(self, key: bytes):
utils._check_byteslike("key", key)
if len(key) not in (16, 24, 32):
raise ValueError("AESGCM key must be 128, 192, or 256 bits.")
self._key = key
@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 (128, 192, 256):
raise ValueError("bit_length must be 128, 192, or 256")
return os.urandom(bit_length // 8)
def encrypt(
self,
nonce: bytes,
data: bytes,
associated_data: typing.Optional[bytes],
) -> bytes:
if associated_data is None:
associated_data = b""
if len(data) > self._MAX_SIZE or len(associated_data) > self._MAX_SIZE:
# This is OverflowError to match what cffi would raise
raise OverflowError(
"Data or associated data too long. Max 2**31 - 1 bytes"
)
self._check_params(nonce, data, associated_data)
return aead._encrypt(backend, self, nonce, data, [associated_data], 16)
def decrypt(
self,
nonce: bytes,
data: bytes,
associated_data: typing.Optional[bytes],
) -> bytes:
if associated_data is None:
associated_data = b""
self._check_params(nonce, data, associated_data)
return aead._decrypt(backend, self, nonce, data, [associated_data], 16)
def _check_params(
self,
nonce: bytes,
data: bytes,
associated_data: bytes,
) -> None:
utils._check_byteslike("nonce", nonce)
utils._check_byteslike("data", data)
utils._check_byteslike("associated_data", associated_data)
if len(nonce) < 8 or len(nonce) > 128:
raise ValueError("Nonce must be between 8 and 128 bytes")
class AESOCB3:
_MAX_SIZE = 2**31 - 1
def __init__(self, key: bytes):
utils._check_byteslike("key", key)
if len(key) not in (16, 24, 32):
raise ValueError("AESOCB3 key must be 128, 192, or 256 bits.")
self._key = key
if not backend.aead_cipher_supported(self):
raise exceptions.UnsupportedAlgorithm(
"OCB3 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 (128, 192, 256):
raise ValueError("bit_length must be 128, 192, or 256")
return os.urandom(bit_length // 8)
def encrypt(
self,
nonce: bytes,
data: bytes,
associated_data: typing.Optional[bytes],
) -> bytes:
if associated_data is None:
associated_data = b""
if len(data) > self._MAX_SIZE or len(associated_data) > self._MAX_SIZE:
# This is OverflowError to match what cffi would raise
raise OverflowError(
"Data or associated data too long. Max 2**31 - 1 bytes"
)
self._check_params(nonce, data, associated_data)
return aead._encrypt(backend, self, nonce, data, [associated_data], 16)
def decrypt(
self,
nonce: bytes,
data: bytes,
associated_data: typing.Optional[bytes],
) -> bytes:
if associated_data is None:
associated_data = b""
self._check_params(nonce, data, associated_data)
return aead._decrypt(backend, self, nonce, data, [associated_data], 16)
def _check_params(
self,
nonce: bytes,
data: bytes,
associated_data: bytes,
) -> None:
utils._check_byteslike("nonce", nonce)
utils._check_byteslike("data", data)
utils._check_byteslike("associated_data", associated_data)
if len(nonce) < 12 or len(nonce) > 15:
raise ValueError("Nonce must be between 12 and 15 bytes")
class AESSIV:
_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[bytes],
) -> None:
utils._check_byteslike("data", data)
if len(data) == 0:
raise ValueError("data must not be zero length")
if not isinstance(associated_data, list):
raise TypeError(
"associated_data must be a list of bytes-like objects or None"
)
for x in associated_data:
utils._check_byteslike("associated_data elements", x)
from cryptography.hazmat.bindings._rust import openssl as rust_openssl
__all__ = [
"AESCCM",
"AESGCM",
"AESGCMSIV",
"AESOCB3",
"AESSIV",
"ChaCha20Poly1305",
]
AESGCM = rust_openssl.aead.AESGCM
ChaCha20Poly1305 = rust_openssl.aead.ChaCha20Poly1305
AESCCM = rust_openssl.aead.AESCCM
AESSIV = rust_openssl.aead.AESSIV
AESOCB3 = rust_openssl.aead.AESOCB3
AESGCMSIV = rust_openssl.aead.AESGCMSIV

View File

@@ -5,33 +5,38 @@
from __future__ import annotations
from cryptography import utils
from cryptography.hazmat.decrepit.ciphers.algorithms import (
ARC4 as ARC4,
)
from cryptography.hazmat.decrepit.ciphers.algorithms import (
CAST5 as CAST5,
)
from cryptography.hazmat.decrepit.ciphers.algorithms import (
IDEA as IDEA,
)
from cryptography.hazmat.decrepit.ciphers.algorithms import (
SEED as SEED,
)
from cryptography.hazmat.decrepit.ciphers.algorithms import (
Blowfish as Blowfish,
)
from cryptography.hazmat.decrepit.ciphers.algorithms import (
TripleDES as TripleDES,
)
from cryptography.hazmat.primitives._cipheralgorithm import _verify_key_size
from cryptography.hazmat.primitives.ciphers import (
BlockCipherAlgorithm,
CipherAlgorithm,
)
def _verify_key_size(algorithm: CipherAlgorithm, key: bytes) -> bytes:
# Verify that the key is instance of bytes
utils._check_byteslike("key", key)
# Verify that the key size matches the expected key size
if len(key) * 8 not in algorithm.key_sizes:
raise ValueError(
"Invalid key size ({}) for {}.".format(
len(key) * 8, algorithm.name
)
)
return key
class AES(BlockCipherAlgorithm):
name = "AES"
block_size = 128
# 512 added to support AES-256-XTS, which uses 512-bit keys
key_sizes = frozenset([128, 192, 256, 512])
def __init__(self, key: bytes):
def __init__(self, key: utils.Buffer):
self.key = _verify_key_size(self, key)
@property
@@ -45,7 +50,7 @@ class AES128(BlockCipherAlgorithm):
key_sizes = frozenset([128])
key_size = 128
def __init__(self, key: bytes):
def __init__(self, key: utils.Buffer):
self.key = _verify_key_size(self, key)
@@ -55,7 +60,7 @@ class AES256(BlockCipherAlgorithm):
key_sizes = frozenset([256])
key_size = 256
def __init__(self, key: bytes):
def __init__(self, key: utils.Buffer):
self.key = _verify_key_size(self, key)
@@ -64,7 +69,7 @@ class Camellia(BlockCipherAlgorithm):
block_size = 128
key_sizes = frozenset([128, 192, 256])
def __init__(self, key: bytes):
def __init__(self, key: utils.Buffer):
self.key = _verify_key_size(self, key)
@property
@@ -72,124 +77,27 @@ class Camellia(BlockCipherAlgorithm):
return len(self.key) * 8
class TripleDES(BlockCipherAlgorithm):
name = "3DES"
block_size = 64
key_sizes = frozenset([64, 128, 192])
def __init__(self, key: bytes):
if len(key) == 8:
key += key + key
elif len(key) == 16:
key += key[:8]
self.key = _verify_key_size(self, key)
@property
def key_size(self) -> int:
return len(self.key) * 8
class Blowfish(BlockCipherAlgorithm):
name = "Blowfish"
block_size = 64
key_sizes = frozenset(range(32, 449, 8))
def __init__(self, key: bytes):
self.key = _verify_key_size(self, key)
@property
def key_size(self) -> int:
return len(self.key) * 8
_BlowfishInternal = Blowfish
utils.deprecated(
Blowfish,
ARC4,
__name__,
"Blowfish has been deprecated",
utils.DeprecatedIn37,
name="Blowfish",
"ARC4 has been moved to "
"cryptography.hazmat.decrepit.ciphers.algorithms.ARC4 and "
"will be removed from "
"cryptography.hazmat.primitives.ciphers.algorithms in 48.0.0.",
utils.DeprecatedIn43,
name="ARC4",
)
class CAST5(BlockCipherAlgorithm):
name = "CAST5"
block_size = 64
key_sizes = frozenset(range(40, 129, 8))
def __init__(self, key: bytes):
self.key = _verify_key_size(self, key)
@property
def key_size(self) -> int:
return len(self.key) * 8
_CAST5Internal = CAST5
utils.deprecated(
CAST5,
TripleDES,
__name__,
"CAST5 has been deprecated",
utils.DeprecatedIn37,
name="CAST5",
)
class ARC4(CipherAlgorithm):
name = "RC4"
key_sizes = frozenset([40, 56, 64, 80, 128, 160, 192, 256])
def __init__(self, key: bytes):
self.key = _verify_key_size(self, key)
@property
def key_size(self) -> int:
return len(self.key) * 8
class IDEA(BlockCipherAlgorithm):
name = "IDEA"
block_size = 64
key_sizes = frozenset([128])
def __init__(self, key: bytes):
self.key = _verify_key_size(self, key)
@property
def key_size(self) -> int:
return len(self.key) * 8
_IDEAInternal = IDEA
utils.deprecated(
IDEA,
__name__,
"IDEA has been deprecated",
utils.DeprecatedIn37,
name="IDEA",
)
class SEED(BlockCipherAlgorithm):
name = "SEED"
block_size = 128
key_sizes = frozenset([128])
def __init__(self, key: bytes):
self.key = _verify_key_size(self, key)
@property
def key_size(self) -> int:
return len(self.key) * 8
_SEEDInternal = SEED
utils.deprecated(
SEED,
__name__,
"SEED has been deprecated",
utils.DeprecatedIn37,
name="SEED",
"TripleDES has been moved to "
"cryptography.hazmat.decrepit.ciphers.algorithms.TripleDES and "
"will be removed from "
"cryptography.hazmat.primitives.ciphers.algorithms in 48.0.0.",
utils.DeprecatedIn43,
name="TripleDES",
)
@@ -197,7 +105,7 @@ class ChaCha20(CipherAlgorithm):
name = "ChaCha20"
key_sizes = frozenset([256])
def __init__(self, key: bytes, nonce: bytes):
def __init__(self, key: utils.Buffer, nonce: utils.Buffer):
self.key = _verify_key_size(self, key)
utils._check_byteslike("nonce", nonce)
@@ -207,7 +115,7 @@ class ChaCha20(CipherAlgorithm):
self._nonce = nonce
@property
def nonce(self) -> bytes:
def nonce(self) -> utils.Buffer:
return self._nonce
@property

View File

@@ -7,30 +7,22 @@ from __future__ import annotations
import abc
import typing
from cryptography.exceptions import (
AlreadyFinalized,
AlreadyUpdated,
NotYetFinalized,
)
from cryptography.hazmat.bindings._rust import openssl as rust_openssl
from cryptography.hazmat.primitives._cipheralgorithm import CipherAlgorithm
from cryptography.hazmat.primitives.ciphers import modes
if typing.TYPE_CHECKING:
from cryptography.hazmat.backends.openssl.ciphers import (
_CipherContext as _BackendCipherContext,
)
from cryptography.utils import Buffer
class CipherContext(metaclass=abc.ABCMeta):
@abc.abstractmethod
def update(self, data: bytes) -> bytes:
def update(self, data: Buffer) -> bytes:
"""
Processes the provided bytes through the cipher and returns the results
as bytes.
"""
@abc.abstractmethod
def update_into(self, data: bytes, buf: bytes) -> int:
def update_into(self, data: Buffer, buf: Buffer) -> int:
"""
Processes the provided bytes and writes the resulting data into the
provided buffer. Returns the number of bytes written.
@@ -42,10 +34,18 @@ class CipherContext(metaclass=abc.ABCMeta):
Returns the results of processing the final block as bytes.
"""
@abc.abstractmethod
def reset_nonce(self, nonce: bytes) -> None:
"""
Resets 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
def authenticate_additional_data(self, data: bytes) -> None:
def authenticate_additional_data(self, data: Buffer) -> None:
"""
Authenticates the provided bytes.
"""
@@ -97,14 +97,12 @@ class Cipher(typing.Generic[Mode]):
@typing.overload
def encryptor(
self: Cipher[modes.ModeWithAuthenticationTag],
) -> AEADEncryptionContext:
...
) -> AEADEncryptionContext: ...
@typing.overload
def encryptor(
self: _CIPHER_TYPE,
) -> CipherContext:
...
) -> CipherContext: ...
def encryptor(self):
if isinstance(self.mode, modes.ModeWithAuthenticationTag):
@@ -112,158 +110,37 @@ class Cipher(typing.Generic[Mode]):
raise ValueError(
"Authentication tag must be None when encrypting."
)
from cryptography.hazmat.backends.openssl.backend import backend
ctx = backend.create_symmetric_encryption_ctx(
return rust_openssl.ciphers.create_encryption_ctx(
self.algorithm, self.mode
)
return self._wrap_ctx(ctx, encrypt=True)
@typing.overload
def decryptor(
self: Cipher[modes.ModeWithAuthenticationTag],
) -> AEADDecryptionContext:
...
) -> AEADDecryptionContext: ...
@typing.overload
def decryptor(
self: _CIPHER_TYPE,
) -> CipherContext:
...
) -> CipherContext: ...
def decryptor(self):
from cryptography.hazmat.backends.openssl.backend import backend
ctx = backend.create_symmetric_decryption_ctx(
return rust_openssl.ciphers.create_decryption_ctx(
self.algorithm, self.mode
)
return self._wrap_ctx(ctx, encrypt=False)
def _wrap_ctx(
self, ctx: _BackendCipherContext, encrypt: bool
) -> typing.Union[
AEADEncryptionContext, AEADDecryptionContext, CipherContext
]:
if isinstance(self.mode, modes.ModeWithAuthenticationTag):
if encrypt:
return _AEADEncryptionContext(ctx)
else:
return _AEADDecryptionContext(ctx)
else:
return _CipherContext(ctx)
_CIPHER_TYPE = Cipher[
typing.Union[
modes.ModeWithNonce,
modes.ModeWithTweak,
None,
modes.ECB,
modes.ModeWithInitializationVector,
None,
]
]
class _CipherContext(CipherContext):
_ctx: typing.Optional[_BackendCipherContext]
def __init__(self, ctx: _BackendCipherContext) -> None:
self._ctx = ctx
def update(self, data: bytes) -> bytes:
if self._ctx is None:
raise AlreadyFinalized("Context was already finalized.")
return self._ctx.update(data)
def update_into(self, data: bytes, buf: bytes) -> int:
if self._ctx is None:
raise AlreadyFinalized("Context was already finalized.")
return self._ctx.update_into(data, buf)
def finalize(self) -> bytes:
if self._ctx is None:
raise AlreadyFinalized("Context was already finalized.")
data = self._ctx.finalize()
self._ctx = None
return data
class _AEADCipherContext(AEADCipherContext):
_ctx: typing.Optional[_BackendCipherContext]
_tag: typing.Optional[bytes]
def __init__(self, ctx: _BackendCipherContext) -> None:
self._ctx = ctx
self._bytes_processed = 0
self._aad_bytes_processed = 0
self._tag = None
self._updated = False
def _check_limit(self, data_size: int) -> None:
if self._ctx is None:
raise AlreadyFinalized("Context was already finalized.")
self._updated = True
self._bytes_processed += data_size
if self._bytes_processed > self._ctx._mode._MAX_ENCRYPTED_BYTES:
raise ValueError(
"{} has a maximum encrypted byte limit of {}".format(
self._ctx._mode.name, self._ctx._mode._MAX_ENCRYPTED_BYTES
)
)
def update(self, data: bytes) -> bytes:
self._check_limit(len(data))
# mypy needs this assert even though _check_limit already checked
assert self._ctx is not None
return self._ctx.update(data)
def update_into(self, data: bytes, buf: bytes) -> int:
self._check_limit(len(data))
# mypy needs this assert even though _check_limit already checked
assert self._ctx is not None
return self._ctx.update_into(data, buf)
def finalize(self) -> bytes:
if self._ctx is None:
raise AlreadyFinalized("Context was already finalized.")
data = self._ctx.finalize()
self._tag = self._ctx.tag
self._ctx = None
return data
def authenticate_additional_data(self, data: bytes) -> None:
if self._ctx is None:
raise AlreadyFinalized("Context was already finalized.")
if self._updated:
raise AlreadyUpdated("Update has been called on this context.")
self._aad_bytes_processed += len(data)
if self._aad_bytes_processed > self._ctx._mode._MAX_AAD_BYTES:
raise ValueError(
"{} has a maximum AAD byte limit of {}".format(
self._ctx._mode.name, self._ctx._mode._MAX_AAD_BYTES
)
)
self._ctx.authenticate_additional_data(data)
class _AEADDecryptionContext(_AEADCipherContext, AEADDecryptionContext):
def finalize_with_tag(self, tag: bytes) -> bytes:
if self._ctx is None:
raise AlreadyFinalized("Context was already finalized.")
data = self._ctx.finalize_with_tag(tag)
self._tag = self._ctx.tag
self._ctx = None
return data
class _AEADEncryptionContext(_AEADCipherContext, AEADEncryptionContext):
@property
def tag(self) -> bytes:
if self._ctx is not None:
raise NotYetFinalized(
"You must finalize encryption before " "getting the tag."
)
assert self._tag is not None
return self._tag
CipherContext.register(rust_openssl.ciphers.CipherContext)
AEADEncryptionContext.register(rust_openssl.ciphers.AEADEncryptionContext)
AEADDecryptionContext.register(rust_openssl.ciphers.AEADDecryptionContext)

View File

@@ -5,7 +5,6 @@
from __future__ import annotations
import abc
import typing
from cryptography import utils
from cryptography.exceptions import UnsupportedAlgorithm, _Reasons
@@ -35,7 +34,7 @@ class Mode(metaclass=abc.ABCMeta):
class ModeWithInitializationVector(Mode, metaclass=abc.ABCMeta):
@property
@abc.abstractmethod
def initialization_vector(self) -> bytes:
def initialization_vector(self) -> utils.Buffer:
"""
The value of the initialization vector for this mode as bytes.
"""
@@ -44,7 +43,7 @@ class ModeWithInitializationVector(Mode, metaclass=abc.ABCMeta):
class ModeWithTweak(Mode, metaclass=abc.ABCMeta):
@property
@abc.abstractmethod
def tweak(self) -> bytes:
def tweak(self) -> utils.Buffer:
"""
The value of the tweak for this mode as bytes.
"""
@@ -53,7 +52,7 @@ class ModeWithTweak(Mode, metaclass=abc.ABCMeta):
class ModeWithNonce(Mode, metaclass=abc.ABCMeta):
@property
@abc.abstractmethod
def nonce(self) -> bytes:
def nonce(self) -> utils.Buffer:
"""
The value of the nonce for this mode as bytes.
"""
@@ -62,7 +61,7 @@ class ModeWithNonce(Mode, metaclass=abc.ABCMeta):
class ModeWithAuthenticationTag(Mode, metaclass=abc.ABCMeta):
@property
@abc.abstractmethod
def tag(self) -> typing.Optional[bytes]:
def tag(self) -> bytes | None:
"""
The value of the tag supplied to the constructor of this mode.
"""
@@ -78,16 +77,13 @@ def _check_aes_key_length(self: Mode, algorithm: CipherAlgorithm) -> None:
def _check_iv_length(
self: ModeWithInitializationVector, algorithm: BlockCipherAlgorithm
) -> None:
if len(self.initialization_vector) * 8 != algorithm.block_size:
raise ValueError(
"Invalid IV size ({}) for {}.".format(
len(self.initialization_vector), self.name
)
)
iv_len = len(self.initialization_vector)
if iv_len * 8 != algorithm.block_size:
raise ValueError(f"Invalid IV size ({iv_len}) for {self.name}.")
def _check_nonce_length(
nonce: bytes, name: str, algorithm: CipherAlgorithm
nonce: utils.Buffer, name: str, algorithm: CipherAlgorithm
) -> None:
if not isinstance(algorithm, BlockCipherAlgorithm):
raise UnsupportedAlgorithm(
@@ -113,12 +109,12 @@ def _check_iv_and_key_length(
class CBC(ModeWithInitializationVector):
name = "CBC"
def __init__(self, initialization_vector: bytes):
def __init__(self, initialization_vector: utils.Buffer):
utils._check_byteslike("initialization_vector", initialization_vector)
self._initialization_vector = initialization_vector
@property
def initialization_vector(self) -> bytes:
def initialization_vector(self) -> utils.Buffer:
return self._initialization_vector
validate_for_algorithm = _check_iv_and_key_length
@@ -127,7 +123,7 @@ class CBC(ModeWithInitializationVector):
class XTS(ModeWithTweak):
name = "XTS"
def __init__(self, tweak: bytes):
def __init__(self, tweak: utils.Buffer):
utils._check_byteslike("tweak", tweak)
if len(tweak) != 16:
@@ -136,7 +132,7 @@ class XTS(ModeWithTweak):
self._tweak = tweak
@property
def tweak(self) -> bytes:
def tweak(self) -> utils.Buffer:
return self._tweak
def validate_for_algorithm(self, algorithm: CipherAlgorithm) -> None:
@@ -162,12 +158,12 @@ class ECB(Mode):
class OFB(ModeWithInitializationVector):
name = "OFB"
def __init__(self, initialization_vector: bytes):
def __init__(self, initialization_vector: utils.Buffer):
utils._check_byteslike("initialization_vector", initialization_vector)
self._initialization_vector = initialization_vector
@property
def initialization_vector(self) -> bytes:
def initialization_vector(self) -> utils.Buffer:
return self._initialization_vector
validate_for_algorithm = _check_iv_and_key_length
@@ -176,12 +172,12 @@ class OFB(ModeWithInitializationVector):
class CFB(ModeWithInitializationVector):
name = "CFB"
def __init__(self, initialization_vector: bytes):
def __init__(self, initialization_vector: utils.Buffer):
utils._check_byteslike("initialization_vector", initialization_vector)
self._initialization_vector = initialization_vector
@property
def initialization_vector(self) -> bytes:
def initialization_vector(self) -> utils.Buffer:
return self._initialization_vector
validate_for_algorithm = _check_iv_and_key_length
@@ -190,12 +186,12 @@ class CFB(ModeWithInitializationVector):
class CFB8(ModeWithInitializationVector):
name = "CFB8"
def __init__(self, initialization_vector: bytes):
def __init__(self, initialization_vector: utils.Buffer):
utils._check_byteslike("initialization_vector", initialization_vector)
self._initialization_vector = initialization_vector
@property
def initialization_vector(self) -> bytes:
def initialization_vector(self) -> utils.Buffer:
return self._initialization_vector
validate_for_algorithm = _check_iv_and_key_length
@@ -204,12 +200,12 @@ class CFB8(ModeWithInitializationVector):
class CTR(ModeWithNonce):
name = "CTR"
def __init__(self, nonce: bytes):
def __init__(self, nonce: utils.Buffer):
utils._check_byteslike("nonce", nonce)
self._nonce = nonce
@property
def nonce(self) -> bytes:
def nonce(self) -> utils.Buffer:
return self._nonce
def validate_for_algorithm(self, algorithm: CipherAlgorithm) -> None:
@@ -224,8 +220,8 @@ class GCM(ModeWithInitializationVector, ModeWithAuthenticationTag):
def __init__(
self,
initialization_vector: bytes,
tag: typing.Optional[bytes] = None,
initialization_vector: utils.Buffer,
tag: bytes | None = None,
min_tag_length: int = 16,
):
# OpenSSL 3.0.0 constrains GCM IVs to [64, 1024] bits inclusive
@@ -243,19 +239,18 @@ class GCM(ModeWithInitializationVector, ModeWithAuthenticationTag):
raise ValueError("min_tag_length must be >= 4")
if len(tag) < min_tag_length:
raise ValueError(
"Authentication tag must be {} bytes or longer.".format(
min_tag_length
)
f"Authentication tag must be {min_tag_length} bytes or "
"longer."
)
self._tag = tag
self._min_tag_length = min_tag_length
@property
def tag(self) -> typing.Optional[bytes]:
def tag(self) -> bytes | None:
return self._tag
@property
def initialization_vector(self) -> bytes:
def initialization_vector(self) -> utils.Buffer:
return self._initialization_vector
def validate_for_algorithm(self, algorithm: CipherAlgorithm) -> None:
@@ -268,7 +263,6 @@ class GCM(ModeWithInitializationVector, ModeWithAuthenticationTag):
block_size_bytes = algorithm.block_size // 8
if self._tag is not None and len(self._tag) > block_size_bytes:
raise ValueError(
"Authentication tag cannot be more than {} bytes.".format(
block_size_bytes
)
f"Authentication tag cannot be more than {block_size_bytes} "
"bytes."
)