Source code for ecs_composex.rds.rds_features_define

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

"""
Module to define and match RDS Features to other AWS resources.
"""


from __future__ import annotations

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from .rds_stack import Rds
    from ecs_composex.common.settings import ComposeXSettings

import re

from troposphere import AWS_NO_VALUE, AWS_PARTITION, Ref, Sub
from troposphere.iam import Policy as IamPolicy

from ecs_composex.common.logging import LOG
from ecs_composex.common.troposphere_tools import add_parameters, add_update_mapping
from ecs_composex.s3.s3_params import S3_BUCKET_ARN

S3_KEY = "x-s3"


[docs]def set_from_x_s3(settings: ComposeXSettings, db: Rds, db_stack, bucket_name: str): """ Function to link the RDS DB to a Bucket defined in x-s3 :param settings: :param db: :param str bucket_name: :return: """ bucket = settings.find_resource(bucket_name) if not bucket: raise LookupError( f"Failed to find bucket {bucket_name} for DB {db.module.res_key}.{db.name}" ) bucket_arn_id = bucket.add_attribute_to_another_stack( db_stack, S3_BUCKET_ARN, settings ) if bucket.cfn_resource: return Sub(f"${{{bucket_arn_id['ImportParameter'].title}}}/*") elif bucket.lookup and bucket.module.mappings: return bucket.lookup_properties[S3_BUCKET_ARN]
[docs]def import_bucket_from_arn(bucket): """ Function to import a bucket defined by its ARN, supports to detect the path in bucket for objects :param str bucket: :return: the bucket ARN to use :rtype: str """ bucket_arn_path_re = re.compile( r"(arn:aws:s3:::(?:[a-zA-Z0-9-.]+))(/path/to/files)?" ) if not bucket_arn_path_re.match(bucket): raise ValueError( "The Bucket ARN provided is not valid. Expecting format", bucket_arn_path_re.pattern, ) elif bucket_arn_path_re.match(bucket): bucket_arn = bucket else: bucket_arn = f"{bucket}/*" return bucket_arn
[docs]def import_raw_bucket_name(bucket) -> Sub: """ Function to import and define a bucket ARN from bucket name alone and support for path to be defined :param str bucket: :return: the bucket ARN :rtype: Sub """ bucket_name_path_re = re.compile(r"(?:[a-zA-Z0-9-.]+)(/path/to/files)?") bucket_arn = Ref(AWS_NO_VALUE) if not bucket_name_path_re.match(bucket): raise ValueError( "Bucket name and path are not valid. Expecting format", bucket_name_path_re.pattern, ) if ( bucket_name_path_re.match(bucket) and len(bucket_name_path_re.match(bucket).groups()) == 1 ): bucket_arn = Sub(f"arn:${{{AWS_PARTITION}}}:s3:::{bucket}/*") elif ( bucket_name_path_re.match(bucket) and len(bucket_name_path_re.match(bucket).groups()) == 2 ): bucket_arn = Sub(f"arn:${{{AWS_PARTITION}}}:s3:::{bucket}") return bucket_arn
[docs]def define_s3_bucket_arns(settings, db, db_stack, config) -> list: """ Function to define the IAM Policy for S3Import access :param settings: :param db: :param db_stack: :param config: :return: """ bucket_arns = [] for bucket in config: bucket_arn = None if not isinstance(bucket, str): raise TypeError( "All buckets defined in IamAccess/Buckets must be of type", str, "Got", type(bucket), ) if bucket.startswith("x-s3"): bucket_arn = set_from_x_s3(settings, db, db_stack, bucket) elif bucket.startswith("arn:aws"): bucket_arn = import_bucket_from_arn(bucket) elif not bucket.startswith("arn:aws"): bucket_arn = import_raw_bucket_name(bucket) if bucket_arn: bucket_arns.append(bucket_arn) return bucket_arns
[docs]def define_s3_export_feature_policy(settings, db, db_stack, config) -> IamPolicy: """ Function to define the IAM Policy for S3Import access :param settings: :param db: :param config: :return: """ bucket_arns = define_s3_bucket_arns(settings, db, db_stack, config) policy = IamPolicy( PolicyName=f"S3AccessFor{db.logical_name}", PolicyDocument={ "Version": "2012-10-17", "Statement": [ { "Sid": "S3WriteObjectsAccess", "Effect": "Allow", "Action": ["s3:PutObject*"], "Resource": bucket_arns, } ], }, ) return policy
[docs]def define_s3_import_feature_policy(settings, db, db_stack, config) -> IamPolicy: """ Function to define the IAM Policy for S3Import access :param settings: :param db: :param db_stack: :param config: :return: """ bucket_arns = define_s3_bucket_arns(settings, db, db_stack, config) policy = IamPolicy( PolicyName=f"S3AccessFor{db.logical_name}", PolicyDocument={ "Version": "2012-10-17", "Statement": [ { "Sid": "S3ReadObjectsAccess", "Effect": "Allow", "Action": ["s3:GetObject*"], "Resource": bucket_arns, } ], }, ) return policy