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.mailgun import MailgunTrackingWebhookView from .webhook_cases import WebhookBasicAuthTestCase, WebhookTestCase TEST_WEBHOOK_SIGNING_KEY = "TEST_WEBHOOK_SIGNING_KEY" def mailgun_signature(timestamp, token, webhook_signing_key): """Generates a Mailgun webhook signature""" # https://documentation.mailgun.com/en/latest/user_manual.html#securing-webhooks return hmac.new( key=webhook_signing_key.encode("ascii"), msg="{timestamp}{token}".format(timestamp=timestamp, token=token).encode( "ascii" ), digestmod=hashlib.sha256, ).hexdigest() def mailgun_sign_payload(data, webhook_signing_key=TEST_WEBHOOK_SIGNING_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, webhook_signing_key=webhook_signing_key ) return data def mailgun_sign_legacy_payload(data, webhook_signing_key=TEST_WEBHOOK_SIGNING_KEY): """Add a Mailgun webhook signature to data dict""" # Modifies the dict in place data.setdefault("timestamp", "1234567890") data.setdefault("token", "1234567890abcdef1234567890abcdef") data["signature"] = mailgun_signature( data["timestamp"], data["token"], webhook_signing_key=webhook_signing_key ) return data def querydict_to_postdict(qd): """Converts a Django QueryDict to a TestClient.post(data)-style dict Single-value fields appear as normal Multi-value fields appear as a list (differs from QueryDict.dict) """ return {key: values if len(values) > 1 else values[0] for key, values in qd.lists()} @tag("mailgun") class MailgunWebhookSettingsTestCase(WebhookTestCase): def test_requires_webhook_signing_key(self): with self.assertRaisesMessage( ImproperlyConfigured, "MAILGUN_WEBHOOK_SIGNING_KEY" ): 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", ANYMAIL_MAILGUN_WEBHOOK_SIGNING_KEY="TEST_WEBHOOK_SIGNING_KEY", ) def test_webhook_signing_is_different_from_api_key(self): """ Webhooks should use MAILGUN_WEBHOOK_SIGNING_KEY, not MAILGUN_API_KEY, if both provided """ payload = json.dumps( mailgun_sign_payload( {"event-data": {"event": "delivered"}}, webhook_signing_key="TEST_WEBHOOK_SIGNING_KEY", ) ) response = self.client.post( "/anymail/mailgun/tracking/", content_type="application/json", data=payload ) self.assertEqual(response.status_code, 200) @override_settings(ANYMAIL_MAILGUN_API_KEY="TEST_API_KEY") def test_defaults_webhook_signing_to_api_key(self): """ Webhooks should default to MAILGUN_API_KEY if MAILGUN_WEBHOOK_SIGNING_KEY not provided """ payload = json.dumps( mailgun_sign_payload( {"event-data": {"event": "delivered"}}, webhook_signing_key="TEST_API_KEY", ) ) response = self.client.post( "/anymail/mailgun/tracking/", content_type="application/json", data=payload ) self.assertEqual(response.status_code, 200) def test_webhook_signing_key_view_params(self): """Webhook signing key can be provided as a view param""" view = MailgunTrackingWebhookView.as_view( webhook_signing_key="VIEW_SIGNING_KEY" ) view_instance = view.view_class(**view.view_initkwargs) self.assertEqual(view_instance.webhook_signing_key, b"VIEW_SIGNING_KEY") # Can also use `api_key` param for backwards compatibility # with earlier Anymail versions view = MailgunTrackingWebhookView.as_view(api_key="VIEW_API_KEY") view_instance = view.view_class(**view.view_initkwargs) self.assertEqual(view_instance.webhook_signing_key, b"VIEW_API_KEY") @tag("mailgun") @override_settings(ANYMAIL_MAILGUN_WEBHOOK_SIGNING_KEY=TEST_WEBHOOK_SIGNING_KEY) class MailgunWebhookSecurityTestCase(WebhookBasicAuthTestCase): should_warn_if_no_auth = False # because we check webhook signature def call_webhook(self): 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 WebhookBasicAuthTestCase def test_verifies_correct_signature(self): 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/", 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_payload( {"event-data": {"event": "delivered"}}, webhook_signing_key="wrong signing key", ) response = self.client.post( "/anymail/mailgun/tracking/", content_type="application/json", data=json.dumps(data), ) self.assertEqual(response.status_code, 400) @tag("mailgun") @override_settings(ANYMAIL_MAILGUN_WEBHOOK_SIGNING_KEY=TEST_WEBHOOK_SIGNING_KEY) class MailgunTestCase(WebhookTestCase): # Tests for Mailgun's new webhooks (announced 2018-06-29) def test_delivered_event(self): # 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=timezone.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, "deferred") self.assertEqual(event.recipient, "undeliverable@nomx.example.com") self.assertEqual(event.reject_reason, "other") self.assertEqual(event.description, "No MX for nomx.example.com") self.assertEqual(event.mta_response, "No MX for nomx.example.com") def test_failed_greylisted_event(self): raw_event = mailgun_sign_payload( { "event-data": { "event": "failed", "severity": "temporary", "reason": "greylisted", "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@mx.example.com", "delivery-status": { "mx-host": "mx.example.com", "attempt-no": 1, "description": "Recipient address rejected: Greylisted", "session-seconds": 0.0, "retry-seconds": 300, "code": 450, "message": "Recipient address rejected: Greylisted", }, }, } ) 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, "deferred") self.assertEqual(event.recipient, "undeliverable@mx.example.com") self.assertEqual(event.reject_reason, "other") self.assertEqual(event.description, "Recipient address rejected: Greylisted") self.assertEqual(event.mta_response, "Recipient address rejected: Greylisted") 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 ", "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 ", "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") def test_delivery_status_is_none_event(self): raw_event = mailgun_sign_payload( { "event-data": { "event": "accepted", "delivery-status": None, } } ) response = self.client.post( "/anymail/mailgun/tracking/", data=json.dumps(raw_event), content_type="application/json", ) self.assertEqual(response.status_code, 200) @tag("mailgun") @override_settings(ANYMAIL_MAILGUN_WEBHOOK_SIGNING_KEY=TEST_WEBHOOK_SIGNING_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"], ["Date", "Thu, 21 Apr 2016 17:55:29 +0000"], [ "X-Mailgun-Sid", "WyIxZmY4ZSIsICJtZWRtdW5kc0BnbWFpbC5jb20iLCAiZjFjNzgyIl0=", ], [ "Received", "by luna.mailgun.net with HTTP;" " Thu, 21 Apr 2016 17:55:29 +0000", ], [ "Message-Id", "<20160421175529.19495.89030.B3AE3728@example.com>", ], ["To", "recipient@example.com"], ["From", "from@example.com"], ["Subject", "Webhook testing"], ["Mime-Version", "1.0"], [ "Content-Type", [ "multipart/alternative", {"boundary": "74fb561763da440d8e6a034054974251"}, ], ], ] ), "X-Mailgun-Sid": "WyIxZmY4ZSIsICJtZWRtdW5kc" "0BnbWFpbC5jb20iLCAiZjFjNzgyIl0=", "token": "06c96bafc3f42a66b9edd546347a2fe18dc23461fe80dc52f0", "timestamp": "1461261330", "Message-Id": "<20160421175529.19495.89030.B3AE3728@example.com>", "recipient": "recipient@example.com", "event": "delivered", } ) response = self.client.post("/anymail/mailgun/tracking/", data=raw_event) 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(2016, 4, 21, 17, 55, 30, tzinfo=timezone.utc) ) self.assertEqual( event.message_id, "<20160421175529.19495.89030.B3AE3728@example.com>" ) self.assertEqual( event.event_id, "06c96bafc3f42a66b9edd546347a2fe18dc23461fe80dc52f0" ) self.assertEqual(event.recipient, "recipient@example.com") self.assertEqual(querydict_to_postdict(event.esp_event), raw_event) self.assertEqual(event.tags, []) self.assertEqual(event.metadata, {}) def test_dropped_bounce(self): raw_event = mailgun_sign_legacy_payload( { "code": "605", "domain": "example.com", "description": "Not delivering to previously bounced address", "attachment-count": "1", "Message-Id": "<20160421180324.70521.79375.96884DDB@example.com>", "reason": "hardfail", "event": "dropped", "message-headers": json.dumps( [ [ "X-Mailgun-Sid", "WyI3Y2VjMyIsICJib3VuY2VAZXhhbXBsZS5jb20iLCAiZjFjNzgyIl0=", ], [ "Received", "by luna.mailgun.net with HTTP;" " Thu, 21 Apr 2016 18:03:24 +0000", ], [ "Message-Id", "<20160421180324.70521.79375.96884DDB@example.com>", ], ["To", "bounce@example.com"], ["From", "from@example.com"], ["Subject", "Webhook testing"], ["Mime-Version", "1.0"], [ "Content-Type", [ "multipart/alternative", {"boundary": "a5b51388a4e3455d8feb8510bb8c9fa2"}, ], ], ] ), "recipient": "bounce@example.com", "timestamp": "1461261330", "X-Mailgun-Sid": "WyI3Y2VjMyIsICJib3VuY2VAZ" "XhhbXBsZS5jb20iLCAiZjFjNzgyIl0=", "token": "a3fe1fa1640349ac552b84ddde373014b4c41645830c8dd3fc", } ) response = self.client.post("/anymail/mailgun/tracking/", data=raw_event) 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, "rejected") self.assertEqual( event.timestamp, datetime(2016, 4, 21, 17, 55, 30, tzinfo=timezone.utc) ) self.assertEqual( event.message_id, "<20160421180324.70521.79375.96884DDB@example.com>" ) self.assertEqual( event.event_id, "a3fe1fa1640349ac552b84ddde373014b4c41645830c8dd3fc" ) self.assertEqual(event.recipient, "bounce@example.com") self.assertEqual(event.reject_reason, "bounced") self.assertEqual( event.description, "Not delivering to previously bounced address" ) self.assertEqual(querydict_to_postdict(event.esp_event), raw_event) def test_dropped_spam(self): raw_event = mailgun_sign_legacy_payload( { "code": "607", "description": "Not delivering to a user who marked" " your messages as spam", "reason": "hardfail", "event": "dropped", "recipient": "complaint@example.com", # (omitting some fields that aren't relevant to the test) } ) response = self.client.post("/anymail/mailgun/tracking/", data=raw_event) 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, "spam") 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_legacy_payload( { "code": "499", "description": "Unable to connect to MX servers: [example.com]", "reason": "old", "event": "dropped", "recipient": "complaint@example.com", # (omitting some fields that aren't relevant to the test) } ) response = self.client.post("/anymail/mailgun/tracking/", data=raw_event) 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, "timed_out") self.assertEqual( event.description, "Unable to connect to MX servers: [example.com]" ) def test_invalid_mailbox(self): 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 " " 5.1.1 unnecessary spaces.", "event": "bounced", "recipient": "noreply@example.com", # (omitting some fields that aren't relevant to the test) } ) response = self.client.post("/anymail/mailgun/tracking/", data=raw_event) 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.reject_reason, "bounced") self.assertIn( "The email account that you tried to reach does not exist", event.mta_response, ) 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_legacy_payload( { "code": "5.1.1", "error": "smtp;550 5.1.1 RESOLVER.ADR.RecipNotFound; not found", "event": "bounced", "recipient": "noreply@example.com", # (omitting some fields that aren't relevant to the test) } ) response = self.client.post("/anymail/mailgun/tracking/", data=raw_event) 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.reject_reason, "bounced") self.assertIn("RecipNotFound", event.mta_response) 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_legacy_payload( { "event": "delivered", "message-headers": json.dumps( [ [ "X-Mailgun-Variables", '{"custom1": "value1",' ' "custom2": "{\\"key\\":\\"value\\"}"}', ], ] ), "custom1": "value1", # you can store JSON, but you'll need to unpack it yourself: "custom2": '{"key":"value"}', } ) self.client.post("/anymail/mailgun/tracking/", data=raw_event) kwargs = self.assert_handler_called_once_with(self.tracking_handler) event = kwargs["event"] self.assertEqual( event.metadata, {"custom1": "value1", "custom2": '{"key":"value"}'} ) def test_metadata_post_fields(self): # Metadata fields are only interspersed with other event params # for opened, clicked, unsubscribed events raw_event = mailgun_sign_legacy_payload( { "event": "clicked", "custom1": "value1", # you can store JSON, but you'll need to unpack it yourself: "custom2": '{"key":"value"}', } ) self.client.post("/anymail/mailgun/tracking/", data=raw_event) kwargs = self.assert_handler_called_once_with(self.tracking_handler) event = kwargs["event"] self.assertEqual( event.metadata, {"custom1": "value1", "custom2": '{"key":"value"}'} ) def test_metadata_key_conflicts(self): # If you happen to name metadata (user-variable) keys the same as Mailgun # event properties, Mailgun will include both in the webhook post. # Make sure we don't confuse them. metadata = { "event": "metadata-event", "recipient": "metadata-recipient", "signature": "metadata-signature", "timestamp": "metadata-timestamp", "token": "metadata-token", "ordinary field": "ordinary metadata value", } raw_event = mailgun_sign_legacy_payload( { "event": "clicked", "recipient": "actual-recipient@example.com", "token": "actual-event-token", "timestamp": "1461261330", "url": "http://clicked.example.com/actual/event/param", "h": "an (undocumented) Mailgun event param", "tag": ["actual-tag-1", "actual-tag-2"], } ) # Simulate how Mailgun merges user-variables fields into event: for key in metadata.keys(): if key in raw_event: if key in {"signature", "timestamp", "token"}: # For these fields, Mailgun's value appears after the metadata value raw_event[key] = [metadata[key], raw_event[key]] elif key == "message-headers": pass # Mailgun won't merge this field into the event else: # For all other fields, the defined event value comes first raw_event[key] = [raw_event[key], metadata[key]] else: raw_event[key] = metadata[key] response = self.client.post("/anymail/mailgun/tracking/", data=raw_event) # if this fails, signature checking is using metadata values: self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.tracking_handler) event = kwargs["event"] self.assertEqual(event.event_type, "clicked") self.assertEqual(event.recipient, "actual-recipient@example.com") self.assertEqual(event.timestamp.isoformat(), "2016-04-21T17:55:30+00:00") self.assertEqual(event.event_id, "actual-event-token") self.assertEqual(event.tags, ["actual-tag-1", "actual-tag-2"]) self.assertEqual(event.metadata, metadata) def test_tags(self): # Most events include multiple 'tag' fields for message's tags raw_event = mailgun_sign_legacy_payload( { # Django TestClient encodes list as multiple field values: "tag": ["tag1", "tag2"], "event": "opened", } ) self.client.post("/anymail/mailgun/tracking/", data=raw_event) kwargs = self.assert_handler_called_once_with(self.tracking_handler) event = kwargs["event"] self.assertEqual(event.tags, ["tag1", "tag2"]) def test_x_tags(self): # Delivery events don't include 'tag', but do include 'X-Mailgun-Tag' fields raw_event = mailgun_sign_legacy_payload( { "X-Mailgun-Tag": ["tag1", "tag2"], "event": "delivered", } ) self.client.post("/anymail/mailgun/tracking/", data=raw_event) kwargs = self.assert_handler_called_once_with(self.tracking_handler) event = kwargs["event"] self.assertEqual(event.tags, ["tag1", "tag2"]) def test_misconfigured_inbound(self): raw_event = mailgun_sign_legacy_payload( { "recipient": "test@inbound.example.com", "sender": "envelope-from@example.org", "message-headers": "[]", "body-plain": "Test body plain", "body-html": "
Test body html
", } ) with self.assertRaisesMessage( AnymailConfigurationError, "You seem to have set Mailgun's *inbound* route" " to Anymail's Mailgun *tracking* webhook URL.", ): self.client.post("/anymail/mailgun/tracking/", data=raw_event)