import hashlib import hmac import json from base64 import b64encode from datetime import datetime, timezone from unittest.mock import ANY from urllib.parse import urljoin from django.core.exceptions import ImproperlyConfigured from django.test import override_settings, tag from anymail.signals import AnymailTrackingEvent from anymail.webhooks.mandrill import ( MandrillCombinedWebhookView, MandrillTrackingWebhookView, ) from .webhook_cases import WebhookBasicAuthTestCase, WebhookTestCase TEST_WEBHOOK_KEY = "TEST_WEBHOOK_KEY" def mandrill_args( events=None, host="http://testserver/", # Django test-client default path="/anymail/mandrill/", # Anymail urlconf default auth="username:password", # WebhookTestCase default key=TEST_WEBHOOK_KEY, ): """Returns TestClient.post kwargs for Mandrill webhook call with events Computes correct signature. """ if events is None: events = [] test_client_path = urljoin(host, path) # https://testserver/anymail/mandrill/ if auth: # we can get away with this simplification in these controlled tests, # but don't ever construct urls like this in production code -- it's not safe! full_url = test_client_path.replace("://", "://" + auth + "@") else: full_url = test_client_path mandrill_events = json.dumps(events) signed_data = full_url + "mandrill_events" + mandrill_events signature = b64encode( hmac.new( key=key.encode("ascii"), msg=signed_data.encode("utf-8"), digestmod=hashlib.sha1, ).digest() ) return { "path": test_client_path, "data": {"mandrill_events": mandrill_events}, "HTTP_X_MANDRILL_SIGNATURE": signature, } @tag("mandrill") class MandrillWebhookSettingsTestCase(WebhookTestCase): def test_requires_webhook_key(self): with self.assertRaisesRegex(ImproperlyConfigured, r"MANDRILL_WEBHOOK_KEY"): self.client.post("/anymail/mandrill/", data={"mandrill_events": "[]"}) def test_head_does_not_require_webhook_key(self): # Mandrill issues an unsigned HEAD request to verify the wehbook url. # Only *after* that succeeds will Mandrill will tell you the webhook key. # So make sure that HEAD request will go through without any key set: response = self.client.head("/anymail/mandrill/") self.assertEqual(response.status_code, 200) @tag("mandrill") @override_settings(ANYMAIL_MANDRILL_WEBHOOK_KEY=TEST_WEBHOOK_KEY) class MandrillWebhookSecurityTestCase(WebhookBasicAuthTestCase): should_warn_if_no_auth = False # because we check webhook signature def call_webhook(self): kwargs = mandrill_args([{"event": "send"}]) return self.client.post(**kwargs) # Additional tests are in WebhookBasicAuthTestCase def test_verifies_correct_signature(self): kwargs = mandrill_args([{"event": "send"}]) response = self.client.post(**kwargs) self.assertEqual(response.status_code, 200) def test_verifies_missing_signature(self): response = self.client.post( "/anymail/mandrill/", data={"mandrill_events": '[{"event":"send"}]'} ) self.assertEqual(response.status_code, 400) def test_verifies_bad_signature(self): kwargs = mandrill_args([{"event": "send"}], key="wrong API key") response = self.client.post(**kwargs) self.assertEqual(response.status_code, 400) @override_settings(ANYMAIL={}) # clear WEBHOOK_SECRET from WebhookTestCase def test_no_basic_auth(self): # Signature validation should work properly if you're not using basic auth self.clear_basic_auth() kwargs = mandrill_args([{"event": "send"}], auth="") response = self.client.post(**kwargs) self.assertEqual(response.status_code, 200) @override_settings( ALLOWED_HOSTS=["127.0.0.1", ".example.com"], ANYMAIL={ "MANDRILL_WEBHOOK_URL": "https://abcde:12345@example.com/anymail/mandrill/", "WEBHOOK_SECRET": "abcde:12345", }, ) def test_webhook_url_setting(self): # If Django can't build_absolute_uri correctly (e.g., because your proxy # frontend isn't setting the proxy headers correctly), you must set # MANDRILL_WEBHOOK_URL to the actual public url where Mandrill calls # the webhook. self.set_basic_auth("abcde", "12345") kwargs = mandrill_args( [{"event": "send"}], host="https://example.com/", auth="abcde:12345" ) response = self.client.post(SERVER_NAME="127.0.0.1", **kwargs) self.assertEqual(response.status_code, 200) # override WebhookBasicAuthTestCase version of this test @override_settings(ANYMAIL={"WEBHOOK_SECRET": ["cred1:pass1", "cred2:pass2"]}) def test_supports_credential_rotation(self): """You can supply a list of basic auth credentials, and any is allowed""" self.set_basic_auth("cred1", "pass1") response = self.client.post(**mandrill_args(auth="cred1:pass1")) self.assertEqual(response.status_code, 200) self.set_basic_auth("cred2", "pass2") response = self.client.post(**mandrill_args(auth="cred2:pass2")) self.assertEqual(response.status_code, 200) self.set_basic_auth("baduser", "wrongpassword") response = self.client.post(**mandrill_args(auth="baduser:wrongpassword")) self.assertEqual(response.status_code, 400) @tag("mandrill") @override_settings(ANYMAIL_MANDRILL_WEBHOOK_KEY=TEST_WEBHOOK_KEY) class MandrillTrackingTestCase(WebhookTestCase): def test_head_request(self): # Mandrill verifies webhooks at config time with a HEAD request # (See MandrillWebhookSettingsTestCase above for equivalent # without the key yet set) response = self.client.head("/anymail/mandrill/tracking/") self.assertEqual(response.status_code, 200) def test_post_request_invalid_json(self): kwargs = mandrill_args() kwargs["data"] = {"mandrill_events": "GARBAGE DATA"} response = self.client.post(**kwargs) self.assertEqual(response.status_code, 400) def test_send_event(self): raw_events = [ { "event": "send", "msg": { "ts": 1461095211, # time send called "subject": "Webhook Test", "email": "recipient@example.com", "sender": "sender@example.com", "tags": ["tag1", "tag2"], "metadata": {"custom1": "value1", "custom2": "value2"}, "_id": "abcdef012345789abcdef012345789", }, "_id": "abcdef012345789abcdef012345789", "ts": 1461095246, # time of event } ] response = self.client.post(**mandrill_args(events=raw_events)) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.tracking_handler, sender=MandrillCombinedWebhookView, event=ANY, esp_name="Mandrill", ) event = kwargs["event"] self.assertIsInstance(event, AnymailTrackingEvent) self.assertEqual(event.event_type, "sent") self.assertEqual( event.timestamp, datetime(2016, 4, 19, 19, 47, 26, tzinfo=timezone.utc) ) self.assertEqual(event.esp_event, raw_events[0]) self.assertEqual(event.message_id, "abcdef012345789abcdef012345789") self.assertEqual(event.recipient, "recipient@example.com") self.assertEqual(event.tags, ["tag1", "tag2"]) self.assertEqual(event.metadata, {"custom1": "value1", "custom2": "value2"}) def test_hard_bounce_event(self): raw_events = [ { "event": "hard_bounce", "msg": { "ts": 1461095211, # time send called "subject": "Webhook Test", "email": "bounce@example.com", "sender": "sender@example.com", "bounce_description": "bad_mailbox", "bgtools_code": 10, "diag": "smtp;550 5.1.1 The email account that you tried" " to reach does not exist.", "_id": "abcdef012345789abcdef012345789", }, "_id": "abcdef012345789abcdef012345789", "ts": 1461095246, # time of event } ] response = self.client.post(**mandrill_args(events=raw_events)) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.tracking_handler, sender=MandrillCombinedWebhookView, event=ANY, esp_name="Mandrill", ) event = kwargs["event"] self.assertIsInstance(event, AnymailTrackingEvent) self.assertEqual(event.event_type, "bounced") self.assertEqual(event.esp_event, raw_events[0]) self.assertEqual(event.message_id, "abcdef012345789abcdef012345789") self.assertEqual(event.recipient, "bounce@example.com") self.assertEqual( event.mta_response, "smtp;550 5.1.1 The email account that you tried to reach does not exist.", ) def test_click_event(self): raw_events = [ { "event": "click", "msg": { "ts": 1461095211, # time send called "subject": "Webhook Test", "email": "recipient@example.com", "sender": "sender@example.com", "opens": [{"ts": 1461095242}], "clicks": [{"ts": 1461095246, "url": "http://example.com"}], "_id": "abcdef012345789abcdef012345789", }, "user_agent": "Mozilla/5.0 (Windows NT 5.1; rv:11.0)" " Gecko Firefox/11.0", "url": "http://example.com", "_id": "abcdef012345789abcdef012345789", "ts": 1461095246, # time of event } ] response = self.client.post(**mandrill_args(events=raw_events)) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.tracking_handler, sender=MandrillCombinedWebhookView, event=ANY, esp_name="Mandrill", ) event = kwargs["event"] self.assertIsInstance(event, AnymailTrackingEvent) self.assertEqual(event.event_type, "clicked") self.assertEqual(event.esp_event, raw_events[0]) self.assertEqual(event.click_url, "http://example.com") self.assertEqual( event.user_agent, "Mozilla/5.0 (Windows NT 5.1; rv:11.0) Gecko Firefox/11.0" ) def test_sync_event(self): # Mandrill sync events use a different format from other events # https://mandrill.zendesk.com/hc/en-us/articles/205583297-Sync-Event-Webhook-format raw_events = [ { "type": "blacklist", "action": "add", "reject": {"email": "recipient@example.com", "reason": "manual edit"}, } ] response = self.client.post(**mandrill_args(events=raw_events)) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with( self.tracking_handler, sender=MandrillCombinedWebhookView, event=ANY, esp_name="Mandrill", ) event = kwargs["event"] self.assertEqual(event.event_type, "unknown") self.assertEqual(event.recipient, "recipient@example.com") self.assertEqual(event.description, "manual edit") def test_old_tracking_url(self): # Earlier versions of Anymail used /mandrill/tracking/ (and didn't support # inbound); make sure that URL continues to work. raw_events = [ { "event": "send", "msg": { "ts": 1461095211, # time send called "subject": "Webhook Test", "email": "recipient@example.com", "sender": "sender@example.com", "tags": ["tag1", "tag2"], "metadata": {"custom1": "value1", "custom2": "value2"}, "_id": "abcdef012345789abcdef012345789", }, "_id": "abcdef012345789abcdef012345789", "ts": 1461095246, # time of event } ] response = self.client.post( **mandrill_args(events=raw_events, path="/anymail/mandrill/tracking/") ) self.assertEqual(response.status_code, 200) self.assert_handler_called_once_with( self.tracking_handler, sender=MandrillTrackingWebhookView, event=ANY, esp_name="Mandrill", )