Source code for ecs_composex.s3.s3_aws

#  -*- coding: utf-8 -*-
# SPDX-License-Identifier: MPL-2.0
# Copyright 2020-2021 John Mille <john@compose-x.io>

"""
Functions to find buckets and identify settings about these.
"""

import re

from botocore.exceptions import ClientError
from compose_x_common.compose_x_common import keyisset

from ecs_composex.common import LOG
from ecs_composex.common.aws import (
    define_lookup_role_from_info,
    find_aws_resource_arn_from_tags_api,
)


[docs]def return_bucket_config(bucket_arn, session): """ :param str bucket_arn: :param boto3.session.Session session: :return: """ bucket_name_finder = re.compile(r"([a-zA-Z0-9.\-_]{1,255}$)") bucket_name = bucket_name_finder.findall(bucket_arn)[-1] bucket_config = {"Name": bucket_name, "Arn": bucket_arn} client = session.client("s3") try: client.head_bucket(Bucket=bucket_name) try: encryption_config_r = client.get_bucket_encryption(Bucket=bucket_name) if keyisset("ServerSideEncryptionConfiguration", encryption_config_r): bucket_config.update( { "ServerSideEncryptionConfiguration": encryption_config_r[ "ServerSideEncryptionConfiguration" ] } ) except ClientError as error: if ( not error.response["Error"]["Code"] == "ServerSideEncryptionConfigurationNotFoundError" ): raise LOG.warning(error.response["Error"]["Message"]) return bucket_config except client.exceptions.NoSuchBucket: return None except ClientError as error: LOG.error(error) raise
[docs]def lookup_bucket_config(lookup, session): """ Function to find the DB in AWS account :param dict lookup: The Lookup definition for DB :param boto3.session.Session session: Boto3 session for clients :return: """ s3_types = { "s3": {"regexp": r"(?:^arn:aws(?:-[a-z]+)?:s3:::)([\S]+)$"}, } lookup_session = define_lookup_role_from_info(lookup, session) bucket_arn = find_aws_resource_arn_from_tags_api( lookup, lookup_session, "s3", types=s3_types, ) if not bucket_arn: return None config = return_bucket_config(bucket_arn, lookup_session) LOG.debug(config) return config