feat: Process sqs:CreateQueue event

Author: rubysoho07

layer/python/cloudtrail_watcher/services/sqs.py

@@ -0,0 +1,35 @@
+import boto3
+
+from .common import *
+
+sqs = boto3.client('sqs')
+
+
+def process_event(event: dict, set_tag: bool = False) -> dict:
+    """ Process CloudTrail event for SQS """
+
+    result = dict()
+
+    if event['eventName'] == 'CreateQueue':
+        result['resource_id'] = _process_create_queue(event, set_tag)
+    else:
+        message = f"Cannot process event: {event['eventName']}, eventID: {event['eventID']}"
+        result['error'] = message
+
+    return result
+
+
+def _process_create_queue(event: dict, set_tag: bool = False) -> list:
+    """ Process CreateQueue event """
+    
+    if set_tag is True:
+        if 'tags' not in event['requestParameters'] or \
+           check_contain_mandatory_tag_dict(event['requestParameters']['tags']) is False:
+            sqs.tag_queue(
+                QueueUrl=event['responseElements']['queueUrl'],
+                Tags={
+                    'User': get_user_identity(event)
+                }
+            )
+
+    return [event['requestParameters']['queueName']]

test/services/samples/sqs_CreateQueue.json

@@ -0,0 +1,64 @@
+{
+  "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-05T10:43:21Z",
+  "eventSource": "sqs.amazonaws.com",
+  "eventName": "CreateQueue",
+  "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": {
+    "queueName": "test-queue",
+    "attributes": {
+      "Policy": "{\"Version\":\"2012-10-17\",\"Id\":\"__default_policy_ID\",\"Statement\":[{\"Sid\":\"__owner_statement\",\"Effect\":\"Allow\",\"Principal\":{\"AWS\":\"000000000000\"},\"Action\":[\"SQS:*\"],\"Resource\":\"arn:aws:sqs:ap-northeast-2:000000000000:test-queue\"}]}",
+      "ReceiveMessageWaitTimeSeconds": "0",
+      "SqsManagedSseEnabled": "true",
+      "DelaySeconds": "0",
+      "KmsMasterKeyId": "",
+      "RedrivePolicy": "",
+      "MessageRetentionPeriod": "345600",
+      "MaximumMessageSize": "1048576",
+      "VisibilityTimeout": "30",
+      "RedriveAllowPolicy": ""
+    },
+    "tags": {
+      "TestKey": "TestValue"
+    }
+  },
+  "responseElements": {
+    "queueUrl": "https://sqs.ap-northeast-2.amazonaws.com/000000000000/test-queue"
+  },
+  "requestID": "ded62e37-7691-5699-bb20-4f03b7984293",
+  "eventID": "b5a8ec24-29c1-4ebd-8dbb-3e328e66818f",
+  "readOnly": false,
+  "resources": [
+    {
+      "accountId": "000000000000",
+      "type": "AWS::SQS::Queue",
+      "ARN": "arn:aws:sqs:ap-northeast-2:000000000000:test-queue"
+    }
+  ],
+  "eventType": "AwsApiCall",
+  "managementEvent": true,
+  "recipientAccountId": "000000000000",
+  "eventCategory": "Management",
+  "tlsDetails": {
+    "tlsVersion": "TLSv1.3",
+    "cipherSuite": "TLS_AES_128_GCM_SHA256",
+    "clientProvidedHostHeader": "sqs.ap-northeast-2.amazonaws.com"
+  },
+  "sessionCredentialFromConsole": "true"
+}

test/services/test_services.py

@@ -413,4 +413,19 @@ def test_create_repository(self):
         self.assertEqual(result['region'], 'ap-northeast-2')
         self.assertEqual(result['event_name'], 'CreateRepository')
         self.assertEqual(result['source_ip_address'], '127.0.0.1')
-        self.assertEqual(result['event_source'], 'ecr')
+        self.assertEqual(result['event_source'], 'ecr')
+
+
+class SQSTest(unittest.TestCase):
+    def test_create_queue(self):
+        with open('./samples/sqs_CreateQueue.json') as f:
+            data = json.loads(f.read())
+
+        result = build_result(data)
+
+        self.assertEqual(result['resource_id'], ['test-queue'])
+        self.assertEqual(result['identity'], 'user/test_user')
+        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')