pacnpal/django-anymail
1
0
Fork 0
mirror of https://github.com/pacnpal/django-anymail.git synced 2025-12-20 03:41:05 -05:00
Code Issues Packages Projects Releases Wiki Activity

Mailgun: Support new (non-legacy) webhooks

Browse Source 
Extend existing Mailgun tracking webhook handler to support both
original (legacy) and new (June, 2018) Mailgun webhooks.

Closes #117
This commit is contained in:
medmunds
2018-08-14 11:53:30 -07:00
parent dacc299e5a
commit 9e7814ad65
5 changed files with 563 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

386
tests/test_mailgun_webhooks.py
View File

@@ -16,14 +16,33 @@ from .webhook_cases import WebhookTestCase, WebhookBasicAuthTestsMixin
TEST_API_KEY = 'TEST_API_KEY'
def mailgun_sign(data, api_key=TEST_API_KEY):
def mailgun_signature(timestamp, token, api_key):
"""Generates a Mailgun webhook signature"""
# https://documentation.mailgun.com/en/latest/user_manual.html#securing-webhooks
return hmac.new(
key=api_key.encode('ascii'),
msg='{timestamp}{token}'.format(timestamp=timestamp, token=token).encode('ascii'),
digestmod=hashlib.sha256).hexdigest()
def mailgun_sign_payload(data, api_key=TEST_API_KEY):
"""Add or complete Mailgun webhook signature block in data dict"""
# Modifies the dict in place
event_data = data.get('event-data', {})
signature = data.setdefault('signature', {})
token = signature.setdefault('token', '1234567890abcdef1234567890abcdef')
timestamp = signature.setdefault('timestamp',
str(int(float(event_data.get('timestamp', '1234567890.123')))))
signature['signature'] = mailgun_signature(timestamp, token, api_key=api_key)
return data
def mailgun_sign_legacy_payload(data, api_key=TEST_API_KEY):
"""Add a Mailgun webhook signature to data dict"""
# Modifies the dict in place
data.setdefault('timestamp', '1234567890')
data.setdefault('token', '1234567890abcdef1234567890abcdef')
data['signature'] = hmac.new(key=api_key.encode('ascii'),
msg='{timestamp}{token}'.format(**data).encode('ascii'),
digestmod=hashlib.sha256).hexdigest()
data['signature'] = mailgun_signature(data['timestamp'], data['token'], api_key=api_key)
return data
@@ -42,8 +61,8 @@ def querydict_to_postdict(qd):
class MailgunWebhookSettingsTestCase(WebhookTestCase):
def test_requires_api_key(self):
with self.assertRaises(ImproperlyConfigured):
self.client.post('/anymail/mailgun/tracking/',
data=mailgun_sign({'event': 'delivered'}))
self.client.post('/anymail/mailgun/tracking/', content_type="application/json",
data=json.dumps(mailgun_sign_payload({'event-data': {'event': 'delivered'}})))
@override_settings(ANYMAIL_MAILGUN_API_KEY=TEST_API_KEY)
@@ -51,32 +70,347 @@ class MailgunWebhookSecurityTestCase(WebhookTestCase, WebhookBasicAuthTestsMixin
should_warn_if_no_auth = False # because we check webhook signature
def call_webhook(self):
return self.client.post('/anymail/mailgun/tracking/',
data=mailgun_sign({'event': 'delivered'}))
return self.client.post('/anymail/mailgun/tracking/', content_type="application/json",
data=json.dumps(mailgun_sign_payload({'event-data': {'event': 'delivered'}})))
# Additional tests are in WebhookBasicAuthTestsMixin
def test_verifies_correct_signature(self):
response = self.client.post('/anymail/mailgun/tracking/',
data=mailgun_sign({'event': 'delivered'}))
response = self.client.post('/anymail/mailgun/tracking/', content_type="application/json",
data=json.dumps(mailgun_sign_payload({'event-data': {'event': 'delivered'}})))
self.assertEqual(response.status_code, 200)
def test_verifies_missing_signature(self):
response = self.client.post('/anymail/mailgun/tracking/',
data={'event': 'delivered'})
response = self.client.post('/anymail/mailgun/tracking/', content_type="application/json",
data=json.dumps({'event-data': {'event': 'delivered'}}))
self.assertEqual(response.status_code, 400)
def test_verifies_bad_signature(self):
data = mailgun_sign({'event': 'delivered'}, api_key="wrong API key")
response = self.client.post('/anymail/mailgun/tracking/', data=data)
data = mailgun_sign_payload({'event-data': {'event': 'delivered'}},
api_key="wrong API key")
response = self.client.post('/anymail/mailgun/tracking/', content_type="application/json",
data=json.dumps(data))
self.assertEqual(response.status_code, 400)
@override_settings(ANYMAIL_MAILGUN_API_KEY=TEST_API_KEY)
class MailgunDeliveryTestCase(WebhookTestCase):
class MailgunTestCase(WebhookTestCase):
# Tests for Mailgun's new webhooks (announced 2018-06-29)
def test_delivered_event(self):
raw_event = mailgun_sign({
# This is an actual, complete (sanitized) "delivered" event as received from Mailgun.
# (For brevity, later tests omit several payload fields that aren't used by Anymail.)
raw_event = mailgun_sign_payload({
"signature": {
"timestamp": "1534108637",
"token": "651869375b9df3c98fc15c4889b102119add1235c38fc92824",
"signature": "...",
},
"event-data": {
"tags": [],
"timestamp": 1534108637.153125,
"storage": {
"url": "https://sw.api.mailgun.net/v3/domains/example.org/messages/eyJwI...",
"key": "eyJwI...",
},
"recipient-domain": "example.com",
"id": "hTWCTD81RtiDN-...",
"campaigns": [],
"user-variables": {},
"flags": {
"is-routed": False,
"is-authenticated": True,
"is-system-test": False,
"is-test-mode": False,
},
"log-level": "info",
"envelope": {
"sending-ip": "333.123.123.200",
"sender": "test@example.org",
"transport": "smtp",
"targets": "recipient@example.com",
},
"message": {
"headers": {
"to": "recipient@example.com",
"message-id": "20180812211713.1.DF5966851B4BAA99@example.org",
"from": "test@example.org",
"subject": "Testing",
},
"attachments": [],
"size": 809,
},
"recipient": "recipient@example.com",
"event": "delivered",
"delivery-status": {
"tls": True,
"mx-host": "smtp-in.example.com",
"attempt-no": 1,
"description": "",
"session-seconds": 3.5700838565826416,
"utf8": True,
"code": 250,
"message": "OK",
"certificate-verified": True,
},
},
})
response = self.client.post('/anymail/mailgun/tracking/',
data=json.dumps(raw_event), content_type='application/json')
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
event=ANY, esp_name='Mailgun')
event = kwargs['event']
self.assertIsInstance(event, AnymailTrackingEvent)
self.assertEqual(event.event_type, "delivered")
self.assertEqual(event.timestamp, datetime(2018, 8, 12, 21, 17, 17, microsecond=153125, tzinfo=utc))
self.assertEqual(event.message_id, "<20180812211713.1.DF5966851B4BAA99@example.org>")
# Note that Anymail uses the "token" as its normalized event_id:
self.assertEqual(event.event_id, "651869375b9df3c98fc15c4889b102119add1235c38fc92824")
# ... if you want the Mailgun "event id", that's available through the raw esp_event:
self.assertEqual(event.esp_event["event-data"]["id"], "hTWCTD81RtiDN-...")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.esp_event, raw_event)
self.assertEqual(event.tags, [])
self.assertEqual(event.metadata, {})
def test_failed_permanent_event(self):
raw_event = mailgun_sign_payload({
"event-data": {
"event": "failed",
"severity": "permanent",
"reason": "bounce",
"recipient": "invalid@example.com",
"timestamp": 1534110422.389832,
"log-level": "error",
"message": {
"headers": {
"to": "invalid@example.com",
"message-id": "20180812214658.1.0DF563D0B3597700@example.org",
"from": "Test Sender ",
},
},
"delivery-status": {
"tls": True,
"mx-host": "aspmx.l.example.org",
"attempt-no": 1,
"description": "",
"session-seconds": 2.952177047729492,
"utf8": True,
"code": 550,
"message": "5.1.1 The email account that you tried to reach does not exist. Please try\n"
"5.1.1 double-checking the recipient's email address for typos",
"certificate-verified": True
}
},
})
response = self.client.post('/anymail/mailgun/tracking/',
data=json.dumps(raw_event), content_type='application/json')
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
event=ANY, esp_name='Mailgun')
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, "")
self.assertEqual(event.mta_response,
"5.1.1 The email account that you tried to reach does not exist. Please try\n"
"5.1.1 double-checking the recipient's email address for typos")
def test_failed_temporary_event(self):
raw_event = mailgun_sign_payload({
"event-data": {
"event": "failed",
"severity": "temporary",
"reason": "generic",
"timestamp": 1534111899.659519,
"log-level": "warn",
"message": {
"headers": {
"to": "undeliverable@nomx.example.com",
"message-id": "20180812214638.1.4A7D468E9BC18C5D@example.org",
"from": "Test Sender ",
"subject": "Testing"
},
},
"recipient": "undeliverable@nomx.example.com",
"delivery-status": {
"attempt-no": 3,
"description": "No MX for nomx.example.com",
"session-seconds": 0.0,
"retry-seconds": 1800,
"code": 498,
"message": "No MX for nomx.example.com"
}
},
})
response = self.client.post('/anymail/mailgun/tracking/',
data=json.dumps(raw_event), content_type='application/json')
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
event=ANY, esp_name='Mailgun')
event = kwargs['event']
self.assertEqual(event.event_type, "bounced")
self.assertEqual(event.recipient, "undeliverable@nomx.example.com")
self.assertEqual(event.reject_reason, "bounced")
self.assertEqual(event.description, "No MX for nomx.example.com")
self.assertEqual(event.mta_response, "No MX for nomx.example.com")
def test_rejected_event(self):
# (The "rejected" event is documented and appears in Mailgun dashboard logs,
# but it doesn't appear to be delivered through webhooks as of 8/2018.)
# Note that this payload lacks the recipient field present in all other events.
raw_event = mailgun_sign_payload({
"event-data": {
"event": "rejected",
"timestamp": 1529704976.104692,
"log-level": "warn",
"reject": {
"reason": "Sandbox subdomains are for test purposes only.",
"description": "",
},
"message": {
"headers": {
"to": "Recipient Name <recipient@example.org>",
"message-id": "20180622220256.1.B31A451A2E5422BB@sandbox55887.mailgun.org",
"from": "test@sandbox55887.mailgun.org",
"subject": "Test Subject"
},
},
},
})
response = self.client.post('/anymail/mailgun/tracking/',
data=json.dumps(raw_event), content_type='application/json')
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
event=ANY, esp_name='Mailgun')
event = kwargs['event']
self.assertEqual(event.event_type, "rejected")
self.assertEqual(event.reject_reason, "other")
self.assertEqual(event.description, "Sandbox subdomains are for test purposes only.")
self.assertEqual(event.recipient, "recipient@example.org")
def test_complained_event(self):
raw_event = mailgun_sign_payload({
"event-data": {
"event": "complained",
"id": "ncV2XwymRUKbPek_MIM-Gw",
"timestamp": 1377214260.049634,
"log-level": "warn",
"recipient": "recipient@example.com",
"message": {
"headers": {
"to": "foo@recipient.com",
"message-id": "20130718032413.263EE2E0926@example.org",
"from": "Sender Name <sender@example.org>",
"subject": "We are not spammer",
},
},
},
})
response = self.client.post('/anymail/mailgun/tracking/',
data=json.dumps(raw_event), content_type='application/json')
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
event=ANY, esp_name='Mailgun')
event = kwargs['event']
self.assertEqual(event.event_type, "complained")
self.assertEqual(event.recipient, "recipient@example.com")
def test_unsubscribed_event(self):
raw_event = mailgun_sign_payload({
"event-data": {
"event": "unsubscribed",
"id": "W3X4JOhFT-OZidZGKKr9iA",
"timestamp": 1377213791.421473,
"log-level": "info",
"recipient": "recipient@example.com",
"message": {
"headers": {
"message-id": "20130822232216.13966.79700@samples.mailgun.org"
}
},
},
})
response = self.client.post('/anymail/mailgun/tracking/',
data=json.dumps(raw_event), content_type='application/json')
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
event=ANY, esp_name='Mailgun')
event = kwargs['event']
self.assertEqual(event.event_type, "unsubscribed")
self.assertEqual(event.recipient, "recipient@example.com")
def test_opened_event(self):
raw_event = mailgun_sign_payload({
"event-data": {
"event": "opened",
"timestamp": 1534109600.089676,
"recipient": "recipient@example.com",
"tags": ["welcome", "variation-A"],
"user-variables": {
"cohort": "2018-08-B",
"user_id": "123456"
},
"message": {
# Mailgun *only* includes the message-id header for opened, clicked events...
"headers": {
"message-id": "20180812213139.1.BC6694A917BB7E6A@example.org"
}
},
"geolocation": {
"country": "US",
"region": "CA",
"city": "San Francisco"
},
"ip": "888.222.444.111",
"client-info": {
"client-type": "browser",
"client-os": "OS X",
"device-type": "desktop",
"client-name": "Chrome",
"user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6)..."
},
}
})
response = self.client.post('/anymail/mailgun/tracking/',
data=json.dumps(raw_event), content_type='application/json')
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
event=ANY, esp_name='Mailgun')
event = kwargs['event']
self.assertEqual(event.event_type, "opened")
self.assertEqual(event.recipient, "recipient@example.com")
self.assertEqual(event.tags, ["welcome", "variation-A"])
self.assertEqual(event.metadata, {"cohort": "2018-08-B", "user_id": "123456"})
def test_clicked_event(self):
raw_event = mailgun_sign_payload({
"event-data": {
"event": "clicked",
"timestamp": 1534109600.089676,
"recipient": "recipient@example.com",
"url": "https://example.com/test"
}
})
response = self.client.post('/anymail/mailgun/tracking/',
data=json.dumps(raw_event), content_type='application/json')
self.assertEqual(response.status_code, 200)
kwargs = self.assert_handler_called_once_with(self.tracking_handler, sender=MailgunTrackingWebhookView,
event=ANY, esp_name='Mailgun')
event = kwargs['event']
self.assertEqual(event.event_type, "clicked")
self.assertEqual(event.click_url, "https://example.com/test")
@override_settings(ANYMAIL_MAILGUN_API_KEY=TEST_API_KEY)
class MailgunLegacyTestCase(WebhookTestCase):
# Tests for Mailgun's "legacy" webhooks
# (which were the only webhooks available prior to Anymail 4.0)
def test_delivered_event(self):
raw_event = mailgun_sign_legacy_payload({
'domain': 'example.com',
'message-headers': json.dumps([
["Sender", "from=example.com"],
@@ -113,7 +447,7 @@ class MailgunDeliveryTestCase(WebhookTestCase):
self.assertEqual(event.metadata, {})
def test_dropped_bounce(self):
raw_event = mailgun_sign({
raw_event = mailgun_sign_legacy_payload({
'code': '605',
'domain': 'example.com',
'description': 'Not delivering to previously bounced address',
@@ -152,7 +486,7 @@ class MailgunDeliveryTestCase(WebhookTestCase):
self.assertEqual(querydict_to_postdict(event.esp_event), raw_event)
def test_dropped_spam(self):
raw_event = mailgun_sign({
raw_event = mailgun_sign_legacy_payload({
'code': '607',
'description': 'Not delivering to a user who marked your messages as spam',
'reason': 'hardfail',
@@ -170,7 +504,7 @@ class MailgunDeliveryTestCase(WebhookTestCase):
self.assertEqual(event.description, 'Not delivering to a user who marked your messages as spam')
def test_dropped_timed_out(self):
raw_event = mailgun_sign({
raw_event = mailgun_sign_legacy_payload({
'code': '499',
'description': 'Unable to connect to MX servers: [example.com]',
'reason': 'old',
@@ -188,7 +522,7 @@ class MailgunDeliveryTestCase(WebhookTestCase):
self.assertEqual(event.description, 'Unable to connect to MX servers: [example.com]')
def test_invalid_mailbox(self):
raw_event = mailgun_sign({
raw_event = mailgun_sign_legacy_payload({
'code': '550',
'error': "550 5.1.1 The email account that you tried to reach does not exist. Please try "
" 5.1.1 double-checking the recipient's email address for typos or "
@@ -209,7 +543,7 @@ class MailgunDeliveryTestCase(WebhookTestCase):
def test_alt_smtp_code(self):
# In some cases, Mailgun uses RFC-3463 extended SMTP status codes (x.y.z, rather than nnn).
# See issue #62.
raw_event = mailgun_sign({
raw_event = mailgun_sign_legacy_payload({
'code': '5.1.1',
'error': 'smtp;550 5.1.1 RESOLVER.ADR.RecipNotFound; not found',
'event': 'bounced',
@@ -228,7 +562,7 @@ class MailgunDeliveryTestCase(WebhookTestCase):
def test_metadata_message_headers(self):
# Metadata fields are interspersed with other data, but also in message-headers
# for delivered, bounced and dropped events
raw_event = mailgun_sign({
raw_event = mailgun_sign_legacy_payload({
'event': 'delivered',
'message-headers': json.dumps([
["X-Mailgun-Variables", "{\"custom1\": \"value1\", \"custom2\": \"{\\\"key\\\":\\\"value\\\"}\"}"],
@@ -244,7 +578,7 @@ class MailgunDeliveryTestCase(WebhookTestCase):
def test_metadata_post_fields(self):
# Metadata fields are only interspersed with other event params
# for opened, clicked, unsubscribed events
raw_event = mailgun_sign({
raw_event = mailgun_sign_legacy_payload({
'event': 'clicked',
'custom1': 'value1',
'custom2': '{"key":"value"}', # you can store JSON, but you'll need to unpack it yourself
@@ -267,7 +601,7 @@ class MailgunDeliveryTestCase(WebhookTestCase):
"ordinary field": "ordinary metadata value",
}
raw_event = mailgun_sign({
raw_event = mailgun_sign_legacy_payload({
'event': 'clicked',
'recipient': 'actual-recipient@example.com',
'token': 'actual-event-token',
@@ -305,7 +639,7 @@ class MailgunDeliveryTestCase(WebhookTestCase):
def test_tags(self):
# Most events include multiple 'tag' fields for message's tags
raw_event = mailgun_sign({
raw_event = mailgun_sign_legacy_payload({
'tag': ['tag1', 'tag2'], # Django TestClient encodes list as multiple field values
'event': 'opened',
})
@@ -316,7 +650,7 @@ class MailgunDeliveryTestCase(WebhookTestCase):
def test_x_tags(self):
# Delivery events don't include 'tag', but do include 'X-Mailgun-Tag' fields
raw_event = mailgun_sign({
raw_event = mailgun_sign_legacy_payload({
'X-Mailgun-Tag': ['tag1', 'tag2'],
'event': 'delivered',
})