<s> import functools <EOL> import os <EOL> import tempfile <EOL> import base64 <EOL> from tempfile import mkstemp <EOL> from OpenSSL import crypto <EOL> import M2Crypto <EOL> from M2Crypto import X509 <EOL> from openflow . optin_manager . sfa . util . faults import CertExpired , CertMissingParent , CertNotSignedByParent <EOL> glo_passphrase_callback = None <EOL> def set_passphrase_callback ( callback_func ) : <EOL> global glo_passphrase_callback <EOL> glo_passphrase_callback = callback_func <EOL> def set_passphrase ( passphrase ) : <EOL> set_passphrase_callback ( lambda k , s , x : passphrase ) <EOL> def test_passphrase ( string , passphrase ) : <EOL> try : <EOL> crypto . load_privatekey ( crypto . FILETYPE_PEM , string , ( lambda x : passphrase ) ) <EOL> return True <EOL> except : <EOL> return False <EOL> def convert_public_key ( key ) : <EOL> keyconvert_path = "<STR_LIT>" <EOL> if not os . path . isfile ( keyconvert_path ) : <EOL> raise IOError , "<STR_LIT>" % keyconvert_path <EOL> if "<STR_LIT>" in key : <EOL> raise Exception , "<STR_LIT>" <EOL> ( ssh_f , ssh_fn ) = tempfile . mkstemp ( ) <EOL> ssl_fn = tempfile . mktemp ( ) <EOL> os . write ( ssh_f , key ) <EOL> os . close ( ssh_f ) <EOL> cmd = keyconvert_path + "<STR_LIT:U+0020>" + ssh_fn + "<STR_LIT:U+0020>" + ssl_fn <EOL> os . system ( cmd ) <EOL> if not os . path . exists ( ssl_fn ) : <EOL> raise Exception , "<STR_LIT>" <EOL> k = Keypair ( ) <EOL> try : <EOL> k . load_pubkey_from_file ( ssl_fn ) <EOL> return k <EOL> except Exception as e : <EOL> raise e <EOL> finally : <EOL> if os . path . exists ( ssh_fn ) : <EOL> os . remove ( ssh_fn ) <EOL> if os . path . exists ( ssl_fn ) : <EOL> os . remove ( ssl_fn ) <EOL> class Keypair : <EOL> key = None <EOL> m2key = None <EOL> def __init__ ( self , create = False , string = None , filename = None ) : <EOL> if create : <EOL> self . create ( ) <EOL> if string : <EOL> self . load_from_string ( string ) <EOL> if filename : <EOL> self . load_from_file ( filename ) <EOL> def create ( self ) : <EOL> self . key = crypto . PKey ( ) <EOL> self . key . generate_key ( crypto . TYPE_RSA , <NUM_LIT> ) <EOL> def save_to_file ( self , filename ) : <EOL> open ( filename , '<STR_LIT:w>' ) . write ( self . as_pem ( ) ) <EOL> self . filename = filename <EOL> def load_from_file ( self , filename ) : <EOL> self . filename = filename <EOL> buffer = open ( filename , '<STR_LIT:r>' ) . read ( ) <EOL> self . load_from_string ( buffer ) <EOL> def load_from_string ( self , string ) : <EOL> if glo_passphrase_callback : <EOL> self . key = crypto . load_privatekey ( crypto . FILETYPE_PEM , string , functools . partial ( glo_passphrase_callback , self , string ) ) <EOL> self . m2key = M2Crypto . EVP . load_key_string ( string , functools . partial ( glo_passphrase_callback , self , string ) ) <EOL> else : <EOL> self . key = crypto . load_privatekey ( crypto . FILETYPE_PEM , string ) <EOL> self . m2key = M2Crypto . EVP . load_key_string ( string ) <EOL> def load_pubkey_from_file ( self , filename ) : <EOL> m2rsakey = M2Crypto . RSA . load_pub_key ( filename ) <EOL> self . m2key = M2Crypto . EVP . PKey ( ) <EOL> self . m2key . assign_rsa ( m2rsakey ) <EOL> m2name = M2Crypto . X509 . X509_Name ( ) <EOL> m2name . add_entry_by_txt ( field = "<STR_LIT>" , type = <NUM_LIT> , entry = "<STR_LIT>" , len = - <NUM_LIT:1> , loc = - <NUM_LIT:1> , set = <NUM_LIT:0> ) <EOL> m2x509 = M2Crypto . X509 . X509 ( ) <EOL> m2x509 . set_pubkey ( self . m2key ) <EOL> m2x509 . set_serial_number ( <NUM_LIT:0> ) <EOL> m2x509 . set_issuer_name ( m2name ) <EOL> m2x509 . set_subject_name ( m2name ) <EOL> ASN1 = M2Crypto . ASN1 . ASN1_UTCTIME ( ) <EOL> ASN1 . set_time ( <NUM_LIT> ) <EOL> m2x509 . set_not_before ( ASN1 ) <EOL> m2x509 . set_not_after ( ASN1 ) <EOL> m2x509 . set_version ( <NUM_LIT:2> ) <EOL> junk_key = Keypair ( create = True ) <EOL> m2x509 . sign ( pkey = junk_key . get_m2_pkey ( ) , md = "<STR_LIT>" ) <EOL> m2pem = m2x509 . as_pem ( ) <EOL> pyx509 = crypto . load_certificate ( crypto . FILETYPE_PEM , m2pem ) <EOL> self . key = pyx509 . get_pubkey ( ) <EOL> self . filename = filename <EOL> def load_pubkey_from_string ( self , string ) : <EOL> ( f , fn ) = tempfile . mkstemp ( ) <EOL> os . write ( f , string ) <EOL> os . close ( f ) <EOL> self . load_pubkey_from_file ( fn ) <EOL> os . remove ( fn ) <EOL> def as_pem ( self ) : <EOL> return crypto . dump_privatekey ( crypto . FILETYPE_PEM , self . key ) <EOL> def get_m2_pkey ( self ) : <EOL> if not self . m2key : <EOL> self . m2key = M2Crypto . EVP . load_key_string ( self . as_pem ( ) ) <EOL> return self . m2key <EOL> def get_pubkey_string ( self ) : <EOL> m2pkey = self . get_m2_pkey ( ) <EOL> return base64 . b64encode ( m2pkey . as_der ( ) ) <EOL> def get_openssl_pkey ( self ) : <EOL> return self . key <EOL> def is_same ( self , pkey ) : <EOL> return self . as_pem ( ) == pkey . as_pem ( ) <EOL> def sign_string ( self , data ) : <EOL> k = self . get_m2_pkey ( ) <EOL> k . sign_init ( ) <EOL> k . sign_update ( data ) <EOL> return base64 . b64encode ( k . sign_final ( ) ) <EOL> def verify_string ( self , data , sig ) : <EOL> k = self . get_m2_pkey ( ) <EOL> k . verify_init ( ) <EOL> k . verify_update ( data ) <EOL> return M2Crypto . m2 . verify_final ( k . ctx , base64 . b64decode ( sig ) , k . pkey ) <EOL> def compute_hash ( self , value ) : <EOL> return self . sign_string ( str ( value ) ) <EOL> def get_filename ( self ) : <EOL> return getattr ( self , '<STR_LIT:filename>' , None ) <EOL> def dump ( self , * args , ** kwargs ) : <EOL> print self . dump_string ( * args , ** kwargs ) <EOL> def dump_string ( self ) : <EOL> result = "<STR_LIT>" <EOL> result += "<STR_LIT>" % self . get_pubkey_string ( ) <EOL> filename = self . get_filename ( ) <EOL> if filename : result += "<STR_LIT>" % filename <EOL> return result <EOL> class Certificate : <EOL> digest = "<STR_LIT>" <EOL> cert = None <EOL> issuerKey = None <EOL> issuerSubject = None <EOL> parent = None <EOL> isCA = None <EOL> separator = "<STR_LIT>" <EOL> def __init__ ( self , lifeDays = <NUM_LIT> , create = False , subject = None , string = None , filename = None , isCA = None ) : <EOL> self . data = { } <EOL> if create or subject : <EOL> self . create ( lifeDays ) <EOL> if subject : <EOL> self . set_subject ( subject ) <EOL> if string : <EOL> self . load_from_string ( string ) <EOL> if filename : <EOL> self . load_from_file ( filename ) <EOL> if isCA != None : <EOL> self . set_is_ca ( isCA ) <EOL> def create ( self , lifeDays = <NUM_LIT> ) : <EOL> self . cert = crypto . X509 ( ) <EOL> self . cert . set_serial_number ( <NUM_LIT:3> ) <EOL> self . cert . gmtime_adj_notBefore ( <NUM_LIT:0> ) <EOL> self . cert . gmtime_adj_notAfter ( lifeDays * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> ) <EOL> self . cert . set_version ( <NUM_LIT:2> ) <EOL> def load_from_pyopenssl_x509 ( self , x509 ) : <EOL> self . cert = x509 <EOL> def load_from_string ( self , string ) : <EOL> string = string . strip ( ) <EOL> if string . count ( '<STR_LIT>' ) == <NUM_LIT:0> : <EOL> string = '<STR_LIT>' % string <EOL> beg = string . find ( '<STR_LIT>' ) <EOL> if beg > <NUM_LIT:0> : <EOL> string = string [ beg : ] <EOL> parts = [ ] <EOL> if string . count ( '<STR_LIT>' ) > <NUM_LIT:1> and string . count ( Certificate . separator ) == <NUM_LIT:0> : <EOL> parts = string . split ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> parts [ <NUM_LIT:0> ] += '<STR_LIT>' <EOL> else : <EOL> parts = string . split ( Certificate . separator , <NUM_LIT:1> ) <EOL> self . cert = crypto . load_certificate ( crypto . FILETYPE_PEM , parts [ <NUM_LIT:0> ] ) <EOL> if len ( parts ) > <NUM_LIT:1> and parts [ <NUM_LIT:1> ] != '<STR_LIT>' : <EOL> self . parent = self . __class__ ( ) <EOL> self . parent . load_from_string ( parts [ <NUM_LIT:1> ] ) <EOL> def load_from_file ( self , filename ) : <EOL> file = open ( filename ) <EOL> string = file . read ( ) <EOL> self . load_from_string ( string ) <EOL> self . filename = filename <EOL> def save_to_string ( self , save_parents = True ) : <EOL> string = crypto . dump_certificate ( crypto . FILETYPE_PEM , self . cert ) <EOL> if save_parents and self . parent : <EOL> string = string + self . parent . save_to_string ( save_parents ) <EOL> return string <EOL> def save_to_file ( self , filename , save_parents = True , filep = None ) : <EOL> string = self . save_to_string ( save_parents = save_parents ) <EOL> if filep : <EOL> f = filep <EOL> else : <EOL> f = open ( filename , '<STR_LIT:w>' ) <EOL> f . write ( string ) <EOL> f . close ( ) <EOL> self . filename = filename <EOL> def save_to_random_tmp_file ( self , save_parents = True ) : <EOL> fp , filename = mkstemp ( suffix = '<STR_LIT>' , text = True ) <EOL> fp = os . fdopen ( fp , "<STR_LIT:w>" ) <EOL> self . save_to_file ( filename , save_parents = True , filep = fp ) <EOL> return filename <EOL> def set_issuer ( self , key , subject = None , cert = None ) : <EOL> self . issuerKey = key <EOL> if subject : <EOL> assert ( not cert ) <EOL> if isinstance ( subject , dict ) or isinstance ( subject , str ) : <EOL> req = crypto . X509Req ( ) <EOL> reqSubject = req . get_subject ( ) <EOL> if ( isinstance ( subject , dict ) ) : <EOL> for key in subject . keys ( ) : <EOL> setattr ( reqSubject , key , subject [ key ] ) <EOL> else : <EOL> setattr ( reqSubject , "<STR_LIT>" , subject ) <EOL> subject = reqSubject <EOL> self . issuerReq = req <EOL> if cert : <EOL> subject = cert . cert . get_subject ( ) <EOL> assert ( subject ) <EOL> self . issuerSubject = subject <EOL> def get_issuer ( self , which = "<STR_LIT>" ) : <EOL> x = self . cert . get_issuer ( ) <EOL> return getattr ( x , which ) <EOL> def set_subject ( self , name ) : <EOL> req = crypto . X509Req ( ) <EOL> subj = req . get_subject ( ) <EOL> if ( isinstance ( name , dict ) ) : <EOL> for key in name . keys ( ) : <EOL> setattr ( subj , key , name [ key ] ) <EOL> else : <EOL> setattr ( subj , "<STR_LIT>" , name ) <EOL> self . cert . set_subject ( subj ) <EOL> def get_subject ( self , which = "<STR_LIT>" ) : <EOL> x = self . cert . get_subject ( ) <EOL> return getattr ( x , which ) <EOL> def get_extended_subject ( self ) : <EOL> x = self . cert . get_subject ( ) <EOL> subject = dict ( ) <EOL> return dict ( x . get_components ( ) ) <EOL> def get_printable_subject ( self ) : <EOL> x = self . cert . get_subject ( ) <EOL> return "<STR_LIT>" % ( getattr ( x , "<STR_LIT>" ) , getattr ( x , "<STR_LIT>" ) , self . get_data ( ) ) <EOL> def set_pubkey ( self , key ) : <EOL> assert ( isinstance ( key , Keypair ) ) <EOL> self . cert . set_pubkey ( key . get_openssl_pkey ( ) ) <EOL> def get_pubkey ( self ) : <EOL> m2x509 = X509 . load_cert_string ( self . save_to_string ( ) ) <EOL> pkey = Keypair ( ) <EOL> pkey . key = self . cert . get_pubkey ( ) <EOL> pkey . m2key = m2x509 . get_pubkey ( ) <EOL> return pkey <EOL> def set_intermediate_ca ( self , val ) : <EOL> return self . set_is_ca ( val ) <EOL> def set_is_ca ( self , val ) : <EOL> if val is None : <EOL> return <EOL> if self . isCA != None : <EOL> raise Exception , "<STR_LIT>" % ( self . isCA , val ) <EOL> self . isCA = val <EOL> if val : <EOL> self . add_extension ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> else : <EOL> self . add_extension ( '<STR_LIT>' , <NUM_LIT:1> , '<STR_LIT>' ) <EOL> def add_extension ( self , name , critical , value ) : <EOL> oldExtVal = None <EOL> try : <EOL> oldExtVal = self . get_extension ( name ) <EOL> except : <EOL> pass <EOL> if oldExtVal and oldExtVal == value : <EOL> return <EOL> ext = crypto . X509Extension ( name , critical , value ) <EOL> self . cert . add_extensions ( [ ext ] ) <EOL> def get_extension ( self , name ) : <EOL> m2x509 = X509 . load_cert_string ( self . save_to_string ( ) ) <EOL> value = m2x509 . get_ext ( name ) . get_value ( ) <EOL> return value <EOL> def set_data ( self , str , field = '<STR_LIT>' ) : <EOL> if self . data . has_key ( field ) : <EOL> raise "<STR_LIT>" , field , "<STR_LIT>" <EOL> self . data [ field ] = str <EOL> self . add_extension ( field , <NUM_LIT:0> , str ) <EOL> def get_data ( self , field = '<STR_LIT>' ) : <EOL> if self . data . has_key ( field ) : <EOL> return self . data [ field ] <EOL> try : <EOL> uri = self . get_extension ( field ) <EOL> self . data [ field ] = uri <EOL> except LookupError : <EOL> return None <EOL> return self . data [ field ] <EOL> def sign ( self ) : <EOL> assert self . cert != None <EOL> assert self . issuerSubject != None <EOL> assert self . issuerKey != None <EOL> self . cert . set_issuer ( self . issuerSubject ) <EOL> self . cert . sign ( self . issuerKey . get_openssl_pkey ( ) , self . digest ) <EOL> def verify ( self , pkey ) : <EOL> m2x509 = X509 . load_cert_string ( self . save_to_string ( ) ) <EOL> m2pkey = pkey . get_m2_pkey ( ) <EOL> return m2x509 . verify ( m2pkey ) <EOL> def is_pubkey ( self , pkey ) : <EOL> return self . get_pubkey ( ) . is_same ( pkey ) <EOL> def is_signed_by_cert ( self , cert ) : <EOL> k = cert . get_pubkey ( ) <EOL> result = self . verify ( k ) <EOL> return result <EOL> def set_parent ( self , p ) : <EOL> self . parent = p <EOL> def get_parent ( self ) : <EOL> return self . parent <EOL> def verify_chain ( self , trusted_certs = None ) : <EOL> if self . cert . has_expired ( ) : <EOL> raise CertExpired ( self . get_printable_subject ( ) , "<STR_LIT>" ) <EOL> for trusted_cert in trusted_certs : <EOL> if self . is_signed_by_cert ( trusted_cert ) : <EOL> if not trusted_cert . cert . has_expired ( ) : <EOL> return trusted_cert <EOL> else : <EOL> raise CertExpired ( self . get_printable_subject ( ) , "<STR_LIT>" % trusted_cert . get_printable_subject ( ) ) <EOL> if not self . parent : <EOL> raise CertMissingParent ( self . get_printable_subject ( ) + "<STR_LIT>" % ( self . get_issuer ( ) , len ( trusted_certs ) ) ) <EOL> if not self . is_signed_by_cert ( self . parent ) : <EOL> raise CertNotSignedByParent ( "<STR_LIT>" % ( self . get_printable_subject ( ) , <EOL> self . parent . get_printable_subject ( ) , <EOL> self . get_issuer ( ) ) ) <EOL> if not self . parent . isCA and not ( self . parent . get_extension ( '<STR_LIT>' ) == '<STR_LIT>' ) : <EOL> raise CertNotSignedByParent ( "<STR_LIT>" % ( self . get_printable_subject ( ) , <EOL> self . parent . get_printable_subject ( ) ) ) <EOL> self . parent . verify_chain ( trusted_certs ) <EOL> return <EOL> def get_extensions ( self ) : <EOL> triples = [ ] <EOL> m2x509 = X509 . load_cert_string ( self . save_to_string ( ) ) <EOL> nb_extensions = m2x509 . get_ext_count ( ) <EOL> for i in range ( nb_extensions ) : <EOL> ext = m2x509 . get_ext_at ( i ) <EOL> triples . append ( ( ext . get_name ( ) , ext . get_value ( ) , ext . get_critical ( ) , ) ) <EOL> return triples <EOL> def get_data_names ( self ) : <EOL> return self . data . keys ( ) <EOL> def get_all_datas ( self ) : <EOL> triples = self . get_extensions ( ) <EOL> for name in self . get_data_names ( ) : <EOL> triples . append ( ( name , self . get_data ( name ) , '<STR_LIT:data>' , ) ) <EOL> return triples <EOL> def get_filename ( self ) : <EOL> return getattr ( self , '<STR_LIT:filename>' , None ) <EOL> def dump ( self , * args , ** kwargs ) : <EOL> print self . dump_string ( * args , ** kwargs ) <EOL> def dump_string ( self , show_extensions = False ) : <EOL> result = "<STR_LIT>" <EOL> result += "<STR_LIT>" % self . get_printable_subject ( ) <EOL> result += "<STR_LIT>" % self . get_issuer ( ) <EOL> filename = self . get_filename ( ) <EOL> if filename : result += "<STR_LIT>" % filename <EOL> if show_extensions : <EOL> all_datas = self . get_all_datas ( ) <EOL> result += "<STR_LIT>" % len ( all_datas ) <EOL> for ( n , v , c ) in all_datas : <EOL> if c == '<STR_LIT:data>' : <EOL> result += "<STR_LIT>" % ( n , v ) <EOL> else : <EOL> result += "<STR_LIT>" % ( n , c , v ) <EOL> return result </s>
