Example Webhook Endpoint Server

This document provides an example implementation of a webhook endpoint server using Python and Flask. This server can receive and validate webhook notifications sent by the Entrupy API.

Dependencies

This code requires Python 3.11 and the following dependencies:

pip install blinker==1.7.0 click==8.1.7 Flask==3.0.0 itsdangerous==2.1.2 Jinja2==3.1.2 MarkupSafe==2.1.3 Werkzeug==3.0.1

Configuration

Before running the server, you need to export the secret key that you configured when creating the webhook in the Entrupy API. This key is used to verify the signature of incoming webhook requests.

export WEBHOOK_ENDPOINT_SECRET_KEY="your-secret-key-here"

Replace "your-secret-key-here" with the actual secret key.

Running the Server

Save the code below as a Python file (e.g., webhook_server.py) and run it:

python webhook_server.py <ADDRESS> <PORT>

For example:

python webhook_server.py localhost 8080

The server will listen for POST requests on the /notify endpoint (e.g., http://localhost:8080/notify).

Example Code

import os
import sys
import hmac
import base64
import hashlib
import flask

"""
This code has been tested to work with python3.11
To run this, you need to first the following dependencies with pip:

pip install blinker==1.7.0
pip install click==8.1.7
pip install Flask==3.0.0
pip install itsdangerous==2.1.2
pip install Jinja2==3.1.2
pip install MarkupSafe==2.1.3
pip install Werkzeug==3.0.1

Also export the secret key you selected for signing messages with
`export WEBHOOK_ENDPOINT_SECRET_KEY="yourkeyhere"`
"""

try:
    # Retrieve the secret key from environment variables
    WEBHOOK_ENDPOINT_SECRET_KEY = os.environ['WEBHOOK_ENDPOINT_SECRET_KEY']
except KeyError:
    print("Error: Environment variable WEBHOOK_ENDPOINT_SECRET_KEY is not set.")
    print("Please export the secret key using:")
    print("export WEBHOOK_ENDPOINT_SECRET_KEY=\"your-secret-key-here\"")
    sys.exit(1)

application = flask.Flask(__name__)
application.config['JSONIFY_PRETTYPRINT_REGULAR'] = False


def sign_payload(string_data, secret_key):
    """Generates the HMAC-SHA256 signature for a given payload and secret key."""
    encoded_secret_key = bytearray(secret_key, 'utf-8')
    encoded_string_data = bytearray(string_data, 'utf-8')
    binary_signature = hmac.new(
        key=encoded_secret_key,
        msg=encoded_string_data,
        digestmod=hashlib.sha256
    ).digest()
    # Return the base64 encoded signature
    return base64.b64encode(binary_signature).decode('utf-8')


def validate_data(string_data, signature):
    """Validates the incoming request signature against a locally calculated signature."""
    # Split the signature header (e.g., "SHA256:abcdef...")
    signature_tokens = signature.split(':')
    if len(signature_tokens) != 2:
        raise AssertionError("Invalid signature format in header.")

    hash_function, hash_from_header = signature_tokens
    # Verify the hash function used (currently always SHA256)
    if hash_function != 'SHA256':
        raise AssertionError(f"Unsupported hash function: {hash_function}")

    # Calculate the expected signature
    correct_hash = sign_payload(string_data, WEBHOOK_ENDPOINT_SECRET_KEY)

    # Securely compare the expected hash with the hash from the header
    if not hmac.compare_digest(correct_hash, hash_from_header):
        raise AssertionError("Signature mismatch! Ensure the correct secret key is configured.")
    # If we reach here, the signature is valid
    print("Signature verified successfully.")


@application.route('/health')
def route__health():
    """A simple health check endpoint."""
    return flask.jsonify(status='ok')


@application.route('/notify', methods=['POST'])
def route__notify():
    """The main endpoint to receive webhook notifications."""
    # Get the raw request body
    string_data = flask.request.data.decode('utf-8')
    try:
        # Get the signature from the request header
        sender_signature = flask.request.headers['Entrupy-Signature']
        # Validate the signature
        validate_data(string_data, sender_signature)
    except (AssertionError, KeyError) as e:
        print(f"Authorization failed: {e}")
        # Return 401 Unauthorized if validation fails or header is missing
        return flask.jsonify(error='unauthorized'), 401

    # --- Process the validated webhook event --- #
    print(f"Received validated data: {string_data}")
    # TODO: Add your code here to process the webhook event.
    # For example, parse the JSON, update your database, trigger other actions, etc.
    # event_data = flask.request.get_json()
    # process_event(event_data)
    # ------------------------------------------- #

    # Return 200 OK to acknowledge receipt
    return flask.jsonify(status='ok'), 200


def main():
    """Parses command-line arguments and starts the Flask server."""
    if len(sys.argv) != 3:
        print(f"Usage: python {sys.argv[0]} <ADDRESS> <PORT>")
        print("Example: python {sys.argv[0]} localhost 8080")
        sys.exit(1)

    address = sys.argv[1]
    try:
        port = int(sys.argv[2])
    except ValueError:
        print(f"Error: Invalid port number '{sys.argv[2]}'")
        sys.exit(1)

    print(f"Starting webhook listener on http://{address}:{port}/notify")
    # Run the Flask development server
    # For production, use a proper WSGI server like Gunicorn or uWSGI
    application.run(host=address, port=port)


if __name__ == "__main__":
    main()