mirror of
https://github.com/pacnpal/django-anymail.git
synced 2025-12-20 03:41:05 -05:00
466 lines
18 KiB
Python
466 lines
18 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import ANY
|
|
|
|
from django.core.exceptions import ImproperlyConfigured
|
|
from django.test import override_settings, tag
|
|
|
|
from anymail.exceptions import AnymailConfigurationError
|
|
from anymail.signals import AnymailTrackingEvent
|
|
from anymail.webhooks.mailersend import MailerSendTrackingWebhookView
|
|
|
|
from .webhook_cases import WebhookBasicAuthTestCase, WebhookTestCase
|
|
|
|
TEST_WEBHOOK_SIGNING_SECRET = "TEST_WEBHOOK_SIGNING_SECRET"
|
|
|
|
|
|
def mailersend_signature(data, secret):
|
|
"""Generate a MailerSend webhook signature for data with secret"""
|
|
# https://developers.mailersend.com/api/v1/webhooks.html#security
|
|
return hmac.new(
|
|
key=secret.encode("ascii"),
|
|
msg=data,
|
|
digestmod=hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
|
|
class MailerSendWebhookTestCase(WebhookTestCase):
|
|
def client_post_signed(self, url, json_data, secret=TEST_WEBHOOK_SIGNING_SECRET):
|
|
"""Return self.client.post(url, serialized json_data) signed with secret"""
|
|
# MailerSend for some reason backslash-escapes all forward slashes ("/")
|
|
# in its webhook payloads ("https:\/\/www..."). This is unnecessary, but
|
|
# harmless. We emulate it here to make sure it won't cause problems.
|
|
data = json.dumps(json_data).replace("/", "\\/").encode("ascii")
|
|
signature = mailersend_signature(data, secret)
|
|
return self.client.post(
|
|
url, content_type="application/json", data=data, HTTP_SIGNATURE=signature
|
|
)
|
|
|
|
|
|
@tag("mailersend")
|
|
class MailerSendWebhookSettingsTestCase(MailerSendWebhookTestCase):
|
|
def test_requires_signing_secret(self):
|
|
with self.assertRaisesMessage(
|
|
ImproperlyConfigured, "MAILERSEND_SIGNING_SECRET"
|
|
):
|
|
self.client_post_signed(
|
|
"/anymail/mailersend/tracking/", {"data": {"type": "sent"}}
|
|
)
|
|
|
|
@override_settings(
|
|
ANYMAIL={
|
|
"MAILERSEND_SIGNING_SECRET": "webhook secret",
|
|
"MAILERSEND_INBOUND_SECRET": "inbound secret",
|
|
}
|
|
)
|
|
def test_inbound_secret_is_different(self):
|
|
response = self.client_post_signed(
|
|
"/anymail/mailersend/tracking/",
|
|
{"data": {"type": "sent"}},
|
|
secret="webhook secret",
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
@override_settings(ANYMAIL_MAILERSEND_SIGNING_SECRET="settings secret")
|
|
def test_signing_secret_view_params(self):
|
|
"""Webhook signing secret can be provided as a view param"""
|
|
view = MailerSendTrackingWebhookView.as_view(signing_secret="view-level secret")
|
|
view_instance = view.view_class(**view.view_initkwargs)
|
|
self.assertEqual(view_instance.signing_secret, b"view-level secret")
|
|
|
|
|
|
@tag("mailersend")
|
|
@override_settings(ANYMAIL_MAILERSEND_SIGNING_SECRET=TEST_WEBHOOK_SIGNING_SECRET)
|
|
class MailerSendWebhookSecurityTestCase(
|
|
MailerSendWebhookTestCase, WebhookBasicAuthTestCase
|
|
):
|
|
should_warn_if_no_auth = False # because we check webhook signature
|
|
|
|
def call_webhook(self):
|
|
return self.client_post_signed(
|
|
"/anymail/mailersend/tracking/",
|
|
{"data": {"type": "sent"}},
|
|
secret=TEST_WEBHOOK_SIGNING_SECRET,
|
|
)
|
|
|
|
# Additional tests are in WebhookBasicAuthTestCase
|
|
|
|
def test_verifies_correct_signature(self):
|
|
response = self.client_post_signed(
|
|
"/anymail/mailersend/tracking/",
|
|
{"data": {"type": "sent"}},
|
|
secret=TEST_WEBHOOK_SIGNING_SECRET,
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_verifies_missing_signature(self):
|
|
response = self.client.post(
|
|
"/anymail/mailersend/tracking/",
|
|
content_type="application/json",
|
|
data=json.dumps({"data": {"type": "sent"}}),
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_verifies_bad_signature(self):
|
|
# This also verifies that the error log references the correct setting to check.
|
|
with self.assertLogs() as logs:
|
|
response = self.client_post_signed(
|
|
"/anymail/mailersend/tracking/",
|
|
{"data": {"type": "sent"}},
|
|
secret="wrong signing key",
|
|
)
|
|
# SuspiciousOperation causes 400 response (even in test client):
|
|
self.assertEqual(response.status_code, 400)
|
|
self.assertIn("check Anymail MAILERSEND_SIGNING_SECRET", logs.output[0])
|
|
|
|
|
|
@tag("mailersend")
|
|
@override_settings(ANYMAIL_MAILERSEND_SIGNING_SECRET=TEST_WEBHOOK_SIGNING_SECRET)
|
|
class MailerSendTestCase(MailerSendWebhookTestCase):
|
|
def test_sent_event(self):
|
|
# This is an actual, complete (sanitized) "sent" event as received from
|
|
# MailerSend. (For brevity, later tests omit several payload fields that
|
|
# Anymail doesn't use.)
|
|
raw_event = {
|
|
"type": "activity.sent",
|
|
"domain_id": "[domain-id-redacted]",
|
|
"created_at": "2023-02-27T21:09:49.520507Z",
|
|
"webhook_id": "[webhook-id-redacted]",
|
|
"url": "https://test.anymail.dev/anymail/mailersend/tracking/",
|
|
"data": {
|
|
"object": "activity",
|
|
"id": "63fd1c1d31b9c750540fe85c",
|
|
"type": "sent",
|
|
"created_at": "2023-02-27T21:09:49.506000Z",
|
|
"email": {
|
|
"object": "email",
|
|
"id": "63fd1c1de225707fa905f0a8",
|
|
"created_at": "2023-02-27T21:09:49.141000Z",
|
|
"from": "sender@mailersend.anymail.dev",
|
|
"subject": "Test webhooks",
|
|
"status": "sent",
|
|
"tags": ["tag1", "Tag 2"],
|
|
"message": {
|
|
"object": "message",
|
|
"id": "63fd1c1d5f010335ed07066b",
|
|
"created_at": "2023-02-27T21:09:49.061000Z",
|
|
},
|
|
"recipient": {
|
|
"object": "recipient",
|
|
"id": "63f3bb1965d98aa98c07d6b7",
|
|
"email": "recipient@example.com",
|
|
"created_at": "2023-02-20T18:25:29.162000Z",
|
|
},
|
|
},
|
|
"morph": None,
|
|
"template_id": "",
|
|
},
|
|
}
|
|
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(
|
|
self.tracking_handler,
|
|
sender=MailerSendTrackingWebhookView,
|
|
event=ANY,
|
|
esp_name="MailerSend",
|
|
)
|
|
event = kwargs["event"]
|
|
self.assertIsInstance(event, AnymailTrackingEvent)
|
|
self.assertEqual(event.event_type, "sent")
|
|
# event.timestamp comes from data.created_at:
|
|
self.assertEqual(
|
|
event.timestamp,
|
|
# "2023-02-27T21:09:49.506000Z"
|
|
datetime(2023, 2, 27, 21, 9, 49, microsecond=506000, tzinfo=timezone.utc),
|
|
)
|
|
# event.message_id matches the message.anymail_status.message_id when the
|
|
# message was sent. It comes from data.email.message.id:
|
|
self.assertEqual(event.message_id, "63fd1c1d5f010335ed07066b")
|
|
# event.event_id is unique for each event, and comes from data.id:
|
|
self.assertEqual(event.event_id, "63fd1c1d31b9c750540fe85c")
|
|
self.assertEqual(event.recipient, "recipient@example.com")
|
|
self.assertEqual(event.tags, ["tag1", "Tag 2"])
|
|
self.assertEqual(event.metadata, {}) # MailerSend doesn't support metadata
|
|
self.assertEqual(event.esp_event, raw_event)
|
|
|
|
# You can construct the sent Message-ID header (which is different from the
|
|
# event.message_id, and is unique per recipient) from esp_event.data.email.id:
|
|
sent_message_id = f"<{event.esp_event['data']['email']['id']}@mailersend.net>"
|
|
self.assertEqual(sent_message_id, "<63fd1c1de225707fa905f0a8@mailersend.net>")
|
|
|
|
def test_delivered_event(self):
|
|
raw_event = {
|
|
"type": "activity.delivered",
|
|
"data": {
|
|
"object": "activity",
|
|
"id": "63fd1c1fcfbe46145d003a7b",
|
|
"type": "delivered",
|
|
"created_at": "2023-02-27T21:09:51.865000Z",
|
|
"email": {
|
|
"status": "delivered",
|
|
"message": {
|
|
"id": "63fd1c1d5f010335ed07066b",
|
|
},
|
|
"recipient": {
|
|
"email": "recipient@example.com",
|
|
},
|
|
},
|
|
"morph": None,
|
|
},
|
|
}
|
|
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(
|
|
self.tracking_handler,
|
|
sender=MailerSendTrackingWebhookView,
|
|
event=ANY,
|
|
esp_name="MailerSend",
|
|
)
|
|
event = kwargs["event"]
|
|
self.assertIsInstance(event, AnymailTrackingEvent)
|
|
self.assertEqual(event.event_type, "delivered")
|
|
self.assertEqual(
|
|
event.timestamp,
|
|
# "2023-02-27T21:09:51.865000Z"
|
|
datetime(2023, 2, 27, 21, 9, 51, microsecond=865000, tzinfo=timezone.utc),
|
|
)
|
|
self.assertEqual(event.message_id, "63fd1c1d5f010335ed07066b")
|
|
self.assertEqual(event.event_id, "63fd1c1fcfbe46145d003a7b")
|
|
self.assertEqual(event.recipient, "recipient@example.com")
|
|
|
|
def test_hard_bounced_event(self):
|
|
raw_event = {
|
|
"type": "activity.hard_bounced",
|
|
"data": {
|
|
"id": "63fd251d5c00f8e52001fce6",
|
|
"type": "hard_bounced",
|
|
"created_at": "2023-02-27T21:48:13.593000Z",
|
|
"email": {
|
|
"status": "rejected",
|
|
"message": {
|
|
"id": "63fd25194d5edba3da09e044",
|
|
},
|
|
"recipient": {
|
|
"email": "invalid@example.com",
|
|
},
|
|
},
|
|
"morph": {
|
|
"object": "recipient_bounce",
|
|
"reason": "Host or domain name not found",
|
|
},
|
|
},
|
|
}
|
|
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(
|
|
self.tracking_handler,
|
|
sender=MailerSendTrackingWebhookView,
|
|
event=ANY,
|
|
esp_name="MailerSend",
|
|
)
|
|
event = kwargs["event"]
|
|
self.assertEqual(event.event_type, "bounced")
|
|
self.assertEqual(event.recipient, "invalid@example.com")
|
|
self.assertEqual(event.reject_reason, "bounced")
|
|
self.assertEqual(event.description, "Host or domain name not found")
|
|
self.assertIsNone(event.mta_response) # raw MTA info not provided
|
|
|
|
def test_soft_bounced_event(self):
|
|
raw_event = {
|
|
"type": "activity.soft_bounced",
|
|
"data": {
|
|
"object": "activity",
|
|
"id": "62f114f8165fe0d8db0288e5",
|
|
"type": "soft_bounced",
|
|
"created_at": "2022-08-08T13:51:52.747000Z",
|
|
"email": {
|
|
"status": "rejected",
|
|
"tags": None,
|
|
"message": {
|
|
"id": "62fb66bef54a112e920b5493",
|
|
},
|
|
"recipient": {
|
|
"email": "notauser@example.com",
|
|
},
|
|
},
|
|
"morph": {
|
|
"object": "recipient_bounce",
|
|
"reason": "Unknown reason",
|
|
},
|
|
},
|
|
}
|
|
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(
|
|
self.tracking_handler,
|
|
sender=MailerSendTrackingWebhookView,
|
|
event=ANY,
|
|
esp_name="MailerSend",
|
|
)
|
|
event = kwargs["event"]
|
|
self.assertEqual(event.event_type, "bounced")
|
|
self.assertEqual(event.recipient, "notauser@example.com")
|
|
self.assertEqual(event.reject_reason, "bounced")
|
|
self.assertEqual(event.description, "Unknown reason")
|
|
self.assertIsNone(event.mta_response) # raw MTA info not provided
|
|
|
|
def test_spam_complaint_event(self):
|
|
raw_event = {
|
|
"type": "activity.spam_complaint",
|
|
"data": {
|
|
"id": "62f114f8165fe0d8db0288e5",
|
|
"type": "spam_complaint",
|
|
"created_at": "2022-08-08T13:51:52.747000Z",
|
|
"email": {
|
|
"status": "delivered",
|
|
"message": {
|
|
"id": "62fb66bef54a112e920b5493",
|
|
},
|
|
"recipient": {
|
|
"email": "recipient@example.com",
|
|
},
|
|
},
|
|
"morph": {
|
|
"object": "spam_complaint",
|
|
"reason": None,
|
|
},
|
|
},
|
|
}
|
|
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(
|
|
self.tracking_handler,
|
|
sender=MailerSendTrackingWebhookView,
|
|
event=ANY,
|
|
esp_name="MailerSend",
|
|
)
|
|
event = kwargs["event"]
|
|
self.assertEqual(event.event_type, "complained")
|
|
self.assertEqual(event.recipient, "recipient@example.com")
|
|
|
|
def test_unsubscribed_event(self):
|
|
raw_event = {
|
|
"type": "activity.unsubscribed",
|
|
"data": {
|
|
"id": "63fd21c23f2bdd360e07d6b2",
|
|
"type": "unsubscribed",
|
|
"created_at": "2023-02-27T21:33:54.791000Z",
|
|
"email": {
|
|
"status": "delivered",
|
|
"message": {
|
|
"id": "63fd1c1d5f010335ed07066b",
|
|
},
|
|
"recipient": {
|
|
"email": "recipient@example.com",
|
|
},
|
|
},
|
|
"morph": {
|
|
"object": "recipient_unsubscribe",
|
|
"reason": "option_3",
|
|
"readable_reason": "I get too many emails",
|
|
},
|
|
},
|
|
}
|
|
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(
|
|
self.tracking_handler,
|
|
sender=MailerSendTrackingWebhookView,
|
|
event=ANY,
|
|
esp_name="MailerSend",
|
|
)
|
|
event = kwargs["event"]
|
|
self.assertEqual(event.event_type, "unsubscribed")
|
|
self.assertEqual(event.recipient, "recipient@example.com")
|
|
self.assertEqual(event.description, "I get too many emails")
|
|
|
|
def test_opened_event(self):
|
|
raw_event = {
|
|
"type": "activity.opened",
|
|
"data": {
|
|
"id": "63fd1e05532bd6cc700a793b",
|
|
"type": "opened",
|
|
"created_at": "2023-02-27T21:17:57.025000Z",
|
|
"email": {
|
|
"status": "delivered",
|
|
"message": {
|
|
"id": "63fd1c1d5f010335ed07066b",
|
|
},
|
|
"recipient": {
|
|
"email": "recipient@example.com",
|
|
},
|
|
},
|
|
"morph": {
|
|
"object": "open",
|
|
"id": "63fd1e05532bd6cc700a793a",
|
|
"created_at": "2023-02-27T21:17:57.018000Z",
|
|
"ip": "10.10.10.10",
|
|
},
|
|
},
|
|
}
|
|
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(
|
|
self.tracking_handler,
|
|
sender=MailerSendTrackingWebhookView,
|
|
event=ANY,
|
|
esp_name="MailerSend",
|
|
)
|
|
event = kwargs["event"]
|
|
self.assertEqual(event.event_type, "opened")
|
|
self.assertEqual(event.recipient, "recipient@example.com")
|
|
|
|
def test_clicked_event(self):
|
|
raw_event = {
|
|
"type": "activity.clicked",
|
|
"data": {
|
|
"id": "63fd1d23afa3c770b00da7d3",
|
|
"type": "clicked",
|
|
"created_at": "2023-02-27T21:14:11.691000Z",
|
|
"email": {
|
|
"status": "delivered",
|
|
"message": {
|
|
"id": "63fd1c1d5f010335ed07066b",
|
|
},
|
|
"recipient": {
|
|
"email": "recipient@example.com",
|
|
},
|
|
},
|
|
"morph": {
|
|
"object": "click",
|
|
"id": "63fd1d23afa3c770b00da7d2",
|
|
"created_at": "2023-02-27T21:14:11.679000Z",
|
|
"ip": "10.10.10.10",
|
|
"url": "https://example.com/test",
|
|
},
|
|
},
|
|
}
|
|
response = self.client_post_signed("/anymail/mailersend/tracking/", raw_event)
|
|
self.assertEqual(response.status_code, 200)
|
|
kwargs = self.assert_handler_called_once_with(
|
|
self.tracking_handler,
|
|
sender=MailerSendTrackingWebhookView,
|
|
event=ANY,
|
|
esp_name="MailerSend",
|
|
)
|
|
event = kwargs["event"]
|
|
self.assertEqual(event.event_type, "clicked")
|
|
self.assertEqual(event.recipient, "recipient@example.com")
|
|
self.assertEqual(event.click_url, "https://example.com/test")
|
|
|
|
def test_misconfigured_inbound(self):
|
|
errmsg = (
|
|
"You seem to have set MailerSend's *inbound* route endpoint"
|
|
" to Anymail's MailerSend *activity tracking* webhook URL."
|
|
)
|
|
with self.assertRaisesMessage(AnymailConfigurationError, errmsg):
|
|
self.client_post_signed(
|
|
"/anymail/mailersend/tracking/",
|
|
{
|
|
"type": "inbound.message",
|
|
"data": {"object": "message", "raw": "..."},
|
|
},
|
|
)
|