Repository: localstack/localstack
Pull request: Fix SNS format when delivering to a DLQ
There was an issue when delivering to a SQS DLQ after failing to deliver an SNS notification. The format wasn't properly handled, and we passed the event directly down to the SQS client.

### Problem
As notified in #5604, the format of the message received from the DLQ after failing to deliver a message with SNS wasn't the same than AWS.
The issue was tracked down to the message formatting of an event to a SNS notification being done in a `try...except` block. When an Exception would rise, we would pass down the raw data to the SQS client, which would then `json.dumps` it directly as a `MessageBody`. 

### Solution
We need to move the creation of the message body outside the `try...except` block, and pass that message down to the SQS client delivering to the DLQ.
A minor issue arising from this fix is a change to the localstack/utils/aws/dead_letter_queue.py file. The utility function `_send_to_dead_letter_queue` expects an event of type `dict`. But in the case of a SNS delivery to a DLQ, the event being passed down is always a `str` (the message body). The event is never used in itself, but passed down all the way to the end to be sent. So I took the liberty to add the type signature to the method, warning that it can be a `str` or a `dict`. (In the past, in case of `RawDelivery` being set, the event passed down was already a `str`).

A small fix I've done is adding a header when `RawDelivery` is `true` in the case of a http endpoint being set for a SNS subscription. While reading the doc, I realised it was missing.
https://docs.aws.amazon.com/sns/latest/dg/sns-large-payload-raw-message-delivery.html

Thanks to the snapshot utility testing, I realised a field of SNS notifications was missing: `UnsubscribeURL`. It has been added as well.

I changed some tests (the one failing because of the right message format) to use the `WaitTimeSeconds` keyword in `sqs_client.receive_message` instead of retrying, it allows better error messages, often in `retry`, assert errors would give the first `assert` when the second one would be the culprit. It also feels better to use the functionality given by the `receive_messages` function, and it seems better suited to test against AWS.

### Tests
I've pushed a bit further and used the new `snapshot` utility. I added the usage in another commit, as I realise it could complicate navigating my tests as I tried to cover well the format of the received messages. 

_Fixes #5604_

Problem statement:
question: Is there any reason behind SNS DLQ message have different format than original AWS?
### Is there an existing issue for this?

- [X] I have searched the existing issues and read the documentation

### Question

I'm trying to implement locally DLQ for SNS with SQS, the problem is, on localstack SQS DLQ message coming from SNS have different format than what you would expect in AWS

When I tested locally I received the message in the following format:

`"{\"Action\": [\"Publish\"], \"Message\": [\"(The original SNS message body here)\"], \"TopicArn\": [\"the topic arn in localstack\"], \"Version\": [\"2010-03-31\"]}"`

I'm using local stack 0.14.0 with docker-compose

my docker-compose file looks like this:


```
 aws:
    image: localstack/localstack:0.14.0
    restart: always
    volumes:
      - ./docker-mounts/localstack:/docker-entrypoint-initaws.d
    environment:
      - SERVICES=sqs,s3,ses,sns
      - DEFAULT_REGION=eu-west-1
      - HOSTNAME_EXTERNAL=aws.localtest.local
      - AWS_ACCESS_KEY_ID=foo
      - AWS_SECRET_ACCESS_KEY=bar
      - AWS_DEFAULT_REGION=eu-west-1
```

### Anything else?

_No response_

Hints from issue discussion:

Commit message: add snapshot to tests

