Verifying signed webhooks

When using the Nexmo APIs (now referred to as the Vonage APIs), your application needs to be notified of certain events. Examples of these events could be an inbound message, an inbound phone call, or the user hanging up and so on. These notifications come from Nexmo to your application via webhooks. Webhooks are essential HTTP requests (GET, POST typically), and are inbound onto the URLs you specified as your webhook URLs for your application. For example, you might have an inbound message webhook of https://www.yourdomain.com/webhooks/inbound. For some APIs, such as our Messages API, customers wanted to have more security on webhooks inbound to their application from Nexmo. In particular they wanted to verify that:

1. The source of the webhook is actually Nexmo
2. The webhook has not been tampered with on route

For Messages API webhooks are now secure by default. This article looks at how this is done, and how you can:

1. Verify the source of the webhook
2. Verify the payload has not been tampered with

If you are using https as your webhook endpoint, step 2 is optional, but does not incur too much overhead.

How Nexmo secures webhooks

For the Messages API, signed webhooks are turned on by default. Webhooks have a JWT in the authorization header, which is signed with your signature secret. Your signature secret can be found in your Nexmo dashboard. The signing algorithm is always SHA-256.

In addition to signing the JWT with your signature secret, the payload is hashed and included in the payload_hash field of the JWT.

How to verify the signed webhook

There are two parts to verifying the signed webhook:

1. Verify the signed webhook using your signature secret. There is a slight complication here in that you might have different API keys and each API key has a different signature secret associated with it. This mainly applies to customers with multiple accounts, or subaccounts.

2. Hash the request payload with SHA-256, and compare the result to the payload_hash contained in the JWT. If they are different it could be the payload has been tampered with.

The code

The example code shown here shows how to verify signed callbacks for the inbound messages webhook only. It would be possible to move this code to a function or two, and then perform the same operation also for the message status webhooks.

This code does the following:

1. It extracts the API key from JWT payload so it can work out which corresponding signature secret to use. This is useful where you have multiple accounts.
2. It verifies the JWT using the signature secret. PyJWT has a nice feature in that it verifies by default as part of the decode() function. In the case where verification fails an exception is generated and caught, and 401 returned.
3. It hashes the request payload using SHA-256 and verifies that this matches payload_hash in the JWT signature of the callback. You would only need to do this if using HTTP rather than HTTPS, so if you want you can make this part optional.

The Python code, with comments, is shown here:

import jwt
import hashlib
import json
import os
from flask import Flask, request, jsonify
from os.path import join, dirname
from dotenv import load_dotenv
from pprint import pprint

#Load the environment
envpath = join(dirname(__file__),'../.env')
load_dotenv(envpath)

app = Flask(__name__)

port = os.getenv('PORT')
api_key = os.getenv('NEXMO_API_KEY') # there may be multiple api_key/sig_secret pairs
sig_secret = os.getenv('NEXMO_SIG_SECRET')

@app.route("/webhooks/inbound", methods=['POST'])
def inbound():
    # Need token after 'Bearer'
    parts = request.headers['authorization'].split() 
    token = parts[1].strip()

    # Extract api_key from token payload
    k = jwt.decode(token, verify=False)["api_key"]
    # Use k to look up corresponding sig secret

    #### 1. Verify request

    try:
        decoded = jwt.decode(token, sig_secret, algorithms='HS256')
    except Exception as e:
        print(e)
        r = '{"msg": "' + str(e) +'"}'
        return (r, 401)

    #### 2. Verify payload (only needed if using HTTP rather than HTTPS)

    # Obtain transmitted payload hash
    payload_hash = decoded["payload_hash"]

    # generate hash of request payload
    payload = request.data # Obtains request data as binary string
    h = hashlib.sha256(payload) # requires binary string
    hd = h.hexdigest() # Use hexdigest() and NOT digest()

    # Check the payload hash matches the one we created ourselves from request data
    if (hd != payload_hash):
        return ('{"msg": "Invalid payload"}', 401)
    else:
        print("Verified payload")
    return "OK"

# You could modify the code to also check this webhook
@app.route("/webhooks/status", methods=['POST'])
def status():
    data = request.get_json()
    return (jsonify(data))

if __name__ == '__main__':
    print("Running locally")
    app.run(host="localhost", port=port)

It's possible to improve this code quite a bit. I kept it as simple as possible to make the code easier to follow.

One modification I would make is to move the signature verification and payload checking to their own functions. This would make it easy to call these functions as required from both inbound messages and message status webhooks.

The other modification is to better handle multiple API keys/signature secret pairs. In this code I just set the API key and signature secret using environment variables, but in a real case you would have a table (Python dictionary perhaps) of API key and signature secret pairs. I did show how you could extract the API key, you could then use that to look up the corresponding signature secret in the table.

Summary

For security certain Nexmo APIs such as the Messages API use signed webhooks. This article is shown how you can verify the source of the webhook, and also check that the request payload has not been tampered with.

Resources

• Nexmo docs on signed webhooks
• PyJWT docs