webhook_receiver.py

"""Receives Subcontext webhooks (the "Webhook endpoint" connection), in Python 3.9
or later with the standard library only.

Each delivery is a POST with a JSON body and these headers:
  X-Subcontext-Signature: t=<unix seconds>,v1=<hex HMAC-SHA256 of "<t>.<raw body>">
  Idempotency-Key: <the same value on every retry of this delivery>

Run it:
  SUBCONTEXT_SIGNING_SECRET=THE_SIGNING_SECRET python3 webhook_receiver.py
Subcontext only sends to https addresses on the public internet, so put it
behind your usual HTTPS web server or load balancer.
"""

import hashlib
import hmac
import json
import os
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

# The Signing secret you entered on the Webhook endpoint connection.
SECRET = os.environ.get("SUBCONTEXT_SIGNING_SECRET")
PORT = int(os.environ.get("PORT", "4100"))
TOLERANCE_SECONDS = 300
if not SECRET:
    raise SystemExit("Set SUBCONTEXT_SIGNING_SECRET to the connection's signing secret.")

SEEN = set()  # Idempotency-Keys already handled. Use your database in real code.


def verify_signature(raw_body: bytes, header: str, secret: str, now: float = None) -> bool:
    """True when the header was made from this exact body with your secret, recently."""
    parts = dict(part.split("=", 1) for part in (header or "").split(",") if "=" in part)
    timestamp, signature = parts.get("t", ""), parts.get("v1", "")
    if not timestamp.isdigit() or not signature:
        return False
    if abs((now or time.time()) - int(timestamp)) > TOLERANCE_SECONDS:
        return False
    signed = timestamp.encode() + b"." + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)


def handle(event: dict) -> None:
    # Your work goes here: create or update the record in your CRM, and so on.
    name = (event.get("contact") or {}).get("name") or "someone"
    print(f"{event.get('event_type') or event.get('type')}: {name} "
          f"({event.get('contact_id') or event.get('id')})")


class Handler(BaseHTTPRequestHandler):
    def reply(self, status: int, text: str) -> None:
        data = text.encode()
        self.send_response(status)
        self.send_header("Content-Type", "text/plain")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def do_POST(self):
        # Read the body exactly as it arrived. The signature covers these bytes.
        raw = self.rfile.read(int(self.headers.get("Content-Length") or 0))

        if not verify_signature(raw, self.headers.get("X-Subcontext-Signature", ""), SECRET):
            # 400, 401 or 403 each refuse only this delivery. Later ones still arrive.
            return self.reply(400, "bad signature")

        key = self.headers.get("Idempotency-Key")
        if key and key in SEEN:
            return self.reply(200, "already had it")

        event = json.loads(raw)
        # Answer quickly (within 15 seconds). In real code, put the slow work on a queue.
        self.reply(200, "ok")
        if key:
            SEEN.add(key)
        handle(event)


if __name__ == "__main__":
    print(f"Webhook receiver on http://localhost:{PORT}")
    ThreadingHTTPServer(("", PORT), Handler).serve_forever()