<s> """<STR_LIT>""" <EOL> import os <EOL> import glob <EOL> from oslo_config import cfg <EOL> from st2common import log as logging <EOL> from st2common . content . loader import MetaLoader <EOL> from st2common . models . api . rbac import RoleDefinitionFileFormatAPI <EOL> from st2common . models . api . rbac import UserRoleAssignmentFileFormatAPI <EOL> from st2common . util . misc import compare_path_file_name <EOL> LOG = logging . getLogger ( __name__ ) <EOL> __all__ = [ <EOL> '<STR_LIT>' <EOL> ] <EOL> class RBACDefinitionsLoader ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> base_path = cfg . CONF . system . base_path <EOL> rbac_definitions_path = os . path . join ( base_path , '<STR_LIT>' ) <EOL> self . _role_definitions_path = os . path . join ( rbac_definitions_path , '<STR_LIT>' ) <EOL> self . _role_assignments_path = os . path . join ( rbac_definitions_path , '<STR_LIT>' ) <EOL> self . _meta_loader = MetaLoader ( ) <EOL> def load ( self ) : <EOL> """<STR_LIT>""" <EOL> result = { } <EOL> result [ '<STR_LIT>' ] = self . load_role_definitions ( ) <EOL> result [ '<STR_LIT>' ] = self . load_user_role_assignments ( ) <EOL> return result <EOL> def load_role_definitions ( self ) : <EOL> """<STR_LIT>""" <EOL> LOG . info ( '<STR_LIT>' % ( self . _role_definitions_path ) ) <EOL> file_paths = self . _get_role_definitions_file_paths ( ) <EOL> result = { } <EOL> for file_path in file_paths : <EOL> LOG . debug ( '<STR_LIT>' % ( file_path ) ) <EOL> role_definition_api = self . load_role_definition_from_file ( file_path = file_path ) <EOL> role_name = role_definition_api . name <EOL> enabled = getattr ( role_definition_api , '<STR_LIT>' , True ) <EOL> if role_name in result : <EOL> raise ValueError ( '<STR_LIT>' % ( role_name ) ) <EOL> if not enabled : <EOL> LOG . debug ( '<STR_LIT>' % ( role_name ) ) <EOL> continue <EOL> result [ role_name ] = role_definition_api <EOL> return result <EOL> def load_user_role_assignments ( self ) : <EOL> """<STR_LIT>""" <EOL> LOG . info ( '<STR_LIT>' % ( self . _role_assignments_path ) ) <EOL> file_paths = self . _get_role_assiginments_file_paths ( ) <EOL> result = { } <EOL> for file_path in file_paths : <EOL> LOG . debug ( '<STR_LIT>' % ( file_path ) ) <EOL> role_assignment_api = self . load_user_role_assignments_from_file ( file_path = file_path ) <EOL> username = role_assignment_api . username <EOL> enabled = getattr ( role_assignment_api , '<STR_LIT>' , True ) <EOL> if username in result : <EOL> raise ValueError ( '<STR_LIT>' % ( username ) ) <EOL> if not enabled : <EOL> LOG . debug ( '<STR_LIT>' % ( username ) ) <EOL> continue <EOL> result [ username ] = role_assignment_api <EOL> return result <EOL> def load_role_definition_from_file ( self , file_path ) : <EOL> """<STR_LIT>""" <EOL> content = self . _meta_loader . load ( file_path ) <EOL> if not content : <EOL> msg = ( '<STR_LIT>' % file_path ) <EOL> raise ValueError ( msg ) <EOL> role_definition_api = RoleDefinitionFileFormatAPI ( ** content ) <EOL> role_definition_api . validate ( ) <EOL> return role_definition_api <EOL> def load_user_role_assignments_from_file ( self , file_path ) : <EOL> """<STR_LIT>""" <EOL> content = self . _meta_loader . load ( file_path ) <EOL> if not content : <EOL> msg = ( '<STR_LIT>' % file_path ) <EOL> raise ValueError ( msg ) <EOL> user_role_assignment_api = UserRoleAssignmentFileFormatAPI ( ** content ) <EOL> user_role_assignment_api . validate ( ) <EOL> return user_role_assignment_api <EOL> def _get_role_definitions_file_paths ( self ) : <EOL> """<STR_LIT>""" <EOL> glob_str = self . _role_definitions_path + '<STR_LIT>' <EOL> file_paths = glob . glob ( glob_str ) <EOL> file_paths = sorted ( file_paths , cmp = compare_path_file_name ) <EOL> return file_paths <EOL> def _get_role_assiginments_file_paths ( self ) : <EOL> """<STR_LIT>""" <EOL> glob_str = self . _role_assignments_path + '<STR_LIT>' <EOL> file_paths = glob . glob ( glob_str ) <EOL> file_paths = sorted ( file_paths , cmp = compare_path_file_name ) <EOL> return file_paths </s>
