<s> """<STR_LIT>""" <EOL> import textwrap <EOL> import _ssl <EOL> from _ssl import OPENSSL_VERSION_NUMBER , OPENSSL_VERSION_INFO , OPENSSL_VERSION <EOL> from _ssl import SSLError <EOL> from _ssl import CERT_NONE , CERT_OPTIONAL , CERT_REQUIRED <EOL> from _ssl import RAND_status , RAND_egd , RAND_add <EOL> from _ssl import SSL_ERROR_ZERO_RETURN , SSL_ERROR_WANT_READ , SSL_ERROR_WANT_WRITE , SSL_ERROR_WANT_X509_LOOKUP , SSL_ERROR_SYSCALL , SSL_ERROR_SSL , SSL_ERROR_WANT_CONNECT , SSL_ERROR_EOF , SSL_ERROR_INVALID_ERROR_CODE <EOL> from _ssl import PROTOCOL_SSLv3 , PROTOCOL_SSLv23 , PROTOCOL_TLSv1 <EOL> _PROTOCOL_NAMES = { <EOL> PROTOCOL_TLSv1 : "<STR_LIT>" , <EOL> PROTOCOL_SSLv23 : "<STR_LIT>" , <EOL> PROTOCOL_SSLv3 : "<STR_LIT>" , <EOL> } <EOL> try : <EOL> from _ssl import PROTOCOL_SSLv2 <EOL> _SSLv2_IF_EXISTS = PROTOCOL_SSLv2 <EOL> except ImportError : <EOL> _SSLv2_IF_EXISTS = None <EOL> else : <EOL> _PROTOCOL_NAMES [ PROTOCOL_SSLv2 ] = "<STR_LIT>" <EOL> from socket import socket , _fileobject , _delegate_methods , error as socket_error <EOL> from socket import getnameinfo as _getnameinfo <EOL> from socket import SOL_SOCKET , SO_TYPE , SOCK_STREAM <EOL> import base64 <EOL> import errno <EOL> _DEFAULT_CIPHERS = '<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 , ciphers = None ) : <EOL> if sock . getsockopt ( SOL_SOCKET , SO_TYPE ) != SOCK_STREAM : <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> socket . __init__ ( self , _sock = sock . _sock ) <EOL> for attr in _delegate_methods : <EOL> try : <EOL> delattr ( self , attr ) <EOL> except AttributeError : <EOL> pass <EOL> if ciphers is None and ssl_version != _SSLv2_IF_EXISTS : <EOL> ciphers = _DEFAULT_CIPHERS <EOL> if certfile and not keyfile : <EOL> keyfile = certfile <EOL> try : <EOL> socket . getpeername ( self ) <EOL> except socket_error , e : <EOL> if e . errno != errno . ENOTCONN : <EOL> raise <EOL> self . _connected = False <EOL> self . _sslobj = None <EOL> else : <EOL> self . _connected = True <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> sock . close ( ) <EOL> def read ( self , len = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . _sslobj . read ( len ) <EOL> except SSLError , x : <EOL> if x . args [ <NUM_LIT:0> ] == SSL_ERROR_EOF and self . suppress_ragged_eofs : <EOL> return '<STR_LIT>' <EOL> else : <EOL> raise <EOL> def write ( self , data ) : <EOL> """<STR_LIT>""" <EOL> return self . _sslobj . write ( data ) <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> ) : <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 , x : <EOL> if x . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_READ : <EOL> return <NUM_LIT:0> <EOL> elif x . args [ <NUM_LIT:0> ] == SSL_ERROR_WANT_WRITE : <EOL> return <NUM_LIT:0> <EOL> else : <EOL> raise <EOL> else : <EOL> return v <EOL> else : <EOL> return self . _sock . send ( data , flags ) <EOL> def sendto ( self , data , flags_or_addr , addr = None ) : <EOL> if self . _sslobj : <EOL> raise ValueError ( "<STR_LIT>" % <EOL> self . __class__ ) <EOL> elif addr is None : <EOL> return self . _sock . sendto ( data , flags_or_addr ) <EOL> else : <EOL> return self . _sock . sendto ( data , flags_or_addr , addr ) <EOL> def sendall ( self , data , flags = <NUM_LIT:0> ) : <EOL> if self . _sslobj : <EOL> if flags != <NUM_LIT:0> : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % <EOL> self . __class__ ) <EOL> amount = len ( data ) <EOL> count = <NUM_LIT:0> <EOL> while ( count < amount ) : <EOL> v = self . send ( data [ count : ] ) <EOL> count += v <EOL> return amount <EOL> else : <EOL> return socket . sendall ( self , data , flags ) <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 self . _sock . recv ( 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> tmp_buffer = self . read ( nbytes ) <EOL> v = len ( tmp_buffer ) <EOL> buffer [ : v ] = tmp_buffer <EOL> return v <EOL> else : <EOL> return self . _sock . recv_into ( buffer , nbytes , flags ) <EOL> def recvfrom ( self , buflen = <NUM_LIT> , flags = <NUM_LIT:0> ) : <EOL> if self . _sslobj : <EOL> raise ValueError ( "<STR_LIT>" % <EOL> self . __class__ ) <EOL> else : <EOL> return self . _sock . recvfrom ( buflen , flags ) <EOL> def recvfrom_into ( self , buffer , nbytes = None , flags = <NUM_LIT:0> ) : <EOL> if self . _sslobj : <EOL> raise ValueError ( "<STR_LIT>" % <EOL> self . __class__ ) <EOL> else : <EOL> return self . _sock . recvfrom_into ( buffer , nbytes , flags ) <EOL> def pending ( self ) : <EOL> if self . _sslobj : <EOL> return self . _sslobj . pending ( ) <EOL> else : <EOL> return <NUM_LIT:0> <EOL> def unwrap ( self ) : <EOL> if self . _sslobj : <EOL> s = self . _sslobj . shutdown ( ) <EOL> self . _sslobj = None <EOL> return 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> def do_handshake ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _sslobj . do_handshake ( ) <EOL> def _real_connect ( self , addr , return_errno ) : <EOL> if self . _connected : <EOL> raise ValueError ( "<STR_LIT>" ) <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> try : <EOL> if return_errno : <EOL> rc = socket . connect_ex ( self , addr ) <EOL> else : <EOL> rc = None <EOL> socket . connect ( self , addr ) <EOL> if not rc : <EOL> if self . do_handshake_on_connect : <EOL> self . do_handshake ( ) <EOL> self . _connected = True <EOL> return rc <EOL> except socket_error : <EOL> self . _sslobj = None <EOL> raise <EOL> def connect ( self , addr ) : <EOL> """<STR_LIT>""" <EOL> self . _real_connect ( addr , False ) <EOL> def connect_ex ( self , addr ) : <EOL> """<STR_LIT>""" <EOL> return self . _real_connect ( addr , True ) <EOL> def accept ( self ) : <EOL> """<STR_LIT>""" <EOL> newsock , addr = socket . accept ( self ) <EOL> try : <EOL> return ( SSLSocket ( newsock , <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> ciphers = self . ciphers , <EOL> do_handshake_on_connect = self . do_handshake_on_connect , <EOL> suppress_ragged_eofs = self . suppress_ragged_eofs ) , <EOL> addr ) <EOL> except socket_error as e : <EOL> newsock . close ( ) <EOL> raise e <EOL> def makefile ( self , mode = '<STR_LIT:r>' , bufsize = - <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> return _fileobject ( self , mode , bufsize , close = True ) <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 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> 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 cert_time_to_seconds ( cert_time ) : <EOL> """<STR_LIT>""" <EOL> import time <EOL> return time . mktime ( time . strptime ( cert_time , "<STR_LIT>" ) ) <EOL> PEM_HEADER = "<STR_LIT>" <EOL> PEM_FOOTER = "<STR_LIT>" <EOL> def DER_cert_to_PEM_cert ( der_cert_bytes ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( base64 , '<STR_LIT>' ) : <EOL> f = base64 . standard_b64encode ( der_cert_bytes ) <EOL> return ( PEM_HEADER + '<STR_LIT:\n>' + <EOL> textwrap . fill ( f , <NUM_LIT:64> ) + '<STR_LIT:\n>' + <EOL> PEM_FOOTER + '<STR_LIT:\n>' ) <EOL> else : <EOL> return ( PEM_HEADER + '<STR_LIT:\n>' + <EOL> base64 . encodestring ( der_cert_bytes ) + <EOL> PEM_FOOTER + '<STR_LIT:\n>' ) <EOL> def PEM_cert_to_DER_cert ( pem_cert_string ) : <EOL> """<STR_LIT>""" <EOL> if not pem_cert_string . startswith ( PEM_HEADER ) : <EOL> raise ValueError ( "<STR_LIT>" <EOL> % PEM_HEADER ) <EOL> if not pem_cert_string . strip ( ) . endswith ( PEM_FOOTER ) : <EOL> raise ValueError ( "<STR_LIT>" <EOL> % PEM_FOOTER ) <EOL> d = pem_cert_string . strip ( ) [ len ( PEM_HEADER ) : - len ( PEM_FOOTER ) ] <EOL> return base64 . decodestring ( d ) <EOL> def get_server_certificate ( addr , ssl_version = PROTOCOL_SSLv3 , ca_certs = None ) : <EOL> """<STR_LIT>""" <EOL> host , port = addr <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 get_protocol_name ( protocol_code ) : <EOL> return _PROTOCOL_NAMES . get ( protocol_code , '<STR_LIT>' ) <EOL> def sslwrap_simple ( sock , keyfile = None , certfile = None ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( sock , "<STR_LIT>" ) : <EOL> sock = sock . _sock <EOL> ssl_sock = _ssl . sslwrap ( sock , <NUM_LIT:0> , keyfile , certfile , CERT_NONE , <EOL> PROTOCOL_SSLv23 , None ) <EOL> try : <EOL> sock . getpeername ( ) <EOL> except socket_error : <EOL> pass <EOL> else : <EOL> ssl_sock . do_handshake ( ) <EOL> return ssl_sock </s>
