django-anymail/anymail/webhooks/postmark.py

import json

from django.utils.dateparse import parse_datetime

from .base import AnymailBaseWebhookView
from ..exceptions import AnymailConfigurationError
from ..inbound import AnymailInboundMessage
from ..signals import inbound, tracking, AnymailInboundEvent, AnymailTrackingEvent, EventType, RejectReason
from ..utils import getfirst, EmailAddress


class PostmarkBaseWebhookView(AnymailBaseWebhookView):
    """Base view class for Postmark webhooks"""

    esp_name = "Postmark"

    def parse_events(self, request):
        esp_event = json.loads(request.body.decode('utf-8'))
        return [self.esp_to_anymail_event(esp_event)]

    def esp_to_anymail_event(self, esp_event):
        raise NotImplementedError()


class PostmarkTrackingWebhookView(PostmarkBaseWebhookView):
    """Handler for Postmark delivery and engagement tracking webhooks"""

    signal = tracking

    event_record_types = {
        # Map Postmark event RecordType --> Anymail normalized event type
        'Bounce': EventType.BOUNCED,  # but check Type field for further info (below)
        'Click': EventType.CLICKED,
        'Delivery': EventType.DELIVERED,
        'Open': EventType.OPENED,
        'SpamComplaint': EventType.COMPLAINED,
        'Inbound': EventType.INBOUND,  # future, probably
    }

    event_types = {
        # Map Postmark bounce/spam event Type --> Anymail normalized (event type, reject reason)
        'HardBounce': (EventType.BOUNCED, RejectReason.BOUNCED),
        'Transient': (EventType.DEFERRED, None),
        'Unsubscribe': (EventType.UNSUBSCRIBED, RejectReason.UNSUBSCRIBED),
        'Subscribe': (EventType.SUBSCRIBED, None),
        'AutoResponder': (EventType.AUTORESPONDED, None),
        'AddressChange': (EventType.AUTORESPONDED, None),
        'DnsError': (EventType.DEFERRED, None),  # "temporary DNS error"
        'SpamNotification': (EventType.COMPLAINED, RejectReason.SPAM),
        'OpenRelayTest': (EventType.DEFERRED, None),  # Receiving MTA is testing Postmark
        'Unknown': (EventType.UNKNOWN, None),
        'SoftBounce': (EventType.BOUNCED, RejectReason.BOUNCED),  # might also receive HardBounce later
        'VirusNotification': (EventType.BOUNCED, RejectReason.OTHER),
        'ChallengeVerification': (EventType.AUTORESPONDED, None),
        'BadEmailAddress': (EventType.REJECTED, RejectReason.INVALID),
        'SpamComplaint': (EventType.COMPLAINED, RejectReason.SPAM),
        'ManuallyDeactivated': (EventType.REJECTED, RejectReason.BLOCKED),
        'Unconfirmed': (EventType.REJECTED, None),
        'Blocked': (EventType.REJECTED, RejectReason.BLOCKED),
        'SMTPApiError': (EventType.FAILED, None),  # could occur if user also using Postmark SMTP directly
        'InboundError': (EventType.INBOUND_FAILED, None),
        'DMARCPolicy': (EventType.REJECTED, RejectReason.BLOCKED),
        'TemplateRenderingFailed': (EventType.FAILED, None),
    }

    def esp_to_anymail_event(self, esp_event):
        reject_reason = None
        try:
            esp_record_type = esp_event["RecordType"]
        except KeyError:
            if 'FromFull' in esp_event:
                # This is an inbound event
                event_type = EventType.INBOUND
            else:
                event_type = EventType.UNKNOWN
        else:
            event_type = self.event_record_types.get(esp_record_type, EventType.UNKNOWN)

        if event_type == EventType.INBOUND:
            raise AnymailConfigurationError(
                "You seem to have set Postmark's *inbound* webhook "
                "to Anymail's Postmark *tracking* webhook URL.")

        if event_type in (EventType.BOUNCED, EventType.COMPLAINED):
            # additional info is in the Type field
            try:
                event_type, reject_reason = self.event_types[esp_event['Type']]
            except KeyError:
                pass

        recipient = getfirst(esp_event, ['Email', 'Recipient'], None)  # Email for bounce; Recipient for open

        try:
            timestr = getfirst(esp_event, ['DeliveredAt', 'BouncedAt', 'ReceivedAt'])
        except KeyError:
            timestamp = None
        else:
            timestamp = parse_datetime(timestr)

        try:
            event_id = str(esp_event['ID'])  # only in bounce events
        except KeyError:
            event_id = None

        try:
            tags = [esp_event['Tag']]
        except KeyError:
            tags = []

        return AnymailTrackingEvent(
            description=esp_event.get('Description', None),
            esp_event=esp_event,
            event_id=event_id,
            event_type=event_type,
            message_id=esp_event.get('MessageID', None),
            mta_response=esp_event.get('Details', None),
            recipient=recipient,
            reject_reason=reject_reason,
            tags=tags,
            timestamp=timestamp,
            user_agent=esp_event.get('UserAgent', None),
            click_url=esp_event.get('OriginalLink', None),
        )


