<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import ssl as __ssl__ <EOL> _ssl = __ssl__ . _ssl <EOL> import sys <EOL> import errno <EOL> from gevent . _socket2 import socket <EOL> from gevent . socket import _fileobject , timeout_default <EOL> from gevent . socket import error as socket_error , EWOULDBLOCK <EOL> from gevent . socket import timeout as _socket_timeout <EOL> from gevent . _compat import PYPY <EOL> from gevent . _util import copy_globals <EOL> __implements__ = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> __imports__ = copy_globals ( __ssl__ , globals ( ) , <EOL> names_to_ignore = __implements__ + [ '<STR_LIT>' ] , <EOL> dunder_names_to_keep = ( ) ) <EOL> __all__ = list ( set ( __implements__ ) | set ( __imports__ ) ) <EOL> if '<STR_LIT>' in __all__ : <EOL> __all__ . remove ( '<STR_LIT>' ) <EOL> class SSLSocket ( socket ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , sock , keyfile = None , certfile = None , <EOL> server_side = False , cert_reqs = CERT_NONE , <EOL> ssl_version = PROTOCOL_SSLv23 , ca_certs = None , <EOL> do_handshake_on_connect = True , <EOL> suppress_ragged_eofs = True , <EOL> ciphers = None ) : <EOL> socket . __init__ ( self , _sock = sock ) <EOL> if PYPY : <EOL> sock . _drop ( ) <EOL> if certfile and not keyfile : <EOL> keyfile = certfile <EOL> try : <EOL> socket . getpeername ( self ) <EOL> except socket_error as e : <EOL> if e . args [ <NUM_LIT:0> ] != errno . ENOTCONN : <EOL> raise <EOL> self . _sslobj = None <EOL> else : <EOL> if ciphers is None : <EOL> self . _sslobj = _ssl . sslwrap ( self . _sock , server_side , <EOL> keyfile , certfile , <EOL> cert_reqs , ssl_version , ca_certs ) <EOL> else : <EOL> self . _sslobj = _ssl . sslwrap ( self . _sock , server_side , <EOL> keyfile , certfile , <EOL> cert_reqs , ssl_version , ca_certs , <EOL> ciphers ) <EOL> if do_handshake_on_connect : <EOL> self . do_handshake ( ) <EOL> self . keyfile = keyfile <EOL> self . certfile = certfile <EOL> self . cert_reqs = cert_reqs <EOL> self . ssl_version = ssl_version <EOL> self . ca_certs = ca_certs <EOL> self . ciphers = ciphers <EOL> self . do_handshake_on_connect = do_handshake_on_connect <EOL> self . suppress_ragged_eofs = suppress_ragged_eofs <EOL> self . _makefile_refs = <NUM_LIT:0> <EOL> def read ( self , len = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> try : <EOL> return self . _sslobj . read ( len ) <EOL> except SSLError as ex : <EOL> if ex . args [ <NUM_LIT:0> ] == SSL_ERROR_EOF and self . suppress_ragged_eofs : <EOL> return '<STR_LIT>' <EOL> elif ex . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_READ : <EOL> if self . timeout == <NUM_LIT:0.0> : <EOL> raise <EOL> sys . exc_clear ( ) <EOL> self . _wait ( self . _read_event , timeout_exc = _SSLErrorReadTimeout ) <EOL> elif ex . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_WRITE : <EOL> if self . timeout == <NUM_LIT:0.0> : <EOL> raise <EOL> sys . exc_clear ( ) <EOL> self . _wait ( self . _write_event , timeout_exc = _SSLErrorReadTimeout ) <EOL> else : <EOL> raise <EOL> def write ( self , data ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> try : <EOL> return self . _sslobj . write ( data ) <EOL> except SSLError as ex : <EOL> if ex . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_READ : <EOL> if self . timeout == <NUM_LIT:0.0> : <EOL> raise <EOL> sys . exc_clear ( ) <EOL> self . _wait ( self . _read_event , timeout_exc = _SSLErrorWriteTimeout ) <EOL> elif ex . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_WRITE : <EOL> if self . timeout == <NUM_LIT:0.0> : <EOL> raise <EOL> sys . exc_clear ( ) <EOL> self . _wait ( self . _write_event , timeout_exc = _SSLErrorWriteTimeout ) <EOL> else : <EOL> raise <EOL> def getpeercert ( self , binary_form = False ) : <EOL> """<STR_LIT>""" <EOL> return self . _sslobj . peer_certificate ( binary_form ) <EOL> def cipher ( self ) : <EOL> if not self . _sslobj : <EOL> return None <EOL> else : <EOL> return self . _sslobj . cipher ( ) <EOL> def send ( self , data , flags = <NUM_LIT:0> , timeout = timeout_default ) : <EOL> if timeout is timeout_default : <EOL> timeout = self . timeout <EOL> if self . _sslobj : <EOL> if flags != <NUM_LIT:0> : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % <EOL> self . __class__ ) <EOL> while True : <EOL> try : <EOL> v = self . _sslobj . write ( data ) <EOL> except SSLError as x : <EOL> if x . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_READ : <EOL> if self . timeout == <NUM_LIT:0.0> : <EOL> return <NUM_LIT:0> <EOL> sys . exc_clear ( ) <EOL> self . _wait ( self . _read_event ) <EOL> elif x . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_WRITE : <EOL> if self . timeout == <NUM_LIT:0.0> : <EOL> return <NUM_LIT:0> <EOL> sys . exc_clear ( ) <EOL> self . _wait ( self . _write_event ) <EOL> else : <EOL> raise <EOL> else : <EOL> return v <EOL> else : <EOL> return socket . send ( self , data , flags , timeout ) <EOL> def sendall ( self , data , flags = <NUM_LIT:0> ) : <EOL> try : <EOL> socket . sendall ( self , data ) <EOL> except _socket_timeout as ex : <EOL> if self . timeout == <NUM_LIT:0.0> : <EOL> raise SSLError ( SSL_ERROR_WANT_WRITE ) <EOL> raise SSLError ( * ex . args ) <EOL> def sendto ( self , * args ) : <EOL> if self . _sslobj : <EOL> raise ValueError ( "<STR_LIT>" % <EOL> self . __class__ ) <EOL> else : <EOL> return socket . sendto ( self , * args ) <EOL> def recv ( self , buflen = <NUM_LIT> , flags = <NUM_LIT:0> ) : <EOL> if self . _sslobj : <EOL> if flags != <NUM_LIT:0> : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % <EOL> self . __class__ ) <EOL> return self . read ( buflen ) <EOL> else : <EOL> return socket . recv ( self , buflen , flags ) <EOL> def recv_into ( self , buffer , nbytes = None , flags = <NUM_LIT:0> ) : <EOL> if buffer and ( nbytes is None ) : <EOL> nbytes = len ( buffer ) <EOL> elif nbytes is None : <EOL> nbytes = <NUM_LIT> <EOL> if self . _sslobj : <EOL> if flags != <NUM_LIT:0> : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % <EOL> self . __class__ ) <EOL> while True : <EOL> try : <EOL> tmp_buffer = self . read ( nbytes ) <EOL> v = len ( tmp_buffer ) <EOL> buffer [ : v ] = tmp_buffer <EOL> return v <EOL> except SSLError as x : <EOL> if x . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_READ : <EOL> if self . timeout == <NUM_LIT:0.0> : <EOL> raise <EOL> sys . exc_clear ( ) <EOL> self . _wait ( self . _read_event ) <EOL> continue <EOL> else : <EOL> raise <EOL> else : <EOL> return socket . recv_into ( self , buffer , nbytes , flags ) <EOL> def recvfrom ( self , * args ) : <EOL> if self . _sslobj : <EOL> raise ValueError ( "<STR_LIT>" % <EOL> self . __class__ ) <EOL> else : <EOL> return socket . recvfrom ( self , * args ) <EOL> def recvfrom_into ( self , * args ) : <EOL> if self . _sslobj : <EOL> raise ValueError ( "<STR_LIT>" % <EOL> self . __class__ ) <EOL> else : <EOL> return socket . recvfrom_into ( self , * args ) <EOL> def pending ( self ) : <EOL> if self . _sslobj : <EOL> return self . _sslobj . pending ( ) <EOL> else : <EOL> return <NUM_LIT:0> <EOL> def _sslobj_shutdown ( self ) : <EOL> while True : <EOL> try : <EOL> return self . _sslobj . shutdown ( ) <EOL> except SSLError as ex : <EOL> if ex . args [ <NUM_LIT:0> ] == SSL_ERROR_EOF and self . suppress_ragged_eofs : <EOL> return '<STR_LIT>' <EOL> elif ex . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_READ : <EOL> if self . timeout == <NUM_LIT:0.0> : <EOL> raise <EOL> sys . exc_clear ( ) <EOL> self . _wait ( self . _read_event , timeout_exc = _SSLErrorReadTimeout ) <EOL> elif ex . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_WRITE : <EOL> if self . timeout == <NUM_LIT:0.0> : <EOL> raise <EOL> sys . exc_clear ( ) <EOL> self . _wait ( self . _write_event , timeout_exc = _SSLErrorWriteTimeout ) <EOL> else : <EOL> raise <EOL> def unwrap ( self ) : <EOL> if self . _sslobj : <EOL> s = self . _sslobj_shutdown ( ) <EOL> self . _sslobj = None <EOL> return socket ( _sock = s ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" + str ( self ) ) <EOL> def shutdown ( self , how ) : <EOL> self . _sslobj = None <EOL> socket . shutdown ( self , how ) <EOL> def close ( self ) : <EOL> if self . _makefile_refs < <NUM_LIT:1> : <EOL> self . _sslobj = None <EOL> socket . close ( self ) <EOL> else : <EOL> self . _makefile_refs -= <NUM_LIT:1> <EOL> if PYPY : <EOL> def _reuse ( self ) : <EOL> self . _makefile_refs += <NUM_LIT:1> <EOL> def _drop ( self ) : <EOL> if self . _makefile_refs < <NUM_LIT:1> : <EOL> self . close ( ) <EOL> else : <EOL> self . _makefile_refs -= <NUM_LIT:1> <EOL> def do_handshake ( self ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> try : <EOL> return self . _sslobj . do_handshake ( ) <EOL> except SSLError as ex : <EOL> if ex . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_READ : <EOL> if self . timeout == <NUM_LIT:0.0> : <EOL> raise <EOL> sys . exc_clear ( ) <EOL> self . _wait ( self . _read_event , timeout_exc = _SSLErrorHandshakeTimeout ) <EOL> elif ex . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_WRITE : <EOL> if self . timeout == <NUM_LIT:0.0> : <EOL> raise <EOL> sys . exc_clear ( ) <EOL> self . _wait ( self . _write_event , timeout_exc = _SSLErrorHandshakeTimeout ) <EOL> else : <EOL> raise <EOL> def connect ( self , addr ) : <EOL> """<STR_LIT>""" <EOL> if self . _sslobj : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> socket . connect ( self , addr ) <EOL> if self . ciphers is None : <EOL> self . _sslobj = _ssl . sslwrap ( self . _sock , False , self . keyfile , self . certfile , <EOL> self . cert_reqs , self . ssl_version , <EOL> self . ca_certs ) <EOL> else : <EOL> self . _sslobj = _ssl . sslwrap ( self . _sock , False , self . keyfile , self . certfile , <EOL> self . cert_reqs , self . ssl_version , <EOL> self . ca_certs , self . ciphers ) <EOL> if self . do_handshake_on_connect : <EOL> self . do_handshake ( ) <EOL> def accept ( self ) : <EOL> """<STR_LIT>""" <EOL> sock = self . _sock <EOL> while True : <EOL> try : <EOL> client_socket , address = sock . accept ( ) <EOL> break <EOL> except socket_error as ex : <EOL> if ex . args [ <NUM_LIT:0> ] != EWOULDBLOCK or self . timeout == <NUM_LIT:0.0> : <EOL> raise <EOL> sys . exc_clear ( ) <EOL> self . _wait ( self . _read_event ) <EOL> sslobj = SSLSocket ( client_socket , <EOL> keyfile = self . keyfile , <EOL> certfile = self . certfile , <EOL> server_side = True , <EOL> cert_reqs = self . cert_reqs , <EOL> ssl_version = self . ssl_version , <EOL> ca_certs = self . ca_certs , <EOL> do_handshake_on_connect = self . do_handshake_on_connect , <EOL> suppress_ragged_eofs = self . suppress_ragged_eofs , <EOL> ciphers = self . ciphers ) <EOL> return sslobj , address <EOL> def makefile ( self , mode = '<STR_LIT:r>' , bufsize = - <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> if not PYPY : <EOL> self . _makefile_refs += <NUM_LIT:1> <EOL> return _fileobject ( self , mode , bufsize , close = True ) <EOL> if PYPY or not hasattr ( SSLSocket , '<STR_LIT>' ) : <EOL> SSLSocket . timeout = property ( lambda self : self . gettimeout ( ) , <EOL> lambda self , value : self . settimeout ( value ) ) <EOL> _SSLErrorReadTimeout = SSLError ( '<STR_LIT>' ) <EOL> _SSLErrorWriteTimeout = SSLError ( '<STR_LIT>' ) <EOL> _SSLErrorHandshakeTimeout = SSLError ( '<STR_LIT>' ) <EOL> def wrap_socket ( sock , keyfile = None , certfile = None , <EOL> server_side = False , cert_reqs = CERT_NONE , <EOL> ssl_version = PROTOCOL_SSLv23 , ca_certs = None , <EOL> do_handshake_on_connect = True , <EOL> suppress_ragged_eofs = True , ciphers = None ) : <EOL> """<STR_LIT>""" <EOL> return SSLSocket ( sock , keyfile = keyfile , certfile = certfile , <EOL> server_side = server_side , cert_reqs = cert_reqs , <EOL> ssl_version = ssl_version , ca_certs = ca_certs , <EOL> do_handshake_on_connect = do_handshake_on_connect , <EOL> suppress_ragged_eofs = suppress_ragged_eofs , <EOL> ciphers = ciphers ) <EOL> def get_server_certificate ( addr , ssl_version = PROTOCOL_SSLv23 , ca_certs = None ) : <EOL> """<STR_LIT>""" <EOL> if ca_certs is not None : <EOL> cert_reqs = CERT_REQUIRED <EOL> else : <EOL> cert_reqs = CERT_NONE <EOL> s = wrap_socket ( socket ( ) , ssl_version = ssl_version , <EOL> cert_reqs = cert_reqs , ca_certs = ca_certs ) <EOL> s . connect ( addr ) <EOL> dercert = s . getpeercert ( True ) <EOL> s . close ( ) <EOL> return DER_cert_to_PEM_cert ( dercert ) <EOL> def sslwrap_simple ( sock , keyfile = None , certfile = None ) : <EOL> """<STR_LIT>""" <EOL> return SSLSocket ( sock , keyfile , certfile ) </s>
