split commoncrypto backend - ciphers
diff --git a/cryptography/hazmat/backends/commoncrypto/backend.py b/cryptography/hazmat/backends/commoncrypto/backend.py
index 6147a99..f78370c 100644
--- a/cryptography/hazmat/backends/commoncrypto/backend.py
+++ b/cryptography/hazmat/backends/commoncrypto/backend.py
@@ -16,8 +16,9 @@
from collections import namedtuple
from cryptography import utils
-from cryptography.exceptions import (
- InternalError, InvalidTag, UnsupportedAlgorithm, _Reasons
+from cryptography.exceptions import InternalError
+from cryptography.hazmat.backends.commoncrypto.ciphers import (
+ _CipherContext, _GCMCipherContext
)
from cryptography.hazmat.backends.commoncrypto.hashes import _HashContext
from cryptography.hazmat.backends.commoncrypto.hmac import _HMACContext
@@ -25,7 +26,6 @@
CipherBackend, HMACBackend, HashBackend, PBKDF2HMACBackend
)
from cryptography.hazmat.bindings.commoncrypto.binding import Binding
-from cryptography.hazmat.primitives import constant_time, interfaces
from cryptography.hazmat.primitives.ciphers.algorithms import (
AES, ARC4, Blowfish, CAST5, TripleDES
)
@@ -250,171 +250,4 @@
ctx[0] = self._ffi.NULL
-@utils.register_interface(interfaces.CipherContext)
-class _CipherContext(object):
- def __init__(self, backend, cipher, mode, operation):
- self._backend = backend
- self._cipher = cipher
- self._mode = mode
- self._operation = operation
- # There is a bug in CommonCrypto where block ciphers do not raise
- # kCCAlignmentError when finalizing if you supply non-block aligned
- # data. To work around this we need to keep track of the block
- # alignment ourselves, but only for alg+mode combos that require
- # block alignment. OFB, CFB, and CTR make a block cipher algorithm
- # into a stream cipher so we don't need to track them (and thus their
- # block size is effectively 1 byte just like OpenSSL/CommonCrypto
- # treat RC4 and other stream cipher block sizes).
- # This bug has been filed as rdar://15589470
- self._bytes_processed = 0
- if (isinstance(cipher, interfaces.BlockCipherAlgorithm) and not
- isinstance(mode, (OFB, CFB, CFB8, CTR))):
- self._byte_block_size = cipher.block_size // 8
- else:
- self._byte_block_size = 1
-
- registry = self._backend._cipher_registry
- try:
- cipher_enum, mode_enum = registry[type(cipher), type(mode)]
- except KeyError:
- raise UnsupportedAlgorithm(
- "cipher {0} in {1} mode is not supported "
- "by this backend.".format(
- cipher.name, mode.name if mode else mode),
- _Reasons.UNSUPPORTED_CIPHER
- )
-
- ctx = self._backend._ffi.new("CCCryptorRef *")
- ctx = self._backend._ffi.gc(ctx, self._backend._release_cipher_ctx)
-
- if isinstance(mode, interfaces.ModeWithInitializationVector):
- iv_nonce = mode.initialization_vector
- elif isinstance(mode, interfaces.ModeWithNonce):
- iv_nonce = mode.nonce
- else:
- iv_nonce = self._backend._ffi.NULL
-
- if isinstance(mode, CTR):
- mode_option = self._backend._lib.kCCModeOptionCTR_BE
- else:
- mode_option = 0
-
- res = self._backend._lib.CCCryptorCreateWithMode(
- operation,
- mode_enum, cipher_enum,
- self._backend._lib.ccNoPadding, iv_nonce,
- cipher.key, len(cipher.key),
- self._backend._ffi.NULL, 0, 0, mode_option, ctx)
- self._backend._check_response(res)
-
- self._ctx = ctx
-
- def update(self, data):
- # Count bytes processed to handle block alignment.
- self._bytes_processed += len(data)
- buf = self._backend._ffi.new(
- "unsigned char[]", len(data) + self._byte_block_size - 1)
- outlen = self._backend._ffi.new("size_t *")
- res = self._backend._lib.CCCryptorUpdate(
- self._ctx[0], data, len(data), buf,
- len(data) + self._byte_block_size - 1, outlen)
- self._backend._check_response(res)
- return self._backend._ffi.buffer(buf)[:outlen[0]]
-
- def finalize(self):
- # Raise error if block alignment is wrong.
- if self._bytes_processed % self._byte_block_size:
- raise ValueError(
- "The length of the provided data is not a multiple of "
- "the block length."
- )
- buf = self._backend._ffi.new("unsigned char[]", self._byte_block_size)
- outlen = self._backend._ffi.new("size_t *")
- res = self._backend._lib.CCCryptorFinal(
- self._ctx[0], buf, len(buf), outlen)
- self._backend._check_response(res)
- self._backend._release_cipher_ctx(self._ctx)
- return self._backend._ffi.buffer(buf)[:outlen[0]]
-
-
-@utils.register_interface(interfaces.AEADCipherContext)
-@utils.register_interface(interfaces.AEADEncryptionContext)
-class _GCMCipherContext(object):
- def __init__(self, backend, cipher, mode, operation):
- self._backend = backend
- self._cipher = cipher
- self._mode = mode
- self._operation = operation
- self._tag = None
-
- registry = self._backend._cipher_registry
- try:
- cipher_enum, mode_enum = registry[type(cipher), type(mode)]
- except KeyError:
- raise UnsupportedAlgorithm(
- "cipher {0} in {1} mode is not supported "
- "by this backend.".format(
- cipher.name, mode.name if mode else mode),
- _Reasons.UNSUPPORTED_CIPHER
- )
-
- ctx = self._backend._ffi.new("CCCryptorRef *")
- ctx = self._backend._ffi.gc(ctx, self._backend._release_cipher_ctx)
-
- self._ctx = ctx
-
- res = self._backend._lib.CCCryptorCreateWithMode(
- operation,
- mode_enum, cipher_enum,
- self._backend._lib.ccNoPadding,
- self._backend._ffi.NULL,
- cipher.key, len(cipher.key),
- self._backend._ffi.NULL, 0, 0, 0, self._ctx)
- self._backend._check_response(res)
-
- res = self._backend._lib.CCCryptorGCMAddIV(
- self._ctx[0],
- mode.initialization_vector,
- len(mode.initialization_vector)
- )
- self._backend._check_response(res)
-
- def update(self, data):
- buf = self._backend._ffi.new("unsigned char[]", len(data))
- args = (self._ctx[0], data, len(data), buf)
- if self._operation == self._backend._lib.kCCEncrypt:
- res = self._backend._lib.CCCryptorGCMEncrypt(*args)
- else:
- res = self._backend._lib.CCCryptorGCMDecrypt(*args)
-
- self._backend._check_response(res)
- return self._backend._ffi.buffer(buf)[:]
-
- def finalize(self):
- tag_size = self._cipher.block_size // 8
- tag_buf = self._backend._ffi.new("unsigned char[]", tag_size)
- tag_len = self._backend._ffi.new("size_t *", tag_size)
- res = self._backend._lib.CCCryptorGCMFinal(
- self._ctx[0], tag_buf, tag_len)
- self._backend._check_response(res)
- self._backend._release_cipher_ctx(self._ctx)
- self._tag = self._backend._ffi.buffer(tag_buf)[:]
- if (self._operation == self._backend._lib.kCCDecrypt and
- not constant_time.bytes_eq(
- self._tag[:len(self._mode.tag)], self._mode.tag
- )):
- raise InvalidTag
- return b""
-
- def authenticate_additional_data(self, data):
- res = self._backend._lib.CCCryptorGCMAddAAD(
- self._ctx[0], data, len(data)
- )
- self._backend._check_response(res)
-
- @property
- def tag(self):
- return self._tag
-
-
backend = Backend()
diff --git a/cryptography/hazmat/backends/commoncrypto/ciphers.py b/cryptography/hazmat/backends/commoncrypto/ciphers.py
new file mode 100644
index 0000000..2820fd0
--- /dev/null
+++ b/cryptography/hazmat/backends/commoncrypto/ciphers.py
@@ -0,0 +1,191 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import absolute_import, division, print_function
+
+from cryptography import utils
+from cryptography.exceptions import (
+ InvalidTag, UnsupportedAlgorithm, _Reasons
+)
+from cryptography.hazmat.primitives import constant_time, interfaces
+from cryptography.hazmat.primitives.ciphers.modes import (
+ CFB, CFB8, CTR, OFB
+)
+
+
+@utils.register_interface(interfaces.CipherContext)
+class _CipherContext(object):
+ def __init__(self, backend, cipher, mode, operation):
+ self._backend = backend
+ self._cipher = cipher
+ self._mode = mode
+ self._operation = operation
+ # There is a bug in CommonCrypto where block ciphers do not raise
+ # kCCAlignmentError when finalizing if you supply non-block aligned
+ # data. To work around this we need to keep track of the block
+ # alignment ourselves, but only for alg+mode combos that require
+ # block alignment. OFB, CFB, and CTR make a block cipher algorithm
+ # into a stream cipher so we don't need to track them (and thus their
+ # block size is effectively 1 byte just like OpenSSL/CommonCrypto
+ # treat RC4 and other stream cipher block sizes).
+ # This bug has been filed as rdar://15589470
+ self._bytes_processed = 0
+ if (isinstance(cipher, interfaces.BlockCipherAlgorithm) and not
+ isinstance(mode, (OFB, CFB, CFB8, CTR))):
+ self._byte_block_size = cipher.block_size // 8
+ else:
+ self._byte_block_size = 1
+
+ registry = self._backend._cipher_registry
+ try:
+ cipher_enum, mode_enum = registry[type(cipher), type(mode)]
+ except KeyError:
+ raise UnsupportedAlgorithm(
+ "cipher {0} in {1} mode is not supported "
+ "by this backend.".format(
+ cipher.name, mode.name if mode else mode),
+ _Reasons.UNSUPPORTED_CIPHER
+ )
+
+ ctx = self._backend._ffi.new("CCCryptorRef *")
+ ctx = self._backend._ffi.gc(ctx, self._backend._release_cipher_ctx)
+
+ if isinstance(mode, interfaces.ModeWithInitializationVector):
+ iv_nonce = mode.initialization_vector
+ elif isinstance(mode, interfaces.ModeWithNonce):
+ iv_nonce = mode.nonce
+ else:
+ iv_nonce = self._backend._ffi.NULL
+
+ if isinstance(mode, CTR):
+ mode_option = self._backend._lib.kCCModeOptionCTR_BE
+ else:
+ mode_option = 0
+
+ res = self._backend._lib.CCCryptorCreateWithMode(
+ operation,
+ mode_enum, cipher_enum,
+ self._backend._lib.ccNoPadding, iv_nonce,
+ cipher.key, len(cipher.key),
+ self._backend._ffi.NULL, 0, 0, mode_option, ctx)
+ self._backend._check_response(res)
+
+ self._ctx = ctx
+
+ def update(self, data):
+ # Count bytes processed to handle block alignment.
+ self._bytes_processed += len(data)
+ buf = self._backend._ffi.new(
+ "unsigned char[]", len(data) + self._byte_block_size - 1)
+ outlen = self._backend._ffi.new("size_t *")
+ res = self._backend._lib.CCCryptorUpdate(
+ self._ctx[0], data, len(data), buf,
+ len(data) + self._byte_block_size - 1, outlen)
+ self._backend._check_response(res)
+ return self._backend._ffi.buffer(buf)[:outlen[0]]
+
+ def finalize(self):
+ # Raise error if block alignment is wrong.
+ if self._bytes_processed % self._byte_block_size:
+ raise ValueError(
+ "The length of the provided data is not a multiple of "
+ "the block length."
+ )
+ buf = self._backend._ffi.new("unsigned char[]", self._byte_block_size)
+ outlen = self._backend._ffi.new("size_t *")
+ res = self._backend._lib.CCCryptorFinal(
+ self._ctx[0], buf, len(buf), outlen)
+ self._backend._check_response(res)
+ self._backend._release_cipher_ctx(self._ctx)
+ return self._backend._ffi.buffer(buf)[:outlen[0]]
+
+
+@utils.register_interface(interfaces.AEADCipherContext)
+@utils.register_interface(interfaces.AEADEncryptionContext)
+class _GCMCipherContext(object):
+ def __init__(self, backend, cipher, mode, operation):
+ self._backend = backend
+ self._cipher = cipher
+ self._mode = mode
+ self._operation = operation
+ self._tag = None
+
+ registry = self._backend._cipher_registry
+ try:
+ cipher_enum, mode_enum = registry[type(cipher), type(mode)]
+ except KeyError:
+ raise UnsupportedAlgorithm(
+ "cipher {0} in {1} mode is not supported "
+ "by this backend.".format(
+ cipher.name, mode.name if mode else mode),
+ _Reasons.UNSUPPORTED_CIPHER
+ )
+
+ ctx = self._backend._ffi.new("CCCryptorRef *")
+ ctx = self._backend._ffi.gc(ctx, self._backend._release_cipher_ctx)
+
+ self._ctx = ctx
+
+ res = self._backend._lib.CCCryptorCreateWithMode(
+ operation,
+ mode_enum, cipher_enum,
+ self._backend._lib.ccNoPadding,
+ self._backend._ffi.NULL,
+ cipher.key, len(cipher.key),
+ self._backend._ffi.NULL, 0, 0, 0, self._ctx)
+ self._backend._check_response(res)
+
+ res = self._backend._lib.CCCryptorGCMAddIV(
+ self._ctx[0],
+ mode.initialization_vector,
+ len(mode.initialization_vector)
+ )
+ self._backend._check_response(res)
+
+ def update(self, data):
+ buf = self._backend._ffi.new("unsigned char[]", len(data))
+ args = (self._ctx[0], data, len(data), buf)
+ if self._operation == self._backend._lib.kCCEncrypt:
+ res = self._backend._lib.CCCryptorGCMEncrypt(*args)
+ else:
+ res = self._backend._lib.CCCryptorGCMDecrypt(*args)
+
+ self._backend._check_response(res)
+ return self._backend._ffi.buffer(buf)[:]
+
+ def finalize(self):
+ tag_size = self._cipher.block_size // 8
+ tag_buf = self._backend._ffi.new("unsigned char[]", tag_size)
+ tag_len = self._backend._ffi.new("size_t *", tag_size)
+ res = self._backend._lib.CCCryptorGCMFinal(
+ self._ctx[0], tag_buf, tag_len
+ )
+ self._backend._check_response(res)
+ self._backend._release_cipher_ctx(self._ctx)
+ self._tag = self._backend._ffi.buffer(tag_buf)[:]
+ if (self._operation == self._backend._lib.kCCDecrypt and
+ not constant_time.bytes_eq(
+ self._tag[:len(self._mode.tag)], self._mode.tag
+ )):
+ raise InvalidTag
+ return b""
+
+ def authenticate_additional_data(self, data):
+ res = self._backend._lib.CCCryptorGCMAddAAD(
+ self._ctx[0], data, len(data)
+ )
+ self._backend._check_response(res)
+
+ @property
+ def tag(self):
+ return self._tag