class PostmarkInboundWebhookView(PostmarkBaseWebhookView):
    """Handler for Postmark inbound webhook"""

    signal = inbound

    def esp_to_anymail_event(self, esp_event):
        if esp_event.get("RecordType", "Inbound") != "Inbound":
            raise AnymailConfigurationError(
                "You seem to have set Postmark's *%s* webhook "
                "to Anymail's Postmark *inbound* webhook URL." % esp_event["RecordType"])

        attachments = [
            AnymailInboundMessage.construct_attachment(
                content_type=attachment["ContentType"],
                content=attachment["Content"], base64=True,
                filename=attachment.get("Name", "") or None,
                content_id=attachment.get("ContentID", "") or None,
            )
            for attachment in esp_event.get("Attachments", [])
        ]

        message = AnymailInboundMessage.construct(
            from_email=self._address(esp_event.get("FromFull")),
            to=', '.join([self._address(to) for to in esp_event.get("ToFull", [])]),
            cc=', '.join([self._address(cc) for cc in esp_event.get("CcFull", [])]),
            # bcc? Postmark specs this for inbound events, but it's unclear how it could occur
            subject=esp_event.get("Subject", ""),
            headers=[(header["Name"], header["Value"]) for header in esp_event.get("Headers", [])],
            text=esp_event.get("TextBody", ""),
            html=esp_event.get("HtmlBody", ""),
            attachments=attachments,
        )

        # Postmark strips these headers and provides them as separate event fields:
        if "Date" in esp_event and "Date" not in message:
            message["Date"] = esp_event["Date"]
        if "ReplyTo" in esp_event and "Reply-To" not in message:
            message["Reply-To"] = esp_event["ReplyTo"]

        # Postmark doesn't have a separate envelope-sender field, but it can be extracted
        # from the Received-SPF header that Postmark will have added:
        if len(message.get_all("Received-SPF", [])) == 1:  # (more than one? someone's up to something weird)
            received_spf = message["Received-SPF"].lower()
            if received_spf.startswith("pass") or received_spf.startswith("neutral"):  # not fail/softfail
                message.envelope_sender = message.get_param("envelope-from", None, header="Received-SPF")

        message.envelope_recipient = esp_event.get("OriginalRecipient", None)
        message.stripped_text = esp_event.get("StrippedTextReply", None)

        message.spam_detected = message.get('X-Spam-Status', 'No').lower() == 'yes'
        try:
            message.spam_score = float(message['X-Spam-Score'])
        except (TypeError, ValueError):
            pass

        return AnymailInboundEvent(
            event_type=EventType.INBOUND,
            timestamp=None,  # Postmark doesn't provide inbound event timestamp
            event_id=esp_event.get("MessageID", None),  # Postmark uuid, different from Message-ID mime header
            esp_event=esp_event,
            message=message,
        )

    @staticmethod
    def _address(full):
        """Return an formatted email address from a Postmark inbound {From,To,Cc}Full dict"""
        if full is None:
            return ""
        return str(EmailAddress(
            display_name=full.get('Name', ""),
            addr_spec=full.get("Email", ""),
        ))