Paint it Black by the Rolling Stones (#920)
diff --git a/src/OpenSSL/SSL.py b/src/OpenSSL/SSL.py
index 25308f1..b4b308f 100644
--- a/src/OpenSSL/SSL.py
+++ b/src/OpenSSL/SSL.py
@@ -23,100 +23,108 @@
)
from OpenSSL.crypto import (
- FILETYPE_PEM, _PassphraseHelper, PKey, X509Name, X509, X509Store)
+ FILETYPE_PEM,
+ _PassphraseHelper,
+ PKey,
+ X509Name,
+ X509,
+ X509Store,
+)
__all__ = [
- 'OPENSSL_VERSION_NUMBER',
- 'SSLEAY_VERSION',
- 'SSLEAY_CFLAGS',
- 'SSLEAY_PLATFORM',
- 'SSLEAY_DIR',
- 'SSLEAY_BUILT_ON',
- 'SENT_SHUTDOWN',
- 'RECEIVED_SHUTDOWN',
- 'SSLv2_METHOD',
- 'SSLv3_METHOD',
- 'SSLv23_METHOD',
- 'TLSv1_METHOD',
- 'TLSv1_1_METHOD',
- 'TLSv1_2_METHOD',
- 'OP_NO_SSLv2',
- 'OP_NO_SSLv3',
- 'OP_NO_TLSv1',
- 'OP_NO_TLSv1_1',
- 'OP_NO_TLSv1_2',
- 'OP_NO_TLSv1_3',
- 'MODE_RELEASE_BUFFERS',
- 'OP_SINGLE_DH_USE',
- 'OP_SINGLE_ECDH_USE',
- 'OP_EPHEMERAL_RSA',
- 'OP_MICROSOFT_SESS_ID_BUG',
- 'OP_NETSCAPE_CHALLENGE_BUG',
- 'OP_NETSCAPE_REUSE_CIPHER_CHANGE_BUG',
- 'OP_SSLREF2_REUSE_CERT_TYPE_BUG',
- 'OP_MICROSOFT_BIG_SSLV3_BUFFER',
- 'OP_MSIE_SSLV2_RSA_PADDING',
- 'OP_SSLEAY_080_CLIENT_DH_BUG',
- 'OP_TLS_D5_BUG',
- 'OP_TLS_BLOCK_PADDING_BUG',
- 'OP_DONT_INSERT_EMPTY_FRAGMENTS',
- 'OP_CIPHER_SERVER_PREFERENCE',
- 'OP_TLS_ROLLBACK_BUG',
- 'OP_PKCS1_CHECK_1',
- 'OP_PKCS1_CHECK_2',
- 'OP_NETSCAPE_CA_DN_BUG',
- 'OP_NETSCAPE_DEMO_CIPHER_CHANGE_BUG',
- 'OP_NO_COMPRESSION',
- 'OP_NO_QUERY_MTU',
- 'OP_COOKIE_EXCHANGE',
- 'OP_NO_TICKET',
- 'OP_ALL',
- 'VERIFY_PEER',
- 'VERIFY_FAIL_IF_NO_PEER_CERT',
- 'VERIFY_CLIENT_ONCE',
- 'VERIFY_NONE',
- 'SESS_CACHE_OFF',
- 'SESS_CACHE_CLIENT',
- 'SESS_CACHE_SERVER',
- 'SESS_CACHE_BOTH',
- 'SESS_CACHE_NO_AUTO_CLEAR',
- 'SESS_CACHE_NO_INTERNAL_LOOKUP',
- 'SESS_CACHE_NO_INTERNAL_STORE',
- 'SESS_CACHE_NO_INTERNAL',
- 'SSL_ST_CONNECT',
- 'SSL_ST_ACCEPT',
- 'SSL_ST_MASK',
- 'SSL_CB_LOOP',
- 'SSL_CB_EXIT',
- 'SSL_CB_READ',
- 'SSL_CB_WRITE',
- 'SSL_CB_ALERT',
- 'SSL_CB_READ_ALERT',
- 'SSL_CB_WRITE_ALERT',
- 'SSL_CB_ACCEPT_LOOP',
- 'SSL_CB_ACCEPT_EXIT',
- 'SSL_CB_CONNECT_LOOP',
- 'SSL_CB_CONNECT_EXIT',
- 'SSL_CB_HANDSHAKE_START',
- 'SSL_CB_HANDSHAKE_DONE',
- 'Error',
- 'WantReadError',
- 'WantWriteError',
- 'WantX509LookupError',
- 'ZeroReturnError',
- 'SysCallError',
- 'SSLeay_version',
- 'Session',
- 'Context',
- 'Connection'
+ "OPENSSL_VERSION_NUMBER",
+ "SSLEAY_VERSION",
+ "SSLEAY_CFLAGS",
+ "SSLEAY_PLATFORM",
+ "SSLEAY_DIR",
+ "SSLEAY_BUILT_ON",
+ "SENT_SHUTDOWN",
+ "RECEIVED_SHUTDOWN",
+ "SSLv2_METHOD",
+ "SSLv3_METHOD",
+ "SSLv23_METHOD",
+ "TLSv1_METHOD",
+ "TLSv1_1_METHOD",
+ "TLSv1_2_METHOD",
+ "OP_NO_SSLv2",
+ "OP_NO_SSLv3",
+ "OP_NO_TLSv1",
+ "OP_NO_TLSv1_1",
+ "OP_NO_TLSv1_2",
+ "OP_NO_TLSv1_3",
+ "MODE_RELEASE_BUFFERS",
+ "OP_SINGLE_DH_USE",
+ "OP_SINGLE_ECDH_USE",
+ "OP_EPHEMERAL_RSA",
+ "OP_MICROSOFT_SESS_ID_BUG",
+ "OP_NETSCAPE_CHALLENGE_BUG",
+ "OP_NETSCAPE_REUSE_CIPHER_CHANGE_BUG",
+ "OP_SSLREF2_REUSE_CERT_TYPE_BUG",
+ "OP_MICROSOFT_BIG_SSLV3_BUFFER",
+ "OP_MSIE_SSLV2_RSA_PADDING",
+ "OP_SSLEAY_080_CLIENT_DH_BUG",
+ "OP_TLS_D5_BUG",
+ "OP_TLS_BLOCK_PADDING_BUG",
+ "OP_DONT_INSERT_EMPTY_FRAGMENTS",
+ "OP_CIPHER_SERVER_PREFERENCE",
+ "OP_TLS_ROLLBACK_BUG",
+ "OP_PKCS1_CHECK_1",
+ "OP_PKCS1_CHECK_2",
+ "OP_NETSCAPE_CA_DN_BUG",
+ "OP_NETSCAPE_DEMO_CIPHER_CHANGE_BUG",
+ "OP_NO_COMPRESSION",
+ "OP_NO_QUERY_MTU",
+ "OP_COOKIE_EXCHANGE",
+ "OP_NO_TICKET",
+ "OP_ALL",
+ "VERIFY_PEER",
+ "VERIFY_FAIL_IF_NO_PEER_CERT",
+ "VERIFY_CLIENT_ONCE",
+ "VERIFY_NONE",
+ "SESS_CACHE_OFF",
+ "SESS_CACHE_CLIENT",
+ "SESS_CACHE_SERVER",
+ "SESS_CACHE_BOTH",
+ "SESS_CACHE_NO_AUTO_CLEAR",
+ "SESS_CACHE_NO_INTERNAL_LOOKUP",
+ "SESS_CACHE_NO_INTERNAL_STORE",
+ "SESS_CACHE_NO_INTERNAL",
+ "SSL_ST_CONNECT",
+ "SSL_ST_ACCEPT",
+ "SSL_ST_MASK",
+ "SSL_CB_LOOP",
+ "SSL_CB_EXIT",
+ "SSL_CB_READ",
+ "SSL_CB_WRITE",
+ "SSL_CB_ALERT",
+ "SSL_CB_READ_ALERT",
+ "SSL_CB_WRITE_ALERT",
+ "SSL_CB_ACCEPT_LOOP",
+ "SSL_CB_ACCEPT_EXIT",
+ "SSL_CB_CONNECT_LOOP",
+ "SSL_CB_CONNECT_EXIT",
+ "SSL_CB_HANDSHAKE_START",
+ "SSL_CB_HANDSHAKE_DONE",
+ "Error",
+ "WantReadError",
+ "WantWriteError",
+ "WantX509LookupError",
+ "ZeroReturnError",
+ "SysCallError",
+ "SSLeay_version",
+ "Session",
+ "Context",
+ "Connection",
]
try:
_buffer = buffer
except NameError:
+
class _buffer(object):
pass
+
OPENSSL_VERSION_NUMBER = _lib.OPENSSL_VERSION_NUMBER
SSLEAY_VERSION = _lib.SSLEAY_VERSION
SSLEAY_CFLAGS = _lib.SSLEAY_CFLAGS
@@ -199,12 +207,9 @@
SSL_ST_BEFORE = _lib.SSL_ST_BEFORE
SSL_ST_OK = _lib.SSL_ST_OK
SSL_ST_RENEGOTIATE = _lib.SSL_ST_RENEGOTIATE
- __all__.extend([
- 'SSL_ST_INIT',
- 'SSL_ST_BEFORE',
- 'SSL_ST_OK',
- 'SSL_ST_RENEGOTIATE',
- ])
+ __all__.extend(
+ ["SSL_ST_INIT", "SSL_ST_BEFORE", "SSL_ST_OK", "SSL_ST_RENEGOTIATE"]
+ )
SSL_CB_LOOP = _lib.SSL_CB_LOOP
SSL_CB_EXIT = _lib.SSL_CB_EXIT
@@ -333,7 +338,8 @@
return 0
self.callback = _ffi.callback(
- "int (*)(int, X509_STORE_CTX *)", wrapper)
+ "int (*)(int, X509_STORE_CTX *)", wrapper
+ )
class _NpnAdvertiseHelper(_CallbackExceptionHelper):
@@ -352,7 +358,7 @@
# Join the protocols into a Python bytestring, length-prefixing
# each element.
- protostr = b''.join(
+ protostr = b"".join(
chain.from_iterable((int2byte(len(p)), p) for p in protos)
)
@@ -373,7 +379,7 @@
self.callback = _ffi.callback(
"int (*)(SSL *, const unsigned char **, unsigned int *, void *)",
- wrapper
+ wrapper,
)
@@ -397,9 +403,9 @@
protolist = []
while instr:
length = indexbytes(instr, 0)
- proto = instr[1:length + 1]
+ proto = instr[1 : length + 1]
protolist.append(proto)
- instr = instr[length + 1:]
+ instr = instr[length + 1 :]
# Call the callback
outstr = callback(conn, protolist)
@@ -420,9 +426,11 @@
return 2 # SSL_TLSEXT_ERR_ALERT_FATAL
self.callback = _ffi.callback(
- ("int (*)(SSL *, unsigned char **, unsigned char *, "
- "const unsigned char *, unsigned int, void *)"),
- wrapper
+ (
+ "int (*)(SSL *, unsigned char **, unsigned char *, "
+ "const unsigned char *, unsigned int, void *)"
+ ),
+ wrapper,
)
@@ -449,15 +457,15 @@
protolist = []
while instr:
encoded_len = indexbytes(instr, 0)
- proto = instr[1:encoded_len + 1]
+ proto = instr[1 : encoded_len + 1]
protolist.append(proto)
- instr = instr[encoded_len + 1:]
+ instr = instr[encoded_len + 1 :]
# Call the callback
outbytes = callback(conn, protolist)
any_accepted = True
if outbytes is NO_OVERLAPPING_PROTOCOLS:
- outbytes = b''
+ outbytes = b""
any_accepted = False
elif not isinstance(outbytes, bytes):
raise TypeError(
@@ -482,9 +490,11 @@
return _lib.SSL_TLSEXT_ERR_ALERT_FATAL
self.callback = _ffi.callback(
- ("int (*)(SSL *, unsigned char **, unsigned char *, "
- "const unsigned char *, unsigned int, void *)"),
- wrapper
+ (
+ "int (*)(SSL *, unsigned char **, unsigned char *, "
+ "const unsigned char *, unsigned int, void *)"
+ ),
+ wrapper,
)
@@ -596,7 +606,7 @@
ocsp_len = _lib.SSL_get_tlsext_status_ocsp_resp(ssl, ocsp_ptr)
if ocsp_len < 0:
# No OCSP data.
- ocsp_data = b''
+ ocsp_data = b""
else:
# Copy the OCSP data, then pass it to the callback.
ocsp_data = _ffi.buffer(ocsp_ptr[0], ocsp_len)[:]
@@ -628,7 +638,8 @@
raise TypeError("argument must be an int, or have a fileno() method.")
elif fd < 0:
raise ValueError(
- "file descriptor cannot be a negative integer (%i)" % (fd,))
+ "file descriptor cannot be a negative integer (%i)" % (fd,)
+ )
return fd
@@ -643,8 +654,11 @@
def _warn_npn():
- warnings.warn("NPN is deprecated. Protocols should switch to using ALPN.",
- DeprecationWarning, stacklevel=3)
+ warnings.warn(
+ "NPN is deprecated. Protocols should switch to using ALPN.",
+ DeprecationWarning,
+ stacklevel=3,
+ )
def _make_requires(flag, error):
@@ -657,11 +671,14 @@
``Cryptography_HAS_NEXTPROTONEG``.
:param error: The string to be used in the exception if the flag is false.
"""
+
def _requires_decorator(func):
if not flag:
+
@wraps(func)
def explode(*args, **kwargs):
raise NotImplementedError(error)
+
return explode
else:
return func
@@ -687,6 +704,7 @@
.. versionadded:: 0.14
"""
+
pass
@@ -698,6 +716,7 @@
:param method: One of SSLv2_METHOD, SSLv3_METHOD, SSLv23_METHOD, or
TLSv1_METHOD.
"""
+
_methods = {
SSLv2_METHOD: "SSLv2_method",
SSLv3_METHOD: "SSLv3_method",
@@ -709,7 +728,8 @@
_methods = dict(
(identifier, getattr(_lib, name))
for (identifier, name) in _methods.items()
- if getattr(_lib, name, None) is not None)
+ if getattr(_lib, name, None) is not None
+ )
def __init__(self, method):
if not isinstance(method, integer_types):
@@ -790,8 +810,10 @@
@wraps(callback)
def wrapper(size, verify, userdata):
return callback(size, verify, self._passphrase_userdata)
+
return _PassphraseHelper(
- FILETYPE_PEM, wrapper, more_args=True, truncate=True)
+ FILETYPE_PEM, wrapper, more_args=True, truncate=True
+ )
def set_passwd_cb(self, callback, userdata=None):
"""
@@ -818,7 +840,8 @@
self._passphrase_helper = self._wrap_callback(callback)
self._passphrase_callback = self._passphrase_helper.callback
_lib.SSL_CTX_set_default_passwd_cb(
- self._context, self._passphrase_callback)
+ self._context, self._passphrase_callback
+ )
self._passphrase_userdata = userdata
def set_default_verify_paths(self):
@@ -848,9 +871,9 @@
# First we'll check to see if any env vars have been set. If so,
# we won't try to do anything else because the user has set the path
# themselves.
- dir_env_var = _ffi.string(
- _lib.X509_get_default_cert_dir_env()
- ).decode("ascii")
+ dir_env_var = _ffi.string(_lib.X509_get_default_cert_dir_env()).decode(
+ "ascii"
+ )
file_env_var = _ffi.string(
_lib.X509_get_default_cert_file_env()
).decode("ascii")
@@ -861,13 +884,12 @@
# to the exact values we use in our manylinux1 builds. If they are
# then we know to load the fallbacks
if (
- default_dir == _CRYPTOGRAPHY_MANYLINUX1_CA_DIR and
- default_file == _CRYPTOGRAPHY_MANYLINUX1_CA_FILE
+ default_dir == _CRYPTOGRAPHY_MANYLINUX1_CA_DIR
+ and default_file == _CRYPTOGRAPHY_MANYLINUX1_CA_FILE
):
# This is manylinux1, let's load our fallback paths
self._fallback_default_verify_paths(
- _CERTIFICATE_FILE_LOCATIONS,
- _CERTIFICATE_PATH_LOCATIONS
+ _CERTIFICATE_FILE_LOCATIONS, _CERTIFICATE_PATH_LOCATIONS
)
def _check_env_vars_set(self, dir_env_var, file_env_var):
@@ -877,8 +899,8 @@
:return: bool
"""
return (
- os.environ.get(file_env_var) is not None or
- os.environ.get(dir_env_var) is not None
+ os.environ.get(file_env_var) is not None
+ or os.environ.get(dir_env_var) is not None
)
def _fallback_default_verify_paths(self, file_path, dir_path):
@@ -996,7 +1018,8 @@
raise TypeError("filetype must be an integer")
use_result = _lib.SSL_CTX_use_PrivateKey_file(
- self._context, keyfile, filetype)
+ self._context, keyfile, filetype
+ )
if not use_result:
self._raise_passphrase_exception()
@@ -1052,11 +1075,8 @@
"""
buf = _text_to_bytes_and_warn("buf", buf)
_openssl_assert(
- _lib.SSL_CTX_set_session_id_context(
- self._context,
- buf,
- len(buf),
- ) == 1
+ _lib.SSL_CTX_set_session_id_context(self._context, buf, len(buf),)
+ == 1
)
def set_session_cache_mode(self, mode):
@@ -1202,19 +1222,17 @@
# invalid cipher string is passed, but without the following check
# for the TLS 1.3 specific cipher suites it would never error.
tmpconn = Connection(self, None)
- if (
- tmpconn.get_cipher_list() == [
- 'TLS_AES_256_GCM_SHA384',
- 'TLS_CHACHA20_POLY1305_SHA256',
- 'TLS_AES_128_GCM_SHA256'
- ]
- ):
+ if tmpconn.get_cipher_list() == [
+ "TLS_AES_256_GCM_SHA384",
+ "TLS_CHACHA20_POLY1305_SHA256",
+ "TLS_AES_128_GCM_SHA256",
+ ]:
raise Error(
[
(
- 'SSL routines',
- 'SSL_CTX_set_cipher_list',
- 'no cipher match',
+ "SSL routines",
+ "SSL_CTX_set_cipher_list",
+ "no cipher match",
),
],
)
@@ -1240,9 +1258,7 @@
if not isinstance(ca_name, X509Name):
raise TypeError(
"client CAs must be X509Name objects, not %s "
- "objects" % (
- type(ca_name).__name__,
- )
+ "objects" % (type(ca_name).__name__,)
)
copy = _lib.X509_NAME_dup(ca_name._name)
_openssl_assert(copy != _ffi.NULL)
@@ -1273,7 +1289,8 @@
raise TypeError("certificate_authority must be an X509 instance")
add_result = _lib.SSL_CTX_add_client_CA(
- self._context, certificate_authority._x509)
+ self._context, certificate_authority._x509
+ )
_openssl_assert(add_result == 1)
def set_timeout(self, timeout):
@@ -1311,11 +1328,14 @@
function call.
:return: None
"""
+
@wraps(callback)
def wrapper(ssl, where, return_code):
callback(Connection._reverse_mapping[ssl], where, return_code)
+
self._info_callback = _ffi.callback(
- "void (*)(const SSL *, int, int)", wrapper)
+ "void (*)(const SSL *, int, int)", wrapper
+ )
_lib.SSL_CTX_set_info_callback(self._context, self._info_callback)
def get_app_data(self):
@@ -1388,15 +1408,18 @@
.. versionadded:: 0.13
"""
+
@wraps(callback)
def wrapper(ssl, alert, arg):
callback(Connection._reverse_mapping[ssl])
return 0
self._tlsext_servername_callback = _ffi.callback(
- "int (*)(SSL *, int *, void *)", wrapper)
+ "int (*)(SSL *, int *, void *)", wrapper
+ )
_lib.SSL_CTX_set_tlsext_servername_callback(
- self._context, self._tlsext_servername_callback)
+ self._context, self._tlsext_servername_callback
+ )
def set_tlsext_use_srtp(self, profiles):
"""
@@ -1431,7 +1454,8 @@
self._npn_advertise_helper = _NpnAdvertiseHelper(callback)
self._npn_advertise_callback = self._npn_advertise_helper.callback
_lib.SSL_CTX_set_next_protos_advertised_cb(
- self._context, self._npn_advertise_callback, _ffi.NULL)
+ self._context, self._npn_advertise_callback, _ffi.NULL
+ )
@_requires_npn
def set_npn_select_callback(self, callback):
@@ -1450,7 +1474,8 @@
self._npn_select_helper = _NpnSelectHelper(callback)
self._npn_select_callback = self._npn_select_helper.callback
_lib.SSL_CTX_set_next_proto_select_cb(
- self._context, self._npn_select_callback, _ffi.NULL)
+ self._context, self._npn_select_callback, _ffi.NULL
+ )
@_requires_alpn
def set_alpn_protos(self, protos):
@@ -1465,7 +1490,7 @@
"""
# Take the list of protocols and join them together, prefixing them
# with their lengths.
- protostr = b''.join(
+ protostr = b"".join(
chain.from_iterable((int2byte(len(p)), p) for p in protos)
)
@@ -1492,7 +1517,8 @@
self._alpn_select_helper = _ALPNSelectHelper(callback)
self._alpn_select_callback = self._alpn_select_helper.callback
_lib.SSL_CTX_set_alpn_select_cb(
- self._context, self._alpn_select_callback, _ffi.NULL)
+ self._context, self._alpn_select_callback, _ffi.NULL
+ )
def _set_ocsp_callback(self, helper, data):
"""
@@ -1556,6 +1582,7 @@
class Connection(object):
"""
"""
+
_reverse_mapping = WeakValueDictionary()
def __init__(self, context, socket=None):
@@ -1609,7 +1636,8 @@
self._from_ssl = None
self._socket = socket
set_result = _lib.SSL_set_fd(
- self._ssl, _asFileDescriptor(self._socket))
+ self._ssl, _asFileDescriptor(self._socket)
+ )
_openssl_assert(set_result == 1)
def __getattr__(self, name):
@@ -1618,9 +1646,10 @@
on the Connection object.
"""
if self._socket is None:
- raise AttributeError("'%s' object has no attribute '%s'" % (
- self.__class__.__name__, name
- ))
+ raise AttributeError(
+ "'%s' object has no attribute '%s'"
+ % (self.__class__.__name__, name)
+ )
else:
return getattr(self._socket, name)
@@ -1777,9 +1806,7 @@
# SSL_write's num arg is an int,
# so we cannot send more than 2**31-1 bytes at once.
result = _lib.SSL_write(
- self._ssl,
- data + total_sent,
- min(left_to_send, 2147483647)
+ self._ssl, data + total_sent, min(left_to_send, 2147483647)
)
self._raise_ssl_error(self._ssl, result)
total_sent += result
@@ -1803,6 +1830,7 @@
result = _lib.SSL_read(self._ssl, buf, bufsiz)
self._raise_ssl_error(self._ssl, result)
return _ffi.buffer(buf, result)[:]
+
read = recv
def recv_into(self, buffer, nbytes=None, flags=None):
@@ -2069,7 +2097,8 @@
:raise: NotImplementedError
"""
raise NotImplementedError(
- "Cannot make file object of OpenSSL.SSL.Connection")
+ "Cannot make file object of OpenSSL.SSL.Connection"
+ )
def get_app_data(self):
"""
@@ -2182,10 +2211,16 @@
context_buf = context
context_len = len(context)
use_context = 1
- success = _lib.SSL_export_keying_material(self._ssl, outp, olen,
- label, len(label),
- context_buf, context_len,
- use_context)
+ success = _lib.SSL_export_keying_material(
+ self._ssl,
+ outp,
+ olen,
+ label,
+ len(label),
+ context_buf,
+ context_len,
+ use_context,
+ )
_openssl_assert(success == 1)
return _ffi.buffer(outp, olen)[:]
@@ -2470,7 +2505,7 @@
"""
# Take the list of protocols and join them together, prefixing them
# with their lengths.
- protostr = b''.join(
+ protostr = b"".join(
chain.from_iterable((int2byte(len(p)), p) for p in protos)
)
@@ -2493,7 +2528,7 @@
_lib.SSL_get0_alpn_selected(self._ssl, data, data_len)
if not data_len:
- return b''
+ return b""
return _ffi.buffer(data[0], data_len[0])[:]