Webhooks

import { createHmac, timingSafeEqual } from 'node:crypto';

// secret: the whsec_… value shown when you created the webhook.
// rawBody: the request body exactly as received (Buffer or string).
// tolerance: the allowed clock difference, in seconds.
export function verifyWebhook(secret, headers, rawBody, tolerance = 300) {
  const id = headers['webhook-id'];
  const timestamp = headers['webhook-timestamp'];
  const signatures = headers['webhook-signature'];
  if (!id || !timestamp || !signatures) {
    throw new Error('missing webhook headers');
  }
  const age = Math.abs(Date.now() / 1000 - Number(timestamp));
  if (!/^\d+$/.test(timestamp) || age > tolerance) {
    throw new Error('timestamp outside tolerance');
  }
  const key = Buffer.from(secret.replace(/^whsec_/, ''), 'base64');
  const expected = createHmac('sha256', key)
    .update(`${id}.${timestamp}.`)
    .update(rawBody)
    .digest();
  const valid = signatures.split(' ').some(entry => {
    const [version, value = ''] = entry.split(',');
    const given = Buffer.from(value, 'base64');
    return version === 'v1'
      && given.length === expected.length
      && timingSafeEqual(given, expected);
  });
  if (!valid) throw new Error('invalid signature');
  return JSON.parse(rawBody);
}

import { createServer } from 'node:http';
import { verifyWebhook } from './verify.mjs';

const secret = process.env.CLITRAIL_WEBHOOK_SECRET;
const seen = new Set(); // Use your database: webhook-id repeats on retries.

createServer(async (req, res) => {
  const chunks = [];
  for await (const chunk of req) chunks.push(chunk);
  let event;
  try { event = verifyWebhook(secret, req.headers, Buffer.concat(chunks)); }
  catch { return res.writeHead(400).end(); }
  const id = req.headers['webhook-id'];
  if (!seen.has(id)) {
    seen.add(id);
    // event.type: cli.install_completed, cli.first_run,
    // cli.install_started or test.ping
    console.log(id, event.type);
  }
  res.writeHead(204).end();
}).listen(Number(process.env.PORT) || 3000);

No dependencies; Node.js 18 or later. With Express, read the body with express.raw({ type: 'application/json' }) and pass req.body.

import base64
import hashlib
import hmac
import json
import time


def verify_webhook(secret, headers, body, tolerance=300):
    """Return the event or raise ValueError. body: raw request bytes."""
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not (msg_id and timestamp and signatures):
        raise ValueError("missing webhook headers")
    now = time.time()
    if not timestamp.isdigit() or abs(now - int(timestamp)) > tolerance:
        raise ValueError("timestamp outside tolerance")
    key = base64.b64decode(secret.removeprefix("whsec_"))
    signed = f"{msg_id}.{timestamp}.".encode() + body
    digest = hmac.new(key, signed, hashlib.sha256).digest()
    expected = base64.b64encode(digest)
    for entry in signatures.split(" "):
        version, _, sig = entry.partition(",")
        if version == "v1" and hmac.compare_digest(sig.encode(), expected):
            return json.loads(body)
    raise ValueError("invalid signature")

import os
from http.server import BaseHTTPRequestHandler, HTTPServer

from verify import verify_webhook

SECRET = os.environ["CLITRAIL_WEBHOOK_SECRET"]
seen = set()  # Use your database: webhook-id repeats on retries.


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
        try:
            event = verify_webhook(SECRET, self.headers, body)
        except ValueError:
            self.send_response(400)
            self.end_headers()
            return
        msg_id = self.headers["webhook-id"]
        if msg_id not in seen:
            seen.add(msg_id)
            print(msg_id, event["type"])
        self.send_response(204)
        self.end_headers()


PORT = int(os.environ.get("PORT", "3000"))
HTTPServer(("", PORT), Handler).serve_forever()

Standard library only; Python 3.9 or later. With Flask, call verify_webhook(SECRET, request.headers, request.get_data()).