Merge pull request #24 from rubysoho07/amazon-q-rule

rubysoho07 · web-flow · commit 308e7c51ad28 · 2025-09-07T20:03:18.000+09:00
feat: Adopt Amazon Q Developer
diff --git a/.amazonq/rules/layer-python-rule.md b/.amazonq/rules/layer-python-rule.md
@@ -0,0 +1,56 @@
+# Lambda Layer in Python
+
+## Purpose
+
+This rule manages features to process CloudTrail log by event name.
+
+## Instructions
+
+* Please refer the rule of a CloudTrail event: https://docs.aws.amazon.com/awscloudtrail/latest/userguide/cloudtrail-events.html
+* Get AWS service name from `eventSource` in the event. If `eventSource` is "cloudtrail.amazonaws.com", use "cloudtrail" before "amazonaws.com". (ID: SERVICE_NAME)
+* Create a file if `layer/python/cloudtrail_watcher/services/SERVICE_NAME.py` doesn't exist. File structure is below. (ID: SERVICE_FILE)
+
+```python
+import boto3
+
+from .common import *
+
+SERVICE_NAME = boto3.client('SERVICE_NAME')
+
+def process_event(event: dict, set_tag: bool = False) -> dict:
+    """ Process CloudTrail event for SERVICE_NAME """
+	
+    result = dict()
+
+    return result
+```
+* If you created a new file, add file name to `__all__` in `lambda/python/cloudtrail_watcher/services/__init__.py` as a new module.
+
+* Get the name of the event from `eventName` in the CloudTrail event. (ID: EVENT_NAME)
+* Create a function in SERVICE_FILE. (ID: PROCESS_EVENT_FUNCTION)
+    * Function name: Start with `_process_`, add EVENT_NAME with snake case.
+    * Arguments: event in dict, set_tag in bool. The default value of set_tag is False.
+    * Return value: resource name in list
+    * If set_tag is True, check whether 'User' tag exists.
+        * If 'User' tag doesn't exists, set 'User' tag by using 'get_user_identity' function in 'common.py'. 
+* Add a rule below in process_event function.
+
+```python
+if event['eventName'] == 'EVENT_NAME':
+    result['resource_id'] = "Call the function created from PROCESS_FUNCTION"
+else:
+    message = f"Cannot process event: {event['eventName']}, eventID: f{event['eventID']}"
+    result['error'] = message
+```
+
+* Add permission for used in PROCESS_EVENT_FUNCTION.
+    * File name: "template.sar.yaml" in the root of the project
+    * Add permission in IAM Policy document of WatcherFunctionPolicy resource.
+
+## Priority
+
+High
+
+## Error Handling
+
+N/A
diff --git a/.amazonq/rules/test-modification-rule.md b/.amazonq/rules/test-modification-rule.md
@@ -0,0 +1,26 @@
+# Test Code Modification
+
+## Purpose
+
+This rule modifies test code.
+
+## Instructions
+
+* If added new file in test/services/samples, mask sensitive data. 
+    * If accountId is not "000000000000", change it to "000000000000".
+    * If values is not "test_user" after "user/", change it to "test_user".
+    * Change same values with sourceIPAddress to "127.0.0.1".
+    * Change principalId to "YOUR_PRINCIPAL_ID".
+    * Change accessKeyId to "YOUR_ACCESS_KEY_ID".
+    * If you discover same values in other places, change them in the same way.
+* Create test case for the action in test/services/test_services.py file.
+    * Create a class and a test case for the service if the class doesn't exist.
+    * If the class exists, just create a test case in the class.
+
+## Priority
+
+Low
+
+## Error Handling
+
+N/A
diff --git a/layer/python/cloudtrail_watcher/services/__init__.py b/layer/python/cloudtrail_watcher/services/__init__.py
@@ -15,5 +15,7 @@
     'airflow',
     'dynamodb',
     'elasticloadbalancing',
-    'cloudfront'
+    'cloudfront',
+    'sqs',
+    'sns'
 ]
diff --git a/layer/python/cloudtrail_watcher/services/sns.py b/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]
diff --git a/layer/python/cloudtrail_watcher/services/sqs.py b/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']]
diff --git a/template.sar.yaml b/template.sar.yaml
@@ -68,7 +68,9 @@ Resources:
               "cloudfront:ListTagsForResource",
               "cloudfront:TagResource",
               "ecr:ListTagsForResource",
-              "ecr:TagResource"
+              "ecr:TagResource",
+              "sqs:TagQueue",
+              "sns:TagResource"
             ]
             Resource: "*"
 
diff --git a/test/services/samples/sns_CreateTopic.json b/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"
+}
diff --git a/test/services/samples/sqs_CreateQueue.json b/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"
+}
diff --git a/test/services/test_process_event_by_service.py b/test/services/test_process_event_by_service.py
@@ -1,7 +1,7 @@
 import json
 import unittest
 
-from layer.python.cloudtrail_watcher.utils import build_result, notify_slack
+from cloudtrail_watcher.utils import build_result, notify_slack
 
 
 class ProcessEventTest(unittest.TestCase):
diff --git a/test/services/test_services.py b/test/services/test_services.py
@@ -1,7 +1,7 @@
 import json
 import unittest
 
-from layer.python.cloudtrail_watcher.utils import build_result
+from cloudtrail_watcher.utils import build_result
 
 
 class EC2Test(unittest.TestCase):
@@ -413,4 +413,34 @@ 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')
+
+
+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')

Original file line number	Diff line number	Diff line change
`@@ -15,5 +15,7 @@`
`15`	`15`	`'airflow',`
`16`	`16`	`'dynamodb',`
`17`	`17`	`'elasticloadbalancing',`
`18`		`- 'cloudfront'`
	`18`	`+ 'cloudfront',`
	`19`	`+ 'sqs',`
	`20`	`+ 'sns'`
`19`	`21`	`]`