<s> import itertools <EOL> from OpenSSL import SSL , crypto <EOL> from twisted . python import reflect , util <EOL> from twisted . python . hashlib import md5 <EOL> from twisted . internet . defer import Deferred <EOL> from twisted . internet . error import VerifyError , CertificateError <EOL> _sessionCounter = itertools . count ( ) . next <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 . iteritems ( ) : <EOL> setattr ( self , k , v ) <EOL> def _copyFrom ( self , x509name ) : <EOL> d = { } <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 . iteritems ( ) : <EOL> setattr ( x509name , k , 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> assert type ( attr ) is str <EOL> if not attr in _x509names : <EOL> raise AttributeError ( "<STR_LIT>" % ( attr , ) ) <EOL> realAttr = _x509names [ attr ] <EOL> value = value . encode ( '<STR_LIT:ascii>' ) <EOL> assert type ( value ) is str <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 . itervalues ( ) ) <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 , 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> 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 _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>' % 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> options = dict ( privateKey = self . privateKey . original , <EOL> certificate = self . original ) <EOL> if authorities : <EOL> options . update ( dict ( verify = True , <EOL> requireCertificate = True , <EOL> caCerts = [ auth . original for auth in authorities ] ) ) <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> def __init__ ( self , osslpkey ) : <EOL> self . original = osslpkey <EOL> req1 = crypto . X509Req ( ) <EOL> req1 . set_pubkey ( osslpkey ) <EOL> self . _emptyReq = crypto . dump_certificate_request ( crypto . FILETYPE_ASN1 , req1 ) <EOL> def matches ( self , otherKey ) : <EOL> return self . _emptyReq == otherKey . _emptyReq <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . __class__ . __name__ , self . keyHash ( ) ) <EOL> def keyHash ( self ) : <EOL> """<STR_LIT>""" <EOL> return md5 ( self . _emptyReq ) . 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> dn = requestObject . getSubject ( ) <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> class OpenSSLCertificateOptions ( object ) : <EOL> """<STR_LIT>""" <EOL> _context = None <EOL> _OP_ALL = getattr ( SSL , '<STR_LIT>' , <NUM_LIT> ) <EOL> _OP_NO_TICKET = <NUM_LIT> <EOL> method = SSL . TLSv1_METHOD <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> """<STR_LIT>""" <EOL> assert ( privateKey is None ) == ( certificate is None ) , "<STR_LIT>" <EOL> self . privateKey = privateKey <EOL> self . certificate = certificate <EOL> if method is not None : <EOL> self . method = method <EOL> self . verify = verify <EOL> assert ( ( verify and caCerts ) or <EOL> ( not verify ) ) , "<STR_LIT>" <EOL> self . caCerts = caCerts <EOL> self . verifyDepth = verifyDepth <EOL> self . requireCertificate = requireCertificate <EOL> self . verifyOnce = verifyOnce <EOL> self . enableSingleUseKeys = enableSingleUseKeys <EOL> self . enableSessions = enableSessions <EOL> self . fixBrokenPeers = fixBrokenPeers <EOL> self . enableSessionTickets = enableSessionTickets <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 = SSL . Context ( self . method ) <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> 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> if self . caCerts : <EOL> store = ctx . get_cert_store ( ) <EOL> for cert in self . caCerts : <EOL> store . add_cert ( cert ) <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 . enableSingleUseKeys : <EOL> ctx . set_options ( SSL . OP_SINGLE_DH_USE ) <EOL> if self . fixBrokenPeers : <EOL> ctx . set_options ( self . _OP_ALL ) <EOL> if self . enableSessions : <EOL> sessionName = md5 ( "<STR_LIT>" % ( reflect . qual ( self . __class__ ) , _sessionCounter ( ) ) ) . hexdigest ( ) <EOL> ctx . set_session_id ( sessionName ) <EOL> if not self . enableSessionTickets : <EOL> ctx . set_options ( self . _OP_NO_TICKET ) <EOL> return ctx </s>
