Webhooks

Creating your own webhooks allows you to respond to events generated by UserHub within your application. Once established, UserHub will deliver events via an HTTP post to your endpoint with the following convention:

/your-webhook-url?action=events.handle

Included in the post will be an Event Object body containing information about the type of event, and details about the entity that changed.

UserHub currently delivers these event actions:

ActionPossible event types
events.handle
  • flows.changed
  • members.changed
  • organizations.changed
  • subscriptions.changed
  • users.changed

And the following user actions when the Custom Users Provider is enabled:

ActionExpected response
users.listList of User Objects
users.createUser Object
users.getUser Object
users.updateUser Object
users.deleteUser Object

This guide will walk you through setting up a handler for the Events Webhook.

Setup Directory

mkdir python-webhooks
cd python-webhooks
python3 -m venv venv
./venv/bin/pip install 'flask>=3,<4'

Create HTTP server

# app.py

import os

from flask import Flask

app = Flask(__name__)

@app.route("/webhook")
def webhook():
    return {}

if __name__ == "__main__":
    port = int(os.environ.get("PORT") or 3000)
    app.run(debug=True, port=port)

Implement Challenge

Webhooks must successfully respond to a challenge request before they can start listening for events.

See Challenge for more information.

Update the HTTP server to listen for the challenge request and set the response body to the request body.

# app.py

import os

from flask import Flask, request

app = Flask(__name__)

@app.route("/webhook", methods=["POST"])
def webhook():
    action = request.args.get("action")

    if action == "challenge":
        return request.json
    else:
        print(f"handler not implemented: {action}: {request.json}")
        return {}

if __name__ == "__main__":
    port = int(os.environ.get("PORT") or 3000)
    app.run(debug=True, port=port)

Setup Tunnel

Make your HTTP server publicly routable.

If you already have a preferred method for doing this you can skip this step, otherwise follow the instructions below to sign up for a free ngrok account.

  1. Sign up for ngrok.
  2. Install the CLI client:
brew install ngrok
  1. Setup the ngrok authtoken:
ngrok config add-authtoken stp1KM0tsJgSZQ...
  1. Create a free static domain via Domains under Cloud Edge in the ngrok dashboard.
  2. Copy and paste the start command into your terminal:
ngrok http --domain=some-wacky-name.ngrok-free.app 3000

Test Access

Start the HTTP server:

./venv/bin/python app.py

Make a request to the public webhook endpoint:

curl https://some-wacky-name.ngrok-free.app/webhook?action=challenge \
  -H 'Content-Type: application/json' -d '{"challenge": "test"}'

Ensure you get the following response:

{"challenge": "test"}

Create Webhook

  1. Go to the Webhooks page under Developers in the Admin Console

  2. Click New webhook

  3. Set the URL field to the public URL for your webhook:

https://some-wacky-name.ngrok-free.app/webhook
  1. Click Save

Test Webhook

To check that the webhook is handling requests correctly, click the Test webhook.

If you see the Test succeeded message everything is setup correctly.

If you get an error, use the message to debug the issue.

Verify Signature

For production deployments you should verify the incoming request using the webhook signing secret and optionally add an authorization header to the webhook settings.

To verify the signature you need to combine the incoming UserHub-Timestamp header with the content of the body and pass it through an HMAC-SHA256 hash function and compare that against the UserHub-Signature header.

The signature function is as follows:

HMAC-SHA256(headers["UserHub-Timestamp"] + "." + body, signingSecret)

Update the server to verify the signature:

# app.py

import hashlib
import hmac
import os
from datetime import datetime, timedelta

from flask import Flask, abort, jsonify, make_response, request

app = Flask(__name__)

def verify():
    error = make_response(jsonify(message="Failed to verify signature"), 500)

    secret = os.environ.get("SIGNING_SECRET")
    if not secret:
        abort(error)

    timestamp = request.headers.get("UserHub-Timestamp")
    if not timestamp or not timestamp.isdigit():
        abort(error)

    dt = datetime.fromtimestamp(int(timestamp))
    if abs(datetime.now() - dt) > timedelta(minutes=5):
        abort(error)

    signature = request.headers.get("UserHub-Signature")
    if not signature:
        abort(error)

    h = hmac.new(
        secret.encode("utf-8"),
        msg=timestamp.encode("utf-8") + b"." + request.data,
        digestmod=hashlib.sha256
    )
    if not hmac.compare_digest(h.hexdigest(), signature):
        abort(error)

@app.route("/webhook", methods=["POST"])
def webhook():
    verify()

    action = request.args.get("action")

    if action == "challenge":
        return request.json
    else:
        print(f"handler not implemented: {action}: {request.json}")
        return {}

if __name__ == "__main__":
    port = int(os.environ.get("PORT") or 3000)
    app.run(debug=True, port=port)

You should copy the signing secret from the Webhooks section in the Admin Console and set the environment variable SIGNING_SECRET to the copied secret.

Restart the HTTP server and then hit the Test webhook button again to ensure the HTTP server is correctly verifying incoming requests.

Handle Action

The final code change is to watch for the webhook action events.handle and handle the incoming events.

Update the server to handle the events.handle action:

# app.py

import hashlib
import hmac
import os
from datetime import datetime, timedelta

from flask import Flask, abort, jsonify, make_response, request

app = Flask(__name__)

def verify():
    error = make_response(jsonify(message="Failed to verify signature"), 500)

    secret = os.environ.get("SIGNING_SECRET")
    if not secret:
        abort(error)

    timestamp = request.headers.get("UserHub-Timestamp")
    if not timestamp or not timestamp.isdigit():
        abort(error)

    dt = datetime.fromtimestamp(int(timestamp))
    if abs(datetime.now() - dt) > timedelta(minutes=5):
        abort(error)

    signature = request.headers.get("UserHub-Signature")
    if not signature:
        abort(error)

    h = hmac.new(
        secret.encode("utf-8"),
        msg=timestamp.encode("utf-8") + b"." + request.data,
        digestmod=hashlib.sha256
    )
    if not hmac.compare_digest(h.hexdigest(), signature):
        abort(error)

@app.route("/webhook", methods=["POST"])
def webhook():
    verify()

    action = request.args.get("action")

    if action == "challenge":
        return request.json
    elif action == "events.handle":
        print(request.json)
        return {}
    else:
        print(f"handler not implemented: {action}: {request.json}")
        return {}

if __name__ == "__main__":
    port = int(os.environ.get("PORT") or 3000)
    app.run(debug=True, port=port)

Subscribe to Event

To receive events you need to subscribe to them.

  1. Go to the Webhooks section of the Admin Console
  2. Click the webhook we created above
  3. Check the organizations.changed checkbox under Send events
  4. Click Save

Send Event

  1. Restart your HTTP server and make sure ngrok is still running
  2. Go to Organizations in the Admin Console
  3. Click the New organization button
  4. Enter a Display name
  5. Click Save

You should see an organizations.changed event delivered to your webhook.

PreviousGet session
NextChallenge

Turn users intorevenue
$

Subscribe to monthly product updates

© 2024 UserHub

Integrations

    UserHub & Auth0UserHub & Stripe BillingUserHub & Google CloudUserHub & FirebaseUserHub & Custom Auth

Resources