<s> import base64 <EOL> from oslo_config import cfg <EOL> from oslo_log import log as oslo_logging <EOL> from cloudbaseinit . osutils import factory as osutils_factory <EOL> from cloudbaseinit . plugins . common import base <EOL> from cloudbaseinit . plugins . common import constants <EOL> from cloudbaseinit . utils import crypt <EOL> CLEAR_TEXT_INJECTED_ONLY = '<STR_LIT>' <EOL> ALWAYS_CHANGE = '<STR_LIT>' <EOL> NEVER_CHANGE = '<STR_LIT>' <EOL> LOGON_PASSWORD_CHANGE_OPTIONS = [ <EOL> CLEAR_TEXT_INJECTED_ONLY , <EOL> NEVER_CHANGE , <EOL> ALWAYS_CHANGE , <EOL> ] <EOL> opts = [ <EOL> cfg . BoolOpt ( '<STR_LIT>' , default = True , help = '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> cfg . StrOpt ( '<STR_LIT>' , <EOL> default = CLEAR_TEXT_INJECTED_ONLY , <EOL> choices = LOGON_PASSWORD_CHANGE_OPTIONS , <EOL> help = '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> ] <EOL> CONF = cfg . CONF <EOL> CONF . register_opts ( opts ) <EOL> CONF . import_opt ( '<STR_LIT:username>' , '<STR_LIT>' ) <EOL> LOG = oslo_logging . getLogger ( __name__ ) <EOL> class SetUserPasswordPlugin ( base . BasePlugin ) : <EOL> def _encrypt_password ( self , ssh_pub_key , password ) : <EOL> cm = crypt . CryptManager ( ) <EOL> with cm . load_ssh_rsa_public_key ( ssh_pub_key ) as rsa : <EOL> enc_password = rsa . public_encrypt ( password . encode ( ) ) <EOL> return base64 . b64encode ( enc_password ) <EOL> def _get_ssh_public_key ( self , service ) : <EOL> public_keys = service . get_public_keys ( ) <EOL> if public_keys : <EOL> return list ( public_keys ) [ <NUM_LIT:0> ] <EOL> def _get_password ( self , service , shared_data ) : <EOL> injected = False <EOL> if CONF . inject_user_password : <EOL> password = service . get_admin_password ( ) <EOL> else : <EOL> password = None <EOL> if password : <EOL> injected = True <EOL> LOG . warn ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> else : <EOL> password = shared_data . get ( constants . SHARED_DATA_PASSWORD ) <EOL> return password , injected <EOL> def _set_metadata_password ( self , password , service ) : <EOL> if service . is_password_set : <EOL> LOG . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return True <EOL> else : <EOL> ssh_pub_key = self . _get_ssh_public_key ( service ) <EOL> if ssh_pub_key : <EOL> enc_password_b64 = self . _encrypt_password ( ssh_pub_key , <EOL> password ) <EOL> return service . post_password ( enc_password_b64 ) <EOL> else : <EOL> LOG . info ( '<STR_LIT>' ) <EOL> return True <EOL> def _set_password ( self , service , osutils , user_name , shared_data ) : <EOL> """<STR_LIT>""" <EOL> if service . can_update_password and not service . is_password_changed ( ) : <EOL> LOG . info ( '<STR_LIT>' ) <EOL> return None <EOL> password , injected = self . _get_password ( service , shared_data ) <EOL> if not password : <EOL> LOG . debug ( '<STR_LIT>' ) <EOL> maximum_length = osutils . get_maximum_password_length ( ) <EOL> password = osutils . generate_random_password ( <EOL> maximum_length ) <EOL> osutils . set_user_password ( user_name , password ) <EOL> self . _change_logon_behaviour ( user_name , password_injected = injected ) <EOL> return password <EOL> def _change_logon_behaviour ( self , username , password_injected = False ) : <EOL> """<STR_LIT>""" <EOL> if CONF . first_logon_behaviour == NEVER_CHANGE : <EOL> return <EOL> clear_text = CONF . first_logon_behaviour == CLEAR_TEXT_INJECTED_ONLY <EOL> always = CONF . first_logon_behaviour == ALWAYS_CHANGE <EOL> if always or ( clear_text and password_injected ) : <EOL> osutils = osutils_factory . get_os_utils ( ) <EOL> osutils . change_password_next_logon ( username ) <EOL> def execute ( self , service , shared_data ) : <EOL> user_name = shared_data . get ( constants . SHARED_DATA_USERNAME , <EOL> CONF . username ) <EOL> osutils = osutils_factory . get_os_utils ( ) <EOL> if osutils . user_exists ( user_name ) : <EOL> password = self . _set_password ( service , osutils , <EOL> user_name , shared_data ) <EOL> if password : <EOL> LOG . info ( '<STR_LIT>' % <EOL> user_name ) <EOL> shared_data [ constants . SHARED_DATA_PASSWORD ] = password <EOL> if not service . can_post_password : <EOL> LOG . info ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> else : <EOL> self . _set_metadata_password ( password , service ) <EOL> if service . can_update_password : <EOL> return base . PLUGIN_EXECUTE_ON_NEXT_BOOT , False <EOL> else : <EOL> return base . PLUGIN_EXECUTION_DONE , False </s>
