<s> from __future__ import division , absolute_import <EOL> import itertools <EOL> import warnings <EOL> from binascii import a2b_base64 <EOL> from hashlib import md5 <EOL> import OpenSSL <EOL> from OpenSSL import SSL , crypto <EOL> try : <EOL> from OpenSSL . SSL import SSL_CB_HANDSHAKE_DONE , SSL_CB_HANDSHAKE_START <EOL> except ImportError : <EOL> SSL_CB_HANDSHAKE_START = <NUM_LIT> <EOL> SSL_CB_HANDSHAKE_DONE = <NUM_LIT> <EOL> from twisted . python import log <EOL> def _cantSetHostnameIndication ( connection , hostname ) : <EOL> """<STR_LIT>""" <EOL> def _setHostNameIndication ( connection , hostname ) : <EOL> """<STR_LIT>""" <EOL> connection . set_tlsext_host_name ( hostname ) <EOL> if getattr ( SSL . Connection , "<STR_LIT>" , None ) is None : <EOL> _maybeSetHostNameIndication = _cantSetHostnameIndication <EOL> else : <EOL> _maybeSetHostNameIndication = _setHostNameIndication <EOL> class SimpleVerificationError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def _idnaBytes ( text ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> import idna <EOL> except ImportError : <EOL> return text . encode ( "<STR_LIT>" ) <EOL> else : <EOL> return idna . encode ( text ) . encode ( "<STR_LIT:ascii>" ) <EOL> def _idnaText ( octets ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> import idna <EOL> except ImportError : <EOL> return octets . decode ( "<STR_LIT>" ) <EOL> else : <EOL> return idna . decode ( octets ) <EOL> def simpleVerifyHostname ( connection , hostname ) : <EOL> """<STR_LIT>""" <EOL> commonName = connection . get_peer_certificate ( ) . get_subject ( ) . commonName <EOL> if commonName != hostname : <EOL> raise SimpleVerificationError ( repr ( commonName ) + "<STR_LIT>" + <EOL> repr ( hostname ) ) <EOL> def _selectVerifyImplementation ( lib ) : <EOL> """<STR_LIT>""" <EOL> whatsWrong = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> major , minor = list ( int ( part ) for part in lib . __version__ . split ( "<STR_LIT:.>" ) ) [ : <NUM_LIT:2> ] <EOL> if ( major , minor ) >= ( <NUM_LIT:0> , <NUM_LIT:12> ) : <EOL> try : <EOL> from service_identity import VerificationError <EOL> from service_identity . pyopenssl import verify_hostname <EOL> return verify_hostname , VerificationError <EOL> except ImportError as e : <EOL> warnings . warn_explicit ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" + str ( e ) + "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> + whatsWrong , <EOL> category = UserWarning , filename = "<STR_LIT>" , lineno = <NUM_LIT:0> ) <EOL> else : <EOL> warnings . warn_explicit ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> . format ( lib . __version__ ) + whatsWrong , <EOL> category = UserWarning , filename = "<STR_LIT>" , lineno = <NUM_LIT:0> ) <EOL> return simpleVerifyHostname , SimpleVerificationError <EOL> verifyHostname , VerificationError = _selectVerifyImplementation ( OpenSSL ) <EOL> from zope . interface import Interface , implementer <EOL> from twisted . internet . defer import Deferred <EOL> from twisted . internet . error import VerifyError , CertificateError <EOL> from twisted . internet . interfaces import ( <EOL> IAcceptableCiphers , ICipher , IOpenSSLClientConnectionCreator <EOL> ) <EOL> from twisted . python import reflect , util <EOL> from twisted . python . deprecate import _mutuallyExclusiveArguments <EOL> from twisted . python . compat import nativeString , networkString , unicode <EOL> from twisted . python . failure import Failure <EOL> from twisted . python . util import FancyEqMixin <EOL> from twisted . python . deprecate import deprecated <EOL> from twisted . python . versions import Version <EOL> def _sessionCounter ( counter = itertools . count ( ) ) : <EOL> """<STR_LIT>""" <EOL> return next ( counter ) <EOL> _x509names = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:O>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:L>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:C>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> class DistinguishedName ( dict ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( ) <EOL> def __init__ ( self , ** kw ) : <EOL> for k , v in kw . items ( ) : <EOL> setattr ( self , k , v ) <EOL> def _copyFrom ( self , x509name ) : <EOL> for name in _x509names : <EOL> value = getattr ( x509name , name , None ) <EOL> if value is not None : <EOL> setattr ( self , name , value ) <EOL> def _copyInto ( self , x509name ) : <EOL> for k , v in self . items ( ) : <EOL> setattr ( x509name , k , nativeString ( v ) ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( dict . __repr__ ( self ) [ <NUM_LIT:1> : - <NUM_LIT:1> ] ) <EOL> def __getattr__ ( self , attr ) : <EOL> try : <EOL> return self [ _x509names [ attr ] ] <EOL> except KeyError : <EOL> raise AttributeError ( attr ) <EOL> def __setattr__ ( self , attr , value ) : <EOL> if attr not in _x509names : <EOL> raise AttributeError ( "<STR_LIT>" % ( attr , ) ) <EOL> realAttr = _x509names [ attr ] <EOL> if not isinstance ( value , bytes ) : <EOL> value = value . encode ( "<STR_LIT:ascii>" ) <EOL> self [ realAttr ] = value <EOL> def inspect ( self ) : <EOL> """<STR_LIT>""" <EOL> l = [ ] <EOL> lablen = <NUM_LIT:0> <EOL> def uniqueValues ( mapping ) : <EOL> return set ( mapping . values ( ) ) <EOL> for k in sorted ( uniqueValues ( _x509names ) ) : <EOL> label = util . nameToLabel ( k ) <EOL> lablen = max ( len ( label ) , lablen ) <EOL> v = getattr ( self , k , None ) <EOL> if v is not None : <EOL> l . append ( ( label , nativeString ( v ) ) ) <EOL> lablen += <NUM_LIT:2> <EOL> for n , ( label , attr ) in enumerate ( l ) : <EOL> l [ n ] = ( label . rjust ( lablen ) + '<STR_LIT>' + attr ) <EOL> return '<STR_LIT:\n>' . join ( l ) <EOL> DN = DistinguishedName <EOL> class CertBase : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , original ) : <EOL> self . original = original <EOL> def _copyName ( self , suffix ) : <EOL> dn = DistinguishedName ( ) <EOL> dn . _copyFrom ( getattr ( self . original , '<STR_LIT>' + suffix ) ( ) ) <EOL> return dn <EOL> def getSubject ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _copyName ( '<STR_LIT>' ) <EOL> def __conform__ ( self , interface ) : <EOL> """<STR_LIT>""" <EOL> if interface is IOpenSSLTrustRoot : <EOL> return OpenSSLCertificateAuthorities ( [ self . original ] ) <EOL> return NotImplemented <EOL> def _handleattrhelper ( Class , transport , methodName ) : <EOL> """<STR_LIT>""" <EOL> method = getattr ( transport . getHandle ( ) , <EOL> "<STR_LIT>" % ( methodName , ) , None ) <EOL> if method is None : <EOL> raise CertificateError ( <EOL> "<STR_LIT>" % ( transport , methodName ) ) <EOL> cert = method ( ) <EOL> if cert is None : <EOL> raise CertificateError ( <EOL> "<STR_LIT>" % ( transport , methodName ) ) <EOL> return Class ( cert ) <EOL> class Certificate ( CertBase ) : <EOL> """<STR_LIT>""" <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , <EOL> self . getSubject ( ) . commonName , <EOL> self . getIssuer ( ) . commonName ) <EOL> def __eq__ ( self , other ) : <EOL> if isinstance ( other , Certificate ) : <EOL> return self . dump ( ) == other . dump ( ) <EOL> return False <EOL> def __ne__ ( self , other ) : <EOL> return not self . __eq__ ( other ) <EOL> def load ( Class , requestData , format = crypto . FILETYPE_ASN1 , args = ( ) ) : <EOL> """<STR_LIT>""" <EOL> return Class ( crypto . load_certificate ( format , requestData ) , * args ) <EOL> load = classmethod ( load ) <EOL> _load = load <EOL> def dumpPEM ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . dump ( crypto . FILETYPE_PEM ) <EOL> def loadPEM ( Class , data ) : <EOL> """<STR_LIT>""" <EOL> return Class . load ( data , crypto . FILETYPE_PEM ) <EOL> loadPEM = classmethod ( loadPEM ) <EOL> def peerFromTransport ( Class , transport ) : <EOL> """<STR_LIT>""" <EOL> return _handleattrhelper ( Class , transport , '<STR_LIT>' ) <EOL> peerFromTransport = classmethod ( peerFromTransport ) <EOL> def hostFromTransport ( Class , transport ) : <EOL> """<STR_LIT>""" <EOL> return _handleattrhelper ( Class , transport , '<STR_LIT:host>' ) <EOL> hostFromTransport = classmethod ( hostFromTransport ) <EOL> def getPublicKey ( self ) : <EOL> """<STR_LIT>""" <EOL> return PublicKey ( self . original . get_pubkey ( ) ) <EOL> def dump ( self , format = crypto . FILETYPE_ASN1 ) : <EOL> return crypto . dump_certificate ( format , self . original ) <EOL> def serialNumber ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . original . get_serial_number ( ) <EOL> def digest ( self , method = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> return self . original . digest ( method ) <EOL> def _inspect ( self ) : <EOL> return '<STR_LIT:\n>' . join ( [ '<STR_LIT>' , <EOL> self . getSubject ( ) . inspect ( ) , <EOL> '<STR_LIT>' , <EOL> self . getIssuer ( ) . inspect ( ) , <EOL> '<STR_LIT>' % self . serialNumber ( ) , <EOL> '<STR_LIT>' % nativeString ( self . digest ( ) ) ] ) <EOL> def inspect ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT:\n>' . join ( ( self . _inspect ( ) , self . getPublicKey ( ) . inspect ( ) ) ) <EOL> def getIssuer ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _copyName ( '<STR_LIT>' ) <EOL> def options ( self , * authorities ) : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> class CertificateRequest ( CertBase ) : <EOL> """<STR_LIT>""" <EOL> def load ( Class , requestData , requestFormat = crypto . FILETYPE_ASN1 ) : <EOL> req = crypto . load_certificate_request ( requestFormat , requestData ) <EOL> dn = DistinguishedName ( ) <EOL> dn . _copyFrom ( req . get_subject ( ) ) <EOL> if not req . verify ( req . get_pubkey ( ) ) : <EOL> raise VerifyError ( "<STR_LIT>" % ( dn , ) ) <EOL> return Class ( req ) <EOL> load = classmethod ( load ) <EOL> def dump ( self , format = crypto . FILETYPE_ASN1 ) : <EOL> return crypto . dump_certificate_request ( format , self . original ) <EOL> class PrivateCertificate ( Certificate ) : <EOL> """<STR_LIT>""" <EOL> def __repr__ ( self ) : <EOL> return Certificate . __repr__ ( self ) + '<STR_LIT>' + repr ( self . privateKey ) <EOL> def _setPrivateKey ( self , privateKey ) : <EOL> if not privateKey . matches ( self . getPublicKey ( ) ) : <EOL> raise VerifyError ( <EOL> "<STR_LIT>" ) <EOL> self . privateKey = privateKey <EOL> return self <EOL> def newCertificate ( self , newCertData , format = crypto . FILETYPE_ASN1 ) : <EOL> """<STR_LIT>""" <EOL> return self . load ( newCertData , self . privateKey , format ) <EOL> def load ( Class , data , privateKey , format = crypto . FILETYPE_ASN1 ) : <EOL> return Class . _load ( data , format ) . _setPrivateKey ( privateKey ) <EOL> load = classmethod ( load ) <EOL> def inspect ( self ) : <EOL> return '<STR_LIT:\n>' . join ( [ Certificate . _inspect ( self ) , <EOL> self . privateKey . inspect ( ) ] ) <EOL> def dumpPEM ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . dump ( crypto . FILETYPE_PEM ) + self . privateKey . dump ( crypto . FILETYPE_PEM ) <EOL> def loadPEM ( Class , data ) : <EOL> """<STR_LIT>""" <EOL> return Class . load ( data , KeyPair . load ( data , crypto . FILETYPE_PEM ) , <EOL> crypto . FILETYPE_PEM ) <EOL> loadPEM = classmethod ( loadPEM ) <EOL> def fromCertificateAndKeyPair ( Class , certificateInstance , privateKey ) : <EOL> privcert = Class ( certificateInstance . original ) <EOL> return privcert . _setPrivateKey ( privateKey ) <EOL> fromCertificateAndKeyPair = classmethod ( fromCertificateAndKeyPair ) <EOL> def options ( self , * authorities ) : <EOL> """<STR_LIT>""" <EOL> options = dict ( privateKey = self . privateKey . original , <EOL> certificate = self . original ) <EOL> if authorities : <EOL> options . update ( dict ( trustRoot = OpenSSLCertificateAuthorities ( <EOL> [ auth . original for auth in authorities ] <EOL> ) ) ) <EOL> return OpenSSLCertificateOptions ( ** options ) <EOL> def certificateRequest ( self , format = crypto . FILETYPE_ASN1 , <EOL> digestAlgorithm = '<STR_LIT>' ) : <EOL> return self . privateKey . certificateRequest ( <EOL> self . getSubject ( ) , <EOL> format , <EOL> digestAlgorithm ) <EOL> def signCertificateRequest ( self , <EOL> requestData , <EOL> verifyDNCallback , <EOL> serialNumber , <EOL> requestFormat = crypto . FILETYPE_ASN1 , <EOL> certificateFormat = crypto . FILETYPE_ASN1 ) : <EOL> issuer = self . getSubject ( ) <EOL> return self . privateKey . signCertificateRequest ( <EOL> issuer , <EOL> requestData , <EOL> verifyDNCallback , <EOL> serialNumber , <EOL> requestFormat , <EOL> certificateFormat ) <EOL> def signRequestObject ( self , certificateRequest , serialNumber , <EOL> secondsToExpiry = <NUM_LIT> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , <EOL> digestAlgorithm = '<STR_LIT>' ) : <EOL> return self . privateKey . signRequestObject ( self . getSubject ( ) , <EOL> certificateRequest , <EOL> serialNumber , <EOL> secondsToExpiry , <EOL> digestAlgorithm ) <EOL> class PublicKey : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , osslpkey ) : <EOL> """<STR_LIT>""" <EOL> self . original = osslpkey <EOL> def matches ( self , otherKey ) : <EOL> """<STR_LIT>""" <EOL> return self . keyHash ( ) == otherKey . keyHash ( ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , self . keyHash ( ) ) <EOL> def keyHash ( self ) : <EOL> """<STR_LIT>""" <EOL> nsspki = crypto . NetscapeSPKI ( ) <EOL> nsspki . set_pubkey ( self . original ) <EOL> encoded = nsspki . b64_encode ( ) <EOL> raw = a2b_base64 ( encoded ) <EOL> h = md5 ( ) <EOL> h . update ( raw ) <EOL> return h . hexdigest ( ) <EOL> def inspect ( self ) : <EOL> return '<STR_LIT>' % ( self . keyHash ( ) , ) <EOL> class KeyPair ( PublicKey ) : <EOL> def load ( Class , data , format = crypto . FILETYPE_ASN1 ) : <EOL> return Class ( crypto . load_privatekey ( format , data ) ) <EOL> load = classmethod ( load ) <EOL> def dump ( self , format = crypto . FILETYPE_ASN1 ) : <EOL> return crypto . dump_privatekey ( format , self . original ) <EOL> def __getstate__ ( self ) : <EOL> return self . dump ( ) <EOL> def __setstate__ ( self , state ) : <EOL> self . __init__ ( crypto . load_privatekey ( crypto . FILETYPE_ASN1 , state ) ) <EOL> def inspect ( self ) : <EOL> t = self . original . type ( ) <EOL> if t == crypto . TYPE_RSA : <EOL> ts = '<STR_LIT>' <EOL> elif t == crypto . TYPE_DSA : <EOL> ts = '<STR_LIT>' <EOL> else : <EOL> ts = '<STR_LIT>' <EOL> L = ( self . original . bits ( ) , ts , self . keyHash ( ) ) <EOL> return '<STR_LIT>' % L <EOL> def generate ( Class , kind = crypto . TYPE_RSA , size = <NUM_LIT> ) : <EOL> pkey = crypto . PKey ( ) <EOL> pkey . generate_key ( kind , size ) <EOL> return Class ( pkey ) <EOL> def newCertificate ( self , newCertData , format = crypto . FILETYPE_ASN1 ) : <EOL> return PrivateCertificate . load ( newCertData , self , format ) <EOL> generate = classmethod ( generate ) <EOL> def requestObject ( self , distinguishedName , digestAlgorithm = '<STR_LIT>' ) : <EOL> req = crypto . X509Req ( ) <EOL> req . set_pubkey ( self . original ) <EOL> distinguishedName . _copyInto ( req . get_subject ( ) ) <EOL> req . sign ( self . original , digestAlgorithm ) <EOL> return CertificateRequest ( req ) <EOL> def certificateRequest ( self , distinguishedName , <EOL> format = crypto . FILETYPE_ASN1 , <EOL> digestAlgorithm = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> return self . requestObject ( distinguishedName , digestAlgorithm ) . dump ( format ) <EOL> def signCertificateRequest ( self , <EOL> issuerDistinguishedName , <EOL> requestData , <EOL> verifyDNCallback , <EOL> serialNumber , <EOL> requestFormat = crypto . FILETYPE_ASN1 , <EOL> certificateFormat = crypto . FILETYPE_ASN1 , <EOL> secondsToExpiry = <NUM_LIT> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , <EOL> digestAlgorithm = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> hlreq = CertificateRequest . load ( requestData , requestFormat ) <EOL> dn = hlreq . getSubject ( ) <EOL> vval = verifyDNCallback ( dn ) <EOL> def verified ( value ) : <EOL> if not value : <EOL> raise VerifyError ( "<STR_LIT>" % ( verifyDNCallback , dn ) ) <EOL> return self . signRequestObject ( issuerDistinguishedName , hlreq , <EOL> serialNumber , secondsToExpiry , digestAlgorithm ) . dump ( certificateFormat ) <EOL> if isinstance ( vval , Deferred ) : <EOL> return vval . addCallback ( verified ) <EOL> else : <EOL> return verified ( vval ) <EOL> def signRequestObject ( self , <EOL> issuerDistinguishedName , <EOL> requestObject , <EOL> serialNumber , <EOL> secondsToExpiry = <NUM_LIT> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , <EOL> digestAlgorithm = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> req = requestObject . original <EOL> cert = crypto . X509 ( ) <EOL> issuerDistinguishedName . _copyInto ( cert . get_issuer ( ) ) <EOL> cert . set_subject ( req . get_subject ( ) ) <EOL> cert . set_pubkey ( req . get_pubkey ( ) ) <EOL> cert . gmtime_adj_notBefore ( <NUM_LIT:0> ) <EOL> cert . gmtime_adj_notAfter ( secondsToExpiry ) <EOL> cert . set_serial_number ( serialNumber ) <EOL> cert . sign ( self . original , digestAlgorithm ) <EOL> return Certificate ( cert ) <EOL> def selfSignedCert ( self , serialNumber , ** kw ) : <EOL> dn = DN ( ** kw ) <EOL> return PrivateCertificate . fromCertificateAndKeyPair ( <EOL> self . signRequestObject ( dn , self . requestObject ( dn ) , serialNumber ) , <EOL> self ) <EOL> KeyPair . __getstate__ = deprecated ( Version ( "<STR_LIT>" , <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:0> ) , <EOL> "<STR_LIT>" ) ( KeyPair . __getstate__ ) <EOL> KeyPair . __setstate__ = deprecated ( Version ( "<STR_LIT>" , <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:0> ) , <EOL> "<STR_LIT>" ) ( KeyPair . __setstate__ ) <EOL> class IOpenSSLTrustRoot ( Interface ) : <EOL> """<STR_LIT>""" <EOL> def _addCACertsToContext ( context ) : <EOL> """<STR_LIT>""" <EOL> @ implementer ( IOpenSSLTrustRoot ) <EOL> class OpenSSLCertificateAuthorities ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , caCerts ) : <EOL> """<STR_LIT>""" <EOL> self . _caCerts = caCerts <EOL> def _addCACertsToContext ( self , context ) : <EOL> store = context . get_cert_store ( ) <EOL> for cert in self . _caCerts : <EOL> store . add_cert ( cert ) <EOL> @ implementer ( IOpenSSLTrustRoot ) <EOL> class OpenSSLDefaultPaths ( object ) : <EOL> """<STR_LIT>""" <EOL> def _addCACertsToContext ( self , context ) : <EOL> context . set_default_verify_paths ( ) <EOL> def platformTrust ( ) : <EOL> """<STR_LIT>""" <EOL> return OpenSSLDefaultPaths ( ) <EOL> def _tolerateErrors ( wrapped ) : <EOL> """<STR_LIT>""" <EOL> def infoCallback ( connection , where , ret ) : <EOL> try : <EOL> return wrapped ( connection , where , ret ) <EOL> except : <EOL> f = Failure ( ) <EOL> log . err ( f , "<STR_LIT>" ) <EOL> connection . get_app_data ( ) . failVerification ( f ) <EOL> return infoCallback <EOL> @ implementer ( IOpenSSLClientConnectionCreator ) <EOL> class ClientTLSOptions ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , hostname , ctx ) : <EOL> """<STR_LIT>""" <EOL> self . _ctx = ctx <EOL> self . _hostname = hostname <EOL> self . _hostnameBytes = _idnaBytes ( hostname ) <EOL> self . _hostnameASCII = self . _hostnameBytes . decode ( "<STR_LIT:ascii>" ) <EOL> ctx . set_info_callback ( <EOL> _tolerateErrors ( self . _identityVerifyingInfoCallback ) <EOL> ) <EOL> def clientConnectionForTLS ( self , tlsProtocol ) : <EOL> """<STR_LIT>""" <EOL> context = self . _ctx <EOL> connection = SSL . Connection ( context , None ) <EOL> connection . set_app_data ( tlsProtocol ) <EOL> return connection <EOL> def _identityVerifyingInfoCallback ( self , connection , where , ret ) : <EOL> """<STR_LIT>""" <EOL> if where & SSL_CB_HANDSHAKE_START : <EOL> _maybeSetHostNameIndication ( connection , self . _hostnameBytes ) <EOL> elif where & SSL_CB_HANDSHAKE_DONE : <EOL> try : <EOL> verifyHostname ( connection , self . _hostnameASCII ) <EOL> except VerificationError : <EOL> f = Failure ( ) <EOL> transport = connection . get_app_data ( ) <EOL> transport . failVerification ( f ) <EOL> def optionsForClientTLS ( hostname , trustRoot = None , clientCertificate = None , <EOL> ** kw ) : <EOL> """<STR_LIT>""" <EOL> extraCertificateOptions = kw . pop ( '<STR_LIT>' , None ) or { } <EOL> if trustRoot is None : <EOL> trustRoot = platformTrust ( ) <EOL> if kw : <EOL> raise TypeError ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> arg = kw . popitem ( ) [ <NUM_LIT:0> ] <EOL> ) <EOL> ) <EOL> if not isinstance ( hostname , unicode ) : <EOL> raise TypeError ( <EOL> "<STR_LIT>" <EOL> + hostname . __class__ . __name__ <EOL> ) <EOL> if clientCertificate : <EOL> extraCertificateOptions . update ( <EOL> privateKey = clientCertificate . privateKey . original , <EOL> certificate = clientCertificate . original <EOL> ) <EOL> certificateOptions = OpenSSLCertificateOptions ( <EOL> trustRoot = trustRoot , <EOL> ** extraCertificateOptions <EOL> ) <EOL> return ClientTLSOptions ( hostname , certificateOptions . getContext ( ) ) <EOL> class OpenSSLCertificateOptions ( object ) : <EOL> """<STR_LIT>""" <EOL> _contextFactory = SSL . Context <EOL> _context = None <EOL> _OP_ALL = getattr ( SSL , '<STR_LIT>' , <NUM_LIT> ) <EOL> _OP_NO_TICKET = getattr ( SSL , '<STR_LIT>' , <NUM_LIT> ) <EOL> _OP_NO_COMPRESSION = getattr ( SSL , '<STR_LIT>' , <NUM_LIT> ) <EOL> _OP_CIPHER_SERVER_PREFERENCE = getattr ( SSL , '<STR_LIT>' , <EOL> <NUM_LIT> ) <EOL> _OP_SINGLE_ECDH_USE = getattr ( SSL , '<STR_LIT>' , <NUM_LIT> ) <EOL> @ _mutuallyExclusiveArguments ( [ <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> ] ) <EOL> def __init__ ( self , <EOL> privateKey = None , <EOL> certificate = None , <EOL> method = None , <EOL> verify = False , <EOL> caCerts = None , <EOL> verifyDepth = <NUM_LIT:9> , <EOL> requireCertificate = True , <EOL> verifyOnce = True , <EOL> enableSingleUseKeys = True , <EOL> enableSessions = True , <EOL> fixBrokenPeers = False , <EOL> enableSessionTickets = False , <EOL> extraCertChain = None , <EOL> acceptableCiphers = None , <EOL> dhParameters = None , <EOL> trustRoot = None ) : <EOL> """<STR_LIT>""" <EOL> if ( privateKey is None ) != ( certificate is None ) : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> self . privateKey = privateKey <EOL> self . certificate = certificate <EOL> self . _options = ( <EOL> SSL . OP_NO_SSLv2 | self . _OP_NO_COMPRESSION | <EOL> self . _OP_CIPHER_SERVER_PREFERENCE <EOL> ) <EOL> if method is None : <EOL> self . method = SSL . SSLv23_METHOD <EOL> self . _options |= SSL . OP_NO_SSLv3 <EOL> else : <EOL> self . method = method <EOL> if verify and not caCerts : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . verify = verify <EOL> if extraCertChain is not None and None in ( privateKey , certificate ) : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if extraCertChain is not None : <EOL> self . extraCertChain = extraCertChain <EOL> else : <EOL> self . extraCertChain = [ ] <EOL> self . caCerts = caCerts <EOL> self . verifyDepth = verifyDepth <EOL> self . requireCertificate = requireCertificate <EOL> self . verifyOnce = verifyOnce <EOL> self . enableSingleUseKeys = enableSingleUseKeys <EOL> if enableSingleUseKeys : <EOL> self . _options |= SSL . OP_SINGLE_DH_USE | self . _OP_SINGLE_ECDH_USE <EOL> self . enableSessions = enableSessions <EOL> self . fixBrokenPeers = fixBrokenPeers <EOL> if fixBrokenPeers : <EOL> self . _options |= self . _OP_ALL <EOL> self . enableSessionTickets = enableSessionTickets <EOL> if not enableSessionTickets : <EOL> self . _options |= self . _OP_NO_TICKET <EOL> self . dhParameters = dhParameters <EOL> try : <EOL> self . _ecCurve = _OpenSSLECCurve ( _defaultCurveName ) <EOL> except NotImplementedError : <EOL> self . _ecCurve = None <EOL> if acceptableCiphers is None : <EOL> acceptableCiphers = defaultCiphers <EOL> self . _cipherString = u'<STR_LIT::>' . join ( <EOL> c . fullName <EOL> for c in acceptableCiphers . selectCiphers ( <EOL> _expandCipherString ( u'<STR_LIT>' , self . method , self . _options ) <EOL> ) <EOL> ) <EOL> if self . _cipherString == u'<STR_LIT>' : <EOL> raise ValueError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) <EOL> if trustRoot is None : <EOL> if self . verify : <EOL> trustRoot = OpenSSLCertificateAuthorities ( caCerts ) <EOL> else : <EOL> self . verify = True <EOL> self . requireCertificate = True <EOL> trustRoot = IOpenSSLTrustRoot ( trustRoot ) <EOL> self . trustRoot = trustRoot <EOL> def __getstate__ ( self ) : <EOL> d = self . __dict__ . copy ( ) <EOL> try : <EOL> del d [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> return d <EOL> def __setstate__ ( self , state ) : <EOL> self . __dict__ = state <EOL> def getContext ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _context is None : <EOL> self . _context = self . _makeContext ( ) <EOL> return self . _context <EOL> def _makeContext ( self ) : <EOL> ctx = self . _contextFactory ( self . method ) <EOL> ctx . set_options ( self . _options ) <EOL> if self . certificate is not None and self . privateKey is not None : <EOL> ctx . use_certificate ( self . certificate ) <EOL> ctx . use_privatekey ( self . privateKey ) <EOL> for extraCert in self . extraCertChain : <EOL> ctx . add_extra_chain_cert ( extraCert ) <EOL> ctx . check_privatekey ( ) <EOL> verifyFlags = SSL . VERIFY_NONE <EOL> if self . verify : <EOL> verifyFlags = SSL . VERIFY_PEER <EOL> if self . requireCertificate : <EOL> verifyFlags |= SSL . VERIFY_FAIL_IF_NO_PEER_CERT <EOL> if self . verifyOnce : <EOL> verifyFlags |= SSL . VERIFY_CLIENT_ONCE <EOL> self . trustRoot . _addCACertsToContext ( ctx ) <EOL> def _verifyCallback ( conn , cert , errno , depth , preverify_ok ) : <EOL> return preverify_ok <EOL> ctx . set_verify ( verifyFlags , _verifyCallback ) <EOL> if self . verifyDepth is not None : <EOL> ctx . set_verify_depth ( self . verifyDepth ) <EOL> if self . enableSessions : <EOL> name = "<STR_LIT>" % ( reflect . qual ( self . __class__ ) , _sessionCounter ( ) ) <EOL> sessionName = md5 ( networkString ( name ) ) . hexdigest ( ) <EOL> ctx . set_session_id ( sessionName ) <EOL> if self . dhParameters : <EOL> ctx . load_tmp_dh ( self . dhParameters . _dhFile . path ) <EOL> ctx . set_cipher_list ( nativeString ( self . _cipherString ) ) <EOL> if self . _ecCurve is not None : <EOL> try : <EOL> self . _ecCurve . addECKeyToContext ( ctx ) <EOL> except BaseException : <EOL> pass <EOL> return ctx <EOL> OpenSSLCertificateOptions . __getstate__ = deprecated ( <EOL> Version ( "<STR_LIT>" , <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:0> ) , <EOL> "<STR_LIT>" ) ( OpenSSLCertificateOptions . __getstate__ ) <EOL> OpenSSLCertificateOptions . __setstate__ = deprecated ( <EOL> Version ( "<STR_LIT>" , <NUM_LIT> , <NUM_LIT:1> , <NUM_LIT:0> ) , <EOL> "<STR_LIT>" ) ( OpenSSLCertificateOptions . __setstate__ ) <EOL> class _OpenSSLECCurve ( FancyEqMixin , object ) : <EOL> """<STR_LIT>""" <EOL> compareAttributes = ( "<STR_LIT>" , ) <EOL> def __init__ ( self , snName ) : <EOL> """<STR_LIT>""" <EOL> self . snName = nativeString ( snName ) <EOL> try : <EOL> binding = self . _getBinding ( ) <EOL> self . _lib = binding . lib <EOL> self . _ffi = binding . ffi <EOL> self . _nid = self . _lib . OBJ_sn2nid ( self . snName . encode ( '<STR_LIT:ascii>' ) ) <EOL> if self . _nid == self . _lib . NID_undef : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> except AttributeError : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" <EOL> ) <EOL> def _getBinding ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> from OpenSSL . _util import binding <EOL> return binding <EOL> except ImportError : <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" <EOL> ) <EOL> def addECKeyToContext ( self , context ) : <EOL> """<STR_LIT>""" <EOL> ecKey = self . _lib . EC_KEY_new_by_curve_name ( self . _nid ) <EOL> if ecKey == self . _ffi . NULL : <EOL> raise EnvironmentError ( "<STR_LIT>" ) <EOL> self . _lib . SSL_CTX_set_tmp_ecdh ( context . _context , ecKey ) <EOL> self . _lib . EC_KEY_free ( ecKey ) <EOL> @ implementer ( ICipher ) <EOL> class OpenSSLCipher ( FancyEqMixin , object ) : <EOL> """<STR_LIT>""" <EOL> compareAttributes = ( '<STR_LIT>' , ) <EOL> def __init__ ( self , fullName ) : <EOL> """<STR_LIT>""" <EOL> self . fullName = fullName <EOL> def __repr__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' . format ( self . fullName ) <EOL> def _expandCipherString ( cipherString , method , options ) : <EOL> """<STR_LIT>""" <EOL> ctx = SSL . Context ( method ) <EOL> ctx . set_options ( options ) <EOL> try : <EOL> ctx . set_cipher_list ( nativeString ( cipherString ) ) <EOL> except SSL . Error as e : <EOL> if e . args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] == '<STR_LIT>' : <EOL> return [ ] <EOL> else : <EOL> raise <EOL> conn = SSL . Connection ( ctx , None ) <EOL> ciphers = conn . get_cipher_list ( ) <EOL> if isinstance ( ciphers [ <NUM_LIT:0> ] , unicode ) : <EOL> return [ OpenSSLCipher ( cipher ) for cipher in ciphers ] <EOL> else : <EOL> return [ OpenSSLCipher ( cipher . decode ( '<STR_LIT:ascii>' ) ) for cipher in ciphers ] <EOL> @ implementer ( IAcceptableCiphers ) <EOL> class OpenSSLAcceptableCiphers ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , ciphers ) : <EOL> self . _ciphers = ciphers <EOL> def selectCiphers ( self , availableCiphers ) : <EOL> return [ cipher <EOL> for cipher in self . _ciphers <EOL> if cipher in availableCiphers ] <EOL> @ classmethod <EOL> def fromOpenSSLCipherString ( cls , cipherString ) : <EOL> """<STR_LIT>""" <EOL> return cls ( _expandCipherString ( <EOL> nativeString ( cipherString ) , <EOL> SSL . SSLv23_METHOD , SSL . OP_NO_SSLv2 | SSL . OP_NO_SSLv3 ) <EOL> ) <EOL> defaultCiphers = OpenSSLAcceptableCiphers . fromOpenSSLCipherString ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> _defaultCurveName = u"<STR_LIT>" <EOL> class OpenSSLDiffieHellmanParameters ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , parameters ) : <EOL> self . _dhFile = parameters <EOL> @ classmethod <EOL> def fromFile ( cls , filePath ) : <EOL> """<STR_LIT>""" <EOL> return cls ( filePath ) </s>
