Catch Mailgun Webhooks with Python3, Flask & SQLAlchemy (with a dash of added security...)

Full project source code available on GitHub

Mailgun is a leading provider of Messaging services for the Internet with a beautifully simple REST API that is a breeze to work with in Python.

Mailgun is similar to SendGrid, Postmark and MailChimp. I have found MG's pricing to be very affordable for start-ups.

In one lead generation system I designed this past year, the application sends approximately ~ 1.5 million emails per month. For about $120/month.

In order to track the email's events and behaviors in order to generate lead follow up reports, we have to use Mailgun's webhooks.

A webhook is a method of modifying the state; or behavior of a web page or web application, with custom callbacks.

Mailgun has several default messaging events that can be tracked.

  1. Delivered
  2. Opened
  3. Bounced
  4. Unsubscribed
  5. Complained (spam)
  6. Clicked
  7. Permanent Fail (dropped)

Mailgun's API automatically registers these events as the data is flowing from our application to their API. Each time a Delivered event occurs, the Mailgun API calls the Delivered webhook, the email recipient address is dumped into a form along with some additional fields; and the data is POSTed to the URL defined in the webhook setup for Delivered.

Once this event fires, Mailgun makes a webhook call to our API; our database is updated with the event that the email we sent via Mailgun has been successfully delivered to the recipient. In order to process these webhooks, we need to build an API that can accept the POST requests and process the email statuses and follow up data.

The Mailgun webhook HTTP POST request looks like so:

form_data = {  
    "message_id": request.form.get('Message-Id', None),
    "x_mail_gun_sid": request.form.get('X-Mailgun-Sid', None),
    "domain": request.form.get('domain', 'email.com'),
    "event": request.form.get('event', 'delivered'),
    "timestamp": request.form.get('timestamp', None),
    "recipient": request.form.get('recipient', None),
    "signature": request.form.get('signature', None),
    "token": request.form.get('token', None)
}

The timestamp, signature and token fields will be used for Signature Verification, which we'll get to shortly.

Let's start with the base Flask app. The single greatest benefit of Flask is it's flexibility. I can design an API that uses a single route to accept a single request, usually a HTTP VERB. like GET, POST, PUT, etc... very much like a RESTful API would respond to requests. While the architecture is not REST, it's REST like.

Let's begin.

The first thing I am going to do is layout the routes for the webhooks and quickly scaffold the most basic of Flask apps. I'm not including the models.py or config.py, please see the source project on GitHub for more details.

Next, create your project directory. I'm working in my default workspace: ~/sites

craig@precision-5810:~/sites$ mkdir mailgun-webhooks  
craig@precision-5810:~/sites$ cd mailgun-webhooks  
craig@precision-5810:~/sites/mailgun-webhooks$ virtualenv .env --python=python3  
craig@precision-5810:~/sites/mailgun-webhooks$ source .env/bin/activate  
craig@precision-5810:~/sites/mailgun-webhooks$ pip3 install Flask Flask-SQLAlchemy  
craig@precision-5810:~/sites/mailgun-webhooks$ nano app.py  

Open app.py in your editor of choice.

Simple enough. Fire up your app and check your routes. Point your browser to http://localhost:5000/api/v1/index

The response should looks like this:

{
  "clicks": "/api/v1/wh/mg/lead/email/click", 
  "delivered": "/api/v1/wh/mg/lead/email/delivered", 
  "dropped": "/api/v1/wh/mg/lead/email/dropped", 
  "bounced": "/api/v1/wh/mg/lead/email/bounced", 
  "opens": "/api/v1/wh/mg/lead/email/open", 
  "complaint": "/api/v1/wh/mg/lead/email/spam/complaint", 
  "unsubscribe": "/api/v1/wh/mg/lead/email/unsubscribe"
}

OK! Perfect. Let's add our first function for the Delivered webhook. I will discuss each section below in more detail.

First, we define the function and the webhook route:

@app.route('/api/v1/wh/mg/lead/email/delivered', methods=['POST'])

Note, this route will only accept POST requests.

Next, we define our function and give it a name:

def lead_email_delivered():  

Don't forget to add your docstrings. It's very important for documentation.

Next, we check the request method and if it's a POST request, we accept the data and setup our form dictionary of elements.

Great! Next, I re-encode the strings we will use for the Signature Verification.

These strings must be properly encoded in UTF-8 format for Signature Verification to work correctly.

Define the fields from our request body and re-encode

# verify the mailgun token and signature with the api_key
token = form_data['token'].encode('utf-8')  
timestamp = form_data['timestamp'].encode('utf-8')  
signature = form_data['signature'].encode('utf-8')  
mg_recipient = form_data['recipient']  
event = form_data['event']  

Now we get to implement our webhook security. It's an often overlooked detail, but we need to secure our webhooks against unauthorized requests. This is done with Signature Verification.

Here is how it works. Mailgun sends us 3 additional pieces of data in our POST request. The token, the timestamp and the signature. Our Mailgun API key is known only to us and Mailgun.

The signature hash is generated by concatenating the token and the timestamp as the Message.

The Message data is then hashed by the Keyed Hashing for Message Authentication (hmac) Python library using our API KEY as the hmac's key.

This produces a SHA256 hash like so...

faa79b8add135baa7d37bdd3808e547910f06e577d83f4194226328f5f74f940  

Our Signature Verification function will then simply compare the hashes. If True, we accept the data and continue to process the webhook.

The Signature Verification function...

Continuing, call the verify function from our lead email delivered function.

if verify(mailgun_api_key, token, timestamp, signature):  

This should return True; next lookup the email recipient in our MySQL database.

If the verification fails, the API will return a HTTP 409 response and die.

EDIT: I realize that I did not include a lot of details on the database portion of this post. Not to worry. I am using the ORM model for interacting with the database through a basic db.session call with SQLAlchemy. For more information, see here

Debugging Webhooks

To test your code and debug the webhooks locally while you are developing, I personally use ngrok to create a HTTPS tunnel to my locally running application. I then use the Tunnel Introspection features to capture and view the webhook POST data.

ngrok console...
ngrok console

and the ngrok status page. I can't tell you how extremely powerful this is when debugging.

ngrok web inspection

Finally, once you have the webhook working as expected, replicate the lead email delivered function for the remaining webhooks outlined above.

I will dig deeper into testing webhooks in a separate chapter and include more screenshots from ngrok.

Full source for this project is available on GitHub

Wrapping Up

Python, along with Flask, SQLAlchemy and ngrok makes working with webhooks a quick, simple and straight-forward process. Debug easily in ngrok's Inspect UI. I use ngrok in my daily work, especially for making local development websites available to remote users for testing. Mailgun is a great SMTP service provider and their documentation for their product offerings are top notch.

Mailgun is sending millions of emails a month for my business applications. How does your service provider measure up?

Craig Derington

Espousing the virtues of Secular Humanism, Libertarianism, Free and Open Source Software, Linux, Ubuntu, CentOS, Terminal Multiplexing, Docker, Python, Flask, Django, MySQL, MongoDB and Git.

comments powered by Disqus