KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use put_resource_policy_response method in localstack
Best Python code snippet using localstack_python
deploy-monitoring.py
Source:deploy-monitoring.py 
1'''2This lambda deploys a small monitoring stack to each region that listens for3EC2 and Lambda creation, and upon detection logs in the detected region and4alerts to a central event bus in the home region.5'''6import boto37import os8config_client = boto3.client('config')9iam_client = boto3.client('iam')10s3_client = boto3.client('s3')11ec2_client = boto3.client('ec2')12logs = {}13logs['errors'] = []14def deploy_logs(region):15 logs[region] = {'region_name': region}16 logs[region]['http_codes'] = {}17 # Creates the log group needed for logging if it doesn't already exist18 logs[region]['logs'] = {}19 logs[region]['logs']['log_group'] = {}20 logs_client = boto3.client('logs', region_name=region)21 lg_list = logs_client.describe_log_groups()22 lg_names = []23 resource_arn = f"arn:aws:logs:{region}:{os.environ['ACCOUNT']}:log-group:/aws/events/*:*"24 policy_doc = '''{25 "Statement": [26 {27 "Action": [28 "logs:CreateLogStream",29 "logs:PutLogEvents"30 ],31 "Effect": "Allow",32 "Principal": {33 "Service": ["events.amazonaws.com", "delivery.logs.amazonaws.com"]34 },35 "Resource": '''36 policy_doc = policy_doc + '\"' + resource_arn + '\"' + ''',37 "Sid": "TrustEventsToStoreLogEvent"38 }39 ],40 "Version": "2012-10-17"41 }'''42 resource_policy_args = {43 'policyName': 'TrustEventsToStoreLogEvents',44 'policyDocument': policy_doc45 }46 lgs = lg_list['logGroups']47 for lg in lgs:48 lg_names.append(lg['logGroupName'])49 if '/aws/events/log-all-unauthorized-activity' in lg_names:50 logs[region]['logs']['log_group'] = 'log_group_exists'51 describe_resource_policies = logs_client.describe_resource_policies()52 logs[region]['logs']['log_resource_policy'] = ''53 for resource in describe_resource_policies['resourcePolicies']:54 if resource['policyName'] == 'TrustEventsToStoreLogEvents':55 logs[region]['logs']['log_resource_policy'] = 'log_resource_policy_exists'56 if hasattr(logs[region]['logs'], 'log_resource_policy') == False:57 put_resource_policy_response = logs_client.put_resource_policy(58 **resource_policy_args)59 logs[region]['http_codes']['put_resource_policy_response'] = put_resource_policy_response['ResponseMetadata']['HTTPStatusCode']60 else:61 create_log_group_response = logs_client.create_log_group(62 logGroupName='/aws/events/log-all-unauthorized-activity'63 )64 logs[region]['http_codes']['create_log_group_response'] = create_log_group_response['ResponseMetadata']['HTTPStatusCode']65 put_resource_policy_response = logs_client.put_resource_policy(66 **resource_policy_args)67 logs[region]['http_codes']['put_resource_policy_response'] = put_resource_policy_response['ResponseMetadata']['HTTPStatusCode']68 return logs69def deploy_config(region, config_client):70 # Deploys config to monitor lambdas71 # Install the config recorder72 config_args = {73 'ConfigurationRecorder': {74 'name': 'config-recorder',75 'roleARN': f"arn:aws:iam::{os.environ['ACCOUNT']}:role/aws-service-role/config.amazonaws.com/AWSServiceRoleForConfig",76 'recordingGroup': {77 'allSupported': False,78 'resourceTypes': [79 'AWS::Lambda::Function'80 ]81 }82 }83 }84 put_config_response = config_client.put_configuration_recorder(85 **config_args)86 logs[region]['http_codes']['put_config_response'] = put_config_response['ResponseMetadata']['HTTPStatusCode']87 # Install the delivery channel88 put_delivery_channel_response = config_client.put_delivery_channel(89 DeliveryChannel={90 'name': 'default',91 's3BucketName': f"config-bucket-{os.environ['ACCOUNT']}",92 'configSnapshotDeliveryProperties': {93 'deliveryFrequency': 'TwentyFour_Hours'94 }95 }96 )97 logs[region]['http_codes']['put_delivery_channel_response'] = put_delivery_channel_response['ResponseMetadata']['HTTPStatusCode']98 # Start the config recorder99 put_config_recorder_response = config_client.start_configuration_recorder(100 ConfigurationRecorderName='config-recorder'101 )102 logs[region]['http_codes']['put_config_recorder_response'] = put_config_recorder_response['ResponseMetadata']['HTTPStatusCode']103 return logs104def deploy_eventbridge(region):105 # Deploys eventbridge rules and targets106 # Deploy the eventbridge EC2 rule107 eventbridge_client = boto3.client('events', region_name=region)108 EC2_rule_name = f"unauthorized-EC2-{region}"109 EC2_rule_args = {110 'Name': EC2_rule_name,111 'EventPattern': '''{112 "source": ["aws.ec2"],113 "detail-type": ["EC2 Instance State-change Notification"],114 "detail": {115 "state": ["running"]116 } 117 }'''118 }119 put_EC2_rule_response = eventbridge_client.put_rule(**EC2_rule_args)120 logs[region]['http_codes']['put_EC2_rule_response'] = put_EC2_rule_response['ResponseMetadata']['HTTPStatusCode']121 # Deploy the eventbridge EC2 targets122 log_arn = f"arn:aws:logs:{region}:{os.environ['ACCOUNT']}:log-group:/aws/events/log-all-unauthorized-activity"123 EC2_target_args = {124 'Rule': EC2_rule_name,125 'Targets': [{126 'Id': 'unauthorized-activity-rule',127 'Arn': f"arn:aws:events:us-east-1:{os.environ['ACCOUNT']}:event-bus/unauthorized-activity",128 'RoleArn': f"arn:aws:iam::{os.environ['ACCOUNT']}:role/service-role/Amazon_EventBridge_Invoke_Event_Bus_2083104951"129 },130 {131 'Id': 'log-all-unauthorized-activity',132 'Arn': log_arn,133 }]134 }135 put_EC2_targets_response = eventbridge_client.put_targets(136 **EC2_target_args)137 logs[region]['http_codes']['put_EC2_targets_response'] = put_EC2_targets_response['ResponseMetadata']['HTTPStatusCode']138 # Deploy the eventbridge lambda rule139 lambda_rule_name = f"unauthorized-lambda-{region}"140 resource_name = f"arn:aws:cloudwatch:{region}:{os.environ['ACCOUNT']}:alarm:unauthorized-lambda-{region}"141 lambda_rule_args = {142 'Name': lambda_rule_name,143 'EventPattern': '''{144 "source": ["aws.config"],145 "detail": {146 "configurationItem": {147 "configurationItemStatus": ["ResourceDiscovered"],148 "resourceType": ["AWS::Lambda::Function"]149 }150 }151 }'''152 }153 put_lambda_rule_response = eventbridge_client.put_rule(**lambda_rule_args)154 logs[region]['http_codes']['put_lambda_rule_response'] = put_lambda_rule_response['ResponseMetadata']['HTTPStatusCode']155 # Deploy the eventbridge lambda targets156 lambda_target_args = {157 'Rule': lambda_rule_name,158 'Targets': [{159 'Id': 'unauthorized-activity-rule',160 'Arn': f"arn:aws:events:us-east-1:{os.environ['ACCOUNT']}:event-bus/unauthorized-activity",161 'RoleArn': f"arn:aws:iam::{os.environ['ACCOUNT']}:role/service-role/Amazon_EventBridge_Invoke_Event_Bus_2083104951"162 },163 {164 'Id': 'log-all-unauthorized-activity',165 'Arn': log_arn,166 }]167 }168 put_lambda_targets_response = eventbridge_client.put_targets(169 **lambda_target_args)170 logs[region]['http_codes']['put_lambda_targets_response'] = put_lambda_targets_response['ResponseMetadata']['HTTPStatusCode']171 return logs172def deploy_monitoring(region):173 # Deploy the cloudwatch log group:174 logs = deploy_logs(region)175 # Deploy the config recorder:176 logs[region]['config'] = {}177 config_client = boto3.client('config', region_name=region)178 config_list_response = config_client.describe_configuration_recorders()179 # If our recorder already exists, log it and do nothing180 if config_list_response['ConfigurationRecorders'] and config_list_response['ConfigurationRecorders'][0]['name'] == 'config-recorder':181 logs[region]['config']['existing_config_recorders'] = config_list_response['ConfigurationRecorders'][0]['name']182 # if there's an existing config recorder other than ours we delete it and deploy ours:183 elif config_list_response['ConfigurationRecorders']:184 delete_configuration_recorder_response = config_client.delete_configuration_recorder(185 ConfigurationRecorderName=config_list_response['ConfigurationRecorders'][0]['name']186 )187 logs[region]['config']['delete_configuration_recorder_response'] = delete_configuration_recorder_response188 logs = deploy_config(region, config_client)189 # if there's no existing recorder we deploy ours:190 else:191 logs = deploy_config(region, config_client)192 # Deploy eventbridge:193 deploy_eventbridge(region)194 # Set the success message, and change it to fail if any response codes for operations were not 2xx:195 logs[region].update({f"{region}-deploy": "succeeded"})196 for msg in logs[region]['http_codes']:197 code = logs[region]['http_codes'][msg]198 if code < 200 or code > 299:199 logs[region][f"{region}-deploy"] = 'failed'200 print(logs[region])201 return logs[region]202 print(logs[region])203 return logs[region]204def lambda_handler(event, context):205 try:206 home_region = os.environ['HOME_REGION']207 except KeyError:208 logs['errors'].append({209 'keyerror': 'env variable HOME_REGION must be set, to avoid deleting resources in home region'210 })211 return {212 'statusCode': 405,213 'body': logs214 }215 # Get the list of all regions, excluding home region216 regions = []217 response = ec2_client.describe_regions(AllRegions=True)218 region_entries = response['Regions']219 for ind_region in region_entries:220 regions.append(ind_region['RegionName'])221 regions.remove(home_region)222 # Deploy in all regions but home region223 for region in regions:224 logs.update(deploy_monitoring(region))225 if logs[region][f"{region}-deploy"] == 'failed':226 return {227 'statusCode': 405,228 'body': logs229 }230 print(logs)231 return {232 'statusCode': 200,233 'body': logs...
deploy-in-home-region.py
Source:deploy-in-home-region.py
1import json2import boto33import os4config_client = boto3.client('config')5iam_client = boto3.client('iam')6s3_client = boto3.client('s3')7ec2_client = boto3.client('ec2')8logs = {}9logs['errors'] = []10def deploy_logs(region):11 logs[region] = {'region_name': region}12 logs[region]['http_codes'] = {}13 # Creates the log group needed for logging if it doesn't already exist14 logs[region]['logs'] = {}15 logs[region]['logs']['log_group'] = {}16 logs_client = boto3.client('logs', region_name=region)17 lg_list = logs_client.describe_log_groups()18 lg_names = []19 resource_arn = 'arn:aws:logs:' + region + ':' + \20 os.environ['ACCOUNT'] + ':log-group:/aws/events/*:*'21 policy_doc = '''{22 "Statement": [23 {24 "Action": [25 "logs:CreateLogStream",26 "logs:PutLogEvents"27 ],28 "Effect": "Allow",29 "Principal": {30 "Service": ["events.amazonaws.com", "delivery.logs.amazonaws.com"]31 },32 "Resource": '''33 policy_doc = policy_doc + '\"' + resource_arn + '\"' + ''',34 "Sid": "TrustEventsToStoreLogEvent"35 }36 ],37 "Version": "2012-10-17"38 }'''39 resource_policy_args = {40 'policyName': 'TrustEventsToStoreLogEvents',41 'policyDocument': policy_doc42 }43 lgs = lg_list['logGroups']44 for lg in lgs:45 lg_names.append(lg['logGroupName'])46 if '/aws/events/log-all-unauthorized-activity' in lg_names:47 print('yes')48 logs[region]['logs']['log_group'] = 'log_group_exists'49 describe_resource_policies = logs_client.describe_resource_policies()50 print(describe_resource_policies)51 logs[region]['logs']['log_resource_policy'] = ''52 for resource in describe_resource_policies['resourcePolicies']:53 if resource['policyName'] == 'TrustEventsToStoreLogEvents':54 logs[region]['logs']['log_resource_policy'] = 'log_resource_policy_exists'55 # We seem to be having issues here:56 if hasattr(logs[region]['logs'], 'log_resource_policy') == False:57 put_resource_policy_response = logs_client.put_resource_policy(58 **resource_policy_args)59 logs[region]['http_codes']['put_resource_policy_response'] = put_resource_policy_response['ResponseMetadata']['HTTPStatusCode']60 else:61 create_log_group_response = logs_client.create_log_group(62 logGroupName='/aws/events/log-all-unauthorized-activity'63 )64 logs[region]['http_codes']['create_log_group_response'] = create_log_group_response['ResponseMetadata']['HTTPStatusCode']65 put_resource_policy_response = logs_client.put_resource_policy(66 **resource_policy_args)67 logs[region]['http_codes']['put_resource_policy_response'] = put_resource_policy_response['ResponseMetadata']['HTTPStatusCode']68 return logs69def deploy_config(region, config_client):70 # Deploys config to monitor lambdas71 # Install the config recorder72 config_args = {73 'ConfigurationRecorder': {74 'name': 'config-recorder',75 'roleARN': 'arn:aws:iam::502245549462:role/aws-service-role/config.amazonaws.com/AWSServiceRoleForConfig',76 'recordingGroup': {77 'allSupported': False,78 'resourceTypes': [79 'AWS::Elasticsearch::Domain', 'AWS::IAM::Group', 'AWS::IAM::Policy', 'AWS::IAM::Role', 'AWS::IAM::User', 'AWS::ElasticLoadBalancingV2::LoadBalancer', 'AWS::ACM::Certificate', 'AWS::RDS::DBInstance', 'AWS::RDS::DBSubnetGroup', 'AWS::RDS::DBSecurityGroup', 'AWS::RDS::DBSnapshot', 'AWS::RDS::DBCluster', 'AWS::RDS::DBClusterSnapshot', 'AWS::RDS::EventSubscription', 'AWS::S3::Bucket', 'AWS::S3::AccountPublicAccessBlock', 'AWS::Redshift::Cluster', 'AWS::Redshift::ClusterSnapshot', 'AWS::Redshift::ClusterParameterGroup', 'AWS::Redshift::ClusterSecurityGroup', 'AWS::Redshift::ClusterSubnetGroup', 'AWS::Redshift::EventSubscription', 'AWS::SSM::ManagedInstanceInventory', 'AWS::CloudWatch::Alarm', 'AWS::CloudFormation::Stack', 'AWS::ElasticLoadBalancing::LoadBalancer', 'AWS::AutoScaling::AutoScalingGroup', 'AWS::AutoScaling::LaunchConfiguration', 'AWS::AutoScaling::ScalingPolicy', 'AWS::AutoScaling::ScheduledAction', 'AWS::DynamoDB::Table', 'AWS::CodeBuild::Project', 'AWS::WAF::RateBasedRule', 'AWS::WAF::Rule', 'AWS::WAF::RuleGroup', 'AWS::WAF::WebACL', 'AWS::WAFRegional::RateBasedRule', 'AWS::WAFRegional::Rule', 'AWS::WAFRegional::RuleGroup', 'AWS::WAFRegional::WebACL', 'AWS::CloudFront::Distribution', 'AWS::CloudFront::StreamingDistribution', 'AWS::Lambda::Function', 'AWS::NetworkFirewall::Firewall', 'AWS::NetworkFirewall::FirewallPolicy', 'AWS::NetworkFirewall::RuleGroup', 'AWS::ElasticBeanstalk::Application', 'AWS::ElasticBeanstalk::ApplicationVersion', 'AWS::ElasticBeanstalk::Environment', 'AWS::WAFv2::WebACL', 'AWS::WAFv2::RuleGroup', 'AWS::WAFv2::IPSet', 'AWS::WAFv2::RegexPatternSet', 'AWS::WAFv2::ManagedRuleSet', 'AWS::XRay::EncryptionConfig', 'AWS::SSM::AssociationCompliance', 'AWS::SSM::PatchCompliance', 'AWS::Shield::Protection', 'AWS::ShieldRegional::Protection', 'AWS::Config::ConformancePackCompliance', 'AWS::Config::ResourceCompliance', 'AWS::ApiGateway::Stage', 'AWS::ApiGateway::RestApi', 'AWS::ApiGatewayV2::Stage', 'AWS::ApiGatewayV2::Api', 'AWS::CodePipeline::Pipeline', 'AWS::ServiceCatalog::CloudFormationProvisionedProduct', 'AWS::ServiceCatalog::CloudFormationProduct', 'AWS::ServiceCatalog::Portfolio', 'AWS::SQS::Queue', 'AWS::KMS::Key', 'AWS::QLDB::Ledger', 'AWS::SecretsManager::Secret', 'AWS::SNS::Topic', 'AWS::SSM::FileData', 'AWS::Backup::BackupPlan', 'AWS::Backup::BackupSelection', 'AWS::Backup::BackupVault', 'AWS::Backup::RecoveryPoint', 'AWS::ECR::Repository', 'AWS::ECS::Cluster', 'AWS::ECS::Service', 'AWS::ECS::TaskDefinition', 'AWS::EFS::AccessPoint', 'AWS::EFS::FileSystem', 'AWS::EKS::Cluster', 'AWS::OpenSearch::Domain', 'AWS::EC2::TransitGateway', 'AWS::Kinesis::Stream', 'AWS::Kinesis::StreamConsumer', 'AWS::CodeDeploy::Application', 'AWS::CodeDeploy::DeploymentConfig', 'AWS::CodeDeploy::DeploymentGroup'80 ]81 }82 }83 }84 put_config_response = config_client.put_configuration_recorder(85 **config_args)86 logs[region]['http_codes']['put_config_response'] = put_config_response['ResponseMetadata']['HTTPStatusCode']87 # Install the delivery channel88 put_delivery_channel_response = config_client.put_delivery_channel(89 DeliveryChannel={90 'name': 'default',91 's3BucketName': 'config-bucket-502245549462',92 'configSnapshotDeliveryProperties': {93 'deliveryFrequency': 'TwentyFour_Hours'94 }95 }96 )97 logs[region]['http_codes']['put_delivery_channel_response'] = put_delivery_channel_response['ResponseMetadata']['HTTPStatusCode']98 # Start the config recorder99 put_config_recorder_response = config_client.start_configuration_recorder(100 ConfigurationRecorderName='config-recorder'101 )102 logs[region]['http_codes']['put_config_recorder_response'] = put_config_recorder_response['ResponseMetadata']['HTTPStatusCode']103 return logs104def lambda_handler(event, context):105 try:106 home_region = os.environ['HOME_REGION']107 except KeyError:108 logs['errors'].append({109 'keyerror': 'env variable HOME_REGION must be set, to avoid deleting resources in home region'110 })111 return {112 'statusCode': 405,113 'body': logs114 }115 logs.update(deploy_monitoring(region))116 if logs[region][region + '-deploy'] == 'failed':117 return {118 'statusCode': 405,119 'body': logs120 }121 print(logs)122 return {123 'statusCode': 200,124 'body': logs...
secretsmanager_starter.py
Source:secretsmanager_starter.py
...73 else:74 raise SecretNotFoundException()75 if not hasattr(SecretsManagerBackend, 'put_resource_policy'):76 setattr(SecretsManagerBackend, 'put_resource_policy', put_resource_policy_model)77 def put_resource_policy_response(self):78 secret_id = self._get_param('SecretId')79 resource_policy = self._get_param('ResourcePolicy')80 return secretsmanager_backends[self.region].put_resource_policy(81 secret_id=secret_id,82 resource_policy=json.loads(resource_policy)83 )84 if not hasattr(SecretsManagerResponse, 'put_resource_policy'):85 setattr(SecretsManagerResponse, 'put_resource_policy', put_resource_policy_response)86def start_secretsmanager(port=None, asynchronous=None, backend_port=None, update_listener=None):87 apply_patches()88 return start_moto_server(89 key='secretsmanager',90 name='Secrets Manager',91 port=port,...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
