Python Webhook ReceiverΒΆ

Download webhook.py

"""
Reference webhook receiver endpoint.

$ openssl req -x509 -newkey rsa:4096 -keyout sslkey.pem -out sslcert.pem -nodes -subj '/CN=localhost'
$ virtualenv env
$ env/bin/pip install uwsgi  # may need to set UWSGI_INCLUDES=/path/to/openssl/include
$ env/bin/uwsgi --master --set api_key=SECRET --wsgi-file webhook.py --https 0.0.0.0:8443,sslcert.pem,sslkey.pem

To test using the default api key of "SECRET":
$ curl -k -v --header "X-Signature: 3q8QXTAGaey18yL8FWTqdVlbMr6hcuNvM4tefa0o9nA=" --data '{}' https://localhost:8443/sent
"""

import base64
import hashlib
import hmac
import logging

try:
    import uwsgi

    logging.basicConfig(level=logging.INFO)
    log = logging.getLogger(__name__)
    api_key = uwsgi.opt.get("api_key")
    if not api_key:
        log.error("Specify api key using uwsgi --set api_key=XXX")
        uwsgi.stop()
except ImportError:
    api_key = None


def verify_signature(environ, body):
    # Fail if no signature supplied
    if "HTTP_X_SIGNATURE" not in environ:
        log.warning("No signature supplied, forbidden")
        return "403 Forbidden"
    signature = environ["HTTP_X_SIGNATURE"]
    computed_signature = base64.b64encode(
        hmac.new(api_key.encode("utf-8"), body, digestmod=hashlib.sha256).digest()
    )
    if signature != computed_signature:
        log.warning("Signature does not match, forbidden (%s != %s)", signature, computed_signature)
        return "403 Forbidden"
    log.info("Signature validated")
    return "204 No Content"


def application(environ, start_response):
    body = environ["wsgi.input"].read(int(environ.get("CONTENT_LENGTH", 0)))
    log.info("Body: %s", body)

    response_status = verify_signature(environ, body)

    start_response(response_status, [])
    return []