How to use iam_client method in localstack

Best Python code snippet using localstack_python

index.py

Source:index.py

1"""2Custom resource function to create/delete a GreengrassGroupRole with specific permissions3"""4import json5import os6import logging7import time8import boto39from botocore.exceptions import ClientError10__copyright__ = (11    "Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved."12)13__license__ = "MIT-0"14logger = logging.getLogger()15logger.setLevel(logging.INFO)16def create_greengrass_group_role(role_name: str, policy: str):17    """Creates a Greengrass Group Role with Greengrass managed role plus the additional18    policy statements from JSON policy statements19    """20    iam_client = boto3.client("iam")21    assume_role_policy_document = json.dumps(22        {23            "Version": "2012-10-17",24            "Statement": [25                {26                    "Effect": "Allow",27                    "Principal": {"Service": "greengrass.amazonaws.com"},28                    "Action": "sts:AssumeRole",29                }30            ],31        }32    )33    # Create IAM role and attach policies (managed and provided)34    try:35        # Create role and allow Greengrass to assume36        response = iam_client.create_role(37            RoleName=role_name, AssumeRolePolicyDocument=assume_role_policy_document38        )39        role_arn = response["Role"]["Arn"]40    except ClientError as e:41        logger.warning(42            f"Error calling iam.create_role() for role {role_name}, error: {e}"43        )44        return False45    try:46        # Apply general resource policy (limited iot, greengrass, and other services)47        iam_client.attach_role_policy(48            RoleName=role_name,49            PolicyArn="arn:aws:iam::aws:policy/service-role/AWSGreengrassResourceAccessRolePolicy",50        )51    except ClientError as e:52        logger.warning(53            f"Error calling iam_client.attach_role_policy() for role {role_name}, error: {e}"54        )55        return False56    try:57        # Apply inline policy to role58        iam_client.put_role_policy(59            RoleName=role_name,60            PolicyName="AdditionalGreengrassGroupPermissions",61            PolicyDocument=json.dumps(policy),62        )63    except ClientError as e:64        logger.warning(65            f"Error calling iam_client.put_role_policy() for role {role_name}, error: {e}"66        )67        return False68    return role_arn69def delete_greengrass_group_role(role_name: str):70    """Delete the GreengrassGroupRole"""71    iam_client = boto3.client("iam")72    try:73        # Get and delete attached inline policies74        policies = iam_client.list_role_policies(RoleName=role_name)["PolicyNames"]75        for policy in policies:76            iam_client.delete_role_policy(RoleName=role_name, PolicyName=policy)77    except ClientError as e:78        logger.warning(79            f"Error deleting inline policies for role {role_name}, error: {e}"80        )81        return False82    try:83        # Get and delete attached managed policies84        policies = iam_client.list_attached_role_policies(RoleName=role_name)[85            "AttachedPolicies"86        ]87        for policy in policies:88            iam_client.detach_role_policy(89                RoleName=role_name, PolicyArn=policy["PolicyArn"]90            )91    except ClientError as e:92        logger.warning(93            f"Error deleting managed policies for role {role_name}, error: {e}"94        )95        return False96    try:97        # Delete the role98        iam_client.delete_role(RoleName=role_name)99    except ClientError as e:100        logger.warning(101            f"Error calling iam_client.delete_role() for role {role_name}, error: {e}"102        )103        return False104    return True105def main(event, context):106    import logging as log107    import cfnresponse108    # NOTE: All ResourceProperties passed will uppercase the first letter109    #       of the property and leave the rest of the case intact.110    physical_id = event["ResourceProperties"]["PhysicalId"]111    cfn_response = cfnresponse.SUCCESS112    try:113        logger.info("Input event: %s", event)114        # Check if this is a Create and we're failing Creates115        if event["RequestType"] == "Create" and event["ResourceProperties"].get(116            "FailCreate", False117        ):118            raise RuntimeError("Create failure requested, logging")119        elif event["RequestType"] == "Create":120            # Operations to perform during Create, then return response_data121            role_arn = create_greengrass_group_role(122                role_name=event["ResourceProperties"]["RoleName"],123                policy=event["ResourceProperties"]["RolePolicy"],124            )125            if role_arn:126                response_data = {"roleArn": role_arn}127                logger.info(f"Created and returning roleArn: {role_arn}")128            else:129                logger.error("should not get here")130                cfn_response = cfnresponse.FAILED131                response_data = {}132        elif event["RequestType"] == "Update":133            # Operations to perform during Update, then return NULL for response data134            response_data = {}135        else:136            # Delete request137            # Operations to perform during Delete, then return response_data138            if not delete_greengrass_group_role(139                role_name=event["ResourceProperties"]["RoleName"]140            ):141                cfn_response = cfnresponse.FAILED142            response_data = {}143        cfnresponse.send(event, context, cfn_response, response_data, physical_id)144    except Exception as e:145        log.exception(e)146        # cfnresponse error message is always "see CloudWatch"...

iam.py

Source:iam.py

1import json2class IAM(object):3    def __init__(self, iam_client):4        self.iam_client = iam_client5    def check_if_role_exists(self, role_name):6        """Method to verify if a particular role exists"""7        try:8            self.iam_client.get_role(RoleName=role_name)9        except self.iam_client.exceptions.NoSuchEntityException:10            return False11        return True12    def check_if_policy_exists(self, policy_arn):13        """Method to verify if a particular policy exists"""14        try:15            self.iam_client.get_policy(PolicyArn=policy_arn)16        except self.iam_client.exceptions.NoSuchEntityException:17            return False18        return True19    def attach_policy_to_role(self, policy_arn, role_name):20        """Method to attach LifecyclePolicy to role specified by role_name"""21        return self.iam_client.attach_role_policy(22            PolicyArn=policy_arn,23            RoleName=role_name24        )25    def create_role_with_trust_policy(self, role_name, assume_role_policy):26        """Method to create role with a given role name27            and assume_role_policy28        """29        return self.iam_client.create_role(30            RoleName=role_name,31            AssumeRolePolicyDocument=json.dumps(assume_role_policy))32    def get_policy(self, arn):33        """Method to get the Policy for a particular ARN34        This is used to display the policy contents to the user35        """36        pol_det = self.iam_client.get_policy(PolicyArn=arn)37        policy_version_details = self.iam_client.get_policy_version(38            PolicyArn=arn,39            VersionId=pol_det.get("Policy", {}).get("DefaultVersionId", "")40        )41        return policy_version_details\42            .get("PolicyVersion", {})\...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.