Source code for idem_aws.states.aws.config.delivery_channel

"""State module for managing Amazon Config Delivery Channel."""
import copy
from dataclasses import field
from dataclasses import make_dataclass
from typing import Any
from typing import Dict

__contracts__ = ["resource"]

TREQ = {
    "present": {
        "require": [
            "aws.ec2.subnet.present",
        ],
    },
}


async def present(
    hub,
    ctx,
    name: str,
    s3_bucket_name: str,
    resource_id: str = None,
    s3_key_prefix: str = None,
    s3_kms_key_arn: str = None,
    sns_topic_arn: str = None,
    config_snapshot_delivery_properties: make_dataclass(
        "ConfigSnapshotDeliveryProperties",
        [("delivery_frequency", str, field(default=None))],
    ) = None,
) -> Dict[str, Any]:
    """Add or update the configuration delivery channel object that delivers the configuration information to an Amazon S3 bucket and to an Amazon SNS topic.

    Args:
        name(str):
            An Idem name of the delivery channel.

        s3_bucket_name (str):
            The name of the Amazon S3 bucket to which Config delivers configuration snapshots and
            configuration history files

        resource_id (str, Optional):
            The resource Id of the delivery channel.

        s3_key_prefix (str, Optional):
            The prefix for the specified Amazon S3 bucket.

        s3_kms_key_arn (str, Optional):
            The Amazon Resource Name (ARN) of the Key Management Service (KMS ) KMS key (KMS key) used to
            encrypt objects delivered by Config.

        sns_topic_arn (str, Optional):
            The Amazon Resource Name (ARN) of the Amazon SNS topic to which Config sends notifications
            about configuration changes.

        config_snapshot_delivery_properties (dict[str, Any], Optional):
            The options for how often Config delivers configuration snapshots to the Amazon S3 bucket.

            * delivery_frequency (str, Optional):
              The frequency with which Config delivers configuration snapshots.

    Request syntax:
        .. code-block:: sls

            [aws-config-delivery_channel]:
              aws.config.delivery_channel.present:
              - name: 'string'
              - resource_id: 'string'
              - s3_bucket_name: 'string'
              - s3_key_prefix: 'string'
              - s3_kms_key_arn: 'string'
              - sns_topic_arn: 'string'
              - config_snapshot_delivery_properties: dict
                delivery_frequency: 'string'

    Returns:
         Dict[str, Any]

    Examples:
        .. code-block:: sls

            aws-config-delivery_channel:
              aws.config.delivery_channel.present:
                - name: 'delivery-channel'
                - resource_id: 'delivery-channel'
                - s3_bucket_name: 'Test-S3'
                - s3_key_prefix: 'S3-prefix'
                - s3_kms_key_arn: 'Test-Kms-Key'
                - sns_topic_arn: 'Test-Topic'
                - config_snapshot_delivery_properties:
                  delivery_frequency: 'One_Hour'
    """
    result = dict(comment=[], old_state=None, new_state=None, name=name, result=True)
    before = None
    resource_updated = False
    update_payload = {}
    update_config_snapshot_delivery_properties = None
    resource_id = resource_id if resource_id else name
    before = await hub.exec.boto3.client.config.describe_delivery_channels(
        ctx, DeliveryChannelNames=[resource_id]
    )
    if before:
        result[
            "old_state"
        ] = hub.tool.aws.config.conversion_utils.convert_raw_config_delivery_channel_to_present(
            ctx,
            raw_resource=before["ret"]["DeliveryChannels"][0],
            idem_resource_name=name,
        )
        plan_state = copy.deepcopy(result["old_state"])
        # updating config_snapshot_delivery_properties
        if config_snapshot_delivery_properties is not None:
            update_config_snapshot_delivery_properties = (
                config_snapshot_delivery_properties
            )
        update_payload = await hub.tool.aws.config.delivery_channel.get_updated_payload_delivery_channel(
            ctx,
            resource_id=resource_id,
            before=before["ret"]["DeliveryChannels"][0],
            s3_bucket_name=s3_bucket_name,
            s3_key_prefix=s3_key_prefix,
            s3_kms_key_arn=s3_kms_key_arn,
            sns_topic_arn=sns_topic_arn,
            config_snapshot_delivery_properties=update_config_snapshot_delivery_properties,
        )
        result["comment"] = hub.tool.aws.comment_utils.already_exists_comment(
            "aws.config.delivery_channel", name
        )

        if update_payload["ret"]:
            resource_updated = True
            if not ctx.get("test", False):
                update_ret = await hub.exec.boto3.client.config.put_delivery_channel(
                    ctx=ctx, DeliveryChannel=update_payload["ret"]
                )
                if not update_ret["result"]:
                    result["comment"] += update_ret["comment"]
                    result["result"] = False
                    return result

                result["comment"] = hub.tool.aws.comment_utils.update_comment(
                    "aws.config.delivery_channel", name
                )
                result["result"] = update_ret["result"]
            else:
                for key, value in update_payload["ret"].items():
                    plan_state[key] = value

    else:
        delivery_props = {}
        dc_payload = {"name": name, "s3BucketName": s3_bucket_name}
        if sns_topic_arn:
            dc_payload["snsTopicARN"] = sns_topic_arn
        if s3_key_prefix:
            dc_payload["s3KeyPrefix"] = s3_key_prefix
        if s3_kms_key_arn:
            dc_payload["s3KmsKeyArn"] = s3_kms_key_arn
        if config_snapshot_delivery_properties:
            delivery_props["deliveryFrequency"] = config_snapshot_delivery_properties[
                "delivery_frequency"
            ]
            dc_payload["configSnapshotDeliveryProperties"] = delivery_props
        if ctx.get("test", False):
            result[
                "new_state"
            ] = hub.tool.aws.config.conversion_utils.convert_raw_config_delivery_channel_to_present(
                ctx,
                raw_resource=dc_payload,
                idem_resource_name=name,
            )
            result["comment"] = hub.tool.aws.comment_utils.would_create_comment(
                "aws.config.delivery_channel", name
            )
            return result

        ret = await hub.exec.boto3.client.config.put_delivery_channel(
            ctx, DeliveryChannel=dc_payload
        )
        result["result"] = ret["result"]
        if not result["result"]:
            result["comment"] = ret["comment"] + ret["comment"]
            return result
            # resource_id = name
        result["comment"] = hub.tool.aws.comment_utils.create_comment(
            "aws.config.delivery_channel", name
        )

    if ctx.get("test", False):
        result[
            "new_state"
        ] = hub.tool.aws.config.conversion_utils.convert_raw_config_delivery_channel_to_present(
            ctx,
            raw_resource=plan_state,
            idem_resource_name=name,
        )
        if resource_updated:
            result["comment"] = hub.tool.aws.comment_utils.would_update_comment(
                "aws.config.delivery_channel", name
            )
    elif (not before) or resource_updated:
        after = await hub.exec.boto3.client.config.describe_delivery_channels(
            ctx, DeliveryChannelNames=[resource_id]
        )
        result[
            "new_state"
        ] = hub.tool.aws.config.conversion_utils.convert_raw_config_delivery_channel_to_present(
            ctx,
            raw_resource=after["ret"]["DeliveryChannels"][0],
            idem_resource_name=name,
        )
    else:
        result["new_state"] = copy.deepcopy(result["old_state"])
    return result


