Source code for ecs_composex.s3.s3_template

# SPDX-License-Identifier: MPL-2.0
# Copyright 2020-2022 John Mille <john@compose-x.io>

from __future__ import annotations

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from ecs_composex.s3.s3_bucket import Bucket
    from troposphere import NoValue
    from troposphere import Template

from compose_x_common.compose_x_common import keyisset
from troposphere import (
    AWS_ACCOUNT_ID,
    AWS_NO_VALUE,
    AWS_PARTITION,
    AWS_REGION,
    MAX_OUTPUTS,
    Ref,
    Sub,
    s3,
)

from ecs_composex.common.logging import LOG
from ecs_composex.common.stacks import ComposeXStack
from ecs_composex.common.troposphere_tools import add_outputs, build_template
from ecs_composex.resource_settings import generate_resource_permissions
from ecs_composex.resources_import import import_record_properties

COMPOSEX_MAX_OUTPUTS = MAX_OUTPUTS - 10


[docs]def define_bucket_name(bucket: Bucket) -> str | NoValue: """ Function to automatically add Region and Account ID to the bucket name. If set, will use a user-defined separator, else, `-` :param bucket: :return: The bucket name :rtype: str """ separator = ( bucket.settings["NameSeparator"] if keyisset("NameSeparator", bucket.parameters) and isinstance(bucket.parameters["NameSeparator"], str) else r"-" ) expand_region_key = "ExpandRegionToBucket" expand_account_id = "ExpandAccountIdToBucket" base_name = ( None if not keyisset("BucketName", bucket.properties) else bucket.properties["BucketName"] ) if base_name: if keyisset(expand_region_key, bucket.parameters) and keyisset( expand_account_id, bucket.parameters ): return f"{base_name}{separator}${{{AWS_ACCOUNT_ID}}}{separator}${{{AWS_REGION}}}" elif keyisset(expand_region_key, bucket.parameters) and not keyisset( expand_account_id, bucket.parameters ): return f"{base_name}{separator}${{{AWS_REGION}}}" elif not keyisset(expand_region_key, bucket.parameters) and keyisset( expand_account_id, bucket.parameters ): return f"{base_name}{separator}${{{AWS_ACCOUNT_ID}}}" elif not keyisset(expand_account_id, bucket.parameters) and not keyisset( expand_region_key, bucket.parameters ): LOG.warning( f"{base_name} - You defined the bucket without any extension. " "Bucket names must be unique. Make sure it is not already in-use" ) return base_name return Ref(AWS_NO_VALUE)
[docs]def generate_bucket(bucket: Bucket) -> None: """ Function to generate the S3 bucket object :param ecs_composex.s3.s3_bucket.Bucket bucket: """ bucket_name = define_bucket_name(bucket) final_bucket_name = ( Sub(bucket_name) if isinstance(bucket_name, str) and (bucket_name.find(AWS_REGION) >= 0 or bucket_name.find(AWS_ACCOUNT_ID) >= 0) else bucket_name ) LOG.debug(bucket_name) LOG.debug(final_bucket_name) props = import_record_properties(bucket.properties, s3.Bucket) props["BucketName"] = final_bucket_name bucket.cfn_resource = s3.Bucket(bucket.logical_name, **props)
[docs]def handle_predefined_policies( bucket: Bucket, param_key: str, managed_policies_key: str, statement: list ) -> None: """ Function to configure and add statements for bucket policy based on predefined Bucket Policies :param bucket: :param str param_key: :param str managed_policies_key: :param list statement: """ unique_policies = list( set(bucket.parameters[param_key]["PredefinedBucketPolicies"]) ) for policy_name in unique_policies: if policy_name not in bucket.module.iam_policies[managed_policies_key].keys(): LOG.error( f"Policy {policy_name} is not defined as part of possible permissions set" ) continue policies = generate_resource_permissions( bucket.logical_name, bucket.module.iam_policies[managed_policies_key], Sub(f"arn:${{{AWS_PARTITION}}}:s3:::${{{bucket.cfn_resource.title}}}"), ) statement += policies[policy_name].PolicyDocument["Statement"]
[docs]def handle_user_defined_policies( bucket: Bucket, param_key: str, user_policies_key: str, statement: list ): """ Function to add user defined policies :param bucket: :param str param_key: :param str user_policies_key: :param list statement: """ policies = bucket.parameters[param_key][user_policies_key] policy_docs = {} for count, policy_doc in enumerate(policies): if keyisset("Sid", policy_doc): name = policy_doc["Sid"] else: name = f"UserDefined{count}" policy_docs[name] = policy_doc generated_policies = generate_resource_permissions( bucket.logical_name, policy_docs, Sub(f"arn:${{{AWS_PARTITION}}}:s3:::${{{bucket.cfn_resource.title}}}"), True, ) for policy in generated_policies.values(): statement += policy.PolicyDocument["Statement"]
[docs]def implement_bucket_policy(bucket: Bucket, param_key: str, bucket_template: Template): """ Function to parse the input parameter for the Bucket Policy, and generate the policy accordingly """ statement = [] managed_policies_key = "PredefinedBucketPolicies" user_policies_key = "Policies" policy_document = {"Version": "2012-10-17", "Statement": statement} if keyisset(managed_policies_key, bucket.parameters[param_key]) and keyisset( managed_policies_key, bucket.module.iam_policies ): handle_predefined_policies(bucket, param_key, managed_policies_key, statement) if keyisset(user_policies_key, bucket.parameters[param_key]): handle_user_defined_policies(bucket, param_key, user_policies_key, statement) bucket_policy = s3.BucketPolicy( f"{bucket.logical_name}BucketPolicy", Bucket=Ref(bucket.cfn_resource), PolicyDocument=policy_document, DependsOn=[bucket.cfn_resource.title], ) bucket_template.add_resource(bucket_policy)
[docs]def evaluate_parameters(bucket, bucket_template): """ Review bucket parameters to configure the bucket and extra properties. """ if bucket.mappings: return if not bucket.parameters: return parameters = {"BucketPolicy": implement_bucket_policy} for name, function in parameters.items(): if keyisset(name, bucket.parameters) and function: function(bucket, name, bucket_template)
[docs]def create_s3_template(new_buckets: list[Bucket], template: Template) -> Template: """ Function to create the root S3 template. """ mono_template = False if len(list(new_buckets)) <= COMPOSEX_MAX_OUTPUTS: mono_template = True for bucket in new_buckets: generate_bucket(bucket) if bucket.cfn_resource: bucket.init_outputs() bucket.generate_outputs() bucket_template = template if mono_template: bucket_template.add_resource(bucket.cfn_resource) add_outputs(bucket_template, bucket.outputs) elif not mono_template: bucket_template = build_template( f"Template for S3 Bucket {bucket.name}" ) bucket_template.add_resource(bucket.cfn_resource) add_outputs(bucket_template, bucket.outputs) bucket_stack = ComposeXStack( bucket.logical_name, stack_template=bucket_template ) template.add_resource(bucket_stack) evaluate_parameters(bucket, bucket_template) return template