AWS - SES - Create new resource type configuration-set for SES

c7n/resources/resource_map.py

@@ -193,6 +193,7 @@
     "aws.scaling-policy": "c7n.resources.asg.ScalingPolicy",
     "aws.secrets-manager": "c7n.resources.secretsmanager.SecretsManager",
     "aws.security-group": "c7n.resources.vpc.SecurityGroup",
+    "aws.ses-configuration-set": "c7n.resources.ses.SESConfigurationSet",
     "aws.serverless-app": "c7n.resources.sar.ServerlessApp",
     "aws.service-quota-request": "c7n.resources.quotas.ServiceQuotaRequest",
     "aws.service-quota": "c7n.resources.quotas.ServiceQuota",

c7n/resources/ses.py

@@ -0,0 +1,32 @@
+# Copyright The Cloud Custodian Authors.
+# SPDX-License-Identifier: Apache-2.0
+
+from c7n.manager import resources
+from c7n.query import DescribeSource, QueryResourceManager, TypeInfo
+from c7n.utils import local_session
+from c7n.tags import universal_augment
+
+class DescribeConfigurationSet(DescribeSource):
+
+    def augment(self, resources):
+        client = local_session(self.manager.session_factory).client('ses')
+        for r in resources:
+            details = client.describe_configuration_set(ConfigurationSetName=r['Name'], 
+                ConfigurationSetAttributeNames=['eventDestinations','trackingOptions','deliveryOptions','reputationOptions'])
+            r.update(details)
+        return universal_augment(self.manager, resources)
+
+
+@resources.register('ses-configuration-set')
+class SESConfigurationSet(QueryResourceManager):
+    class resource_type(TypeInfo):
+        service = 'ses'
+        enum_spec = ('list_configuration_sets', 'ConfigurationSets', None)
+        name = id = 'Name'
+        arn_type = 'configuration-set'
+        universal_taggable = object()
+
+    source_mapping = {
+        'describe': DescribeConfigurationSet
+    }
+

tests/data/placebo/test_ses_configuration_set_query/email.DescribeConfigurationSet_1.json

@@ -0,0 +1,16 @@
+{
+    "status_code": 200,
+    "data": {
+        "ConfigurationSet": {
+            "Name": "c7n-test"
+        },
+        "DeliveryOptions": {
+            "TlsPolicy": "Require"
+        },
+        "ReputationOptions": {
+            "SendingEnabled": true,
+            "ReputationMetricsEnabled": true
+        },
+        "ResponseMetadata": {}
+    }
+}

tests/data/placebo/test_ses_configuration_set_query/email.ListConfigurationSets_1.json

@@ -0,0 +1,11 @@
+{
+    "status_code": 200,
+    "data": {
+        "ConfigurationSets": [
+            {
+                "Name": "c7n-test"
+            }
+        ],
+        "ResponseMetadata": {}
+    }
+}

tests/data/placebo/test_ses_configuration_set_query/tagging.GetResources_1.json

@@ -0,0 +1,18 @@
+{
+    "status_code": 200,
+    "data": {
+        "PaginationToken": "",
+        "ResourceTagMappingList": [
+            {
+                "ResourceARN": "arn:aws:ses:us-east-1:644160558196:configuration-set/c7n-test",
+                "Tags": [
+                    {
+                        "Key": "Owner",
+                        "Value": "Pratyush"
+                    }
+                ]
+            }
+        ],
+        "ResponseMetadata": {}
+    }
+}

tests/test_ses.py

@@ -0,0 +1,16 @@
+# Copyright The Cloud Custodian Authors.
+# SPDX-License-Identifier: Apache-2.0
+from .common import BaseTest
+
+class SESTest(BaseTest):
+
+    def test_ses_configuration_set_query(self):
+        session_factory = self.replay_flight_data("test_ses_configuration_set_query")
+        p = self.load_policy(
+            {
+                "name": "ses-configuration-set-query-test",
+                "resource": "ses-configuration-set"
+            }, session_factory=session_factory
+        )
+        resources = p.run()
+        self.assertEqual(len(resources), 1)