import json import warnings from datetime import datetime, timezone from unittest.mock import ANY, patch from django.test import SimpleTestCase, override_settings, tag from anymail.exceptions import AnymailConfigurationError, AnymailInsecureWebhookWarning from anymail.signals import AnymailTrackingEvent from anymail.webhooks.amazon_ses import AmazonSESTrackingWebhookView from .webhook_cases import WebhookBasicAuthTestCase, WebhookTestCase class AmazonSESWebhookTestsMixin(SimpleTestCase): def post_from_sns(self, path, raw_sns_message, **kwargs): return self.client.post( path, content_type="text/plain; charset=UTF-8", # SNS posts JSON as text/plain data=json.dumps(raw_sns_message), HTTP_X_AMZ_SNS_MESSAGE_ID=raw_sns_message["MessageId"], HTTP_X_AMZ_SNS_MESSAGE_TYPE=raw_sns_message["Type"], # Anymail doesn't use other x-amz-sns-* headers **kwargs, ) @tag("amazon_ses") class AmazonSESWebhookSecurityTests( AmazonSESWebhookTestsMixin, WebhookBasicAuthTestCase ): def call_webhook(self): return self.post_from_sns( "/anymail/amazon_ses/tracking/", {"Type": "Notification", "MessageId": "123", "Message": "{}"}, ) # Most actual tests are in WebhookBasicAuthTestCase def test_verifies_missing_auth(self): # Must handle missing auth header slightly differently from Anymail default 400 # SuspiciousOperation: SNS will only send basic auth after missing auth responds # 401 WWW-Authenticate: Basic realm="..." self.clear_basic_auth() response = self.call_webhook() self.assertEqual(response.status_code, 401) self.assertEqual( response["WWW-Authenticate"], 'Basic realm="Anymail WEBHOOK_SECRET"' ) @tag("amazon_ses") class AmazonSESNotificationsTests(WebhookTestCase, AmazonSESWebhookTestsMixin): def test_bounce_event(self): # This test includes a complete Amazon SES example event. # (Later tests omit some payload for brevity.) # https://docs.aws.amazon.com/ses/latest/DeveloperGuide/notification-examples.html#notification-examples-bounce raw_ses_event = { "notificationType": "Bounce", "bounce": { "bounceType": "Permanent", "reportingMTA": "dns; email.example.com", "bouncedRecipients": [ { "emailAddress": "jane@example.com", "status": "5.1.1", "action": "failed", "diagnosticCode": "smtp; 550 5.1.1 ..." " User unknown", } ], "bounceSubType": "General", # when bounce sent (by receiving ISP): "timestamp": "2016-01-27T14:59:44.101Z", # unique id for bounce: "feedbackId": "00000138111222aa-44455566-cccc" "-cccc-cccc-ddddaaaa068a-000000", "remoteMtaIp": "127.0.2.0", }, "mail": { "timestamp": "2016-01-27T14:59:38.237Z", # when message sent "source": "john@example.com", "sourceArn": "arn:aws:ses:us-west-2:888888888888:identity/example.com", "sourceIp": "127.0.3.0", "sendingAccountId": "123456789012", "messageId": "00000138111222aa-33322211-cccc" "-cccc-cccc-ddddaaaa0680-000000", "destination": [ "jane@example.com", "mary@example.com", "richard@example.com", ], "headersTruncated": False, "headers": [ {"name": "From", "value": '"John Doe" '}, { "name": "To", "value": '"Jane Doe" ,' ' "Mary Doe" ,' ' "Richard Doe" ', }, {"name": "Message-ID", "value": "custom-message-ID"}, {"name": "Subject", "value": "Hello"}, {"name": "Content-Type", "value": 'text/plain; charset="UTF-8"'}, {"name": "Content-Transfer-Encoding", "value": "base64"}, {"name": "Date", "value": "Wed, 27 Jan 2016 14:05:45 +0000"}, {"name": "X-Tag", "value": "tag 1"}, {"name": "X-Tag", "value": "tag 2"}, {"name": "X-Metadata", "value": '{"meta1":"string","meta2":2}'}, ], "commonHeaders": { "from": ["John Doe "], "date": "Wed, 27 Jan 2016 14:05:45 +0000", "to": [ "Jane Doe , Mary Doe ," " Richard Doe " ], "messageId": "custom-message-ID", "subject": "Hello", }, }, } raw_sns_message = { "Type": "Notification", # unique id for SNS event: "MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc", "TopicArn": "arn:aws:sns:us-east-1:1234567890:SES_Events", "Subject": "Amazon SES Email Event Notification", "Message": json.dumps(raw_ses_event) + "\n", "Timestamp": "2018-03-26T17:58:59.675Z", "SignatureVersion": "1", "Signature": "EXAMPLE-SIGNATURE==", "SigningCertURL": "https://sns.us-east-1.amazonaws.com" "/SimpleNotificationService-12345abcde.pem", "UnsubscribeURL": "https://sns.us-east-1.amazonaws.com" "/?Action=Unsubscribe&SubscriptionArn=arn...", } response = self.post_from_sns("/anymail/amazon_ses/tracking/", raw_sns_message) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.tracking_handler, sender=AmazonSESTrackingWebhookView, event=ANY, esp_name="Amazon SES", ) event = kwargs["event"] self.assertIsInstance(event, AnymailTrackingEvent) self.assertEqual(event.event_type, "bounced") self.assertEqual(event.esp_event, raw_ses_event) # timestamp from SNS: self.assertEqual( event.timestamp, datetime(2018, 3, 26, 17, 58, 59, microsecond=675000, tzinfo=timezone.utc), ) self.assertEqual( event.message_id, "00000138111222aa-33322211-cccc-cccc-cccc-ddddaaaa0680-000000", ) self.assertEqual(event.event_id, "19ba9823-d7f2-53c1-860e-cb10e0d13dfc") self.assertEqual(event.recipient, "jane@example.com") self.assertEqual(event.reject_reason, "bounced") self.assertEqual(event.description, "Permanent: General") self.assertEqual( event.mta_response, "smtp; 550 5.1.1 ... User unknown" ) self.assertEqual(event.tags, ["tag 1", "tag 2"]) self.assertEqual(event.metadata, {"meta1": "string", "meta2": 2}) # For brevity, remaining tests omit some event fields that aren't used by Anymail def test_multiple_bounce_event(self): """Amazon SES notification can cover multiple recipients""" raw_ses_event = { "notificationType": "Bounce", "bounce": { "bounceType": "Permanent", "bounceSubType": "General", "bouncedRecipients": [ {"emailAddress": "jane@example.com"}, {"emailAddress": "richard@example.com"}, ], }, "mail": { "messageId": "00000137860315fd-34208509-5b74" "-41f3-95c5-22c1edc3c924-000000", "destination": [ "jane@example.com", "mary@example.com", "richard@example.com", ], }, } raw_sns_message = { "Type": "Notification", "MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc", "Message": json.dumps(raw_ses_event) + "\n", "Timestamp": "2018-03-26T17:58:59.675Z", } response = self.post_from_sns("/anymail/amazon_ses/tracking/", raw_sns_message) self.assertEqual(response.status_code, 200) # tracking handler should be called twice -- once for each bounced recipient # (but not for the third, non-bounced recipient) self.assertEqual(self.tracking_handler.call_count, 2) _, kwargs = self.tracking_handler.call_args_list[0] event = kwargs["event"] self.assertEqual(event.event_type, "bounced") self.assertEqual(event.recipient, "jane@example.com") self.assertEqual(event.description, "Permanent: General") self.assertIsNone(event.mta_response) _, kwargs = self.tracking_handler.call_args_list[1] event = kwargs["event"] self.assertEqual(event.esp_event, raw_ses_event) self.assertEqual(event.recipient, "richard@example.com") def test_complaint_event(self): raw_ses_event = { "notificationType": "Complaint", "complaint": { "userAgent": "AnyCompany Feedback Loop (V0.01)", "complainedRecipients": [{"emailAddress": "richard@example.com"}], "complaintFeedbackType": "abuse", }, "mail": { "messageId": "000001378603177f-7a5433e7-8edb" "-42ae-af10-f0181f34d6ee-000000", "destination": [ "jane@example.com", "mary@example.com", "richard@example.com", ], }, } raw_sns_message = { "Type": "Notification", "MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc", "Message": json.dumps(raw_ses_event) + "\n", "Timestamp": "2018-03-26T17:58:59.675Z", } response = self.post_from_sns("/anymail/amazon_ses/tracking/", raw_sns_message) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.tracking_handler, sender=AmazonSESTrackingWebhookView, event=ANY, esp_name="Amazon SES", ) event = kwargs["event"] self.assertEqual(event.event_type, "complained") self.assertEqual(event.recipient, "richard@example.com") self.assertEqual(event.reject_reason, "spam") self.assertEqual(event.description, "abuse") self.assertEqual(event.user_agent, "AnyCompany Feedback Loop (V0.01)") def test_delivery_event(self): raw_ses_event = { "notificationType": "Delivery", "mail": { "timestamp": "2016-01-27T14:59:38.237Z", "messageId": "0000014644fe5ef6-9a483358-9170" "-4cb4-a269-f5dcdf415321-000000", "destination": [ "jane@example.com", "mary@example.com", "richard@example.com", ], }, "delivery": { "timestamp": "2016-01-27T14:59:38.237Z", "recipients": ["jane@example.com"], "processingTimeMillis": 546, "reportingMTA": "a8-70.smtp-out.amazonses.com", "smtpResponse": "250 ok: Message 64111812 accepted", "remoteMtaIp": "127.0.2.0", }, } raw_sns_message = { "Type": "Notification", "MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc", "Message": json.dumps(raw_ses_event) + "\n", "Timestamp": "2018-03-26T17:58:59.675Z", } response = self.post_from_sns("/anymail/amazon_ses/tracking/", raw_sns_message) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.tracking_handler, sender=AmazonSESTrackingWebhookView, event=ANY, esp_name="Amazon SES", ) event = kwargs["event"] self.assertEqual(event.event_type, "delivered") self.assertEqual(event.recipient, "jane@example.com") self.assertEqual(event.mta_response, "250 ok: Message 64111812 accepted") def test_send_event(self): raw_ses_event = { "eventType": "Send", "mail": { "timestamp": "2016-10-14T05:02:16.645Z", "messageId": "7c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000", "destination": ["recipient@example.com"], "tags": { "ses:configuration-set": ["ConfigSet"], "ses:source-ip": ["192.0.2.0"], "ses:from-domain": ["example.com"], "ses:caller-identity": ["ses_user"], "myCustomTag1": ["myCustomTagValue1"], "myCustomTag2": ["myCustomTagValue2"], }, }, "send": {}, } raw_sns_message = { "Type": "Notification", "MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc", "Message": json.dumps(raw_ses_event) + "\n", "Timestamp": "2018-03-26T17:58:59.675Z", } response = self.post_from_sns("/anymail/amazon_ses/tracking/", raw_sns_message) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.tracking_handler, sender=AmazonSESTrackingWebhookView, event=ANY, esp_name="Amazon SES", ) event = kwargs["event"] self.assertIsInstance(event, AnymailTrackingEvent) self.assertEqual(event.event_type, "sent") self.assertEqual(event.esp_event, raw_ses_event) # timestamp from SNS: self.assertEqual( event.timestamp, datetime(2018, 3, 26, 17, 58, 59, microsecond=675000, tzinfo=timezone.utc), ) self.assertEqual( event.message_id, "7c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000" ) self.assertEqual(event.event_id, "19ba9823-d7f2-53c1-860e-cb10e0d13dfc") self.assertEqual(event.recipient, "recipient@example.com") # Anymail doesn't load Amazon SES "Message Tags": self.assertEqual(event.tags, []) self.assertEqual(event.metadata, {}) def test_reject_event(self): raw_ses_event = { "eventType": "Reject", "mail": { "timestamp": "2016-10-14T17:38:15.211Z", "messageId": "7c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000", "destination": ["recipient@example.com"], }, "reject": {"reason": "Bad content"}, } raw_sns_message = { "Type": "Notification", "MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc", "Message": json.dumps(raw_ses_event) + "\n", "Timestamp": "2018-03-26T17:58:59.675Z", } response = self.post_from_sns("/anymail/amazon_ses/tracking/", raw_sns_message) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.tracking_handler, sender=AmazonSESTrackingWebhookView, event=ANY, esp_name="Amazon SES", ) event = kwargs["event"] self.assertEqual(event.event_type, "rejected") self.assertEqual(event.reject_reason, "blocked") self.assertEqual(event.description, "Bad content") self.assertEqual(event.recipient, "recipient@example.com") def test_open_event(self): raw_ses_event = { "eventType": "Open", "mail": { "destination": ["recipient@example.com"], "messageId": "7c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000", }, "open": { "ipAddress": "192.0.2.1", "timestamp": "2017-08-09T22:00:19.652Z", "userAgent": "Mozilla/5.0" " (iPhone; CPU iPhone OS 10_3_3 like Mac OS X)...", }, } raw_sns_message = { "Type": "Notification", "MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc", "Message": json.dumps(raw_ses_event) + "\n", "Timestamp": "2018-03-26T17:58:59.675Z", } response = self.post_from_sns("/anymail/amazon_ses/tracking/", raw_sns_message) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.tracking_handler, sender=AmazonSESTrackingWebhookView, event=ANY, esp_name="Amazon SES", ) event = kwargs["event"] self.assertEqual(event.event_type, "opened") self.assertEqual(event.recipient, "recipient@example.com") self.assertEqual( event.user_agent, "Mozilla/5.0 (iPhone; CPU iPhone OS 10_3_3 like Mac OS X)...", ) def test_click_event(self): raw_ses_event = { "eventType": "Click", "click": { "ipAddress": "192.0.2.1", "link": "https://docs.aws.amazon.com/ses/latest/DeveloperGuide/", "linkTags": { "samplekey0": ["samplevalue0"], "samplekey1": ["samplevalue1"], }, "timestamp": "2017-08-09T23:51:25.570Z", "userAgent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" " AppleWebKit/537.36...", }, "mail": { "destination": ["recipient@example.com"], "messageId": "7c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000", }, } raw_sns_message = { "Type": "Notification", "MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc", "Message": json.dumps(raw_ses_event) + "\n", "Timestamp": "2018-03-26T17:58:59.675Z", } response = self.post_from_sns("/anymail/amazon_ses/tracking/", raw_sns_message) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.tracking_handler, sender=AmazonSESTrackingWebhookView, event=ANY, esp_name="Amazon SES", ) event = kwargs["event"] self.assertEqual(event.event_type, "clicked") self.assertEqual(event.recipient, "recipient@example.com") self.assertEqual( event.user_agent, "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36...", ) self.assertEqual( event.click_url, "https://docs.aws.amazon.com/ses/latest/DeveloperGuide/" ) def test_rendering_failure_event(self): raw_ses_event = { "eventType": "Rendering Failure", "mail": { "messageId": "c191be45-e9aedb9a-02f9-4d12-a87d-dd0099a07f8a-000000", "destination": ["recipient@example.com"], }, "failure": { "errorMessage": "Attribute 'attributeName' is not present" " in the rendering data.", "templateName": "MyTemplate", }, } raw_sns_message = { "Type": "Notification", "MessageId": "19ba9823-d7f2-53c1-860e-cb10e0d13dfc", "Message": json.dumps(raw_ses_event) + "\n", "Timestamp": "2018-03-26T17:58:59.675Z", } response = self.post_from_sns("/anymail/amazon_ses/tracking/", raw_sns_message) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.tracking_handler, sender=AmazonSESTrackingWebhookView, event=ANY, esp_name="Amazon SES", ) event = kwargs["event"] self.assertEqual(event.event_type, "failed") self.assertEqual(event.recipient, "recipient@example.com") self.assertEqual( event.description, "Attribute 'attributeName' is not present in the rendering data.", ) def test_incorrect_received_event(self): """The tracking webhook should warn if it receives inbound events""" raw_sns_message = { "Type": "Notification", "MessageId": "8f6dee70-c885-558a-be7d-bd48bbf5335e", "TopicArn": "arn:aws:sns:us-east-1:111111111111:SES_Inbound", "Message": '{"notificationType": "Received"}', } with self.assertRaisesMessage( AnymailConfigurationError, "You seem to have set an Amazon SES *inbound* receipt rule to publish to an" " SNS Topic that posts to Anymail's *tracking* webhook URL. (SNS TopicArn" " arn:aws:sns:us-east-1:111111111111:SES_Inbound)", ): self.post_from_sns("/anymail/amazon_ses/tracking/", raw_sns_message) @tag("amazon_ses") class AmazonSESSubscriptionManagementTests(WebhookTestCase, AmazonSESWebhookTestsMixin): # Anymail will automatically respond to SNS subscription notifications # if Anymail is configured to require basic auth via WEBHOOK_SECRET. # (Note that WebhookTestCase sets up ANYMAIL WEBHOOK_SECRET.) def setUp(self): super().setUp() # Mock boto3.session.Session().client('sns').confirm_subscription (and any # other client operations). (We could also use botocore.stub.Stubber, but mock # works well with our test structure.) self.patch_boto3_session = patch( "anymail.webhooks.amazon_ses.boto3.session.Session", autospec=True ) #: boto3.session.Session self.mock_session = self.patch_boto3_session.start() self.addCleanup(self.patch_boto3_session.stop) #: boto3.session.Session().client self.mock_client = self.mock_session.return_value.client #: boto3.session.Session().client('sns', ...) self.mock_client_instance = self.mock_client.return_value self.mock_client_instance.confirm_subscription.return_value = { "SubscriptionArn": "arn:aws:sns:us-west-2:123456789012:SES_Notifications" ":aaaaaaa-..." } SNS_SUBSCRIPTION_CONFIRMATION = { "Type": "SubscriptionConfirmation", "MessageId": "165545c9-2a5c-472c-8df2-7ff2be2b3b1b", "Token": "EXAMPLE_TOKEN", "TopicArn": "arn:aws:sns:us-west-2:123456789012:SES_Notifications", "Message": "You have chosen to subscribe ...\n" "To confirm..., visit the SubscribeURL included in this message.", "SubscribeURL": "https://sns.us-west-2.amazonaws.com/" "?Action=ConfirmSubscription&TopicArn=...", "Timestamp": "2012-04-26T20:45:04.751Z", "SignatureVersion": "1", "Signature": "EXAMPLE-SIGNATURE==", "SigningCertURL": "https://sns.us-west-2.amazonaws.com/" "SimpleNotificationService-12345abcde.pem", } def test_sns_subscription_auto_confirmation(self): """Anymail webhook will auto-confirm SNS topic subscriptions""" response = self.post_from_sns( "/anymail/amazon_ses/tracking/", self.SNS_SUBSCRIPTION_CONFIRMATION ) self.assertEqual(response.status_code, 200) # auto-confirmed: self.mock_client.assert_called_once_with( "sns", config=ANY, region_name="us-west-2" ) self.mock_client_instance.confirm_subscription.assert_called_once_with( TopicArn="arn:aws:sns:us-west-2:123456789012:SES_Notifications", Token="EXAMPLE_TOKEN", AuthenticateOnUnsubscribe="true", ) # didn't notify receivers: self.assertEqual(self.tracking_handler.call_count, 0) self.assertEqual(self.inbound_handler.call_count, 0) def test_sns_subscription_confirmation_failure(self): """Auto-confirmation allows error through if confirm call fails""" from botocore.exceptions import ClientError self.mock_client_instance.confirm_subscription.side_effect = ClientError( { "Error": { "Type": "Sender", "Code": "InternalError", "Message": "Gremlins!", }, "ResponseMetadata": { "RequestId": "aaaaaaaa-2222-1111-8888-bbbb3333bbbb", "HTTPStatusCode": 500, }, }, operation_name="confirm_subscription", ) with self.assertRaisesMessage(ClientError, "Gremlins!"): self.post_from_sns( "/anymail/amazon_ses/tracking/", self.SNS_SUBSCRIPTION_CONFIRMATION ) # didn't notify receivers: self.assertEqual(self.tracking_handler.call_count, 0) self.assertEqual(self.inbound_handler.call_count, 0) @override_settings(ANYMAIL_AMAZON_SES_CLIENT_PARAMS={"region_name": "us-east-1"}) def test_sns_subscription_confirmation_different_region(self): """ Anymail confirms the subscription in the SNS Topic's own region, rather than any default region """ # (The SNS_SUBSCRIPTION_CONFIRMATION above has a TopicArn in region us-west-2) self.post_from_sns( "/anymail/amazon_ses/tracking/", self.SNS_SUBSCRIPTION_CONFIRMATION ) self.mock_client.assert_called_once_with( "sns", config=ANY, region_name="us-west-2" ) # clear WEBHOOK_SECRET setting from base WebhookTestCase @override_settings(ANYMAIL={}) def test_sns_subscription_confirmation_auth_disabled(self): """ Anymail *won't* auto-confirm SNS subscriptions if WEBHOOK_SECRET isn't in use """ # (warning gets tested elsewhere) warnings.simplefilter("ignore", AnymailInsecureWebhookWarning) with self.assertLogs("django.security.AnymailWebhookValidationFailure") as cm: response = self.post_from_sns( "/anymail/amazon_ses/tracking/", self.SNS_SUBSCRIPTION_CONFIRMATION ) self.assertEqual(response.status_code, 400) # bad request self.assertEqual( [ "Anymail received an unexpected SubscriptionConfirmation request for" " Amazon SNS topic" " 'arn:aws:sns:us-west-2:123456789012:SES_Notifications'." " (Anymail can automatically confirm SNS subscriptions if you set" " a WEBHOOK_SECRET and use that in your SNS notification url." " Or you can manually confirm this subscription in the SNS dashboard" " with token 'EXAMPLE_TOKEN'.)" ], [record.getMessage() for record in cm.records], ) # *didn't* try to confirm the subscription: self.assertEqual(self.mock_client_instance.confirm_subscription.call_count, 0) # didn't notify receivers: self.assertEqual(self.tracking_handler.call_count, 0) self.assertEqual(self.inbound_handler.call_count, 0) def test_sns_confirmation_success_notification(self): """ Anymail ignores the 'Successfully validated' notification after confirming an SNS subscription """ response = self.post_from_sns( "/anymail/amazon_ses/tracking/", { "Type": "Notification", "MessageId": "7fbca0d9-eeab-5285-ae27-f3f57f2e84b0", "TopicArn": "arn:aws:sns:us-west-2:123456789012:SES_Notifications", "Message": "Successfully validated SNS topic" " for Amazon SES event publishing.", "Timestamp": "2018-03-21T16:58:45.077Z", "SignatureVersion": "1", "Signature": "EXAMPLE_SIGNATURE==", "SigningCertURL": "https://sns.us-east-1.amazonaws.com" "/SimpleNotificationService-12345abcde.pem", "UnsubscribeURL": "https://sns.us-east-1.amazonaws.com" "/?Action=Unsubscribe...", }, ) self.assertEqual(response.status_code, 200) # didn't notify receivers: self.assertEqual(self.tracking_handler.call_count, 0) self.assertEqual(self.inbound_handler.call_count, 0) def test_sns_unsubscribe_confirmation(self): """ Anymail ignores the UnsubscribeConfirmation SNS message after deleting a subscription """ response = self.post_from_sns( "/anymail/amazon_ses/tracking/", { "Type": "UnsubscribeConfirmation", "MessageId": "47138184-6831-46b8-8f7c-afc488602d7d", "Token": "EXAMPLE_TOKEN", "TopicArn": "arn:aws:sns:us-west-2:123456789012:SES_Notifications", "Message": "You have chosen to deactivate subscription ...\n" "To cancel ... visit the SubscribeURL...", "SubscribeURL": "https://sns.us-west-2.amazonaws.com" "/?Action=ConfirmSubscription&TopicArn=...", "Timestamp": "2012-04-26T20:06:41.581Z", "SignatureVersion": "1", "Signature": "EXAMPLE_SIGNATURE==", "SigningCertURL": "https://sns.us-east-1.amazonaws.com" "/SimpleNotificationService-12345abcde.pem", }, ) self.assertEqual(response.status_code, 200) # *didn't* try to use the Token to re-enable the subscription: self.assertEqual(self.mock_client_instance.confirm_subscription.call_count, 0) # didn't notify receivers: self.assertEqual(self.tracking_handler.call_count, 0) self.assertEqual(self.inbound_handler.call_count, 0) @override_settings(ANYMAIL_AMAZON_SES_AUTO_CONFIRM_SNS_SUBSCRIPTIONS=False) def test_disable_auto_confirmation(self): """ The ANYMAIL setting AMAZON_SES_AUTO_CONFIRM_SNS_SUBSCRIPTIONS will disable this feature """ response = self.post_from_sns( "/anymail/amazon_ses/tracking/", self.SNS_SUBSCRIPTION_CONFIRMATION ) self.assertEqual(response.status_code, 200) # *didn't* try to subscribe: self.assertEqual(self.mock_session.call_count, 0) self.assertEqual(self.mock_client.call_count, 0) # didn't notify receivers: self.assertEqual(self.tracking_handler.call_count, 0) self.assertEqual(self.inbound_handler.call_count, 0)