Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.