async def absent(
    hub,
    ctx,
    name: str,
    resource_id: str = None,
) -> Dict[str, Any]:
    """Deletes the specified delivery channel.

    Before you can delete the delivery channel, you must stop the configuration recorder by using the StopConfigurationRecorder action.

    Args:
        name(str): An Idem name of the rule.
        resource_id(str, Optional): AWS Config delivery channel Name. Idem automatically considers this resource being absent
         if this field is not specified.

    Returns:
          Dict[str, Any]

    Examples:
          .. code-block:: sls

            ec2-instance-no-public-ip:
              aws.config.delivery_channel.absent:
              - name: ec2-instance-no-public-ip
              - resource_id: ec2-instance-no-public-ip

    """
    result = dict(comment=[], old_state=None, new_state=None, name=name, result=True)
    resource_id = resource_id if resource_id else name
    # if not resource_id:
    #     result["comment"] = hub.tool.aws.comment_utils.already_absent_comment(
    #         resource_type="aws.config.delivery_channel", name=name
    #     )
    #     return result
    before = await hub.exec.boto3.client.config.describe_delivery_channels(
        ctx, DeliveryChannelNames=[resource_id]
    )
    if not before:
        result["comment"] = hub.tool.aws.comment_utils.already_absent_comment(
            resource_type="aws.config.delivery_channel", name=name
        )
    elif ctx.get("test", False):
        result[
            "old_state"
        ] = hub.tool.aws.config.conversion_utils.convert_raw_config_delivery_channel_to_present(
            ctx,
            raw_resource=before["ret"]["DeliveryChannels"][0],
            idem_resource_name=resource_id,
        )
        result["comment"] = hub.tool.aws.comment_utils.would_delete_comment(
            resource_type="aws.config.delivery_channel", name=resource_id
        )
        return result
    else:
        try:
            result[
                "old_state"
            ] = hub.tool.aws.config.conversion_utils.convert_raw_config_delivery_channel_to_present(
                ctx,
                raw_resource=before["ret"]["DeliveryChannels"][0],
                idem_resource_name=resource_id,
            )

            ret = await hub.exec.boto3.client.config.delete_delivery_channel(
                ctx, DeliveryChannelName=resource_id
            )
            result["result"] = ret["result"]
            if not result["result"]:
                result["comment"] = ret["comment"]
                return result
            result["comment"] = hub.tool.aws.comment_utils.delete_comment(
                resource_type="aws.config.delivery_channel", name=name
            )
        except hub.tool.boto3.exception.ClientError as e:
            result["comment"] += [f"{e.__class__.__name__}: {e}"]
            result["result"] = False
    return result


async def describe(hub, ctx) -> Dict[str, Dict[str, Any]]:
    """Describe the resource in a way that can be recreated/managed with the corresponding "present" function.

    Return details about your delivery channel.

    Returns:
        Dict[str, Any]

    Examples:
        .. code-block:: bash

            $ idem describe aws.config.delivery_channel
    """
    result = {}
    ret = await hub.exec.boto3.client.config.describe_delivery_channels(ctx)
    if not ret["result"]:
        hub.log.warning(f"Could not describe Delivery Channels {ret['comment']}")
        return {}

    for resource in ret["ret"]["DeliveryChannels"]:
        resource_id = resource.get("name")
        resource_translated = hub.tool.aws.config.conversion_utils.convert_raw_config_delivery_channel_to_present(
            ctx, raw_resource=resource, idem_resource_name=resource_id
        )
        result[resource_id] = {
            "aws.config.delivery_channel.present": [
                {parameter_key: parameter_value}
                for parameter_key, parameter_value in resource_translated.items()
            ]
        }
    return result