Added support for AWS Config Retention Configuration

- Added AWS Config Retention Configuration support
- Also added S3 KMS Key ARN support for the Delivery Channel
- Updated the supported functions page for Config

docs/docs/services/config.rst

@@ -62,7 +62,7 @@ config
 - [ ] delete_remediation_configuration
 - [ ] delete_remediation_exceptions
 - [ ] delete_resource_config
-- [ ] delete_retention_configuration
+- [X] delete_retention_configuration
 - [ ] delete_stored_query
 - [ ] deliver_config_snapshot
 - [ ] describe_aggregate_compliance_by_config_rules
@@ -91,7 +91,7 @@ config
 - [ ] describe_remediation_configurations
 - [ ] describe_remediation_exceptions
 - [ ] describe_remediation_execution_status
-- [ ] describe_retention_configurations
+- [X] describe_retention_configurations
 - [ ] get_aggregate_compliance_details_by_config_rule
 - [ ] get_aggregate_config_rule_compliance_summary
 - [ ] get_aggregate_conformance_pack_compliance_summary
@@ -179,7 +179,7 @@ config
 - [ ] put_remediation_configurations
 - [ ] put_remediation_exceptions
 - [ ] put_resource_config
-- [ ] put_retention_configuration
+- [X] put_retention_configuration
 - [ ] put_stored_query
 - [ ] select_aggregate_resource_config
 - [ ] select_resource_config

moto/config/exceptions.py

@@ -114,6 +114,14 @@ def __init__(self) -> None:
         super().__init__("InvalidS3KeyPrefixException", message)
 
 
+class InvalidS3KmsKeyArnException(JsonRESTError):
+    code = 400
+
+    def __init__(self) -> None:
+        message = "The arn '' is not a valid kms key or alias arn."
+        super().__init__("InvalidS3KmsKeyArnException", message)
+
+
 class InvalidSNSTopicARNException(JsonRESTError):
     """We are *only* validating that there is value that is not '' here."""
 
@@ -373,3 +381,13 @@ class MissingRequiredConfigRuleParameterException(JsonRESTError):
 
     def __init__(self, message: str):
         super().__init__("ParamValidationError", message)
+
+
+class NoSuchRetentionConfigurationException(JsonRESTError):
+    code = 400
+
+    def __init__(self, name: str):
+        message = (
+            f"Cannot find retention configuration with the specified name '{name}'."
+        )
+        super().__init__("NoSuchRetentionConfigurationException", message)

moto/config/models.py

@@ -18,6 +18,7 @@
     InvalidDeliveryChannelNameException,
     NoSuchBucketException,
     InvalidS3KeyPrefixException,
+    InvalidS3KmsKeyArnException,
     InvalidSNSTopicARNException,
     MaxNumberOfDeliveryChannelsExceededException,
     NoAvailableDeliveryChannelException,
@@ -44,6 +45,7 @@
     MaxNumberOfConfigRulesExceededException,
     InsufficientPermissionsException,
     NoSuchConfigRuleException,
+    NoSuchRetentionConfigurationException,
     ResourceInUseException,
     MissingRequiredConfigRuleParameterException,
 )
@@ -259,16 +261,29 @@ def __init__(
         s3_bucket_name: str,
         prefix: Optional[str] = None,
         sns_arn: Optional[str] = None,
+        s3_kms_key_arn: Optional[str] = None,
         snapshot_properties: Optional[ConfigDeliverySnapshotProperties] = None,
     ):
         super().__init__()
 
         self.name = name
         self.s3_bucket_name = s3_bucket_name
         self.s3_key_prefix = prefix
+        self.s3_kms_key_arn = s3_kms_key_arn
         self.sns_topic_arn = sns_arn
         self.config_snapshot_delivery_properties = snapshot_properties
 
