Mandrill webhook decorator

mandrill-hook-utils.py

"""
Utilities for use with Mandrill Webhooks

requirements:
- six
- email-reply-parser (git+https://github.com/atc0m/email-reply-parser.git#egg=email_reply_parser)

"""

import base64
import functools
import hashlib
import hmac
import json

import six
from email_reply_parser import EmailReplyParser
from flask import request, current_app


def calculate_mandrill_signature(secret, *, data=None, url=None):
    """Return a calculated signature if a secret key was provided"""

    if not secret:
        return None

    if not data:
        data = request.form
    if not url:
        url = request.url

    signable = url
    for key in sorted(data.keys()):
        signable += key
        signable += data[key]

    if not isinstance(secret, six.binary_type):
        secret = secret.encode('utf-8')

    return hmac.new(secret, signable.encode('utf-8'), hashlib.sha1).digest()


def validate_mandrill_request(secret):
    """Validate that the request is an authorized webhook from Mandrill"""

    signature = request.headers.get('X-Mandrill-Signature')
    if not signature:
        return False

    digest = calculate_mandrill_signature(secret)
    return digest and hmac.compare_digest(base64.b64decode(signature), digest)


def mandrill_verify(secret_name):
    """
    Simple decorator to extract the real payload from a mandrill webhook
    and verify the signature of the request
    """

    def with_verifier(f):
        """Functional wrapper for accepting args to the decorator"""

        @functools.wraps(f)
        def verifier(*args, **kwargs):

            if request.method != 'POST':
                return '', 204

            data = request.form.get('mandrill_events')
            if not data:
                raise Exception('Invalid POST body')

            try:
                secret = current_app.config[secret_name]
            except KeyError:
                secret = secret_name

            if not validate_mandrill_request(secret):
                raise Exception('Invalid sginature')

            try:
                data = json.loads(data)
            except json.JSONDecodeError as exc:
                current_app.logger.exception(exc)
                raise Exception('Not a JSON payload')

            return f(data, *args, **kwargs)

        return verifier
    return with_verifier


class MandrillEventMessage:
    """An interface to parsing a Mandrill event message"""

    event_msg = None
    event_type = None
    subject = None
    sender = None
    recipient = None
    html = None
    text = None

    def __init__(self, event_data):
        """Instantiate the MandrillEventMessage with the event data"""

        self.event_type = event_data.get('event')
        self.event_msg = event_data.get('msg', {})
        self.recipient = self.event_msg.get('email')
        # Inbound email uses `from_email`, other events use `sender`
        self.sender = self.event_msg.get('from_email', self.event_msg.get('sender'))
        self.subject = self.event_msg.get('subject')
        self.html = self.event_msg.get('html')
        self.text = self.event_msg.get('text')

    @property
    def reply_text(self):
        """Return the parsed reply from the message"""

        return EmailReplyParser().parse_reply(self.event_msg.get('raw_msg'))


# Example 1: Define the Mandrill signature key in the app's config
# Add this to the app's routing however you like
@mandrill_verify('CONFIG_KEY_NAME_FOR_MANDRILL_SIGNATURE_KEY')
def mandrill_conversation(data):
    """
    Accepts Mandrill webhook to process reply messages
    """

    for item in data:
        msg = MandrillEventMessage(item)
        if msg.event_type != 'inbound':
            current_app.logger.info('Received a non-acceptable mandrill event type on the '
                                    'inbound hook: {}'.format(msg.event_type))
            continue

        if not msg.reply_text:
            current_app.logger.info('No message content was found')
            continue

        # Do something with the reply text
        text = msg.reply_text

    return '', 204


# Example 2: Use the specific value of the Mandrill signature key
# Add this to the app's routing however you like
@mandrill_verify('actual-signature-key-hardcoded')
def mandrill_outbound_status(data):
    """
    Receives outbound email event statuses from Mandrill
    """

    for item in data:
        msg = MandrillEventMessage(item)
        if msg.event_type not in ['hard_bounce', 'reject', 'unsub']:
            current_app.logger.info('Received a non-acceptable mandrill event type on the '
                                    'bounce hook: {}'.format(msg.event_type))
            continue

        # Unsubscribe is sent "from" a user, every other event is sent "to" a user
        user_email = msg.sender
        if msg.event_type != 'unsub':
            user_email = msg.recipient

        # Handle user's unsubscription or bounce based on event type

    return '', 204