Webhooks and signatures

import crypto from 'node:crypto';

const TOLERANCE_SECONDS = 5 * 60;

/**
 * Verifies a Codzen webhook and returns the generation it carries.
 * rawBody: the request body as a string or Buffer, exactly as received.
 * headers: the request headers, with lowercase names.
 * secret: your webhook signing secret, `whsec_…`.
 */
function verifyWebhook(rawBody, headers, secret) {
  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 now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - Number(timestamp)) > TOLERANCE_SECONDS) {
    throw new Error('Webhook timestamp is outside the tolerance');
  }

  const key = Buffer.from(secret.slice('whsec_'.length), 'base64');
  const expected = crypto
    .createHmac('sha256', key)
    .update(`${id}.${timestamp}.`)
    .update(rawBody)
    .digest();

  const valid = signatures.split(' ').some((signature) => {
    const [version, value] = signature.split(',');
    if (version !== 'v1' || !value) return false;
    const received = Buffer.from(value, 'base64');
    return (
      received.length === expected.length &&
      crypto.timingSafeEqual(received, expected)
    );
  });
  if (!valid) {
    throw new Error('Invalid webhook signature');
  }

  return JSON.parse(rawBody.toString());
}

import base64
import hashlib
import hmac
import json
import time

TOLERANCE_SECONDS = 5 * 60


def verify_webhook(raw_body: bytes, headers, secret: str):
    """Verifies a Codzen webhook and returns the generation it carries.

    raw_body: the request body as bytes, exactly as received.
    headers: the request headers (a dict with lowercase names, or any
        mapping with case-insensitive lookup, such as Flask's).
    secret: your webhook signing secret, "whsec_...".
    """
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not msg_id or not timestamp or not signatures:
        raise ValueError("Missing webhook headers")

    if abs(int(time.time()) - int(timestamp)) > TOLERANCE_SECONDS:
        raise ValueError("Webhook timestamp is outside the tolerance")

    key = base64.b64decode(secret.removeprefix("whsec_"))
    signed = f"{msg_id}.{timestamp}.".encode() + raw_body
    expected = base64.b64encode(
        hmac.new(key, signed, hashlib.sha256).digest()
    ).decode()

    for signature in signatures.split(" "):
        version, _, value = signature.partition(",")
        if version == "v1" and hmac.compare_digest(value, expected):
            return json.loads(raw_body)
    raise ValueError("Invalid webhook signature")

import express from 'express';

const app = express();

app.post(
  '/webhooks/codzen',
  express.raw({ type: 'application/json' }),
  (req, res) => {
    let generation;
    try {
      generation = verifyWebhook(
        req.body,
        req.headers,
        process.env.CODZEN_WEBHOOK_SECRET
      );
    } catch {
      return res.sendStatus(400);
    }
    // Skip it if you have already handled req.headers['webhook-id'].
    console.log(generation.id, generation.status);
    res.sendStatus(200);
  }
);

import os

from flask import Flask, request

app = Flask(__name__)


@app.post("/webhooks/codzen")
def codzen_webhook():
    try:
        generation = verify_webhook(
            request.get_data(),
            request.headers,
            os.environ["CODZEN_WEBHOOK_SECRET"],
        )
    except ValueError:
        return "", 400
    # Skip it if you have already handled request.headers["webhook-id"].
    print(generation["id"], generation["status"])
    return "", 200