Patch to review:
diff --git a/localstack/services/sns/provider.py b/localstack/services/sns/provider.py
index c2b1efb69665a..c7a3988022df9 100644
--- a/localstack/services/sns/provider.py
+++ b/localstack/services/sns/provider.py
@@ -804,7 +804,7 @@ async def message_to_subscriber(
 
     elif subscriber["Protocol"] == "sqs":
         queue_url = None
-
+        message_body = create_sns_message_body(subscriber, req_data, message_id)
         try:
             endpoint = subscriber["Endpoint"]
 
@@ -813,8 +813,7 @@ async def message_to_subscriber(
             elif "://" in endpoint:
                 queue_url = endpoint
             else:
-                queue_name = endpoint.split(":")[5]
-                queue_url = aws_stack.get_sqs_queue_url(queue_name)
+                queue_url = aws_stack.get_sqs_queue_url(endpoint)
                 subscriber["sqs_queue_url"] = queue_url
 
             message_group_id = req_data.get("MessageGroupId", [""])[0]
@@ -831,7 +830,7 @@ async def message_to_subscriber(
 
             sqs_client.send_message(
                 QueueUrl=queue_url,
-                MessageBody=create_sns_message_body(subscriber, req_data, message_id),
+                MessageBody=message_body,
                 MessageAttributes=create_sqs_message_attributes(subscriber, message_attributes),
                 MessageSystemAttributes=create_sqs_system_attributes(headers),
                 **kwargs,
@@ -840,7 +839,7 @@ async def message_to_subscriber(
         except Exception as exc:
             LOG.info("Unable to forward SNS message to SQS: %s %s", exc, traceback.format_exc())
             store_delivery_log(subscriber, False, message, message_id)
-            sns_error_to_dead_letter_queue(subscriber["SubscriptionArn"], req_data, str(exc))
+            sns_error_to_dead_letter_queue(subscriber["SubscriptionArn"], message_body, str(exc))
             if "NonExistentQueue" in str(exc):
                 LOG.info(
                     'Removing non-existent queue "%s" subscribed to topic "%s"',
@@ -888,7 +887,8 @@ async def message_to_subscriber(
                 "Unable to run Lambda function on SNS message: %s %s", exc, traceback.format_exc()
             )
             store_delivery_log(subscriber, False, message, message_id)
-            sns_error_to_dead_letter_queue(subscriber["SubscriptionArn"], req_data, str(exc))
+            message_body = create_sns_message_body(subscriber, req_data, message_id)
+            sns_error_to_dead_letter_queue(subscriber["SubscriptionArn"], message_body, str(exc))
         return
 
     elif subscriber["Protocol"] in ["http", "https"]:
@@ -898,17 +898,24 @@ async def message_to_subscriber(
         except Exception:
             return
         try:
+            headers = {
+                "Content-Type": "text/plain",
+                # AWS headers according to
+                # https://docs.aws.amazon.com/sns/latest/dg/sns-message-and-json-formats.html#http-header
+                "x-amz-sns-message-type": msg_type,
+                "x-amz-sns-topic-arn": subscriber["TopicArn"],
+                "x-amz-sns-subscription-arn": subscriber["SubscriptionArn"],
+                "User-Agent": "Amazon Simple Notification Service Agent",
+            }
+            # When raw message delivery is enabled, x-amz-sns-rawdelivery needs to be set to 'true'
+            # indicating that the message has been published without JSON formatting.
+            # https://docs.aws.amazon.com/sns/latest/dg/sns-large-payload-raw-message-delivery.html
+            if is_raw_message_delivery(subscriber):
+                headers["x-amz-sns-rawdelivery"] = "true"
+
             response = requests.post(
                 subscriber["Endpoint"],
-                headers={
-                    "Content-Type": "text/plain",
-                    # AWS headers according to
-                    # https://docs.aws.amazon.com/sns/latest/dg/sns-message-and-json-formats.html#http-header
-                    "x-amz-sns-message-type": msg_type,
-                    "x-amz-sns-topic-arn": subscriber["TopicArn"],
-                    "x-amz-sns-subscription-arn": subscriber["SubscriptionArn"],
-                    "User-Agent": "Amazon Simple Notification Service Agent",
-                },
+                headers=headers,
                 data=message_body,
                 verify=False,
             )
@@ -925,7 +932,7 @@ async def message_to_subscriber(
                 "Received error on sending SNS message, putting to DLQ (if configured): %s", exc
             )
             store_delivery_log(subscriber, False, message, message_id)
-            sns_error_to_dead_letter_queue(subscriber["SubscriptionArn"], req_data, str(exc))
+            sns_error_to_dead_letter_queue(subscriber["SubscriptionArn"], message_body, str(exc))
         return
 
     elif subscriber["Protocol"] == "application":
@@ -940,6 +947,7 @@ async def message_to_subscriber(
                 traceback.format_exc(),
             )
             store_delivery_log(subscriber, False, message, message_id)
+            # check AWS format of DLQ message after application delivery failure, maybe format is wrong
             sns_error_to_dead_letter_queue(subscriber["SubscriptionArn"], req_data, str(exc))
         return
 
@@ -996,6 +1004,11 @@ def create_sns_message_body(subscriber, req_data, message_id=None):
             message["MessageAttributes"] = attributes
         return message
 
+    external_url = external_service_url("sns")
+    unsubscribe_url = (
+        f"{external_url}/?Action=Unsubscribe&SubscriptionArn={subscriber['SubscriptionArn']}"
+    )
+
     data = {
         "Type": req_data.get("Type", ["Notification"])[0],
         "MessageId": message_id,
@@ -1007,10 +1020,11 @@ def create_sns_message_body(subscriber, req_data, message_id=None):
         # Hardcoded
         "Signature": "EXAMPLEpH+..",
         "SigningCertURL": "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-0000000000000000000000.pem",
+        "UnsubscribeURL": unsubscribe_url,
     }
 
     for key in ["Subject", "SubscribeURL", "Token"]:
-        if req_data.get(key):
+        if key in req_data and req_data[key][0]:
             data[key] = req_data[key][0]
 
     for key in HTTP_SUBSCRIPTION_ATTRIBUTES:
diff --git a/localstack/utils/aws/dead_letter_queue.py b/localstack/utils/aws/dead_letter_queue.py
index 9f045cf85dbc5..1e89d1448ae94 100644
--- a/localstack/utils/aws/dead_letter_queue.py
+++ b/localstack/utils/aws/dead_letter_queue.py
@@ -2,7 +2,7 @@
 import logging
 import uuid
 from json import JSONDecodeError
-from typing import Dict, List
+from typing import Dict, List, Union
 
 from localstack.utils.aws import aws_stack
 from localstack.utils.aws.aws_models import LambdaFunction
@@ -32,7 +32,8 @@ def sqs_error_to_dead_letter_queue(queue_arn: str, event: Dict, error):
     return _send_to_dead_letter_queue("SQS", queue_arn, target_arn, event, error)
 
 
-def sns_error_to_dead_letter_queue(sns_subscriber_arn: str, event: Dict, error):
+def sns_error_to_dead_letter_queue(sns_subscriber_arn: str, event: Union[Dict, str], error):
+    # event should be of type str if coming from SNS, as it represents the message body being passed down
     client = aws_stack.connect_to_service("sns")
     attrs = client.get_subscription_attributes(SubscriptionArn=sns_subscriber_arn)
     attrs = attrs.get("Attributes", {})
@@ -49,7 +50,9 @@ def lambda_error_to_dead_letter_queue(func_details: LambdaFunction, event: Dict,
     return _send_to_dead_letter_queue("Lambda", source_arn, dlq_arn, event, error)
 
 
-def _send_to_dead_letter_queue(source_type: str, source_arn: str, dlq_arn: str, event: Dict, error):
+def _send_to_dead_letter_queue(
+    source_type: str, source_arn: str, dlq_arn: str, event: Union[Dict, str], error
+):
     if not dlq_arn:
         return
     LOG.info("Sending failed execution %s to dead letter queue %s", source_arn, dlq_arn)
@@ -87,7 +90,7 @@ def _send_to_dead_letter_queue(source_type: str, source_arn: str, dlq_arn: str,
     return dlq_arn
 
 
-def _prepare_messages_to_dlq(source_arn: str, event: Dict, error) -> List[Dict]:
+def _prepare_messages_to_dlq(source_arn: str, event: Union[Dict, str], error) -> List[Dict]:
     messages = []
     custom_attrs = {
         "RequestID": {"DataType": "String", "StringValue": str(uuid.uuid4())},
@@ -110,7 +113,7 @@ def _prepare_messages_to_dlq(source_arn: str, event: Dict, error) -> List[Dict]:
         messages.append(
             {
                 "Id": str(uuid.uuid4()),
-                "MessageBody": json.dumps(event),
+                "MessageBody": event,
                 "MessageAttributes": custom_attrs,
             }
         )
diff --git a/tests/integration/test_sns.py b/tests/integration/test_sns.py
index 660132e43c955..725fae7a8ad3a 100644
--- a/tests/integration/test_sns.py
+++ b/tests/integration/test_sns.py
@@ -2,6 +2,7 @@
 import json
 import queue
 import random
+import re
 import time
 
 import pytest
@@ -21,6 +22,7 @@
 from localstack.utils.common import (
     get_free_tcp_port,
     get_service_protocol,
+    poll_condition,
     retry,
     short_uid,
     to_str,
@@ -46,6 +48,24 @@
 PUBLICATION_RETRIES = 4
 
 
+# copy/pasted from test_sqs.py, utility function
+def queue_exists(sqs_client, queue_url: str) -> bool:
+    """
+    Checks whether a queue with the given queue URL exists.
+
+    :param sqs_client: the botocore client
+    :param queue_url: the queue URL
+    :return: true if the queue exists, false otherwise
+    """
+    try:
+        result = sqs_client.get_queue_url(QueueName=queue_url.split("/")[-1])
+        return result.get("QueueUrl") == queue_url
+    except ClientError as e:
+        if "NonExistentQueue" in e.response["Error"]["Code"]:
+            return False
+        raise
+
+
 class TestSNSSubscription:
     def test_python_lambda_subscribe_sns_topic(
         self,
@@ -677,15 +697,13 @@ def forward_request(self, method, path, data, headers):
             Message=json.dumps({"message": "test_redrive_policy"}),
         )
 
-        def receive_dlq():
-            result = sqs_client.receive_message(QueueUrl=dlq_url, MessageAttributeNames=["All"])
-            assert len(result["Messages"]) > 0
-            assert (
-                json.loads(json.loads(result["Messages"][0]["Body"])["Message"][0])["message"]
-                == "test_redrive_policy"
-            )
-
-        retry(receive_dlq, retries=7, sleep=2.5)
+        response = sqs_client.receive_message(QueueUrl=dlq_url, WaitTimeSeconds=10)
+        assert (
+            len(response["Messages"]) == 1
+        ), f"invalid number of messages in DLQ response {response}"
+        message = json.loads(response["Messages"][0]["Body"])
+        assert message["Type"] == "Notification"
+        assert json.loads(message["Message"])["message"] == "test_redrive_policy"
 
     def test_redrive_policy_lambda_subscription(
         self,
@@ -725,15 +743,13 @@ def test_redrive_policy_lambda_subscription(
             Message=json.dumps({"message": "test_redrive_policy"}),
         )
 
-        def receive_dlq():
-            result = sqs_client.receive_message(QueueUrl=dlq_url, MessageAttributeNames=["All"])
-            assert len(result["Messages"]) > 0
-            assert (
-                json.loads(json.loads(result["Messages"][0]["Body"])["Message"][0])["message"]
-                == "test_redrive_policy"
-            )
-
-        retry(receive_dlq, retries=10, sleep=2)
+        response = sqs_client.receive_message(QueueUrl=dlq_url, WaitTimeSeconds=10)
+        assert (
+            len(response["Messages"]) == 1
+        ), f"invalid number of messages in DLQ response {response}"
+        message = json.loads(response["Messages"][0]["Body"])
+        assert message["Type"] == "Notification"
+        assert json.loads(message["Message"])["message"] == "test_redrive_policy"
 
     def test_redrive_policy_queue_subscription(
         self,
@@ -766,15 +782,13 @@ def test_redrive_policy_queue_subscription(
             TopicArn=topic_arn, Message=json.dumps({"message": "test_redrive_policy"})
         )
 
-        def receive_dlq():
-            result = sqs_client.receive_message(QueueUrl=dlq_url, MessageAttributeNames=["All"])
-            assert len(result["Messages"]) > 0
-            assert (
-                json.loads(json.loads(result["Messages"][0]["Body"])["Message"][0])["message"]
-                == "test_redrive_policy"
-            )
-
-        retry(receive_dlq, retries=10, sleep=2)
+        response = sqs_client.receive_message(QueueUrl=dlq_url, WaitTimeSeconds=10)
+        assert (
+            len(response["Messages"]) == 1
+        ), f"invalid number of messages in DLQ response {response}"
+        message = json.loads(response["Messages"][0]["Body"])
+        assert message["Type"] == "Notification"
+        assert json.loads(message["Message"])["message"] == "test_redrive_policy"
 
     def test_publish_with_empty_subject(self, sns_client, sns_create_topic):
         topic_arn = sns_create_topic()["TopicArn"]
@@ -1510,3 +1524,96 @@ def test_empty_sns_message(
             )["Attributes"]["ApproximateNumberOfMessages"]
             == "0"
         )
+
+    @pytest.mark.parametrize("raw_message_delivery", [True, False])
+    @pytest.mark.aws_validated
+    def test_dead_letter_queue_with_deleted_sqs_queue(
+        self,
+        sns_client,
+        sqs_client,
+        sns_create_topic,
+        sqs_create_queue,
+        sqs_queue_arn,
+        sns_create_sqs_subscription,
+        raw_message_delivery,
+        snapshot,
+    ):
+        topic_arn = sns_create_topic()["TopicArn"]
+        queue_url = sqs_create_queue()
+
+        subscription = sns_create_sqs_subscription(topic_arn=topic_arn, queue_url=queue_url)
+
+        dlq_url = sqs_create_queue()
+        dlq_arn = sqs_queue_arn(dlq_url)
+
+        sns_client.set_subscription_attributes(
+            SubscriptionArn=subscription["SubscriptionArn"],
+            AttributeName="RedrivePolicy",
+            AttributeValue=json.dumps({"deadLetterTargetArn": dlq_arn}),
+        )
+
+        if raw_message_delivery:
+            sns_client.set_subscription_attributes(
+                SubscriptionArn=subscription["SubscriptionArn"],
+                AttributeName="RawMessageDelivery",
+                AttributeValue="true",
+            )
+
+        # allow topic to write to the DLQ sqs queue
+        # taken from localstack.testing.pytest.fixtures.sns_create_sqs_subscription#494
+        sqs_client.set_queue_attributes(
+            QueueUrl=dlq_url,
+            Attributes={
+                "Policy": json.dumps(
+                    {
+                        "Statement": [
+                            {
+                                "Effect": "Allow",
+                                "Principal": {"Service": "sns.amazonaws.com"},
+                                "Action": "sqs:SendMessage",
+                                "Resource": dlq_arn,
+                                "Condition": {"ArnEquals": {"aws:SourceArn": topic_arn}},
+                            }
+                        ]
+                    }
+                )
+            },
+        )
+
+        sqs_client.delete_queue(QueueUrl=queue_url)
+
+        # AWS takes some time to delete the queue, which make the test fails as it delivers the message correctly
+        assert poll_condition(lambda: not queue_exists(sqs_client, queue_url), timeout=5)
+
+        message = "test_dlq_after_sqs_endpoint_deleted"
+        sns_client.publish(TopicArn=topic_arn, Message=message)
+
+        response = sqs_client.receive_message(QueueUrl=dlq_url, WaitTimeSeconds=10)
+        assert (
+            len(response["Messages"]) == 1
+        ), f"invalid number of messages in DLQ response {response}"
+
+        snapshot.skip_key(re.compile(r"^ReceiptHandle$"), "<receipt-handle>")
+
+        if raw_message_delivery:
+            assert response["Messages"][0]["Body"] == message
+            snapshot.match("raw_message_delivery", response)
+        else:
+            received_message = json.loads(response["Messages"][0]["Body"])
+            assert received_message["Type"] == "Notification"
+            assert received_message["Message"] == message
+
+            # Set the decoded JSON Body to be able to skip keys directly
+            response["Messages"][0]["Body"] = received_message
+            # Need to skip the MD5OfBody/Signature, because it contains a timestamp
+            snapshot.skip_key(re.compile(r"^Signature$"), "<signature>")
+            snapshot.skip_key(re.compile(r"^MD5OfBody$"), "<md5-hash>")
+            snapshot.register_replacement(
+                pattern=re.compile(r"(?<=\bSimpleNotificationService-)(.*?)(?=\.)"),
+                value="<signing-cert-file>",
+            )
+            snapshot.register_replacement(
+                pattern=re.compile(r"(?<=://)(.*?)(?=/\?Action=Unsubscribe)"),
+                value="<unsubscribe-domain>",
+            )
+            snapshot.match("json_encoded_delivery", response)
diff --git a/tests/integration/test_sns.snapshot.json b/tests/integration/test_sns.snapshot.json
new file mode 100644
index 0000000000000..97c64c17cccc7
--- /dev/null
+++ b/tests/integration/test_sns.snapshot.json
@@ -0,0 +1,37 @@
+{
+  "tests/integration/test_sns.py::TestSNSProvider::test_dead_letter_queue_with_deleted_sqs_queue[True]": {
+    "raw_message_delivery": {
+      "Messages": [
+        {
+          "MessageId": "<uuid>",
+          "ReceiptHandle": "<receipt-handle>",
+          "MD5OfBody": "2040fa62e6ff90d21d5fe319d5e65443",
+          "Body": "test_dlq_after_sqs_endpoint_deleted"
+        }
+      ],
+      "ResponseMetadata": "<response-metadata>"
+    }
+  },
+  "tests/integration/test_sns.py::TestSNSProvider::test_dead_letter_queue_with_deleted_sqs_queue[False]": {
+    "json_encoded_delivery": {
+      "Messages": [
+        {
+          "MessageId": "<uuid>",
+          "ReceiptHandle": "<receipt-handle>",
+          "MD5OfBody": "<md5-hash>",
+          "Body": {
+            "Type": "Notification",
+            "MessageId": "<uuid>",
+            "TopicArn": "<arn>",
+            "Message": "test_dlq_after_sqs_endpoint_deleted",
+            "Timestamp": "<timestamp>",
+            "SignatureVersion": "1",
+            "Signature": "<signature>",
+            "SigningCertURL": "https://<unsubscribe-domain>/?Action=Unsubscribe&SubscriptionArn=<arn>"
+          }
+        }
+      ],
+      "ResponseMetadata": "<response-metadata>"
+    }
+  }
+}


Code review comments:
diff_hunk: @@ -46,6 +48,24 @@
 PUBLICATION_RETRIES = 4
 
 
+# copy/pasted from test_sqs.py, utility function
+def queue_exists(sqs_client, queue_url: str) -> bool:
+    """
+    Checks whether a queue with the given queue URL exists.
+
+    :param sqs_client: the botocore client
+    :param queue_url: the queue URL
+    :return: true if the queue exists, false otherwise
+    """
+    try:
+        result = sqs_client.get_queue_url(QueueName=queue_url.split("/")[-1])
+        return result.get("QueueUrl") == queue_url
+    except ClientError as e:
+        if "NonExistentQueue" in e.response["Error"]["Code"]:
+            return False
+        raise
path: tests/integration/test_sns.py
text: @user1:
nit: Instead of copying the function, we usually extract them and migrate them to pytest fixtures. They can then be easily used across all tests by using pytest's fixture injection. You can find lots of examples in `localstack/testing/pytest/fixtures.py`.

@author:
Thanks a lot, makes sense! It's done!

diff_hunk: @@ -1007,10 +1020,11 @@ def create_sns_message_body(subscriber, req_data, message_id=None):
         # Hardcoded
         "Signature": "EXAMPLEpH+..",
         "SigningCertURL": "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-0000000000000000000000.pem",
+        "UnsubscribeURL": unsubscribe_url,
     }
 
     for key in ["Subject", "SubscribeURL", "Token"]:
-        if req_data.get(key):
+        if key in req_data and req_data[key][0]:
path: localstack/services/sns/provider.py
text: @user1:
The previous condition checked if it is available and truthy. Your changed code could raise a `TypeError` or an `IndexError` if `req_data` holds the key with value `None` or an empty list.


@author:
The previous condition was always right, because it contains a list with None inside, which is not an empty list. It would add a `Subject: null` to the response.
But I see the issue now. This code was quite smart. I'll let the first check be `req_data.get(key)` but also check the first entry in the list is truthy.

diff_hunk: @@ -940,6 +947,7 @@ async def message_to_subscriber(
                 traceback.format_exc(),
             )
             store_delivery_log(subscriber, False, message, message_id)
+            # check AWS format of DLQ message after application delivery failure, maybe format is wrong
path: localstack/services/sns/provider.py
text: @user1:
q: Is this a leftover from analyzing the issue?

@author:
Sorry, yes, I didn't know how to properly test PushNotification with AWS directly, but after reading the docs it should be the same format, so it will be fixed as well.