KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_ebs_default_kms_key_id method in localstack

Best Python code snippet using localstack_python

account.py

Source:account.py

...1042                   self.data['key'] or {'c7n:AliasName': self.data['key']})1043            vf = KmsRelatedFilter(vfd, self.manager)1044            vf.RelatedIdsExpression = 'KmsKeyId'1045            vf.annotate = False1046            key = client.get_ebs_default_kms_key_id().get('KmsKeyId')1047            if not vf.process([{'KmsKeyId': key}]):1048                return []1049        return resources1050@actions.register('set-ebs-encryption')1051class SetEbsEncryption(BaseAction):1052    """Set AWS EBS default encryption on an account1053    :example:1054    .. code-block:: yaml1055       policies:1056         - name: set-default-ebs-encryption1057           resource: aws.account1058           filters:1059            - type: default-ebs-encryption1060              state: false...

ec2.py

Source:ec2.py

1import logging2from typing import Any, Dict, Generator, Iterator, List, Tuple3import botocore.exceptions4from introspector import ImportWriter, PathStack5from introspector.aws.fetch import Proxy, ServiceProxy6from introspector.aws.svc import RegionalService, ServiceSpec, resource_gate7_log = logging.getLogger(__name__)8def _synthesize_defaults(proxy: ServiceProxy,9                         region: str) -> Iterator[Tuple[str, Any]]:10  defaults = {'name': 'Defaults', 'uri': f'ec2/defaults/{region}'}11  # TODO: update permissions12  # key_id_resp = proxy.get('get_ebs_default_kms_key_id')13  # defaults['EbsDefaultKmsKeyId'] = key_id_resp.get('KmsKeyId')14  encryption_resp = proxy.get('get_ebs_encryption_by_default')15  defaults['EbsEncryptionByDefault'] = encryption_resp[16      'EbsEncryptionByDefault']17  yield 'Defaults', defaults18def _add_security_group_references(proxy: ServiceProxy, response: Dict):19  security_groups = response.get('SecurityGroups', [])20  for security_group in security_groups:21    group_id = security_group['GroupId']22    result = proxy.list('describe_security_group_references', GroupId=group_id)23    if result is not None:24      # everything is mutable, sigh...25      security_group['references'] = result[1].get('SecurityGroupReferenceSet',26                                                   [])27def _add_user_data(proxy: ServiceProxy, response: Dict):28  reservations = response.get('Reservations', [])29  for reservation in reservations:30    instances = reservation.get('Instances', [])31    for instance in instances:32      instance_id = instance['InstanceId']33      try:34        user_data = proxy.get('describe_instance_attribute',35                              InstanceId=instance_id,36                              Attribute='userData')37        if user_data is not None:38          instance['UserData'] = user_data.get('UserData', {}).get('Value')39      except botocore.exceptions.ClientError:40        pass41def _add_launch_permissions(proxy: ServiceProxy, response: Dict):42  snapshots = response.get('Snapshots', [])43  for snapshot in snapshots:44    snapshot_id = snapshot['SnapshotId']45    permission_resp = proxy.get('describe_snapshot_attribute',46                                SnapshotId=snapshot_id,47                                Attribute='createVolumePermission')48    permissions = permission_resp.get('CreateVolumePermissions', [])49    snapshot['CreateVolumePermissions'] = permissions50def _add_image_attributes(proxy: ServiceProxy, response: Dict[str, Any]):51  images = response.get('Images', [])52  for image in images:53    launch_permission = proxy.get('describe_image_attribute',54                                  Attribute='launchPermission',55                                  ImageId=image['ImageId'])56    if launch_permission is not None:57      image['LaunchPermissions'] = launch_permission.get(58          'LaunchPermissions', [])59    else:60      image['LaunchPermissions'] = []61RESOURCES = [62    'Addresses', 'FlowLogs', 'Images', 'Instances', 'KeyPairs',63    'NetworkInterfaces', 'RouteTables', 'SecurityGroups', 'Snapshots',64    'Subnets', 'Volumes', 'VpcPeeringConnections', 'Vpcs', 'VpcEndpoints'65]66def _import_ec2_region(67    proxy: ServiceProxy, region: str,68    spec: ServiceSpec) -> Generator[Tuple[str, Any], None, None]:69  for resource in proxy.resource_names():70    _log.info(f'importing {resource}')71    op_name = proxy._impl._client._PY_TO_OP_NAME[resource]72    op_resource = op_name[len('Describe'):]73    if op_resource in RESOURCES:74      if resource_gate(spec, op_resource):75        result = proxy.list(resource)76        if result is not None:77          if resource == 'describe_instances':78            _add_user_data(proxy, result[1])79          elif resource == 'describe_snapshots':80            _add_launch_permissions(proxy, result[1])81          elif resource == 'describe_images':82            _add_image_attributes(proxy, result[1])83          elif resource == 'describe_security_groups':84            _add_security_group_references(proxy, result[1])85          yield result[0], result[1]86        _log.info(f'done with {resource}')87  if resource_gate(spec, 'Defaults'):88    yield from _synthesize_defaults(proxy, region)89SVC = RegionalService('ec2', _import_ec2_region)90def add_amis_to_import_job(proxy: Proxy, writer: ImportWriter, ps: PathStack,91                           region: str, amis: List[str]) -> str:92  ps = ps.scope(region)93  service_proxy = proxy.service('ec2', region)94  result = service_proxy.list(95      'describe_images',96      ImageIds=amis,97      # Remove the default filters98      Filters=[])99  _log.debug(f'describe images result {result}')100  if result is not None:101    resource_name = result[0]102    raw_resources = result[1]103    # We can't add launch permissions here because we don't own the images104    #_add_image_attributes(service_proxy, raw_resources)105    writer(ps, resource_name, raw_resources, {'region': region})...

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.