fix: fix parsing of non-JSON messages

This commit fixes the parsing of messages that aren't JSON-formatted.

In specific, any non-JSON-formatted message that happened to contain the
string `"attachments"` or `"text"` was incorrectly treated as a
dictionary, and failed to parse.

This commit also makes the typing a little stricter.

Author: cdepillabout

functions/messages/text_message.json

@@ -8,7 +8,7 @@
       "MessageId": "f86e3c5b-cd17-1ab8-80e9-c0776d4f1e7a",
       "TopicArn": "arn:aws:sns:us-gov-west-1:123456789012:ExampleTopic",
       "Subject": "All Fine",
-      "Message": "This\nis\na typical multi-line\nmessage from SNS!\n\nHave a ~good~ amazing day! :)",
+      "Message": "This\nis\na typical multi-line\ntext-formatted message from SNS!\n\nHave a ~good~ amazing day! :)",
       "Timestamp": "2019-02-12T15:45:24.091Z",
       "SignatureVersion": "1",
       "Signature": "EXAMPLE",

functions/notify_slack.py

@@ -574,7 +574,7 @@ def format_default(
     return attachments
 
 
-def parse_notification(message: Dict[str, Any], subject: Optional[str], region: str) -> Optional[Dict]:
+def parse_notification(message: Union[str, Dict[str, Any]], subject: Optional[str], region: str) -> Dict[str, Any]:
     """
     Parse notification message and format into Slack message payload
 
@@ -583,7 +583,7 @@ def parse_notification(message: Dict[str, Any], subject: Optional[str], region:
     :params region: AWS region where the event originated from
     :returns: Slack message payload
     """
-    if "AlarmName" in message:
+    if isinstance(message, Dict) and "AlarmName" in message:
         return format_cloudwatch_alarm(message=message, region=region)
     if isinstance(message, Dict) and message.get("detail-type") == "GuardDuty Finding":
         return format_guardduty_finding(message=message, region=message["region"])
@@ -619,23 +619,18 @@ def get_slack_message_payload(
         "username": slack_username,
         "icon_emoji": slack_emoji,
     }
-    attachment = None
 
     if isinstance(message, str):
         try:
             message = json.loads(message)
         except json.JSONDecodeError:
             logging.info("Not a structured payload, just a string message")
 
-    message = cast(Dict[str, Any], message)
-
-    if "attachments" in message or "text" in message:
+    if isinstance(message, Dict) and ("attachments" in message or "text" in message):
+        message = cast(Dict[str, Any], message)
         payload = {**payload, **message}
     else:
-        attachment = parse_notification(message, subject, region)
-
-    if attachment:
-        payload["attachments"] = [attachment]  # type: ignore
+        payload["attachments"] = [parse_notification(message, subject, region)]
 
     return payload
 

functions/snapshots/snap_notify_slack_test.py

@@ -645,7 +645,7 @@
                         'value': '''This
 is
 a typical multi-line
-message from SNS!
+text-formatted message from SNS!
 
 Have a ~good~ amazing day! :)'''
                     }