import json from base64 import b64encode from django.test import tag from mock import ANY from anymail.exceptions import AnymailConfigurationError from anymail.inbound import AnymailInboundMessage from anymail.signals import AnymailInboundEvent from anymail.webhooks.postmark import PostmarkInboundWebhookView from .utils import sample_image_content, sample_email_content from .webhook_cases import WebhookTestCase @tag('postmark') class PostmarkInboundTestCase(WebhookTestCase): def test_inbound_basics(self): raw_event = { "FromFull": { "Email": "from+test@example.org", "Name": "Displayed From", "MailboxHash": "test" }, "ToFull": [{ "Email": "test@inbound.example.com", "Name": "Test Inbound", "MailboxHash": "" }, { "Email": "other@example.com", "Name": "", "MailboxHash": "" }], "CcFull": [{ "Email": "cc@example.com", "Name": "", "MailboxHash": "" }], "BccFull": [{ "Email": "bcc@example.com", "Name": "Postmark documents blind cc on inbound email (?)", "MailboxHash": "" }], "OriginalRecipient": "test@inbound.example.com", "ReplyTo": "from+test@milter.example.org", "Subject": "Test subject", "MessageID": "22c74902-a0c1-4511-804f2-341342852c90", "Date": "Wed, 11 Oct 2017 18:31:04 -0700", "TextBody": "Test body plain", "HtmlBody": "
Test body html
", "StrippedTextReply": "stripped plaintext body", "Tag": "", "Headers": [{ "Name": "Received", "Value": "from mail.example.org by inbound.postmarkapp.com ..." }, { "Name": "X-Spam-Checker-Version", "Value": "SpamAssassin 3.4.0 (2014-02-07) onp-pm-smtp-inbound01b-aws-useast2b" }, { "Name": "X-Spam-Status", "Value": "No" }, { "Name": "X-Spam-Score", "Value": "1.7" }, { "Name": "X-Spam-Tests", "Value": "SPF_PASS" }, { "Name": "Received-SPF", "Value": "Pass (sender SPF authorized) identity=mailfrom; client-ip=333.3.3.3;" " helo=mail-02.example.org; envelope-from=envelope-from@example.org;" " receiver=test@inbound.example.com" }, { "Name": "Received", "Value": "by mail.example.org for ..." }, { "Name": "Received", "Value": "by 10.10.1.71 with HTTP; Wed, 11 Oct 2017 18:31:04 -0700 (PDT)" }, { "Name": "MIME-Version", "Value": "1.0" }, { "Name": "Message-ID", "Value": "" }], } response = self.client.post('/anymail/postmark/inbound/', content_type='application/json', data=json.dumps(raw_event)) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=PostmarkInboundWebhookView, event=ANY, esp_name='Postmark') # AnymailInboundEvent event = kwargs['event'] self.assertIsInstance(event, AnymailInboundEvent) self.assertEqual(event.event_type, 'inbound') self.assertIsNone(event.timestamp) # Postmark doesn't provide inbound event timestamp self.assertEqual(event.event_id, "22c74902-a0c1-4511-804f2-341342852c90") self.assertIsInstance(event.message, AnymailInboundMessage) self.assertEqual(event.esp_event, raw_event) # AnymailInboundMessage - convenience properties message = event.message self.assertEqual(message.from_email.display_name, 'Displayed From') self.assertEqual(message.from_email.addr_spec, 'from+test@example.org') self.assertEqual([str(e) for e in message.to], ['Test Inbound ', 'other@example.com']) self.assertEqual([str(e) for e in message.cc], ['cc@example.com']) self.assertEqual(message.subject, 'Test subject') self.assertEqual(message.date.isoformat(" "), "2017-10-11 18:31:04-07:00") self.assertEqual(message.text, 'Test body plain') self.assertEqual(message.html, '
Test body html
') self.assertEqual(message.envelope_sender, 'envelope-from@example.org') self.assertEqual(message.envelope_recipient, 'test@inbound.example.com') self.assertEqual(message.stripped_text, 'stripped plaintext body') self.assertIsNone(message.stripped_html) # Postmark doesn't provide stripped html self.assertIs(message.spam_detected, False) self.assertEqual(message.spam_score, 1.7) # AnymailInboundMessage - other headers self.assertEqual(message['Message-ID'], "") self.assertEqual(message['Reply-To'], "from+test@milter.example.org") self.assertEqual(message.get_all('Received'), [ "from mail.example.org by inbound.postmarkapp.com ...", "by mail.example.org for ...", "by 10.10.1.71 with HTTP; Wed, 11 Oct 2017 18:31:04 -0700 (PDT)", ]) def test_attachments(self): image_content = sample_image_content() email_content = sample_email_content() raw_event = { "Attachments": [{ "Name": "test.txt", "Content": b64encode('test attachment'.encode('utf-8')).decode('ascii'), "ContentType": "text/plain", "ContentLength": len('test attachment') }, { "Name": "image.png", "Content": b64encode(image_content).decode('ascii'), "ContentType": "image/png", "ContentID": "abc123", "ContentLength": len(image_content) }, { "Name": "bounce.txt", "Content": b64encode(email_content).decode('ascii'), "ContentType": 'message/rfc822; charset="us-ascii"', "ContentLength": len(email_content) }] } response = self.client.post('/anymail/postmark/inbound/', content_type='application/json', data=json.dumps(raw_event)) self.assertEqual(response.status_code, 200) kwargs = self.assert_handler_called_once_with(self.inbound_handler, sender=PostmarkInboundWebhookView, event=ANY, esp_name='Postmark') event = kwargs['event'] message = event.message attachments = message.attachments # AnymailInboundMessage convenience accessor self.assertEqual(len(attachments), 2) self.assertEqual(attachments[0].get_filename(), 'test.txt') self.assertEqual(attachments[0].get_content_type(), 'text/plain') self.assertEqual(attachments[0].get_content_text(), u'test attachment') self.assertEqual(attachments[1].get_content_type(), 'message/rfc822') self.assertEqualIgnoringHeaderFolding(attachments[1].get_content_bytes(), email_content) inlines = message.inline_attachments self.assertEqual(len(inlines), 1) inline = inlines['abc123'] self.assertEqual(inline.get_filename(), 'image.png') self.assertEqual(inline.get_content_type(), 'image/png') self.assertEqual(inline.get_content_bytes(), image_content) def test_envelope_sender(self): # Anymail extracts envelope-sender from Postmark Received-SPF header raw_event = { "Headers": [{ "Name": "Received-SPF", "Value": "Pass (sender SPF authorized) identity=mailfrom; client-ip=333.3.3.3;" " helo=mail-02.example.org; envelope-from=envelope-from@example.org;" " receiver=test@inbound.example.com" }], } response = self.client.post('/anymail/postmark/inbound/', content_type='application/json', data=json.dumps(raw_event)) self.assertEqual(response.status_code, 200) self.assertEqual(self.get_kwargs(self.inbound_handler)['event'].message.envelope_sender, "envelope-from@example.org") # Allow neutral SPF response self.client.post( '/anymail/postmark/inbound/', content_type='application/json', data=json.dumps({"Headers": [{ "Name": "Received-SPF", "Value": "Neutral (no SPF record exists) identity=mailfrom; envelope-from=envelope-from@example.org" }]})) self.assertEqual(self.get_kwargs(self.inbound_handler)['event'].message.envelope_sender, "envelope-from@example.org") # Ignore fail/softfail self.client.post( '/anymail/postmark/inbound/', content_type='application/json', data=json.dumps({"Headers": [{ "Name": "Received-SPF", "Value": "Fail (sender not SPF authorized) identity=mailfrom; envelope-from=spoofed@example.org" }]})) self.assertIsNone(self.get_kwargs(self.inbound_handler)['event'].message.envelope_sender) # Ignore garbage self.client.post( '/anymail/postmark/inbound/', content_type='application/json', data=json.dumps({"Headers": [{ "Name": "Received-SPF", "Value": "ThisIsNotAValidReceivedSPFHeader@example.org" }]})) self.assertIsNone(self.get_kwargs(self.inbound_handler)['event'].message.envelope_sender) # Ignore multiple Received-SPF headers self.client.post( '/anymail/postmark/inbound/', content_type='application/json', data=json.dumps({"Headers": [{ "Name": "Received-SPF", "Value": "Fail (sender not SPF authorized) identity=mailfrom; envelope-from=spoofed@example.org" }, { "Name": "Received-SPF", "Value": "Pass (malicious sender added this) identity=mailfrom; envelope-from=spoofed@example.org" }]})) self.assertIsNone(self.get_kwargs(self.inbound_handler)['event'].message.envelope_sender) def test_misconfigured_tracking(self): errmsg = "You seem to have set Postmark's *Delivery* webhook to Anymail's Postmark *inbound* webhook URL." with self.assertRaisesMessage(AnymailConfigurationError, errmsg): self.client.post('/anymail/postmark/inbound/', content_type='application/json', data=json.dumps({"RecordType": "Delivery"}))