<s> """<STR_LIT>""" <EOL> import array <EOL> import base64 <EOL> import binascii <EOL> from barbicanclient import client as barbican_client <EOL> from keystoneauth1 import loading as ks_loading <EOL> from keystoneauth1 import session <EOL> from oslo_log import log as logging <EOL> from oslo_utils import excutils <EOL> import nova . conf <EOL> from nova import exception <EOL> from nova . i18n import _ <EOL> from nova . i18n import _LE <EOL> from nova . keymgr import key as keymgr_key <EOL> from nova . keymgr import key_mgr <EOL> CONF = nova . conf . CONF <EOL> LOG = logging . getLogger ( __name__ ) <EOL> class BarbicanKeyManager ( key_mgr . KeyManager ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _barbican_client = None <EOL> self . _current_context = None <EOL> self . _base_url = None <EOL> def _get_barbican_client ( self , ctxt ) : <EOL> """<STR_LIT>""" <EOL> if not ctxt : <EOL> msg = _ ( "<STR_LIT>" ) <EOL> LOG . error ( msg ) <EOL> raise exception . Forbidden ( msg ) <EOL> if not hasattr ( ctxt , '<STR_LIT>' ) or ctxt . project_id is None : <EOL> msg = _ ( "<STR_LIT>" ) <EOL> LOG . error ( msg ) <EOL> raise exception . KeyManagerError ( msg ) <EOL> if self . _barbican_client and self . _current_context == ctxt : <EOL> return self . _barbican_client <EOL> try : <EOL> _SESSION = ks_loading . load_session_from_conf_options ( <EOL> CONF , <EOL> nova . conf . barbican . barbican_group . name ) <EOL> auth = ctxt . get_auth_plugin ( ) <EOL> service_type , service_name , interface = ( CONF . <EOL> barbican . <EOL> catalog_info . <EOL> split ( '<STR_LIT::>' ) ) <EOL> region_name = CONF . barbican . os_region_name <EOL> service_parameters = { '<STR_LIT>' : service_type , <EOL> '<STR_LIT>' : service_name , <EOL> '<STR_LIT>' : interface , <EOL> '<STR_LIT>' : region_name } <EOL> if CONF . barbican . endpoint_template : <EOL> self . _base_url = ( CONF . barbican . endpoint_template % <EOL> ctxt . to_dict ( ) ) <EOL> else : <EOL> self . _base_url = _SESSION . get_endpoint ( <EOL> auth , ** service_parameters ) <EOL> self . _barbican_endpoint = self . _base_url . rpartition ( '<STR_LIT:/>' ) [ <NUM_LIT:0> ] <EOL> sess = session . Session ( auth = auth ) <EOL> self . _barbican_client = barbican_client . Client ( <EOL> session = sess , <EOL> endpoint = self . _barbican_endpoint ) <EOL> self . _current_context = ctxt <EOL> except Exception as e : <EOL> with excutils . save_and_reraise_exception ( ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , e ) <EOL> return self . _barbican_client <EOL> def create_key ( self , ctxt , expiration = None , name = '<STR_LIT>' , <EOL> payload_content_type = '<STR_LIT>' , mode = '<STR_LIT>' , <EOL> algorithm = '<STR_LIT>' , length = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> barbican_client = self . _get_barbican_client ( ctxt ) <EOL> try : <EOL> key_order = barbican_client . orders . create_key ( <EOL> name , <EOL> algorithm , <EOL> length , <EOL> mode , <EOL> payload_content_type , <EOL> expiration ) <EOL> order_ref = key_order . submit ( ) <EOL> order = barbican_client . orders . get ( order_ref ) <EOL> return self . _retrieve_secret_uuid ( order . secret_ref ) <EOL> except Exception as e : <EOL> with excutils . save_and_reraise_exception ( ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , e ) <EOL> def store_key ( self , ctxt , key , expiration = None , name = '<STR_LIT>' , <EOL> payload_content_type = '<STR_LIT>' , <EOL> payload_content_encoding = '<STR_LIT>' , algorithm = '<STR_LIT>' , <EOL> bit_length = <NUM_LIT> , mode = '<STR_LIT>' , from_copy = False ) : <EOL> """<STR_LIT>""" <EOL> barbican_client = self . _get_barbican_client ( ctxt ) <EOL> try : <EOL> if key . get_algorithm ( ) : <EOL> algorithm = key . get_algorithm ( ) <EOL> if payload_content_type == '<STR_LIT>' : <EOL> payload_content_encoding = None <EOL> encoded_key = key . get_encoded ( ) <EOL> elif ( payload_content_type == '<STR_LIT>' and <EOL> not from_copy ) : <EOL> key_list = key . get_encoded ( ) <EOL> string_key = '<STR_LIT>' . join ( map ( lambda byte : "<STR_LIT>" % byte , key_list ) ) <EOL> encoded_key = base64 . b64encode ( binascii . unhexlify ( string_key ) ) <EOL> else : <EOL> encoded_key = key . get_encoded ( ) <EOL> secret = barbican_client . secrets . create ( name , <EOL> encoded_key , <EOL> payload_content_type , <EOL> payload_content_encoding , <EOL> algorithm , <EOL> bit_length , <EOL> mode , <EOL> expiration ) <EOL> secret_ref = secret . store ( ) <EOL> return self . _retrieve_secret_uuid ( secret_ref ) <EOL> except Exception as e : <EOL> with excutils . save_and_reraise_exception ( ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , e ) <EOL> def copy_key ( self , ctxt , key_id ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> secret = self . _get_secret ( ctxt , key_id ) <EOL> con_type = secret . content_types [ '<STR_LIT:default>' ] <EOL> secret_data = self . _get_secret_data ( secret , <EOL> payload_content_type = con_type ) <EOL> key = keymgr_key . SymmetricKey ( secret . algorithm , secret_data ) <EOL> copy_uuid = self . store_key ( ctxt , key , secret . expiration , <EOL> secret . name , con_type , <EOL> '<STR_LIT>' , <EOL> secret . algorithm , secret . bit_length , <EOL> secret . mode , True ) <EOL> return copy_uuid <EOL> except Exception as e : <EOL> with excutils . save_and_reraise_exception ( ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , e ) <EOL> def _create_secret_ref ( self , key_id ) : <EOL> """<STR_LIT>""" <EOL> if not key_id : <EOL> msg = "<STR_LIT>" <EOL> raise exception . KeyManagerError ( msg ) <EOL> return self . _base_url + "<STR_LIT>" + key_id <EOL> def _retrieve_secret_uuid ( self , secret_ref ) : <EOL> """<STR_LIT>""" <EOL> return secret_ref . rpartition ( '<STR_LIT:/>' ) [ <NUM_LIT:2> ] <EOL> def _get_secret_data ( self , <EOL> secret , <EOL> payload_content_type = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> generated_data = secret . payload <EOL> if payload_content_type == '<STR_LIT>' : <EOL> secret_data = base64 . b64encode ( generated_data ) <EOL> else : <EOL> secret_data = generated_data <EOL> return secret_data <EOL> except Exception as e : <EOL> with excutils . save_and_reraise_exception ( ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , e ) <EOL> def _get_secret ( self , ctxt , key_id ) : <EOL> """<STR_LIT>""" <EOL> barbican_client = self . _get_barbican_client ( ctxt ) <EOL> try : <EOL> secret_ref = self . _create_secret_ref ( key_id ) <EOL> return barbican_client . secrets . get ( secret_ref ) <EOL> except Exception as e : <EOL> with excutils . save_and_reraise_exception ( ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , e ) <EOL> def get_key ( self , ctxt , key_id , <EOL> payload_content_type = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> secret = self . _get_secret ( ctxt , key_id ) <EOL> secret_data = self . _get_secret_data ( secret , <EOL> payload_content_type ) <EOL> if payload_content_type == '<STR_LIT>' : <EOL> key_data = array . array ( '<STR_LIT:B>' , <EOL> base64 . b64decode ( secret_data ) ) . tolist ( ) <EOL> else : <EOL> key_data = secret_data <EOL> key = keymgr_key . SymmetricKey ( secret . algorithm , key_data ) <EOL> return key <EOL> except Exception as e : <EOL> with excutils . save_and_reraise_exception ( ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , e ) <EOL> def delete_key ( self , ctxt , key_id ) : <EOL> """<STR_LIT>""" <EOL> barbican_client = self . _get_barbican_client ( ctxt ) <EOL> try : <EOL> secret_ref = self . _create_secret_ref ( key_id ) <EOL> barbican_client . secrets . delete ( secret_ref ) <EOL> except Exception as e : <EOL> with excutils . save_and_reraise_exception ( ) : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) , e ) </s>
