django-anymail/tests/test_postmark_inbound.py

import json
from base64 import b64encode

from mock import ANY

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)