feat: Support creating a SNS topic

Author: rubysoho07

layer/python/cloudtrail_watcher/services/__init__.py

@@ -16,5 +16,6 @@
     'dynamodb',
     'elasticloadbalancing',
     'cloudfront',
-    'sqs'
+    'sqs',
+    'sns'
 ]

layer/python/cloudtrail_watcher/services/sns.py

@@ -0,0 +1,34 @@
+import boto3
+
+from .common import *
+
+sns = boto3.client('sns')
+
+def process_event(event: dict, set_tag: bool = False) -> dict:
+    """ Process CloudTrail event for sns """
+	
+    result = dict()
+
+    if event['eventName'] == 'CreateTopic':
+        result['resource_id'] = _process_create_topic(event, set_tag)
+    else:
+        message = f"Cannot process event: {event['eventName']}, eventID: {event['eventID']}"
+        result['error'] = message
+
+    return result
+
+
+def _process_create_topic(event: dict, set_tag: bool = False) -> list:
+    """ Process CreateTopic event """
+    
+    topic_arn = event['responseElements']['topicArn']
+    topic_name = event['requestParameters']['name']
+    
+    if set_tag:
+        if not check_contain_mandatory_tag_list(event['requestParameters']['tags']):
+            sns.tag_resource(
+                ResourceArn=topic_arn,
+                Tags=[{'Key': 'User', 'Value': get_user_identity(event)}]
+            )
+    
+    return [topic_name]

template.sar.yaml

@@ -69,7 +69,8 @@ Resources:
               "cloudfront:TagResource",
               "ecr:ListTagsForResource",
               "ecr:TagResource",
-              "sqs:TagQueue"
+              "sqs:TagQueue",
+              "sns:TagResource"
             ]
             Resource: "*"
 

test/services/samples/sns_CreateTopic.json

@@ -0,0 +1,49 @@
+{
+  "eventVersion": "1.11",
+  "userIdentity": {
+    "type": "IAMUser",
+    "principalId": "YOUR_PRINCIPAL_ID",
+    "arn": "arn:aws:iam::000000000000:user/test_user",
+    "accountId": "000000000000",
+    "accessKeyId": "YOUR_ACCESS_KEY_ID",
+    "userName": "test_user",
+    "sessionContext": {
+      "attributes": {
+        "creationDate": "2025-09-05T02:33:23Z",
+        "mfaAuthenticated": "true"
+      }
+    }
+  },
+  "eventTime": "2025-09-05T13:15:27Z",
+  "eventSource": "sns.amazonaws.com",
+  "eventName": "CreateTopic",
+  "awsRegion": "ap-northeast-2",
+  "sourceIPAddress": "127.0.0.1",
+  "userAgent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36",
+  "requestParameters": {
+    "name": "sns-test.fifo",
+    "attributes": {
+      "Policy": "{\"Version\":\"2008-10-17\",\"Id\":\"__default_policy_ID\",\"Statement\":[{\"Sid\":\"__default_statement_ID\",\"Effect\":\"Allow\",\"Principal\":{\"AWS\":\"*\"},\"Action\":[\"SNS:Publish\",\"SNS:RemovePermission\",\"SNS:SetTopicAttributes\",\"SNS:DeleteTopic\",\"SNS:ListSubscriptionsByTopic\",\"SNS:GetTopicAttributes\",\"SNS:AddPermission\",\"SNS:Subscribe\"],\"Resource\":\"arn:aws:sns:ap-northeast-2:000000000000:sns-test.fifo\",\"Condition\":{\"StringEquals\":{\"AWS:SourceAccount\":\"000000000000\"}}}]}",
+      "TracingConfig": "PassThrough",
+      "FifoTopic": "true",
+      "FifoThroughputScope": "MessageGroup"
+    },
+    "tags": []
+  },
+  "responseElements": {
+    "topicArn": "arn:aws:sns:ap-northeast-2:000000000000:sns-test.fifo"
+  },
+  "requestID": "e9345ab2-abc8-5d6e-a882-3b864f67588a",
+  "eventID": "ae241752-2a62-4166-9991-c1854aee9d36",
+  "readOnly": false,
+  "eventType": "AwsApiCall",
+  "managementEvent": true,
+  "recipientAccountId": "000000000000",
+  "eventCategory": "Management",
+  "tlsDetails": {
+    "tlsVersion": "TLSv1.3",
+    "cipherSuite": "TLS_AES_128_GCM_SHA256",
+    "clientProvidedHostHeader": "sns.ap-northeast-2.amazonaws.com"
+  },
+  "sessionCredentialFromConsole": "true"
+}

test/services/test_services.py

@@ -428,4 +428,19 @@ def test_create_queue(self):
         self.assertEqual(result['region'], 'ap-northeast-2')
         self.assertEqual(result['event_name'], 'CreateQueue')
         self.assertEqual(result['source_ip_address'], '127.0.0.1')
-        self.assertEqual(result['event_source'], 'sqs')
+        self.assertEqual(result['event_source'], 'sqs')
+
+
+class SNSTest(unittest.TestCase):
+    def test_create_topic(self):
+        with open('./samples/sns_CreateTopic.json') as f:
+            data = json.loads(f.read())
+
+        result = build_result(data)
+
+        self.assertEqual(result['resource_id'], ['sns-test.fifo'])
+        self.assertEqual(result['identity'], 'user/test_user')
+        self.assertEqual(result['region'], 'ap-northeast-2')
+        self.assertEqual(result['event_name'], 'CreateTopic')
+        self.assertEqual(result['source_ip_address'], '127.0.0.1')
+        self.assertEqual(result['event_source'], 'sns')