<s> import socket <EOL> from sys import platform <EOL> from functools import wraps , partial <EOL> from itertools import count , chain <EOL> from weakref import WeakValueDictionary <EOL> from errno import errorcode <EOL> from six import binary_type as _binary_type <EOL> from six import integer_types as integer_types <EOL> from six import int2byte , indexbytes <EOL> from OpenSSL . _util import ( <EOL> UNSPECIFIED as _UNSPECIFIED , <EOL> exception_from_error_queue as _exception_from_error_queue , <EOL> ffi as _ffi , <EOL> lib as _lib , <EOL> make_assert as _make_assert , <EOL> native as _native , <EOL> path_string as _path_string , <EOL> text_to_bytes_and_warn as _text_to_bytes_and_warn , <EOL> ) <EOL> from OpenSSL . crypto import ( <EOL> FILETYPE_PEM , _PassphraseHelper , PKey , X509Name , X509 , X509Store ) <EOL> try : <EOL> _memoryview = memoryview <EOL> except NameError : <EOL> class _memoryview ( object ) : <EOL> pass <EOL> try : <EOL> _buffer = buffer <EOL> except NameError : <EOL> class _buffer ( object ) : <EOL> pass <EOL> OPENSSL_VERSION_NUMBER = _lib . OPENSSL_VERSION_NUMBER <EOL> SSLEAY_VERSION = _lib . SSLEAY_VERSION <EOL> SSLEAY_CFLAGS = _lib . SSLEAY_CFLAGS <EOL> SSLEAY_PLATFORM = _lib . SSLEAY_PLATFORM <EOL> SSLEAY_DIR = _lib . SSLEAY_DIR <EOL> SSLEAY_BUILT_ON = _lib . SSLEAY_BUILT_ON <EOL> SENT_SHUTDOWN = _lib . SSL_SENT_SHUTDOWN <EOL> RECEIVED_SHUTDOWN = _lib . SSL_RECEIVED_SHUTDOWN <EOL> SSLv2_METHOD = <NUM_LIT:1> <EOL> SSLv3_METHOD = <NUM_LIT:2> <EOL> SSLv23_METHOD = <NUM_LIT:3> <EOL> TLSv1_METHOD = <NUM_LIT:4> <EOL> TLSv1_1_METHOD = <NUM_LIT:5> <EOL> TLSv1_2_METHOD = <NUM_LIT:6> <EOL> OP_NO_SSLv2 = _lib . SSL_OP_NO_SSLv2 <EOL> OP_NO_SSLv3 = _lib . SSL_OP_NO_SSLv3 <EOL> OP_NO_TLSv1 = _lib . SSL_OP_NO_TLSv1 <EOL> OP_NO_TLSv1_1 = getattr ( _lib , "<STR_LIT>" , <NUM_LIT:0> ) <EOL> OP_NO_TLSv1_2 = getattr ( _lib , "<STR_LIT>" , <NUM_LIT:0> ) <EOL> try : <EOL> MODE_RELEASE_BUFFERS = _lib . SSL_MODE_RELEASE_BUFFERS <EOL> except AttributeError : <EOL> pass <EOL> OP_SINGLE_DH_USE = _lib . SSL_OP_SINGLE_DH_USE <EOL> OP_SINGLE_ECDH_USE = _lib . SSL_OP_SINGLE_ECDH_USE <EOL> OP_EPHEMERAL_RSA = _lib . SSL_OP_EPHEMERAL_RSA <EOL> OP_MICROSOFT_SESS_ID_BUG = _lib . SSL_OP_MICROSOFT_SESS_ID_BUG <EOL> OP_NETSCAPE_CHALLENGE_BUG = _lib . SSL_OP_NETSCAPE_CHALLENGE_BUG <EOL> OP_NETSCAPE_REUSE_CIPHER_CHANGE_BUG = ( <EOL> _lib . SSL_OP_NETSCAPE_REUSE_CIPHER_CHANGE_BUG <EOL> ) <EOL> OP_SSLREF2_REUSE_CERT_TYPE_BUG = _lib . SSL_OP_SSLREF2_REUSE_CERT_TYPE_BUG <EOL> OP_MICROSOFT_BIG_SSLV3_BUFFER = _lib . SSL_OP_MICROSOFT_BIG_SSLV3_BUFFER <EOL> try : <EOL> OP_MSIE_SSLV2_RSA_PADDING = _lib . SSL_OP_MSIE_SSLV2_RSA_PADDING <EOL> except AttributeError : <EOL> pass <EOL> OP_SSLEAY_080_CLIENT_DH_BUG = _lib . SSL_OP_SSLEAY_080_CLIENT_DH_BUG <EOL> OP_TLS_D5_BUG = _lib . SSL_OP_TLS_D5_BUG <EOL> OP_TLS_BLOCK_PADDING_BUG = _lib . SSL_OP_TLS_BLOCK_PADDING_BUG <EOL> OP_DONT_INSERT_EMPTY_FRAGMENTS = _lib . SSL_OP_DONT_INSERT_EMPTY_FRAGMENTS <EOL> OP_CIPHER_SERVER_PREFERENCE = _lib . SSL_OP_CIPHER_SERVER_PREFERENCE <EOL> OP_TLS_ROLLBACK_BUG = _lib . SSL_OP_TLS_ROLLBACK_BUG <EOL> OP_PKCS1_CHECK_1 = _lib . SSL_OP_PKCS1_CHECK_1 <EOL> OP_PKCS1_CHECK_2 = _lib . SSL_OP_PKCS1_CHECK_2 <EOL> OP_NETSCAPE_CA_DN_BUG = _lib . SSL_OP_NETSCAPE_CA_DN_BUG <EOL> OP_NETSCAPE_DEMO_CIPHER_CHANGE_BUG = ( <EOL> _lib . SSL_OP_NETSCAPE_DEMO_CIPHER_CHANGE_BUG <EOL> ) <EOL> try : <EOL> OP_NO_COMPRESSION = _lib . SSL_OP_NO_COMPRESSION <EOL> except AttributeError : <EOL> pass <EOL> OP_NO_QUERY_MTU = _lib . SSL_OP_NO_QUERY_MTU <EOL> OP_COOKIE_EXCHANGE = _lib . SSL_OP_COOKIE_EXCHANGE <EOL> try : <EOL> OP_NO_TICKET = _lib . SSL_OP_NO_TICKET <EOL> except AttributeError : <EOL> pass <EOL> OP_ALL = _lib . SSL_OP_ALL <EOL> VERIFY_PEER = _lib . SSL_VERIFY_PEER <EOL> VERIFY_FAIL_IF_NO_PEER_CERT = _lib . SSL_VERIFY_FAIL_IF_NO_PEER_CERT <EOL> VERIFY_CLIENT_ONCE = _lib . SSL_VERIFY_CLIENT_ONCE <EOL> VERIFY_NONE = _lib . SSL_VERIFY_NONE <EOL> SESS_CACHE_OFF = _lib . SSL_SESS_CACHE_OFF <EOL> SESS_CACHE_CLIENT = _lib . SSL_SESS_CACHE_CLIENT <EOL> SESS_CACHE_SERVER = _lib . SSL_SESS_CACHE_SERVER <EOL> SESS_CACHE_BOTH = _lib . SSL_SESS_CACHE_BOTH <EOL> SESS_CACHE_NO_AUTO_CLEAR = _lib . SSL_SESS_CACHE_NO_AUTO_CLEAR <EOL> SESS_CACHE_NO_INTERNAL_LOOKUP = _lib . SSL_SESS_CACHE_NO_INTERNAL_LOOKUP <EOL> SESS_CACHE_NO_INTERNAL_STORE = _lib . SSL_SESS_CACHE_NO_INTERNAL_STORE <EOL> SESS_CACHE_NO_INTERNAL = _lib . SSL_SESS_CACHE_NO_INTERNAL <EOL> SSL_ST_CONNECT = _lib . SSL_ST_CONNECT <EOL> SSL_ST_ACCEPT = _lib . SSL_ST_ACCEPT <EOL> SSL_ST_MASK = _lib . SSL_ST_MASK <EOL> SSL_ST_INIT = _lib . SSL_ST_INIT <EOL> SSL_ST_BEFORE = _lib . SSL_ST_BEFORE <EOL> SSL_ST_OK = _lib . SSL_ST_OK <EOL> SSL_ST_RENEGOTIATE = _lib . SSL_ST_RENEGOTIATE <EOL> SSL_CB_LOOP = _lib . SSL_CB_LOOP <EOL> SSL_CB_EXIT = _lib . SSL_CB_EXIT <EOL> SSL_CB_READ = _lib . SSL_CB_READ <EOL> SSL_CB_WRITE = _lib . SSL_CB_WRITE <EOL> SSL_CB_ALERT = _lib . SSL_CB_ALERT <EOL> SSL_CB_READ_ALERT = _lib . SSL_CB_READ_ALERT <EOL> SSL_CB_WRITE_ALERT = _lib . SSL_CB_WRITE_ALERT <EOL> SSL_CB_ACCEPT_LOOP = _lib . SSL_CB_ACCEPT_LOOP <EOL> SSL_CB_ACCEPT_EXIT = _lib . SSL_CB_ACCEPT_EXIT <EOL> SSL_CB_CONNECT_LOOP = _lib . SSL_CB_CONNECT_LOOP <EOL> SSL_CB_CONNECT_EXIT = _lib . SSL_CB_CONNECT_EXIT <EOL> SSL_CB_HANDSHAKE_START = _lib . SSL_CB_HANDSHAKE_START <EOL> SSL_CB_HANDSHAKE_DONE = _lib . SSL_CB_HANDSHAKE_DONE <EOL> class Error ( Exception ) : <EOL> """<STR_LIT>""" <EOL> _raise_current_error = partial ( _exception_from_error_queue , Error ) <EOL> _openssl_assert = _make_assert ( Error ) <EOL> class WantReadError ( Error ) : <EOL> pass <EOL> class WantWriteError ( Error ) : <EOL> pass <EOL> class WantX509LookupError ( Error ) : <EOL> pass <EOL> class ZeroReturnError ( Error ) : <EOL> pass <EOL> class SysCallError ( Error ) : <EOL> pass <EOL> class _CallbackExceptionHelper ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _problems = [ ] <EOL> def raise_if_problem ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _problems : <EOL> try : <EOL> _raise_current_error ( ) <EOL> except Error : <EOL> pass <EOL> raise self . _problems . pop ( <NUM_LIT:0> ) <EOL> class _VerifyHelper ( _CallbackExceptionHelper ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , callback ) : <EOL> _CallbackExceptionHelper . __init__ ( self ) <EOL> @ wraps ( callback ) <EOL> def wrapper ( ok , store_ctx ) : <EOL> cert = X509 . __new__ ( X509 ) <EOL> cert . _x509 = _lib . X509_STORE_CTX_get_current_cert ( store_ctx ) <EOL> error_number = _lib . X509_STORE_CTX_get_error ( store_ctx ) <EOL> error_depth = _lib . X509_STORE_CTX_get_error_depth ( store_ctx ) <EOL> index = _lib . SSL_get_ex_data_X509_STORE_CTX_idx ( ) <EOL> ssl = _lib . X509_STORE_CTX_get_ex_data ( store_ctx , index ) <EOL> connection = Connection . _reverse_mapping [ ssl ] <EOL> try : <EOL> result = callback ( <EOL> connection , cert , error_number , error_depth , ok <EOL> ) <EOL> except Exception as e : <EOL> self . _problems . append ( e ) <EOL> return <NUM_LIT:0> <EOL> else : <EOL> if result : <EOL> _lib . X509_STORE_CTX_set_error ( store_ctx , _lib . X509_V_OK ) <EOL> return <NUM_LIT:1> <EOL> else : <EOL> return <NUM_LIT:0> <EOL> self . callback = _ffi . callback ( <EOL> "<STR_LIT>" , wrapper ) <EOL> class _NpnAdvertiseHelper ( _CallbackExceptionHelper ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , callback ) : <EOL> _CallbackExceptionHelper . __init__ ( self ) <EOL> @ wraps ( callback ) <EOL> def wrapper ( ssl , out , outlen , arg ) : <EOL> try : <EOL> conn = Connection . _reverse_mapping [ ssl ] <EOL> protos = callback ( conn ) <EOL> protostr = b'<STR_LIT>' . join ( <EOL> chain . from_iterable ( ( int2byte ( len ( p ) ) , p ) for p in protos ) <EOL> ) <EOL> conn . _npn_advertise_callback_args = [ <EOL> _ffi . new ( "<STR_LIT>" , len ( protostr ) ) , <EOL> _ffi . new ( "<STR_LIT>" , protostr ) , <EOL> ] <EOL> outlen [ <NUM_LIT:0> ] = conn . _npn_advertise_callback_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> out [ <NUM_LIT:0> ] = conn . _npn_advertise_callback_args [ <NUM_LIT:1> ] <EOL> return <NUM_LIT:0> <EOL> except Exception as e : <EOL> self . _problems . append ( e ) <EOL> return <NUM_LIT:2> <EOL> self . callback = _ffi . callback ( <EOL> "<STR_LIT>" , <EOL> wrapper <EOL> ) <EOL> class _NpnSelectHelper ( _CallbackExceptionHelper ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , callback ) : <EOL> _CallbackExceptionHelper . __init__ ( self ) <EOL> @ wraps ( callback ) <EOL> def wrapper ( ssl , out , outlen , in_ , inlen , arg ) : <EOL> try : <EOL> conn = Connection . _reverse_mapping [ ssl ] <EOL> instr = _ffi . buffer ( in_ , inlen ) [ : ] <EOL> protolist = [ ] <EOL> while instr : <EOL> l = indexbytes ( instr , <NUM_LIT:0> ) <EOL> proto = instr [ <NUM_LIT:1> : l + <NUM_LIT:1> ] <EOL> protolist . append ( proto ) <EOL> instr = instr [ l + <NUM_LIT:1> : ] <EOL> outstr = callback ( conn , protolist ) <EOL> conn . _npn_select_callback_args = [ <EOL> _ffi . new ( "<STR_LIT>" , len ( outstr ) ) , <EOL> _ffi . new ( "<STR_LIT>" , outstr ) , <EOL> ] <EOL> outlen [ <NUM_LIT:0> ] = conn . _npn_select_callback_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> out [ <NUM_LIT:0> ] = conn . _npn_select_callback_args [ <NUM_LIT:1> ] <EOL> return <NUM_LIT:0> <EOL> except Exception as e : <EOL> self . _problems . append ( e ) <EOL> return <NUM_LIT:2> <EOL> self . callback = _ffi . callback ( <EOL> ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> wrapper <EOL> ) <EOL> class _ALPNSelectHelper ( _CallbackExceptionHelper ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , callback ) : <EOL> _CallbackExceptionHelper . __init__ ( self ) <EOL> @ wraps ( callback ) <EOL> def wrapper ( ssl , out , outlen , in_ , inlen , arg ) : <EOL> try : <EOL> conn = Connection . _reverse_mapping [ ssl ] <EOL> instr = _ffi . buffer ( in_ , inlen ) [ : ] <EOL> protolist = [ ] <EOL> while instr : <EOL> encoded_len = indexbytes ( instr , <NUM_LIT:0> ) <EOL> proto = instr [ <NUM_LIT:1> : encoded_len + <NUM_LIT:1> ] <EOL> protolist . append ( proto ) <EOL> instr = instr [ encoded_len + <NUM_LIT:1> : ] <EOL> outstr = callback ( conn , protolist ) <EOL> if not isinstance ( outstr , _binary_type ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> conn . _alpn_select_callback_args = [ <EOL> _ffi . new ( "<STR_LIT>" , len ( outstr ) ) , <EOL> _ffi . new ( "<STR_LIT>" , outstr ) , <EOL> ] <EOL> outlen [ <NUM_LIT:0> ] = conn . _alpn_select_callback_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> out [ <NUM_LIT:0> ] = conn . _alpn_select_callback_args [ <NUM_LIT:1> ] <EOL> return <NUM_LIT:0> <EOL> except Exception as e : <EOL> self . _problems . append ( e ) <EOL> return <NUM_LIT:2> <EOL> self . callback = _ffi . callback ( <EOL> ( "<STR_LIT>" <EOL> "<STR_LIT>" ) , <EOL> wrapper <EOL> ) <EOL> def _asFileDescriptor ( obj ) : <EOL> fd = None <EOL> if not isinstance ( obj , integer_types ) : <EOL> meth = getattr ( obj , "<STR_LIT>" , None ) <EOL> if meth is not None : <EOL> obj = meth ( ) <EOL> if isinstance ( obj , integer_types ) : <EOL> fd = obj <EOL> if not isinstance ( fd , integer_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> elif fd < <NUM_LIT:0> : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % ( fd , ) ) <EOL> return fd <EOL> def SSLeay_version ( type ) : <EOL> """<STR_LIT>""" <EOL> return _ffi . string ( _lib . SSLeay_version ( type ) ) <EOL> def _make_requires ( flag , error ) : <EOL> """<STR_LIT>""" <EOL> def _requires_decorator ( func ) : <EOL> if not flag : <EOL> @ wraps ( func ) <EOL> def explode ( * args , ** kwargs ) : <EOL> raise NotImplementedError ( error ) <EOL> return explode <EOL> else : <EOL> return func <EOL> return _requires_decorator <EOL> _requires_npn = _make_requires ( <EOL> _lib . Cryptography_HAS_NEXTPROTONEG , "<STR_LIT>" <EOL> ) <EOL> _requires_alpn = _make_requires ( <EOL> _lib . Cryptography_HAS_ALPN , "<STR_LIT>" <EOL> ) <EOL> _requires_sni = _make_requires ( <EOL> _lib . Cryptography_HAS_TLSEXT_HOSTNAME , "<STR_LIT>" <EOL> ) <EOL> class Session ( object ) : <EOL> pass <EOL> class Context ( object ) : <EOL> """<STR_LIT>""" <EOL> _methods = { <EOL> SSLv2_METHOD : "<STR_LIT>" , <EOL> SSLv3_METHOD : "<STR_LIT>" , <EOL> SSLv23_METHOD : "<STR_LIT>" , <EOL> TLSv1_METHOD : "<STR_LIT>" , <EOL> TLSv1_1_METHOD : "<STR_LIT>" , <EOL> TLSv1_2_METHOD : "<STR_LIT>" , <EOL> } <EOL> _methods = dict ( <EOL> ( identifier , getattr ( _lib , name ) ) <EOL> for ( identifier , name ) in _methods . items ( ) <EOL> if getattr ( _lib , name , None ) is not None ) <EOL> def __init__ ( self , method ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( method , integer_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> try : <EOL> method_func = self . _methods [ method ] <EOL> except KeyError : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> method_obj = method_func ( ) <EOL> if method_obj == _ffi . NULL : <EOL> _raise_current_error ( ) <EOL> context = _lib . SSL_CTX_new ( method_obj ) <EOL> if context == _ffi . NULL : <EOL> _raise_current_error ( ) <EOL> context = _ffi . gc ( context , _lib . SSL_CTX_free ) <EOL> self . _context = context <EOL> self . _passphrase_helper = None <EOL> self . _passphrase_callback = None <EOL> self . _passphrase_userdata = None <EOL> self . _verify_helper = None <EOL> self . _verify_callback = None <EOL> self . _info_callback = None <EOL> self . _tlsext_servername_callback = None <EOL> self . _app_data = None <EOL> self . _npn_advertise_helper = None <EOL> self . _npn_advertise_callback = None <EOL> self . _npn_select_helper = None <EOL> self . _npn_select_callback = None <EOL> self . _alpn_select_helper = None <EOL> self . _alpn_select_callback = None <EOL> self . set_mode ( _lib . SSL_MODE_ENABLE_PARTIAL_WRITE ) <EOL> def load_verify_locations ( self , cafile , capath = None ) : <EOL> """<STR_LIT>""" <EOL> if cafile is None : <EOL> cafile = _ffi . NULL <EOL> else : <EOL> cafile = _path_string ( cafile ) <EOL> if capath is None : <EOL> capath = _ffi . NULL <EOL> else : <EOL> capath = _path_string ( capath ) <EOL> load_result = _lib . SSL_CTX_load_verify_locations ( <EOL> self . _context , cafile , capath <EOL> ) <EOL> if not load_result : <EOL> _raise_current_error ( ) <EOL> def _wrap_callback ( self , callback ) : <EOL> @ wraps ( callback ) <EOL> def wrapper ( size , verify , userdata ) : <EOL> return callback ( size , verify , self . _passphrase_userdata ) <EOL> return _PassphraseHelper ( <EOL> FILETYPE_PEM , wrapper , more_args = True , truncate = True ) <EOL> def set_passwd_cb ( self , callback , userdata = None ) : <EOL> """<STR_LIT>""" <EOL> if not callable ( callback ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . _passphrase_helper = self . _wrap_callback ( callback ) <EOL> self . _passphrase_callback = self . _passphrase_helper . callback <EOL> _lib . SSL_CTX_set_default_passwd_cb ( <EOL> self . _context , self . _passphrase_callback ) <EOL> self . _passphrase_userdata = userdata <EOL> def set_default_verify_paths ( self ) : <EOL> """<STR_LIT>""" <EOL> set_result = _lib . SSL_CTX_set_default_verify_paths ( self . _context ) <EOL> if not set_result : <EOL> _raise_current_error ( ) <EOL> def use_certificate_chain_file ( self , certfile ) : <EOL> """<STR_LIT>""" <EOL> certfile = _path_string ( certfile ) <EOL> result = _lib . SSL_CTX_use_certificate_chain_file ( <EOL> self . _context , certfile <EOL> ) <EOL> if not result : <EOL> _raise_current_error ( ) <EOL> def use_certificate_file ( self , certfile , filetype = FILETYPE_PEM ) : <EOL> """<STR_LIT>""" <EOL> certfile = _path_string ( certfile ) <EOL> if not isinstance ( filetype , integer_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> use_result = _lib . SSL_CTX_use_certificate_file ( <EOL> self . _context , certfile , filetype <EOL> ) <EOL> if not use_result : <EOL> _raise_current_error ( ) <EOL> def use_certificate ( self , cert ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( cert , X509 ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> use_result = _lib . SSL_CTX_use_certificate ( self . _context , cert . _x509 ) <EOL> if not use_result : <EOL> _raise_current_error ( ) <EOL> def add_extra_chain_cert ( self , certobj ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( certobj , X509 ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> copy = _lib . X509_dup ( certobj . _x509 ) <EOL> add_result = _lib . SSL_CTX_add_extra_chain_cert ( self . _context , copy ) <EOL> if not add_result : <EOL> _lib . X509_free ( copy ) <EOL> _raise_current_error ( ) <EOL> def _raise_passphrase_exception ( self ) : <EOL> if self . _passphrase_helper is None : <EOL> _raise_current_error ( ) <EOL> exception = self . _passphrase_helper . raise_if_problem ( Error ) <EOL> if exception is not None : <EOL> raise exception <EOL> def use_privatekey_file ( self , keyfile , filetype = _UNSPECIFIED ) : <EOL> """<STR_LIT>""" <EOL> keyfile = _path_string ( keyfile ) <EOL> if filetype is _UNSPECIFIED : <EOL> filetype = FILETYPE_PEM <EOL> elif not isinstance ( filetype , integer_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> use_result = _lib . SSL_CTX_use_PrivateKey_file ( <EOL> self . _context , keyfile , filetype ) <EOL> if not use_result : <EOL> self . _raise_passphrase_exception ( ) <EOL> def use_privatekey ( self , pkey ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( pkey , PKey ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> use_result = _lib . SSL_CTX_use_PrivateKey ( self . _context , pkey . _pkey ) <EOL> if not use_result : <EOL> self . _raise_passphrase_exception ( ) <EOL> def check_privatekey ( self ) : <EOL> """<STR_LIT>""" <EOL> if not _lib . SSL_CTX_check_private_key ( self . _context ) : <EOL> _raise_current_error ( ) <EOL> def load_client_ca ( self , cafile ) : <EOL> """<STR_LIT>""" <EOL> ca_list = _lib . SSL_load_client_CA_file ( <EOL> _text_to_bytes_and_warn ( "<STR_LIT>" , cafile ) <EOL> ) <EOL> _openssl_assert ( ca_list != _ffi . NULL ) <EOL> _lib . SSL_CTX_set_client_CA_list ( self . _context , ca_list ) <EOL> def set_session_id ( self , buf ) : <EOL> """<STR_LIT>""" <EOL> buf = _text_to_bytes_and_warn ( "<STR_LIT>" , buf ) <EOL> _openssl_assert ( <EOL> _lib . SSL_CTX_set_session_id_context ( <EOL> self . _context , <EOL> buf , <EOL> len ( buf ) , <EOL> ) == <NUM_LIT:1> <EOL> ) <EOL> def set_session_cache_mode ( self , mode ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( mode , integer_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> return _lib . SSL_CTX_set_session_cache_mode ( self . _context , mode ) <EOL> def get_session_cache_mode ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . SSL_CTX_get_session_cache_mode ( self . _context ) <EOL> def set_verify ( self , mode , callback ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( mode , integer_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if not callable ( callback ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> self . _verify_helper = _VerifyHelper ( callback ) <EOL> self . _verify_callback = self . _verify_helper . callback <EOL> _lib . SSL_CTX_set_verify ( self . _context , mode , self . _verify_callback ) <EOL> def set_verify_depth ( self , depth ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( depth , integer_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> _lib . SSL_CTX_set_verify_depth ( self . _context , depth ) <EOL> def get_verify_mode ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . SSL_CTX_get_verify_mode ( self . _context ) <EOL> def get_verify_depth ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . SSL_CTX_get_verify_depth ( self . _context ) <EOL> def load_tmp_dh ( self , dhfile ) : <EOL> """<STR_LIT>""" <EOL> dhfile = _path_string ( dhfile ) <EOL> bio = _lib . BIO_new_file ( dhfile , b"<STR_LIT:r>" ) <EOL> if bio == _ffi . NULL : <EOL> _raise_current_error ( ) <EOL> bio = _ffi . gc ( bio , _lib . BIO_free ) <EOL> dh = _lib . PEM_read_bio_DHparams ( bio , _ffi . NULL , _ffi . NULL , _ffi . NULL ) <EOL> dh = _ffi . gc ( dh , _lib . DH_free ) <EOL> _lib . SSL_CTX_set_tmp_dh ( self . _context , dh ) <EOL> def set_tmp_ecdh ( self , curve ) : <EOL> """<STR_LIT>""" <EOL> _lib . SSL_CTX_set_tmp_ecdh ( self . _context , curve . _to_EC_KEY ( ) ) <EOL> def set_cipher_list ( self , cipher_list ) : <EOL> """<STR_LIT>""" <EOL> cipher_list = _text_to_bytes_and_warn ( "<STR_LIT>" , cipher_list ) <EOL> if not isinstance ( cipher_list , bytes ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> _openssl_assert ( <EOL> _lib . SSL_CTX_set_cipher_list ( self . _context , cipher_list ) == <NUM_LIT:1> <EOL> ) <EOL> def set_client_ca_list ( self , certificate_authorities ) : <EOL> """<STR_LIT>""" <EOL> name_stack = _lib . sk_X509_NAME_new_null ( ) <EOL> if name_stack == _ffi . NULL : <EOL> _raise_current_error ( ) <EOL> try : <EOL> for ca_name in certificate_authorities : <EOL> if not isinstance ( ca_name , X509Name ) : <EOL> raise TypeError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( <EOL> type ( ca_name ) . __name__ , <EOL> ) <EOL> ) <EOL> copy = _lib . X509_NAME_dup ( ca_name . _name ) <EOL> if copy == _ffi . NULL : <EOL> _raise_current_error ( ) <EOL> push_result = _lib . sk_X509_NAME_push ( name_stack , copy ) <EOL> if not push_result : <EOL> _lib . X509_NAME_free ( copy ) <EOL> _raise_current_error ( ) <EOL> except : <EOL> _lib . sk_X509_NAME_free ( name_stack ) <EOL> raise <EOL> _lib . SSL_CTX_set_client_CA_list ( self . _context , name_stack ) <EOL> def add_client_ca ( self , certificate_authority ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( certificate_authority , X509 ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> add_result = _lib . SSL_CTX_add_client_CA ( <EOL> self . _context , certificate_authority . _x509 ) <EOL> if not add_result : <EOL> _raise_current_error ( ) <EOL> def set_timeout ( self , timeout ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( timeout , integer_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> return _lib . SSL_CTX_set_timeout ( self . _context , timeout ) <EOL> def get_timeout ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . SSL_CTX_get_timeout ( self . _context ) <EOL> def set_info_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ wraps ( callback ) <EOL> def wrapper ( ssl , where , return_code ) : <EOL> callback ( Connection . _reverse_mapping [ ssl ] , where , return_code ) <EOL> self . _info_callback = _ffi . callback ( <EOL> "<STR_LIT>" , wrapper ) <EOL> _lib . SSL_CTX_set_info_callback ( self . _context , self . _info_callback ) <EOL> def get_app_data ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _app_data <EOL> def set_app_data ( self , data ) : <EOL> """<STR_LIT>""" <EOL> self . _app_data = data <EOL> def get_cert_store ( self ) : <EOL> """<STR_LIT>""" <EOL> store = _lib . SSL_CTX_get_cert_store ( self . _context ) <EOL> if store == _ffi . NULL : <EOL> return None <EOL> pystore = X509Store . __new__ ( X509Store ) <EOL> pystore . _store = store <EOL> return pystore <EOL> def set_options ( self , options ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( options , integer_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> return _lib . SSL_CTX_set_options ( self . _context , options ) <EOL> def set_mode ( self , mode ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( mode , integer_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> return _lib . SSL_CTX_set_mode ( self . _context , mode ) <EOL> @ _requires_sni <EOL> def set_tlsext_servername_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> @ wraps ( callback ) <EOL> def wrapper ( ssl , alert , arg ) : <EOL> callback ( Connection . _reverse_mapping [ ssl ] ) <EOL> return <NUM_LIT:0> <EOL> self . _tlsext_servername_callback = _ffi . callback ( <EOL> "<STR_LIT>" , wrapper ) <EOL> _lib . SSL_CTX_set_tlsext_servername_callback ( <EOL> self . _context , self . _tlsext_servername_callback ) <EOL> @ _requires_npn <EOL> def set_npn_advertise_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> self . _npn_advertise_helper = _NpnAdvertiseHelper ( callback ) <EOL> self . _npn_advertise_callback = self . _npn_advertise_helper . callback <EOL> _lib . SSL_CTX_set_next_protos_advertised_cb ( <EOL> self . _context , self . _npn_advertise_callback , _ffi . NULL ) <EOL> @ _requires_npn <EOL> def set_npn_select_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> self . _npn_select_helper = _NpnSelectHelper ( callback ) <EOL> self . _npn_select_callback = self . _npn_select_helper . callback <EOL> _lib . SSL_CTX_set_next_proto_select_cb ( <EOL> self . _context , self . _npn_select_callback , _ffi . NULL ) <EOL> @ _requires_alpn <EOL> def set_alpn_protos ( self , protos ) : <EOL> """<STR_LIT>""" <EOL> protostr = b'<STR_LIT>' . join ( <EOL> chain . from_iterable ( ( int2byte ( len ( p ) ) , p ) for p in protos ) <EOL> ) <EOL> input_str = _ffi . new ( "<STR_LIT>" , protostr ) <EOL> input_str_len = _ffi . cast ( "<STR_LIT>" , len ( protostr ) ) <EOL> _lib . SSL_CTX_set_alpn_protos ( self . _context , input_str , input_str_len ) <EOL> @ _requires_alpn <EOL> def set_alpn_select_callback ( self , callback ) : <EOL> """<STR_LIT>""" <EOL> self . _alpn_select_helper = _ALPNSelectHelper ( callback ) <EOL> self . _alpn_select_callback = self . _alpn_select_helper . callback <EOL> _lib . SSL_CTX_set_alpn_select_cb ( <EOL> self . _context , self . _alpn_select_callback , _ffi . NULL ) <EOL> ContextType = Context <EOL> class Connection ( object ) : <EOL> """<STR_LIT:U+0020>""" <EOL> _reverse_mapping = WeakValueDictionary ( ) <EOL> def __init__ ( self , context , socket = None ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( context , Context ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> ssl = _lib . SSL_new ( context . _context ) <EOL> self . _ssl = _ffi . gc ( ssl , _lib . SSL_free ) <EOL> self . _context = context <EOL> self . _app_data = None <EOL> self . _npn_advertise_callback_args = None <EOL> self . _npn_select_callback_args = None <EOL> self . _alpn_select_callback_args = None <EOL> self . _reverse_mapping [ self . _ssl ] = self <EOL> if socket is None : <EOL> self . _socket = None <EOL> self . _into_ssl = _lib . BIO_new ( _lib . BIO_s_mem ( ) ) <EOL> self . _from_ssl = _lib . BIO_new ( _lib . BIO_s_mem ( ) ) <EOL> if self . _into_ssl == _ffi . NULL or self . _from_ssl == _ffi . NULL : <EOL> _raise_current_error ( ) <EOL> _lib . SSL_set_bio ( self . _ssl , self . _into_ssl , self . _from_ssl ) <EOL> else : <EOL> self . _into_ssl = None <EOL> self . _from_ssl = None <EOL> self . _socket = socket <EOL> set_result = _lib . SSL_set_fd ( <EOL> self . _ssl , _asFileDescriptor ( self . _socket ) ) <EOL> if not set_result : <EOL> _raise_current_error ( ) <EOL> def __getattr__ ( self , name ) : <EOL> """<STR_LIT>""" <EOL> if self . _socket is None : <EOL> raise AttributeError ( "<STR_LIT>" % ( <EOL> self . __class__ . __name__ , name <EOL> ) ) <EOL> else : <EOL> return getattr ( self . _socket , name ) <EOL> def _raise_ssl_error ( self , ssl , result ) : <EOL> if self . _context . _verify_helper is not None : <EOL> self . _context . _verify_helper . raise_if_problem ( ) <EOL> if self . _context . _npn_advertise_helper is not None : <EOL> self . _context . _npn_advertise_helper . raise_if_problem ( ) <EOL> if self . _context . _npn_select_helper is not None : <EOL> self . _context . _npn_select_helper . raise_if_problem ( ) <EOL> if self . _context . _alpn_select_helper is not None : <EOL> self . _context . _alpn_select_helper . raise_if_problem ( ) <EOL> error = _lib . SSL_get_error ( ssl , result ) <EOL> if error == _lib . SSL_ERROR_WANT_READ : <EOL> raise WantReadError ( ) <EOL> elif error == _lib . SSL_ERROR_WANT_WRITE : <EOL> raise WantWriteError ( ) <EOL> elif error == _lib . SSL_ERROR_ZERO_RETURN : <EOL> raise ZeroReturnError ( ) <EOL> elif error == _lib . SSL_ERROR_WANT_X509_LOOKUP : <EOL> raise WantX509LookupError ( ) <EOL> elif error == _lib . SSL_ERROR_SYSCALL : <EOL> if _lib . ERR_peek_error ( ) == <NUM_LIT:0> : <EOL> if result < <NUM_LIT:0> : <EOL> if platform == "<STR_LIT:win32>" : <EOL> errno = _ffi . getwinerror ( ) [ <NUM_LIT:0> ] <EOL> else : <EOL> errno = _ffi . errno <EOL> raise SysCallError ( errno , errorcode . get ( errno ) ) <EOL> else : <EOL> raise SysCallError ( - <NUM_LIT:1> , "<STR_LIT>" ) <EOL> else : <EOL> _raise_current_error ( ) <EOL> elif error == _lib . SSL_ERROR_NONE : <EOL> pass <EOL> else : <EOL> _raise_current_error ( ) <EOL> def get_context ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _context <EOL> def set_context ( self , context ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( context , Context ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> _lib . SSL_set_SSL_CTX ( self . _ssl , context . _context ) <EOL> self . _context = context <EOL> @ _requires_sni <EOL> def get_servername ( self ) : <EOL> """<STR_LIT>""" <EOL> name = _lib . SSL_get_servername ( <EOL> self . _ssl , _lib . TLSEXT_NAMETYPE_host_name <EOL> ) <EOL> if name == _ffi . NULL : <EOL> return None <EOL> return _ffi . string ( name ) <EOL> @ _requires_sni <EOL> def set_tlsext_host_name ( self , name ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( name , bytes ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> elif b"<STR_LIT>" in name : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> _lib . SSL_set_tlsext_host_name ( self . _ssl , name ) <EOL> def pending ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . SSL_pending ( self . _ssl ) <EOL> def send ( self , buf , flags = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> buf = _text_to_bytes_and_warn ( "<STR_LIT>" , buf ) <EOL> if isinstance ( buf , _memoryview ) : <EOL> buf = buf . tobytes ( ) <EOL> if isinstance ( buf , _buffer ) : <EOL> buf = str ( buf ) <EOL> if not isinstance ( buf , bytes ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> result = _lib . SSL_write ( self . _ssl , buf , len ( buf ) ) <EOL> self . _raise_ssl_error ( self . _ssl , result ) <EOL> return result <EOL> write = send <EOL> def sendall ( self , buf , flags = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> buf = _text_to_bytes_and_warn ( "<STR_LIT>" , buf ) <EOL> if isinstance ( buf , _memoryview ) : <EOL> buf = buf . tobytes ( ) <EOL> if isinstance ( buf , _buffer ) : <EOL> buf = str ( buf ) <EOL> if not isinstance ( buf , bytes ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> left_to_send = len ( buf ) <EOL> total_sent = <NUM_LIT:0> <EOL> data = _ffi . new ( "<STR_LIT>" , buf ) <EOL> while left_to_send : <EOL> result = _lib . SSL_write ( self . _ssl , data + total_sent , left_to_send ) <EOL> self . _raise_ssl_error ( self . _ssl , result ) <EOL> total_sent += result <EOL> left_to_send -= result <EOL> def recv ( self , bufsiz , flags = None ) : <EOL> """<STR_LIT>""" <EOL> buf = _ffi . new ( "<STR_LIT>" , bufsiz ) <EOL> if flags is not None and flags & socket . MSG_PEEK : <EOL> result = _lib . SSL_peek ( self . _ssl , buf , bufsiz ) <EOL> else : <EOL> result = _lib . SSL_read ( self . _ssl , buf , bufsiz ) <EOL> self . _raise_ssl_error ( self . _ssl , result ) <EOL> return _ffi . buffer ( buf , result ) [ : ] <EOL> read = recv <EOL> def recv_into ( self , buffer , nbytes = None , flags = None ) : <EOL> """<STR_LIT>""" <EOL> if nbytes is None : <EOL> nbytes = len ( buffer ) <EOL> else : <EOL> nbytes = min ( nbytes , len ( buffer ) ) <EOL> buf = _ffi . new ( "<STR_LIT>" , nbytes ) <EOL> if flags is not None and flags & socket . MSG_PEEK : <EOL> result = _lib . SSL_peek ( self . _ssl , buf , nbytes ) <EOL> else : <EOL> result = _lib . SSL_read ( self . _ssl , buf , nbytes ) <EOL> self . _raise_ssl_error ( self . _ssl , result ) <EOL> try : <EOL> buffer [ : result ] = memoryview ( _ffi . buffer ( buf , result ) ) <EOL> except NameError : <EOL> buffer [ : result ] = _ffi . buffer ( buf , result ) <EOL> return result <EOL> def _handle_bio_errors ( self , bio , result ) : <EOL> if _lib . BIO_should_retry ( bio ) : <EOL> if _lib . BIO_should_read ( bio ) : <EOL> raise WantReadError ( ) <EOL> elif _lib . BIO_should_write ( bio ) : <EOL> raise WantWriteError ( ) <EOL> elif _lib . BIO_should_io_special ( bio ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> else : <EOL> _raise_current_error ( ) <EOL> def bio_read ( self , bufsiz ) : <EOL> """<STR_LIT>""" <EOL> if self . _from_ssl is None : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> if not isinstance ( bufsiz , integer_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> buf = _ffi . new ( "<STR_LIT>" , bufsiz ) <EOL> result = _lib . BIO_read ( self . _from_ssl , buf , bufsiz ) <EOL> if result <= <NUM_LIT:0> : <EOL> self . _handle_bio_errors ( self . _from_ssl , result ) <EOL> return _ffi . buffer ( buf , result ) [ : ] <EOL> def bio_write ( self , buf ) : <EOL> """<STR_LIT>""" <EOL> buf = _text_to_bytes_and_warn ( "<STR_LIT>" , buf ) <EOL> if self . _into_ssl is None : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> result = _lib . BIO_write ( self . _into_ssl , buf , len ( buf ) ) <EOL> if result <= <NUM_LIT:0> : <EOL> self . _handle_bio_errors ( self . _into_ssl , result ) <EOL> return result <EOL> def renegotiate ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . renegotiate_pending ( ) : <EOL> _openssl_assert ( _lib . SSL_renegotiate ( self . _ssl ) == <NUM_LIT:1> ) <EOL> return True <EOL> return False <EOL> def do_handshake ( self ) : <EOL> """<STR_LIT>""" <EOL> result = _lib . SSL_do_handshake ( self . _ssl ) <EOL> self . _raise_ssl_error ( self . _ssl , result ) <EOL> def renegotiate_pending ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . SSL_renegotiate_pending ( self . _ssl ) == <NUM_LIT:1> <EOL> def total_renegotiations ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . SSL_total_renegotiations ( self . _ssl ) <EOL> def connect ( self , addr ) : <EOL> """<STR_LIT>""" <EOL> _lib . SSL_set_connect_state ( self . _ssl ) <EOL> return self . _socket . connect ( addr ) <EOL> def connect_ex ( self , addr ) : <EOL> """<STR_LIT>""" <EOL> connect_ex = self . _socket . connect_ex <EOL> self . set_connect_state ( ) <EOL> return connect_ex ( addr ) <EOL> def accept ( self ) : <EOL> """<STR_LIT>""" <EOL> client , addr = self . _socket . accept ( ) <EOL> conn = Connection ( self . _context , client ) <EOL> conn . set_accept_state ( ) <EOL> return ( conn , addr ) <EOL> def bio_shutdown ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _from_ssl is None : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> _lib . BIO_set_mem_eof_return ( self . _into_ssl , <NUM_LIT:0> ) <EOL> def shutdown ( self ) : <EOL> """<STR_LIT>""" <EOL> result = _lib . SSL_shutdown ( self . _ssl ) <EOL> if result < <NUM_LIT:0> : <EOL> self . _raise_ssl_error ( self . _ssl , result ) <EOL> elif result > <NUM_LIT:0> : <EOL> return True <EOL> else : <EOL> return False <EOL> def get_cipher_list ( self ) : <EOL> """<STR_LIT>""" <EOL> ciphers = [ ] <EOL> for i in count ( ) : <EOL> result = _lib . SSL_get_cipher_list ( self . _ssl , i ) <EOL> if result == _ffi . NULL : <EOL> break <EOL> ciphers . append ( _native ( _ffi . string ( result ) ) ) <EOL> return ciphers <EOL> def get_client_ca_list ( self ) : <EOL> """<STR_LIT>""" <EOL> ca_names = _lib . SSL_get_client_CA_list ( self . _ssl ) <EOL> if ca_names == _ffi . NULL : <EOL> return [ ] <EOL> result = [ ] <EOL> for i in range ( _lib . sk_X509_NAME_num ( ca_names ) ) : <EOL> name = _lib . sk_X509_NAME_value ( ca_names , i ) <EOL> copy = _lib . X509_NAME_dup ( name ) <EOL> if copy == _ffi . NULL : <EOL> _raise_current_error ( ) <EOL> pyname = X509Name . __new__ ( X509Name ) <EOL> pyname . _name = _ffi . gc ( copy , _lib . X509_NAME_free ) <EOL> result . append ( pyname ) <EOL> return result <EOL> def makefile ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" ) <EOL> def get_app_data ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _app_data <EOL> def set_app_data ( self , data ) : <EOL> """<STR_LIT>""" <EOL> self . _app_data = data <EOL> def get_shutdown ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . SSL_get_shutdown ( self . _ssl ) <EOL> def set_shutdown ( self , state ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( state , integer_types ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> _lib . SSL_set_shutdown ( self . _ssl , state ) <EOL> def get_state_string ( self ) : <EOL> """<STR_LIT>""" <EOL> return _ffi . string ( _lib . SSL_state_string_long ( self . _ssl ) ) <EOL> def server_random ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _ssl . session == _ffi . NULL : <EOL> return None <EOL> return _ffi . buffer ( <EOL> self . _ssl . s3 . server_random , <EOL> _lib . SSL3_RANDOM_SIZE ) [ : ] <EOL> def client_random ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _ssl . session == _ffi . NULL : <EOL> return None <EOL> return _ffi . buffer ( <EOL> self . _ssl . s3 . client_random , <EOL> _lib . SSL3_RANDOM_SIZE ) [ : ] <EOL> def master_key ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _ssl . session == _ffi . NULL : <EOL> return None <EOL> return _ffi . buffer ( <EOL> self . _ssl . session . master_key , <EOL> self . _ssl . session . master_key_length ) [ : ] <EOL> def sock_shutdown ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . _socket . shutdown ( * args , ** kwargs ) <EOL> def get_peer_certificate ( self ) : <EOL> """<STR_LIT>""" <EOL> cert = _lib . SSL_get_peer_certificate ( self . _ssl ) <EOL> if cert != _ffi . NULL : <EOL> pycert = X509 . __new__ ( X509 ) <EOL> pycert . _x509 = _ffi . gc ( cert , _lib . X509_free ) <EOL> return pycert <EOL> return None <EOL> def get_peer_cert_chain ( self ) : <EOL> """<STR_LIT>""" <EOL> cert_stack = _lib . SSL_get_peer_cert_chain ( self . _ssl ) <EOL> if cert_stack == _ffi . NULL : <EOL> return None <EOL> result = [ ] <EOL> for i in range ( _lib . sk_X509_num ( cert_stack ) ) : <EOL> cert = _lib . X509_dup ( _lib . sk_X509_value ( cert_stack , i ) ) <EOL> pycert = X509 . __new__ ( X509 ) <EOL> pycert . _x509 = _ffi . gc ( cert , _lib . X509_free ) <EOL> result . append ( pycert ) <EOL> return result <EOL> def want_read ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . SSL_want_read ( self . _ssl ) <EOL> def want_write ( self ) : <EOL> """<STR_LIT>""" <EOL> return _lib . SSL_want_write ( self . _ssl ) <EOL> def set_accept_state ( self ) : <EOL> """<STR_LIT>""" <EOL> _lib . SSL_set_accept_state ( self . _ssl ) <EOL> def set_connect_state ( self ) : <EOL> """<STR_LIT>""" <EOL> _lib . SSL_set_connect_state ( self . _ssl ) <EOL> def get_session ( self ) : <EOL> """<STR_LIT>""" <EOL> session = _lib . SSL_get1_session ( self . _ssl ) <EOL> if session == _ffi . NULL : <EOL> return None <EOL> pysession = Session . __new__ ( Session ) <EOL> pysession . _session = _ffi . gc ( session , _lib . SSL_SESSION_free ) <EOL> return pysession <EOL> def set_session ( self , session ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( session , Session ) : <EOL> raise TypeError ( "<STR_LIT>" ) <EOL> result = _lib . SSL_set_session ( self . _ssl , session . _session ) <EOL> if not result : <EOL> _raise_current_error ( ) <EOL> def _get_finished_message ( self , function ) : <EOL> """<STR_LIT>""" <EOL> empty = _ffi . new ( "<STR_LIT>" , <NUM_LIT:0> ) <EOL> size = function ( self . _ssl , empty , <NUM_LIT:0> ) <EOL> if size == <NUM_LIT:0> : <EOL> return None <EOL> buf = _ffi . new ( "<STR_LIT>" , size ) <EOL> function ( self . _ssl , buf , size ) <EOL> return _ffi . buffer ( buf , size ) [ : ] <EOL> def get_finished ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_finished_message ( _lib . SSL_get_finished ) <EOL> def get_peer_finished ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_finished_message ( _lib . SSL_get_peer_finished ) <EOL> def get_cipher_name ( self ) : <EOL> """<STR_LIT>""" <EOL> cipher = _lib . SSL_get_current_cipher ( self . _ssl ) <EOL> if cipher == _ffi . NULL : <EOL> return None <EOL> else : <EOL> name = _ffi . string ( _lib . SSL_CIPHER_get_name ( cipher ) ) <EOL> return name . decode ( "<STR_LIT:utf-8>" ) <EOL> def get_cipher_bits ( self ) : <EOL> """<STR_LIT>""" <EOL> cipher = _lib . SSL_get_current_cipher ( self . _ssl ) <EOL> if cipher == _ffi . NULL : <EOL> return None <EOL> else : <EOL> return _lib . SSL_CIPHER_get_bits ( cipher , _ffi . NULL ) <EOL> def get_cipher_version ( self ) : <EOL> """<STR_LIT>""" <EOL> cipher = _lib . SSL_get_current_cipher ( self . _ssl ) <EOL> if cipher == _ffi . NULL : <EOL> return None <EOL> else : <EOL> version = _ffi . string ( _lib . SSL_CIPHER_get_version ( cipher ) ) <EOL> return version . decode ( "<STR_LIT:utf-8>" ) <EOL> def get_protocol_version_name ( self ) : <EOL> """<STR_LIT>""" <EOL> version = _ffi . string ( _lib . SSL_get_version ( self . _ssl ) ) <EOL> return version . decode ( "<STR_LIT:utf-8>" ) <EOL> def get_protocol_version ( self ) : <EOL> """<STR_LIT>""" <EOL> version = _lib . SSL_version ( self . _ssl ) <EOL> return version <EOL> @ _requires_npn <EOL> def get_next_proto_negotiated ( self ) : <EOL> """<STR_LIT>""" <EOL> data = _ffi . new ( "<STR_LIT>" ) <EOL> data_len = _ffi . new ( "<STR_LIT>" ) <EOL> _lib . SSL_get0_next_proto_negotiated ( self . _ssl , data , data_len ) <EOL> return _ffi . buffer ( data [ <NUM_LIT:0> ] , data_len [ <NUM_LIT:0> ] ) [ : ] <EOL> @ _requires_alpn <EOL> def set_alpn_protos ( self , protos ) : <EOL> """<STR_LIT>""" <EOL> protostr = b'<STR_LIT>' . join ( <EOL> chain . from_iterable ( ( int2byte ( len ( p ) ) , p ) for p in protos ) <EOL> ) <EOL> input_str = _ffi . new ( "<STR_LIT>" , protostr ) <EOL> input_str_len = _ffi . cast ( "<STR_LIT>" , len ( protostr ) ) <EOL> _lib . SSL_set_alpn_protos ( self . _ssl , input_str , input_str_len ) <EOL> @ _requires_alpn <EOL> def get_alpn_proto_negotiated ( self ) : <EOL> """<STR_LIT>""" <EOL> data = _ffi . new ( "<STR_LIT>" ) <EOL> data_len = _ffi . new ( "<STR_LIT>" ) <EOL> _lib . SSL_get0_alpn_selected ( self . _ssl , data , data_len ) <EOL> if not data_len : <EOL> return b'<STR_LIT>' <EOL> return _ffi . buffer ( data [ <NUM_LIT:0> ] , data_len [ <NUM_LIT:0> ] ) [ : ] <EOL> ConnectionType = Connection <EOL> _lib . SSL_library_init ( ) </s>
