<s> from __future__ import unicode_literals <EOL> from moto . core import BaseBackend <EOL> from . exceptions import IAMNotFoundException , IAMConflictException , IAMReportNotPresentException <EOL> from . utils import random_access_key , random_alphanumeric , random_resource_id <EOL> from datetime import datetime <EOL> import base64 <EOL> class Role ( object ) : <EOL> def __init__ ( self , role_id , name , assume_role_policy_document , path ) : <EOL> self . id = role_id <EOL> self . name = name <EOL> self . assume_role_policy_document = assume_role_policy_document <EOL> self . path = path <EOL> self . policies = { } <EOL> @ classmethod <EOL> def create_from_cloudformation_json ( cls , resource_name , cloudformation_json , region_name ) : <EOL> properties = cloudformation_json [ '<STR_LIT>' ] <EOL> role = iam_backend . create_role ( <EOL> role_name = resource_name , <EOL> assume_role_policy_document = properties [ '<STR_LIT>' ] , <EOL> path = properties [ '<STR_LIT>' ] , <EOL> ) <EOL> policies = properties . get ( '<STR_LIT>' , [ ] ) <EOL> for policy in policies : <EOL> policy_name = policy [ '<STR_LIT>' ] <EOL> policy_json = policy [ '<STR_LIT>' ] <EOL> role . put_policy ( policy_name , policy_json ) <EOL> return role <EOL> def put_policy ( self , policy_name , policy_json ) : <EOL> self . policies [ policy_name ] = policy_json <EOL> @ property <EOL> def physical_resource_id ( self ) : <EOL> return self . id <EOL> def get_cfn_attribute ( self , attribute_name ) : <EOL> from moto . cloudformation . exceptions import UnformattedGetAttTemplateException <EOL> if attribute_name == '<STR_LIT>' : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> raise UnformattedGetAttTemplateException ( ) <EOL> class InstanceProfile ( object ) : <EOL> def __init__ ( self , instance_profile_id , name , path , roles ) : <EOL> self . id = instance_profile_id <EOL> self . name = name <EOL> self . path = path <EOL> self . roles = roles if roles else [ ] <EOL> @ classmethod <EOL> def create_from_cloudformation_json ( cls , resource_name , cloudformation_json , region_name ) : <EOL> properties = cloudformation_json [ '<STR_LIT>' ] <EOL> role_ids = properties [ '<STR_LIT>' ] <EOL> return iam_backend . create_instance_profile ( <EOL> name = resource_name , <EOL> path = properties [ '<STR_LIT>' ] , <EOL> role_ids = role_ids , <EOL> ) <EOL> @ property <EOL> def physical_resource_id ( self ) : <EOL> return self . name <EOL> def get_cfn_attribute ( self , attribute_name ) : <EOL> from moto . cloudformation . exceptions import UnformattedGetAttTemplateException <EOL> if attribute_name == '<STR_LIT>' : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> raise UnformattedGetAttTemplateException ( ) <EOL> class Certificate ( object ) : <EOL> def __init__ ( self , cert_name , cert_body , private_key , cert_chain = None , path = None ) : <EOL> self . cert_name = cert_name <EOL> self . cert_body = cert_body <EOL> self . private_key = private_key <EOL> self . path = path <EOL> self . cert_chain = cert_chain <EOL> @ property <EOL> def physical_resource_id ( self ) : <EOL> return self . name <EOL> class AccessKey ( object ) : <EOL> def __init__ ( self , user_name ) : <EOL> self . user_name = user_name <EOL> self . access_key_id = random_access_key ( ) <EOL> self . secret_access_key = random_alphanumeric ( <NUM_LIT:32> ) <EOL> self . status = '<STR_LIT>' <EOL> self . create_date = datetime . strftime ( <EOL> datetime . utcnow ( ) , <EOL> "<STR_LIT>" <EOL> ) <EOL> def get_cfn_attribute ( self , attribute_name ) : <EOL> from moto . cloudformation . exceptions import UnformattedGetAttTemplateException <EOL> if attribute_name == '<STR_LIT>' : <EOL> return self . secret_access_key <EOL> raise UnformattedGetAttTemplateException ( ) <EOL> class Group ( object ) : <EOL> def __init__ ( self , name , path = '<STR_LIT:/>' ) : <EOL> self . name = name <EOL> self . id = random_resource_id ( ) <EOL> self . path = path <EOL> self . created = datetime . strftime ( <EOL> datetime . utcnow ( ) , <EOL> "<STR_LIT>" <EOL> ) <EOL> self . users = [ ] <EOL> def get_cfn_attribute ( self , attribute_name ) : <EOL> from moto . cloudformation . exceptions import UnformattedGetAttTemplateException <EOL> if attribute_name == '<STR_LIT>' : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> raise UnformattedGetAttTemplateException ( ) <EOL> class User ( object ) : <EOL> def __init__ ( self , name , path = '<STR_LIT:/>' ) : <EOL> self . name = name <EOL> self . id = random_resource_id ( ) <EOL> self . path = path <EOL> self . created = datetime . strftime ( <EOL> datetime . utcnow ( ) , <EOL> "<STR_LIT>" <EOL> ) <EOL> self . arn = '<STR_LIT>' . format ( name ) <EOL> self . policies = { } <EOL> self . access_keys = [ ] <EOL> self . password = None <EOL> def get_policy ( self , policy_name ) : <EOL> policy_json = None <EOL> try : <EOL> policy_json = self . policies [ policy_name ] <EOL> except KeyError : <EOL> raise IAMNotFoundException ( "<STR_LIT>" . format ( policy_name ) ) <EOL> return { <EOL> '<STR_LIT>' : policy_name , <EOL> '<STR_LIT>' : policy_json , <EOL> '<STR_LIT>' : self . name , <EOL> } <EOL> def put_policy ( self , policy_name , policy_json ) : <EOL> self . policies [ policy_name ] = policy_json <EOL> def delete_policy ( self , policy_name ) : <EOL> if policy_name not in self . policies : <EOL> raise IAMNotFoundException ( "<STR_LIT>" . format ( policy_name ) ) <EOL> del self . policies [ policy_name ] <EOL> def create_access_key ( self ) : <EOL> access_key = AccessKey ( self . name ) <EOL> self . access_keys . append ( access_key ) <EOL> return access_key <EOL> def get_all_access_keys ( self ) : <EOL> return self . access_keys <EOL> def delete_access_key ( self , access_key_id ) : <EOL> for key in self . access_keys : <EOL> if key . access_key_id == access_key_id : <EOL> self . access_keys . remove ( key ) <EOL> break <EOL> else : <EOL> raise IAMNotFoundException ( "<STR_LIT>" . format ( access_key_id ) ) <EOL> def get_cfn_attribute ( self , attribute_name ) : <EOL> from moto . cloudformation . exceptions import UnformattedGetAttTemplateException <EOL> if attribute_name == '<STR_LIT>' : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> raise UnformattedGetAttTemplateException ( ) <EOL> def to_csv ( self ) : <EOL> date_format = '<STR_LIT>' <EOL> date_created = datetime . strptime ( self . created , '<STR_LIT>' ) <EOL> if not self . password : <EOL> password_enabled = '<STR_LIT:false>' <EOL> password_last_used = '<STR_LIT>' <EOL> else : <EOL> password_enabled = '<STR_LIT:true>' <EOL> password_last_used = '<STR_LIT>' <EOL> if len ( self . access_keys ) == <NUM_LIT:0> : <EOL> access_key_1_active = '<STR_LIT:false>' <EOL> access_key_1_last_rotated = '<STR_LIT>' <EOL> access_key_2_active = '<STR_LIT:false>' <EOL> access_key_2_last_rotated = '<STR_LIT>' <EOL> elif len ( self . access_keys ) == <NUM_LIT:1> : <EOL> access_key_1_active = '<STR_LIT:true>' <EOL> access_key_1_last_rotated = date_created . strftime ( date_format ) <EOL> access_key_2_active = '<STR_LIT:false>' <EOL> access_key_2_last_rotated = '<STR_LIT>' <EOL> else : <EOL> access_key_1_active = '<STR_LIT:true>' <EOL> access_key_1_last_rotated = date_created . strftime ( date_format ) <EOL> access_key_2_active = '<STR_LIT:true>' <EOL> access_key_2_last_rotated = date_created . strftime ( date_format ) <EOL> return '<STR_LIT>' . format ( self . name , <EOL> self . arn , <EOL> date_created . strftime ( date_format ) , <EOL> password_enabled , <EOL> password_last_used , <EOL> date_created . strftime ( date_format ) , <EOL> access_key_1_active , <EOL> access_key_1_last_rotated , <EOL> access_key_2_active , <EOL> access_key_2_last_rotated <EOL> ) <EOL> class IAMBackend ( BaseBackend ) : <EOL> def __init__ ( self ) : <EOL> self . instance_profiles = { } <EOL> self . roles = { } <EOL> self . certificates = { } <EOL> self . groups = { } <EOL> self . users = { } <EOL> self . credential_report = None <EOL> super ( IAMBackend , self ) . __init__ ( ) <EOL> def create_role ( self , role_name , assume_role_policy_document , path ) : <EOL> role_id = random_resource_id ( ) <EOL> role = Role ( role_id , role_name , assume_role_policy_document , path ) <EOL> self . roles [ role_id ] = role <EOL> return role <EOL> def get_role_by_id ( self , role_id ) : <EOL> return self . roles . get ( role_id ) <EOL> def get_role ( self , role_name ) : <EOL> for role in self . get_roles ( ) : <EOL> if role . name == role_name : <EOL> return role <EOL> raise IAMNotFoundException ( "<STR_LIT>" . format ( role_name ) ) <EOL> def get_roles ( self ) : <EOL> return self . roles . values ( ) <EOL> def put_role_policy ( self , role_name , policy_name , policy_json ) : <EOL> role = self . get_role ( role_name ) <EOL> role . put_policy ( policy_name , policy_json ) <EOL> def get_role_policy ( self , role_name , policy_name ) : <EOL> role = self . get_role ( role_name ) <EOL> for p , d in role . policies . items ( ) : <EOL> if p == policy_name : <EOL> return p , d <EOL> def list_role_policies ( self , role_name ) : <EOL> role = self . get_role ( role_name ) <EOL> return role . policies . keys ( ) <EOL> def create_instance_profile ( self , name , path , role_ids ) : <EOL> instance_profile_id = random_resource_id ( ) <EOL> roles = [ iam_backend . get_role_by_id ( role_id ) for role_id in role_ids ] <EOL> instance_profile = InstanceProfile ( instance_profile_id , name , path , roles ) <EOL> self . instance_profiles [ instance_profile_id ] = instance_profile <EOL> return instance_profile <EOL> def get_instance_profile ( self , profile_name ) : <EOL> for profile in self . get_instance_profiles ( ) : <EOL> if profile . name == profile_name : <EOL> return profile <EOL> def get_instance_profiles ( self ) : <EOL> return self . instance_profiles . values ( ) <EOL> def get_instance_profiles_for_role ( self , role_name ) : <EOL> found_profiles = [ ] <EOL> for profile in self . get_instance_profiles ( ) : <EOL> if len ( profile . roles ) > <NUM_LIT:0> : <EOL> if profile . roles [ <NUM_LIT:0> ] . name == role_name : <EOL> found_profiles . append ( profile ) <EOL> return found_profiles <EOL> def add_role_to_instance_profile ( self , profile_name , role_name ) : <EOL> profile = self . get_instance_profile ( profile_name ) <EOL> role = self . get_role ( role_name ) <EOL> profile . roles . append ( role ) <EOL> def get_all_server_certs ( self , marker = None ) : <EOL> return self . certificates . values ( ) <EOL> def upload_server_cert ( self , cert_name , cert_body , private_key , cert_chain = None , path = None ) : <EOL> certificate_id = random_resource_id ( ) <EOL> cert = Certificate ( cert_name , cert_body , private_key , cert_chain , path ) <EOL> self . certificates [ certificate_id ] = cert <EOL> return cert <EOL> def get_server_certificate ( self , name ) : <EOL> for key , cert in self . certificates . items ( ) : <EOL> if name == cert . cert_name : <EOL> return cert <EOL> def create_group ( self , group_name , path = '<STR_LIT:/>' ) : <EOL> if group_name in self . groups : <EOL> raise IAMConflictException ( "<STR_LIT>" . format ( group_name ) ) <EOL> group = Group ( group_name , path ) <EOL> self . groups [ group_name ] = group <EOL> return group <EOL> def get_group ( self , group_name , marker = None , max_items = None ) : <EOL> group = None <EOL> try : <EOL> group = self . groups [ group_name ] <EOL> except KeyError : <EOL> raise IAMNotFoundException ( "<STR_LIT>" . format ( group_name ) ) <EOL> return group <EOL> def list_groups ( self ) : <EOL> return self . groups . values ( ) <EOL> def get_groups_for_user ( self , user_name ) : <EOL> user = self . get_user ( user_name ) <EOL> groups = [ ] <EOL> for group in self . list_groups ( ) : <EOL> if user in group . users : <EOL> groups . append ( group ) <EOL> return groups <EOL> def create_user ( self , user_name , path = '<STR_LIT:/>' ) : <EOL> if user_name in self . users : <EOL> raise IAMConflictException ( "<STR_LIT>" , "<STR_LIT>" . format ( user_name ) ) <EOL> user = User ( user_name , path ) <EOL> self . users [ user_name ] = user <EOL> return user <EOL> def get_user ( self , user_name ) : <EOL> user = None <EOL> try : <EOL> user = self . users [ user_name ] <EOL> except KeyError : <EOL> raise IAMNotFoundException ( "<STR_LIT>" . format ( user_name ) ) <EOL> return user <EOL> def create_login_profile ( self , user_name , password ) : <EOL> user = self . get_user ( user_name ) <EOL> if user . password : <EOL> raise IAMConflictException ( "<STR_LIT>" . format ( user_name ) ) <EOL> user . password = password <EOL> def add_user_to_group ( self , group_name , user_name ) : <EOL> user = self . get_user ( user_name ) <EOL> group = self . get_group ( group_name ) <EOL> group . users . append ( user ) <EOL> def remove_user_from_group ( self , group_name , user_name ) : <EOL> group = self . get_group ( group_name ) <EOL> user = self . get_user ( user_name ) <EOL> try : <EOL> group . users . remove ( user ) <EOL> except ValueError : <EOL> raise IAMNotFoundException ( "<STR_LIT>" . format ( user_name , group_name ) ) <EOL> def get_user_policy ( self , user_name , policy_name ) : <EOL> user = self . get_user ( user_name ) <EOL> policy = user . get_policy ( policy_name ) <EOL> return policy <EOL> def put_user_policy ( self , user_name , policy_name , policy_json ) : <EOL> user = self . get_user ( user_name ) <EOL> user . put_policy ( policy_name , policy_json ) <EOL> def delete_user_policy ( self , user_name , policy_name ) : <EOL> user = self . get_user ( user_name ) <EOL> user . delete_policy ( policy_name ) <EOL> def create_access_key ( self , user_name = None ) : <EOL> user = self . get_user ( user_name ) <EOL> key = user . create_access_key ( ) <EOL> return key <EOL> def get_all_access_keys ( self , user_name , marker = None , max_items = None ) : <EOL> user = self . get_user ( user_name ) <EOL> keys = user . get_all_access_keys ( ) <EOL> return keys <EOL> def delete_access_key ( self , access_key_id , user_name ) : <EOL> user = self . get_user ( user_name ) <EOL> user . delete_access_key ( access_key_id ) <EOL> def delete_user ( self , user_name ) : <EOL> try : <EOL> del self . users [ user_name ] <EOL> except KeyError : <EOL> raise IAMNotFoundException ( "<STR_LIT>" . format ( user_name ) ) <EOL> def report_generated ( self ) : <EOL> return self . credential_report <EOL> def generate_report ( self ) : <EOL> self . credential_report = True <EOL> def get_credential_report ( self ) : <EOL> if not self . credential_report : <EOL> raise IAMReportNotPresentException ( "<STR_LIT>" ) <EOL> report = '<STR_LIT>' <EOL> for user in self . users : <EOL> report += self . users [ user ] . to_csv ( ) <EOL> return base64 . b64encode ( report . encode ( '<STR_LIT:ascii>' ) ) . decode ( '<STR_LIT:ascii>' ) <EOL> iam_backend = IAMBackend ( ) </s>