+    def to_dict(self) -> Dict[str, Any]:
+        """Need to override this function because the KMS Key ARN is written as `Arn` vs. SNS which is `ARN`."""
+        data = super().to_dict()
+
+        # Fix the KMS ARN if it's here:
+        kms_arn = data.pop("s3KmsKeyARN", None)
+        if kms_arn:
+            data["s3KmsKeyArn"] = kms_arn
+
+        return data
+
 
 class RecordingGroup(ConfigEmptyDictable):
     def __init__(
@@ -883,6 +898,14 @@ def validate_managed_rule(self) -> None:
         # Verify the rule is allowed for this region -- not yet implemented.
 
 
+class RetentionConfiguration(ConfigEmptyDictable):
+    def __init__(self, retention_period_in_days: int, name: Optional[str] = None):
+        super().__init__(capitalize_start=True, capitalize_arn=False)
+
+        self.name = name or "default"
+        self.retention_period_in_days = retention_period_in_days
+
+
 class ConfigBackend(BaseBackend):
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
@@ -893,6 +916,7 @@ def __init__(self, region_name: str, account_id: str):
         self.organization_conformance_packs: Dict[str, OrganizationConformancePack] = {}
         self.config_rules: Dict[str, ConfigRule] = {}
         self.config_schema: Optional[AWSServiceSpec] = None
+        self.retention_configuration: Optional[RetentionConfiguration] = None
 
     @staticmethod
     def default_vpc_endpoint_service(service_region: str, zones: List[str]) -> List[Dict[str, Any]]:  # type: ignore[misc]
@@ -1264,9 +1288,15 @@ def put_delivery_channel(self, delivery_channel: Dict[str, Any]) -> None:
 
         # Ditto for SNS -- Only going to assume that the ARN provided is not
         # an empty string:
+        # NOTE: SNS "ARN" is all caps, but KMS "Arn" is UpperCamelCase!
         if delivery_channel.get("snsTopicARN", None) == "":
             raise InvalidSNSTopicARNException()
 
+        # Ditto for S3 KMS Key ARN -- Only going to assume that the ARN provided is not
+        # an empty string:
+        if delivery_channel.get("s3KmsKeyArn", None) == "":
+            raise InvalidS3KmsKeyArnException()
+
         # Config currently only allows 1 delivery channel for an account:
         if len(self.delivery_channels) == 1 and not self.delivery_channels.get(
             delivery_channel["name"]
@@ -1292,6 +1322,7 @@ def put_delivery_channel(self, delivery_channel: Dict[str, Any]) -> None:
             delivery_channel["name"],
             delivery_channel["s3BucketName"],
             prefix=delivery_channel.get("s3KeyPrefix", None),
+            s3_kms_key_arn=delivery_channel.get("s3KmsKeyArn", None),
             sns_arn=delivery_channel.get("snsTopicARN", None),
             snapshot_properties=dprop,
         )
@@ -2012,5 +2043,66 @@ def delete_config_rule(self, rule_name: str) -> None:
         rule.config_rule_state = "DELETING"
         self.config_rules.pop(rule_name)
 
+    def put_retention_configuration(
+        self, retention_period_in_days: int
+    ) -> Dict[str, Any]:
+        """Creates a Retention Configuration."""
+        if retention_period_in_days < 30:
+            raise ValidationException(
+                f"Value '{retention_period_in_days}' at 'retentionPeriodInDays' failed to satisfy constraint: Member must have value greater than or equal to 30"
+            )
+
+        if retention_period_in_days > 2557:
+            raise ValidationException(
+                f"Value '{retention_period_in_days}' at 'retentionPeriodInDays' failed to satisfy constraint: Member must have value less than or equal to 2557"
+            )
+
+        self.retention_configuration = RetentionConfiguration(retention_period_in_days)
+        return {"RetentionConfiguration": self.retention_configuration.to_dict()}
+
+    def describe_retention_configurations(
+        self, retention_configuration_names: Optional[List[str]]
+    ) -> List[Dict[str, Any]]:
+        """
+        This will return the retention configuration if one is present.
+
+        This should only receive at most 1 name in. It will raise a ValidationException if more than 1 is supplied.
+        """
+        # Handle the cases where we get a retention name to search for:
+        if retention_configuration_names:
+            if len(retention_configuration_names) > 1:
+                raise ValidationException(
+                    f"Value '{retention_configuration_names}' at 'retentionConfigurationNames' failed to satisfy constraint: Member must have length less than or equal to 1"
+                )
+
+            # If we get a retention name to search for, and we don't have it, then we need to raise an exception:
+            if (
+                not self.retention_configuration
+                or not self.retention_configuration.name
+                == retention_configuration_names[0]
+            ):
+                raise NoSuchRetentionConfigurationException(
+                    retention_configuration_names[0]
+                )
+
+            # If we found it, then return it:
+            return [self.retention_configuration.to_dict()]
+
+        # If no name was supplied:
+        if self.retention_configuration:
+            return [self.retention_configuration.to_dict()]
+
+        return []
+
+    def delete_retention_configuration(self, retention_configuration_name: str) -> None:
+        """This will delete the retention configuration if one is present with the provided name."""
+        if (
+            not self.retention_configuration
+            or not self.retention_configuration.name == retention_configuration_name
+        ):
+            raise NoSuchRetentionConfigurationException(retention_configuration_name)
+
+        self.retention_configuration = None
+
 
 config_backends = BackendDict(ConfigBackend, "config")

moto/config/responses.py

@@ -235,3 +235,24 @@ def describe_config_rules(self) -> str:
     def delete_config_rule(self) -> str:
         self.config_backend.delete_config_rule(self._get_param("ConfigRuleName"))
         return ""
+
+    def put_retention_configuration(self) -> str:
+        retention_configuration = self.config_backend.put_retention_configuration(
+            self._get_param("RetentionPeriodInDays")
+        )
+        return json.dumps(retention_configuration)
+
+    def describe_retention_configurations(self) -> str:
+        retention_configurations = (
+            self.config_backend.describe_retention_configurations(
+                self._get_param("RetentionConfigurationNames")
+            )
+        )
+        schema = {"RetentionConfigurations": retention_configurations}
+        return json.dumps(schema)
+
+    def delete_retention_configuration(self) -> str:
+        self.config_backend.delete_retention_configuration(
+            self._get_param("RetentionConfigurationName")
+        )
+        return ""