mirror of
https://github.com/pacnpal/django-anymail.git
synced 2025-12-20 03:41:05 -05:00
0ded9f7529
Simplify Postmark tracking webhook code by using new "RecordType" field introduced with Postmark "modular webhooks". (Rather than looking for fields that are probably only in certain events.) Also issue configuration error on inbound url installed as tracking webhook (and vice versa).
232 lines
11 KiB
Python
232 lines
11 KiB
Python
import json
|
|
from base64 import b64encode
|
|
|
|
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
|
|
|
|
|
|
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": "<div>Test body html</div>",
|
|
"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 <test@inbound.example.com> ..."
|
|
}, {
|
|
"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": "<CAEPk3R+4Zr@mail.example.org>"
|
|
}],
|
|
}
|
|
|
|
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 <test@inbound.example.com>', '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, '<div>Test body html</div>')
|
|
|
|
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'], "<CAEPk3R+4Zr@mail.example.org>")
|
|
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 <test@inbound.example.com> ...",
|
|
"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"}))
|