aboutsummaryrefslogtreecommitdiffstats
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/cryptography/hazmat/backends/openssl/backend.py34
-rw-r--r--src/cryptography/hazmat/backends/openssl/ciphers.py21
-rw-r--r--src/cryptography/hazmat/backends/openssl/cmac.py8
-rw-r--r--src/cryptography/hazmat/backends/openssl/dsa.py10
-rw-r--r--src/cryptography/hazmat/backends/openssl/ec.py24
-rw-r--r--src/cryptography/hazmat/backends/openssl/hashes.py12
-rw-r--r--src/cryptography/hazmat/backends/openssl/hmac.py10
-rw-r--r--src/cryptography/hazmat/backends/openssl/rsa.py62
-rw-r--r--src/cryptography/hazmat/backends/openssl/x509.py36
-rw-r--r--src/cryptography/hazmat/bindings/openssl/binding.py51
10 files changed, 141 insertions, 127 deletions
diff --git a/src/cryptography/hazmat/backends/openssl/backend.py b/src/cryptography/hazmat/backends/openssl/backend.py
index 18592893..c4cac1a6 100644
--- a/src/cryptography/hazmat/backends/openssl/backend.py
+++ b/src/cryptography/hazmat/backends/openssl/backend.py
@@ -42,7 +42,7 @@ from cryptography.hazmat.backends.openssl.x509 import (
_Certificate, _CertificateRevocationList, _CertificateSigningRequest,
_DISTPOINT_TYPE_FULLNAME, _DISTPOINT_TYPE_RELATIVENAME
)
-from cryptography.hazmat.bindings.openssl.binding import Binding
+from cryptography.hazmat.bindings.openssl import binding
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import dsa, ec, rsa
from cryptography.hazmat.primitives.asymmetric.padding import (
@@ -58,14 +58,6 @@ from cryptography.x509.oid import ExtensionOID
_MemoryBIO = collections.namedtuple("_MemoryBIO", ["bio", "char_ptr"])
-_OpenSSLError = collections.namedtuple("_OpenSSLError",
- ["code", "lib", "func", "reason"])
-
-
-class UnhandledOpenSSLError(Exception):
- def __init__(self, msg, errors):
- super(UnhandledOpenSSLError, self).__init__(msg)
- self.errors = errors
def _encode_asn1_int(backend, x):
@@ -524,7 +516,7 @@ class Backend(object):
name = "openssl"
def __init__(self):
- self._binding = Binding()
+ self._binding = binding.Binding()
self._ffi = self._binding.ffi
self._lib = self._binding.lib
@@ -541,14 +533,7 @@ class Backend(object):
self.activate_osrandom_engine()
def openssl_assert(self, ok):
- if not ok:
- errors = self._consume_errors()
- raise UnhandledOpenSSLError(
- "Unknown OpenSSL error. Please file an issue at https://github"
- ".com/pyca/cryptography/issues with information on how to "
- "reproduce this.",
- errors
- )
+ return binding._openssl_assert(self._lib, ok)
def activate_builtin_random(self):
# Obtain a new structural reference.
@@ -759,18 +744,7 @@ class Backend(object):
return self._ffi.string(err_buf, 256)[:]
def _consume_errors(self):
- errors = []
- while True:
- code = self._lib.ERR_get_error()
- if code == 0:
- break
-
- lib = self._lib.ERR_GET_LIB(code)
- func = self._lib.ERR_GET_FUNC(code)
- reason = self._lib.ERR_GET_REASON(code)
-
- errors.append(_OpenSSLError(code, lib, func, reason))
- return errors
+ return binding._consume_errors(self._lib)
def _unknown_error(self, error):
return InternalError(
diff --git a/src/cryptography/hazmat/backends/openssl/ciphers.py b/src/cryptography/hazmat/backends/openssl/ciphers.py
index 64097c7b..4c1c7bc9 100644
--- a/src/cryptography/hazmat/backends/openssl/ciphers.py
+++ b/src/cryptography/hazmat/backends/openssl/ciphers.py
@@ -66,24 +66,24 @@ class _CipherContext(object):
self._backend._ffi.NULL,
self._backend._ffi.NULL,
operation)
- assert res != 0
+ self._backend.openssl_assert(res != 0)
# set the key length to handle variable key ciphers
res = self._backend._lib.EVP_CIPHER_CTX_set_key_length(
ctx, len(cipher.key)
)
- assert res != 0
+ self._backend.openssl_assert(res != 0)
if isinstance(mode, modes.GCM):
res = self._backend._lib.EVP_CIPHER_CTX_ctrl(
ctx, self._backend._lib.EVP_CTRL_GCM_SET_IVLEN,
len(iv_nonce), self._backend._ffi.NULL
)
- assert res != 0
+ self._backend.openssl_assert(res != 0)
if operation == self._DECRYPT:
res = self._backend._lib.EVP_CIPHER_CTX_ctrl(
ctx, self._backend._lib.EVP_CTRL_GCM_SET_TAG,
len(mode.tag), mode.tag
)
- assert res != 0
+ self._backend.openssl_assert(res != 0)
# pass key/iv
res = self._backend._lib.EVP_CipherInit_ex(
@@ -94,7 +94,7 @@ class _CipherContext(object):
iv_nonce,
operation
)
- assert res != 0
+ self._backend.openssl_assert(res != 0)
# We purposely disable padding here as it's handled higher up in the
# API.
self._backend._lib.EVP_CIPHER_CTX_set_padding(ctx, 0)
@@ -115,7 +115,7 @@ class _CipherContext(object):
outlen = self._backend._ffi.new("int *")
res = self._backend._lib.EVP_CipherUpdate(self._ctx, buf, outlen, data,
len(data))
- assert res != 0
+ self._backend.openssl_assert(res != 0)
return self._backend._ffi.buffer(buf)[:outlen[0]]
def finalize(self):
@@ -164,11 +164,11 @@ class _CipherContext(object):
self._ctx, self._backend._lib.EVP_CTRL_GCM_GET_TAG,
block_byte_size, tag_buf
)
- assert res != 0
+ self._backend.openssl_assert(res != 0)
self._tag = self._backend._ffi.buffer(tag_buf)[:]
res = self._backend._lib.EVP_CIPHER_CTX_cleanup(self._ctx)
- assert res == 1
+ self._backend.openssl_assert(res == 1)
return self._backend._ffi.buffer(buf)[:outlen[0]]
def authenticate_additional_data(self, data):
@@ -176,7 +176,7 @@ class _CipherContext(object):
res = self._backend._lib.EVP_CipherUpdate(
self._ctx, self._backend._ffi.NULL, outlen, data, len(data)
)
- assert res != 0
+ self._backend.openssl_assert(res != 0)
tag = utils.read_only_property("_tag")
@@ -191,11 +191,10 @@ class _AESCTRCipherContext(object):
self._backend = backend
self._key = self._backend._ffi.new("AES_KEY *")
- assert self._key != self._backend._ffi.NULL
res = self._backend._lib.AES_set_encrypt_key(
cipher.key, len(cipher.key) * 8, self._key
)
- assert res == 0
+ self._backend.openssl_assert(res == 0)
self._ecount = self._backend._ffi.new("char[]", 16)
self._nonce = self._backend._ffi.new("char[16]", mode.nonce)
self._num = self._backend._ffi.new("unsigned int *", 0)
diff --git a/src/cryptography/hazmat/backends/openssl/cmac.py b/src/cryptography/hazmat/backends/openssl/cmac.py
index 6c48ac34..eaefc276 100644
--- a/src/cryptography/hazmat/backends/openssl/cmac.py
+++ b/src/cryptography/hazmat/backends/openssl/cmac.py
@@ -33,7 +33,7 @@ class _CMACContext(object):
ctx = self._backend._lib.CMAC_CTX_new()
- assert ctx != self._backend._ffi.NULL
+ self._backend.openssl_assert(ctx != self._backend._ffi.NULL)
ctx = self._backend._ffi.gc(ctx, self._backend._lib.CMAC_CTX_free)
self._backend._lib.CMAC_Init(
@@ -47,7 +47,7 @@ class _CMACContext(object):
def update(self, data):
res = self._backend._lib.CMAC_Update(self._ctx, data, len(data))
- assert res == 1
+ self._backend.openssl_assert(res == 1)
def finalize(self):
buf = self._backend._ffi.new("unsigned char[]", self._output_length)
@@ -55,7 +55,7 @@ class _CMACContext(object):
res = self._backend._lib.CMAC_Final(
self._ctx, buf, length
)
- assert res == 1
+ self._backend.openssl_assert(res == 1)
self._ctx = None
@@ -69,7 +69,7 @@ class _CMACContext(object):
res = self._backend._lib.CMAC_CTX_copy(
copied_ctx, self._ctx
)
- assert res == 1
+ self._backend.openssl_assert(res == 1)
return _CMACContext(
self._backend, self._algorithm, ctx=copied_ctx
)
diff --git a/src/cryptography/hazmat/backends/openssl/dsa.py b/src/cryptography/hazmat/backends/openssl/dsa.py
index ec056767..9b4c1aff 100644
--- a/src/cryptography/hazmat/backends/openssl/dsa.py
+++ b/src/cryptography/hazmat/backends/openssl/dsa.py
@@ -82,8 +82,8 @@ class _DSASignatureContext(object):
res = self._backend._lib.DSA_sign(
0, data_to_sign, len(data_to_sign), sig_buf,
buflen, self._private_key._dsa_cdata)
- assert res == 1
- assert buflen[0]
+ self._backend.openssl_assert(res == 1)
+ self._backend.openssl_assert(buflen[0])
return self._backend._ffi.buffer(sig_buf)[:buflen[0]]
@@ -133,7 +133,7 @@ class _DSAPrivateKey(object):
def public_key(self):
dsa_cdata = self._backend._lib.DSA_new()
- assert dsa_cdata != self._backend._ffi.NULL
+ self._backend.openssl_assert(dsa_cdata != self._backend._ffi.NULL)
dsa_cdata = self._backend._ffi.gc(
dsa_cdata, self._backend._lib.DSA_free
)
@@ -146,7 +146,7 @@ class _DSAPrivateKey(object):
def parameters(self):
dsa_cdata = self._backend._lib.DSA_new()
- assert dsa_cdata != self._backend._ffi.NULL
+ self._backend.openssl_assert(dsa_cdata != self._backend._ffi.NULL)
dsa_cdata = self._backend._ffi.gc(
dsa_cdata, self._backend._lib.DSA_free
)
@@ -195,7 +195,7 @@ class _DSAPublicKey(object):
def parameters(self):
dsa_cdata = self._backend._lib.DSA_new()
- assert dsa_cdata != self._backend._ffi.NULL
+ self._backend.openssl_assert(dsa_cdata != self._backend._ffi.NULL)
dsa_cdata = self._backend._ffi.gc(
dsa_cdata, self._backend._lib.DSA_free
)
diff --git a/src/cryptography/hazmat/backends/openssl/ec.py b/src/cryptography/hazmat/backends/openssl/ec.py
index 6764416d..939a3f90 100644
--- a/src/cryptography/hazmat/backends/openssl/ec.py
+++ b/src/cryptography/hazmat/backends/openssl/ec.py
@@ -30,10 +30,10 @@ def _truncate_digest_for_ecdsa(ec_key_cdata, digest, backend):
with backend._tmp_bn_ctx() as bn_ctx:
order = _lib.BN_CTX_get(bn_ctx)
- assert order != _ffi.NULL
+ backend.openssl_assert(order != _ffi.NULL)
res = _lib.EC_GROUP_get_order(group, order, bn_ctx)
- assert res == 1
+ backend.openssl_assert(res == 1)
order_bits = _lib.BN_num_bits(order)
@@ -42,7 +42,7 @@ def _truncate_digest_for_ecdsa(ec_key_cdata, digest, backend):
def _ec_key_curve_sn(backend, ec_key):
group = backend._lib.EC_KEY_get0_group(ec_key)
- assert group != backend._ffi.NULL
+ backend.openssl_assert(group != backend._ffi.NULL)
nid = backend._lib.EC_GROUP_get_curve_name(group)
# The following check is to find EC keys with unnamed curves and raise
@@ -54,7 +54,7 @@ def _ec_key_curve_sn(backend, ec_key):
)
curve_name = backend._lib.OBJ_nid2sn(nid)
- assert curve_name != backend._ffi.NULL
+ backend.openssl_assert(curve_name != backend._ffi.NULL)
sn = backend._ffi.string(curve_name).decode('ascii')
return sn
@@ -100,7 +100,7 @@ class _ECDSASignatureContext(object):
digest = _truncate_digest_for_ecdsa(ec_key, digest, self._backend)
max_size = self._backend._lib.ECDSA_size(ec_key)
- assert max_size > 0
+ self._backend.openssl_assert(max_size > 0)
sigbuf = self._backend._ffi.new("char[]", max_size)
siglen_ptr = self._backend._ffi.new("unsigned int[]", 1)
@@ -112,7 +112,7 @@ class _ECDSASignatureContext(object):
siglen_ptr,
ec_key
)
- assert res == 1
+ self._backend.openssl_assert(res == 1)
return self._backend._ffi.buffer(sigbuf)[:siglen_ptr[0]]
@@ -173,21 +173,21 @@ class _EllipticCurvePrivateKey(object):
def public_key(self):
group = self._backend._lib.EC_KEY_get0_group(self._ec_key)
- assert group != self._backend._ffi.NULL
+ self._backend.openssl_assert(group != self._backend._ffi.NULL)
curve_nid = self._backend._lib.EC_GROUP_get_curve_name(group)
public_ec_key = self._backend._lib.EC_KEY_new_by_curve_name(curve_nid)
- assert public_ec_key != self._backend._ffi.NULL
+ self._backend.openssl_assert(public_ec_key != self._backend._ffi.NULL)
public_ec_key = self._backend._ffi.gc(
public_ec_key, self._backend._lib.EC_KEY_free
)
point = self._backend._lib.EC_KEY_get0_public_key(self._ec_key)
- assert point != self._backend._ffi.NULL
+ self._backend.openssl_assert(point != self._backend._ffi.NULL)
res = self._backend._lib.EC_KEY_set_public_key(public_ec_key, point)
- assert res == 1
+ self._backend.openssl_assert(res == 1)
evp_pkey = self._backend._ec_cdata_to_evp_pkey(public_ec_key)
@@ -242,14 +242,14 @@ class _EllipticCurvePublicKey(object):
self._backend._ec_key_determine_group_get_set_funcs(self._ec_key)
)
point = self._backend._lib.EC_KEY_get0_public_key(self._ec_key)
- assert point != self._backend._ffi.NULL
+ self._backend.openssl_assert(point != self._backend._ffi.NULL)
with self._backend._tmp_bn_ctx() as bn_ctx:
bn_x = self._backend._lib.BN_CTX_get(bn_ctx)
bn_y = self._backend._lib.BN_CTX_get(bn_ctx)
res = get_func(group, point, bn_x, bn_y, bn_ctx)
- assert res == 1
+ self._backend.openssl_assert(res == 1)
x = self._backend._bn_to_int(bn_x)
y = self._backend._bn_to_int(bn_y)
diff --git a/src/cryptography/hazmat/backends/openssl/hashes.py b/src/cryptography/hazmat/backends/openssl/hashes.py
index 2c1702f8..02ce5f0d 100644
--- a/src/cryptography/hazmat/backends/openssl/hashes.py
+++ b/src/cryptography/hazmat/backends/openssl/hashes.py
@@ -31,7 +31,7 @@ class _HashContext(object):
)
res = self._backend._lib.EVP_DigestInit_ex(ctx, evp_md,
self._backend._ffi.NULL)
- assert res != 0
+ self._backend.openssl_assert(res != 0)
self._ctx = ctx
@@ -43,20 +43,20 @@ class _HashContext(object):
copied_ctx, self._backend._lib.EVP_MD_CTX_destroy
)
res = self._backend._lib.EVP_MD_CTX_copy_ex(copied_ctx, self._ctx)
- assert res != 0
+ self._backend.openssl_assert(res != 0)
return _HashContext(self._backend, self.algorithm, ctx=copied_ctx)
def update(self, data):
res = self._backend._lib.EVP_DigestUpdate(self._ctx, data, len(data))
- assert res != 0
+ self._backend.openssl_assert(res != 0)
def finalize(self):
buf = self._backend._ffi.new("unsigned char[]",
self._backend._lib.EVP_MAX_MD_SIZE)
outlen = self._backend._ffi.new("unsigned int *")
res = self._backend._lib.EVP_DigestFinal_ex(self._ctx, buf, outlen)
- assert res != 0
- assert outlen[0] == self.algorithm.digest_size
+ self._backend.openssl_assert(res != 0)
+ self._backend.openssl_assert(outlen[0] == self.algorithm.digest_size)
res = self._backend._lib.EVP_MD_CTX_cleanup(self._ctx)
- assert res == 1
+ self._backend.openssl_assert(res == 1)
return self._backend._ffi.buffer(buf)[:outlen[0]]
diff --git a/src/cryptography/hazmat/backends/openssl/hmac.py b/src/cryptography/hazmat/backends/openssl/hmac.py
index dfe9d93b..dcf2fbaf 100644
--- a/src/cryptography/hazmat/backends/openssl/hmac.py
+++ b/src/cryptography/hazmat/backends/openssl/hmac.py
@@ -36,7 +36,7 @@ class _HMACContext(object):
res = self._backend._lib.Cryptography_HMAC_Init_ex(
ctx, key, len(key), evp_md, self._backend._ffi.NULL
)
- assert res != 0
+ self._backend.openssl_assert(res != 0)
self._ctx = ctx
self._key = key
@@ -52,7 +52,7 @@ class _HMACContext(object):
res = self._backend._lib.Cryptography_HMAC_CTX_copy(
copied_ctx, self._ctx
)
- assert res != 0
+ self._backend.openssl_assert(res != 0)
return _HMACContext(
self._backend, self._key, self.algorithm, ctx=copied_ctx
)
@@ -61,7 +61,7 @@ class _HMACContext(object):
res = self._backend._lib.Cryptography_HMAC_Update(
self._ctx, data, len(data)
)
- assert res != 0
+ self._backend.openssl_assert(res != 0)
def finalize(self):
buf = self._backend._ffi.new("unsigned char[]",
@@ -70,8 +70,8 @@ class _HMACContext(object):
res = self._backend._lib.Cryptography_HMAC_Final(
self._ctx, buf, outlen
)
- assert res != 0
- assert outlen[0] == self.algorithm.digest_size
+ self._backend.openssl_assert(res != 0)
+ self._backend.openssl_assert(outlen[0] == self.algorithm.digest_size)
self._backend._lib.HMAC_CTX_cleanup(self._ctx)
return self._backend._ffi.buffer(buf)[:outlen[0]]
diff --git a/src/cryptography/hazmat/backends/openssl/rsa.py b/src/cryptography/hazmat/backends/openssl/rsa.py
index 7da42292..664f6d35 100644
--- a/src/cryptography/hazmat/backends/openssl/rsa.py
+++ b/src/cryptography/hazmat/backends/openssl/rsa.py
@@ -89,15 +89,15 @@ def _enc_dec_rsa_pkey_ctx(backend, key, data, padding_enum):
pkey_ctx = backend._lib.EVP_PKEY_CTX_new(
key._evp_pkey, backend._ffi.NULL
)
- assert pkey_ctx != backend._ffi.NULL
+ backend.openssl_assert(pkey_ctx != backend._ffi.NULL)
pkey_ctx = backend._ffi.gc(pkey_ctx, backend._lib.EVP_PKEY_CTX_free)
res = init(pkey_ctx)
- assert res == 1
+ backend.openssl_assert(res == 1)
res = backend._lib.EVP_PKEY_CTX_set_rsa_padding(
pkey_ctx, padding_enum)
- assert res > 0
+ backend.openssl_assert(res > 0)
buf_size = backend._lib.EVP_PKEY_size(key._evp_pkey)
- assert buf_size > 0
+ backend.openssl_assert(buf_size > 0)
outlen = backend._ffi.new("size_t *", buf_size)
buf = backend._ffi.new("char[]", buf_size)
res = crypt(pkey_ctx, buf, outlen, data, len(data))
@@ -114,7 +114,7 @@ def _enc_dec_rsa_098(backend, key, data, padding_enum):
crypt = backend._lib.RSA_private_decrypt
key_size = backend._lib.RSA_size(key._rsa_cdata)
- assert key_size > 0
+ backend.openssl_assert(key_size > 0)
buf = backend._ffi.new("unsigned char[]", key_size)
res = crypt(len(data), data, buf, key._rsa_cdata, padding_enum)
if res < 0:
@@ -158,6 +158,7 @@ class _RSASignatureContext(object):
self._pkey_size = self._backend._lib.EVP_PKEY_size(
self._private_key._evp_pkey
)
+ self._backend.openssl_assert(self._pkey_size > 0)
if isinstance(padding, PKCS1v15):
if self._backend._lib.Cryptography_HAS_PKEY_CTX:
@@ -174,7 +175,6 @@ class _RSASignatureContext(object):
# Size of key in bytes - 2 is the maximum
# PSS signature length (salt length is checked later)
- assert self._pkey_size > 0
if self._pkey_size - algorithm.digest_size - 2 < 0:
raise ValueError("Digest too large for key size. Use a larger "
"key.")
@@ -207,7 +207,7 @@ class _RSASignatureContext(object):
def finalize(self):
evp_md = self._backend._lib.EVP_get_digestbyname(
self._algorithm.name.encode("ascii"))
- assert evp_md != self._backend._ffi.NULL
+ self._backend.openssl_assert(evp_md != self._backend._ffi.NULL)
return self._finalize_method(evp_md)
@@ -215,18 +215,18 @@ class _RSASignatureContext(object):
pkey_ctx = self._backend._lib.EVP_PKEY_CTX_new(
self._private_key._evp_pkey, self._backend._ffi.NULL
)
- assert pkey_ctx != self._backend._ffi.NULL
+ self._backend.openssl_assert(pkey_ctx != self._backend._ffi.NULL)
pkey_ctx = self._backend._ffi.gc(pkey_ctx,
self._backend._lib.EVP_PKEY_CTX_free)
res = self._backend._lib.EVP_PKEY_sign_init(pkey_ctx)
- assert res == 1
+ self._backend.openssl_assert(res == 1)
res = self._backend._lib.EVP_PKEY_CTX_set_signature_md(
pkey_ctx, evp_md)
- assert res > 0
+ self._backend.openssl_assert(res > 0)
res = self._backend._lib.EVP_PKEY_CTX_set_rsa_padding(
pkey_ctx, self._padding_enum)
- assert res > 0
+ self._backend.openssl_assert(res > 0)
if isinstance(self._padding, PSS):
res = self._backend._lib.EVP_PKEY_CTX_set_rsa_pss_saltlen(
pkey_ctx,
@@ -236,17 +236,19 @@ class _RSASignatureContext(object):
self._hash_ctx.algorithm.digest_size
)
)
- assert res > 0
+ self._backend.openssl_assert(res > 0)
if self._backend._lib.Cryptography_HAS_MGF1_MD:
# MGF1 MD is configurable in OpenSSL 1.0.1+
mgf1_md = self._backend._lib.EVP_get_digestbyname(
self._padding._mgf._algorithm.name.encode("ascii"))
- assert mgf1_md != self._backend._ffi.NULL
+ self._backend.openssl_assert(
+ mgf1_md != self._backend._ffi.NULL
+ )
res = self._backend._lib.EVP_PKEY_CTX_set_rsa_mgf1_md(
pkey_ctx, mgf1_md
)
- assert res > 0
+ self._backend.openssl_assert(res > 0)
data_to_sign = self._hash_ctx.finalize()
buflen = self._backend._ffi.new("size_t *")
res = self._backend._lib.EVP_PKEY_sign(
@@ -256,7 +258,7 @@ class _RSASignatureContext(object):
data_to_sign,
len(data_to_sign)
)
- assert res == 1
+ self._backend.openssl_assert(res == 1)
buf = self._backend._ffi.new("unsigned char[]", buflen[0])
res = self._backend._lib.EVP_PKEY_sign(
pkey_ctx, buf, buflen, data_to_sign, len(data_to_sign))
@@ -330,7 +332,7 @@ class _RSASignatureContext(object):
self._private_key._rsa_cdata,
self._backend._lib.RSA_NO_PADDING
)
- assert sig_len != -1
+ self._backend.openssl_assert(sig_len != -1)
return self._backend._ffi.buffer(sig_buf)[:sig_len]
@@ -347,6 +349,7 @@ class _RSAVerificationContext(object):
self._pkey_size = self._backend._lib.EVP_PKEY_size(
self._public_key._evp_pkey
)
+ self._backend.openssl_assert(self._pkey_size > 0)
if isinstance(padding, PKCS1v15):
if self._backend._lib.Cryptography_HAS_PKEY_CTX:
@@ -363,7 +366,6 @@ class _RSAVerificationContext(object):
# Size of key in bytes - 2 is the maximum
# PSS signature length (salt length is checked later)
- assert self._pkey_size > 0
if self._pkey_size - algorithm.digest_size - 2 < 0:
raise ValueError(
"Digest too large for key size. Check that you have the "
@@ -398,7 +400,7 @@ class _RSAVerificationContext(object):
def verify(self):
evp_md = self._backend._lib.EVP_get_digestbyname(
self._algorithm.name.encode("ascii"))
- assert evp_md != self._backend._ffi.NULL
+ self._backend.openssl_assert(evp_md != self._backend._ffi.NULL)
self._verify_method(evp_md)
@@ -406,18 +408,18 @@ class _RSAVerificationContext(object):
pkey_ctx = self._backend._lib.EVP_PKEY_CTX_new(
self._public_key._evp_pkey, self._backend._ffi.NULL
)
- assert pkey_ctx != self._backend._ffi.NULL
+ self._backend.openssl_assert(pkey_ctx != self._backend._ffi.NULL)
pkey_ctx = self._backend._ffi.gc(pkey_ctx,
self._backend._lib.EVP_PKEY_CTX_free)
res = self._backend._lib.EVP_PKEY_verify_init(pkey_ctx)
- assert res == 1
+ self._backend.openssl_assert(res == 1)
res = self._backend._lib.EVP_PKEY_CTX_set_signature_md(
pkey_ctx, evp_md)
- assert res > 0
+ self._backend.openssl_assert(res > 0)
res = self._backend._lib.EVP_PKEY_CTX_set_rsa_padding(
pkey_ctx, self._padding_enum)
- assert res > 0
+ self._backend.openssl_assert(res > 0)
if isinstance(self._padding, PSS):
res = self._backend._lib.EVP_PKEY_CTX_set_rsa_pss_saltlen(
pkey_ctx,
@@ -427,16 +429,18 @@ class _RSAVerificationContext(object):
self._hash_ctx.algorithm.digest_size
)
)
- assert res > 0
+ self._backend.openssl_assert(res > 0)
if self._backend._lib.Cryptography_HAS_MGF1_MD:
# MGF1 MD is configurable in OpenSSL 1.0.1+
mgf1_md = self._backend._lib.EVP_get_digestbyname(
self._padding._mgf._algorithm.name.encode("ascii"))
- assert mgf1_md != self._backend._ffi.NULL
+ self._backend.openssl_assert(
+ mgf1_md != self._backend._ffi.NULL
+ )
res = self._backend._lib.EVP_PKEY_CTX_set_rsa_mgf1_md(
pkey_ctx, mgf1_md
)
- assert res > 0
+ self._backend.openssl_assert(res > 0)
data_to_verify = self._hash_ctx.finalize()
res = self._backend._lib.EVP_PKEY_verify(
@@ -449,7 +453,7 @@ class _RSAVerificationContext(object):
# The previous call can return negative numbers in the event of an
# error. This is not a signature failure but we need to fail if it
# occurs.
- assert res >= 0
+ self._backend.openssl_assert(res >= 0)
if res == 0:
errors = self._backend._consume_errors()
assert errors
@@ -469,7 +473,7 @@ class _RSAVerificationContext(object):
# The previous call can return negative numbers in the event of an
# error. This is not a signature failure but we need to fail if it
# occurs.
- assert res >= 0
+ self._backend.openssl_assert(res >= 0)
if res == 0:
errors = self._backend._consume_errors()
assert errors
@@ -530,12 +534,12 @@ class _RSAPrivateKey(object):
def public_key(self):
ctx = self._backend._lib.RSA_new()
- assert ctx != self._backend._ffi.NULL
+ self._backend.openssl_assert(ctx != self._backend._ffi.NULL)
ctx = self._backend._ffi.gc(ctx, self._backend._lib.RSA_free)
ctx.e = self._backend._lib.BN_dup(self._rsa_cdata.e)
ctx.n = self._backend._lib.BN_dup(self._rsa_cdata.n)
res = self._backend._lib.RSA_blinding_on(ctx, self._backend._ffi.NULL)
- assert res == 1
+ self._backend.openssl_assert(res == 1)
evp_pkey = self._backend._rsa_cdata_to_evp_pkey(ctx)
return _RSAPublicKey(self._backend, ctx, evp_pkey)
diff --git a/src/cryptography/hazmat/backends/openssl/x509.py b/src/cryptography/hazmat/backends/openssl/x509.py
index 79b516cc..8a13aa5d 100644
--- a/src/cryptography/hazmat/backends/openssl/x509.py
+++ b/src/cryptography/hazmat/backends/openssl/x509.py
@@ -27,15 +27,15 @@ def _obj2txt(backend, obj):
buf_len = 80
buf = backend._ffi.new("char[]", buf_len)
res = backend._lib.OBJ_obj2txt(buf, buf_len, obj, 1)
- assert res > 0
+ backend.openssl_assert(res > 0)
return backend._ffi.buffer(buf, res)[:].decode()
def _decode_x509_name_entry(backend, x509_name_entry):
obj = backend._lib.X509_NAME_ENTRY_get_object(x509_name_entry)
- assert obj != backend._ffi.NULL
+ backend.openssl_assert(obj != backend._ffi.NULL)
data = backend._lib.X509_NAME_ENTRY_get_data(x509_name_entry)
- assert data != backend._ffi.NULL
+ backend.openssl_assert(data != backend._ffi.NULL)
value = backend._asn1_string_to_utf8(data)
oid = _obj2txt(backend, obj)
@@ -57,7 +57,7 @@ def _decode_general_names(backend, gns):
names = []
for i in range(num):
gn = backend._lib.sk_GENERAL_NAME_value(gns, i)
- assert gn != backend._ffi.NULL
+ backend.openssl_assert(gn != backend._ffi.NULL)
names.append(_decode_general_name(backend, gn))
return names
@@ -183,7 +183,7 @@ class _X509ExtensionParser(object):
seen_oids = set()
for i in range(self.ext_count(backend, x509_obj)):
ext = self.get_ext(backend, x509_obj, i)
- assert ext != backend._ffi.NULL
+ backend.openssl_assert(ext != backend._ffi.NULL)
crit = backend._lib.X509_EXTENSION_get_critical(ext)
critical = crit == 1
oid = x509.ObjectIdentifier(_obj2txt(backend, ext.object))
@@ -257,12 +257,12 @@ class _Certificate(object):
@property
def serial(self):
asn1_int = self._backend._lib.X509_get_serialNumber(self._x509)
- assert asn1_int != self._backend._ffi.NULL
+ self._backend.openssl_assert(asn1_int != self._backend._ffi.NULL)
return self._backend._asn1_integer_to_int(asn1_int)
def public_key(self):
pkey = self._backend._lib.X509_get_pubkey(self._x509)
- assert pkey != self._backend._ffi.NULL
+ self._backend.openssl_assert(pkey != self._backend._ffi.NULL)
pkey = self._backend._ffi.gc(pkey, self._backend._lib.EVP_PKEY_free)
return self._backend._evp_pkey_to_public_key(pkey)
@@ -280,13 +280,13 @@ class _Certificate(object):
@property
def issuer(self):
issuer = self._backend._lib.X509_get_issuer_name(self._x509)
- assert issuer != self._backend._ffi.NULL
+ self._backend.openssl_assert(issuer != self._backend._ffi.NULL)
return _decode_x509_name(self._backend, issuer)
@property
def subject(self):
subject = self._backend._lib.X509_get_subject_name(self._x509)
- assert subject != self._backend._ffi.NULL
+ self._backend.openssl_assert(subject != self._backend._ffi.NULL)
return _decode_x509_name(self._backend, subject)
@property
@@ -312,7 +312,7 @@ class _Certificate(object):
else:
raise TypeError("encoding must be an item from the Encoding enum")
- assert res == 1
+ self._backend.openssl_assert(res == 1)
return self._backend._read_mem_bio(bio)
@@ -443,9 +443,9 @@ def _decode_authority_information_access(backend, aia):
access_descriptions = []
for i in range(num):
ad = backend._lib.sk_ACCESS_DESCRIPTION_value(aia, i)
- assert ad.method != backend._ffi.NULL
+ backend.openssl_assert(ad.method != backend._ffi.NULL)
oid = x509.ObjectIdentifier(_obj2txt(backend, ad.method))
- assert ad.location != backend._ffi.NULL
+ backend.openssl_assert(ad.location != backend._ffi.NULL)
gn = _decode_general_name(backend, ad.location)
access_descriptions.append(x509.AccessDescription(oid, gn))
@@ -516,7 +516,7 @@ def _decode_general_subtrees(backend, stack_subtrees):
for i in range(num):
obj = backend._lib.sk_GENERAL_SUBTREE_value(stack_subtrees, i)
- assert obj != backend._ffi.NULL
+ backend.openssl_assert(obj != backend._ffi.NULL)
name = _decode_general_name(backend, obj.base)
subtrees.append(name)
@@ -531,7 +531,7 @@ def _decode_extended_key_usage(backend, sk):
for i in range(num):
obj = backend._lib.sk_ASN1_OBJECT_value(sk, i)
- assert obj != backend._ffi.NULL
+ backend.openssl_assert(obj != backend._ffi.NULL)
oid = x509.ObjectIdentifier(_obj2txt(backend, obj))
ekus.append(oid)
@@ -616,7 +616,7 @@ def _decode_crl_distribution_points(backend, cdps):
rn = backend._lib.sk_X509_NAME_ENTRY_value(
rns, i
)
- assert rn != backend._ffi.NULL
+ backend.openssl_assert(rn != backend._ffi.NULL)
attributes.append(
_decode_x509_name_entry(backend, rn)
)
@@ -917,14 +917,14 @@ class _CertificateSigningRequest(object):
def public_key(self):
pkey = self._backend._lib.X509_REQ_get_pubkey(self._x509_req)
- assert pkey != self._backend._ffi.NULL
+ self._backend.openssl_assert(pkey != self._backend._ffi.NULL)
pkey = self._backend._ffi.gc(pkey, self._backend._lib.EVP_PKEY_free)
return self._backend._evp_pkey_to_public_key(pkey)
@property
def subject(self):
subject = self._backend._lib.X509_REQ_get_subject_name(self._x509_req)
- assert subject != self._backend._ffi.NULL
+ self._backend.openssl_assert(subject != self._backend._ffi.NULL)
return _decode_x509_name(self._backend, subject)
@property
@@ -953,7 +953,7 @@ class _CertificateSigningRequest(object):
else:
raise TypeError("encoding must be an item from the Encoding enum")
- assert res == 1
+ self._backend.openssl_assert(res == 1)
return self._backend._read_mem_bio(bio)
diff --git a/src/cryptography/hazmat/bindings/openssl/binding.py b/src/cryptography/hazmat/bindings/openssl/binding.py
index 50d7f6d5..a5635f7d 100644
--- a/src/cryptography/hazmat/bindings/openssl/binding.py
+++ b/src/cryptography/hazmat/bindings/openssl/binding.py
@@ -4,6 +4,7 @@
from __future__ import absolute_import, division, print_function
+import collections
import os
import threading
import types
@@ -12,6 +13,42 @@ from cryptography.hazmat.bindings._openssl import ffi, lib
from cryptography.hazmat.bindings.openssl._conditional import CONDITIONAL_NAMES
+_OpenSSLError = collections.namedtuple("_OpenSSLError",
+ ["code", "lib", "func", "reason"])
+
+
+class UnhandledOpenSSLError(Exception):
+ def __init__(self, msg, errors):
+ super(UnhandledOpenSSLError, self).__init__(msg)
+ self.errors = errors
+
+
+def _consume_errors(lib):
+ errors = []
+ while True:
+ code = lib.ERR_get_error()
+ if code == 0:
+ break
+
+ err_lib = lib.ERR_GET_LIB(code)
+ err_func = lib.ERR_GET_FUNC(code)
+ err_reason = lib.ERR_GET_REASON(code)
+
+ errors.append(_OpenSSLError(code, err_lib, err_func, err_reason))
+ return errors
+
+
+def _openssl_assert(lib, ok):
+ if not ok:
+ errors = _consume_errors(lib)
+ raise UnhandledOpenSSLError(
+ "Unknown OpenSSL error. Please file an issue at https://github.com"
+ "/pyca/cryptography/issues with information on how to reproduce "
+ "this.",
+ errors
+ )
+
+
@ffi.callback("int (*)(unsigned char *, int)", error=-1)
def _osrandom_rand_bytes(buf, size):
signed = ffi.cast("char *", buf)
@@ -64,7 +101,7 @@ class Binding(object):
@classmethod
def _register_osrandom_engine(cls):
- assert cls.lib.ERR_peek_error() == 0
+ _openssl_assert(cls.lib, cls.lib.ERR_peek_error() == 0)
looked_up_engine = cls.lib.ENGINE_by_id(cls._osrandom_engine_id)
if looked_up_engine != ffi.NULL:
raise RuntimeError("osrandom engine already registered")
@@ -72,19 +109,19 @@ class Binding(object):
cls.lib.ERR_clear_error()
engine = cls.lib.ENGINE_new()
- assert engine != cls.ffi.NULL
+ _openssl_assert(cls.lib, engine != cls.ffi.NULL)
try:
result = cls.lib.ENGINE_set_id(engine, cls._osrandom_engine_id)
- assert result == 1
+ _openssl_assert(cls.lib, result == 1)
result = cls.lib.ENGINE_set_name(engine, cls._osrandom_engine_name)
- assert result == 1
+ _openssl_assert(cls.lib, result == 1)
result = cls.lib.ENGINE_set_RAND(engine, cls._osrandom_method)
- assert result == 1
+ _openssl_assert(cls.lib, result == 1)
result = cls.lib.ENGINE_add(engine)
- assert result == 1
+ _openssl_assert(cls.lib, result == 1)
finally:
result = cls.lib.ENGINE_free(engine)
- assert result == 1
+ _openssl_assert(cls.lib, result == 1)
@classmethod
def _ensure_ffi_initialized(cls):
generated by cgit v1.2.3 (git 2.25.1) at 2024-07-17 06:27:09 +0000