Untitled

"""Create s3 resources and tasks related to s3"""

from aws_cdk import RemovalPolicy
from aws_cdk import Stack
from aws_cdk import aws_iam as iam
from aws_cdk import aws_s3
from aws_cdk import aws_kms as kms
from aws_cdk import Duration
from principal_environment import PrincipalEnvironment


def get_s3_general_policy(
    *bucket_names: str, kms_key_arn: str
) -> iam.PolicyStatement:
    """Create policy statement for reading and writing S3 objects."""
    policy_statement = iam.PolicyStatement()
    policy_statement.effect = iam.Effect.ALLOW
    policy_statement.add_actions(
        "s3:GetObject*",
        "s3:GetBucket*",
        "s3:List*",
        "s3:DeleteObject*",
        "s3:PutObject*",
        "s3:Abort*",
        "kms:Encrypt",
        "kms:Decrypt",
        "kms:ListAliases",
        "kms:ListKeys",
        "kms:GenerateDataKey",
    )
    for bucket in bucket_names:
        policy_statement.add_resources(
            f"arn:aws:s3:::{bucket}/*", f"arn:aws:s3:::{bucket}"
        )

    policy_statement.add_resources(kms_key_arn)

    return policy_statement


def create_bucket(
    scope: Stack, bucket: str, env: PrincipalEnvironment, encryption_key: kms.Ikey
) -> aws_s3.Bucket:

    bucket_obj = aws_s3.Bucket(
        scope,
        f"Bucket-{bucket}",
        block_public_access=aws_s3.BlockPublicAccess.BLOCK_ALL,
        encryption=aws_s3.BucketEncryption.KMS,
        enforce_ssl=True,
        bucket_name=f"{bucket}-{env.aws_environment_name}-{env.region}",
        versioned=False,
        removal_policy=RemovalPolicy.DESTROY,
        encryption_key=scope.kms,
        lifecycle_rules=[
            aws_s3.LifecycleRule(
                id="bucket_lifecycle", expiration=Duration.days(365 * 7)
            )
        ],
    )

    s3_policy_statement = get_s3_general_policy(bucket, encryption_key.key_arn)

    bucket_obj.add_to_resource_policy(s3_policy_statement)

    return bucket_obj