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", {})\...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!