Verify webhook signatures

const crypto = require("node:crypto");
const express = require("express");

const TOLERANCE_SECONDS = 300;

function verifyEmbersWebhook(secret, headers, rawBody) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatures = headers["webhook-signature"];
  if (!id || !timestamp || !signatures || !Buffer.isBuffer(rawBody)) return false;

  const age = Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp));
  if (!Number.isFinite(age) || age > TOLERANCE_SECONDS) return false;

  const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");
  const expected = crypto
    .createHmac("sha256", key)
    .update(`${id}.${timestamp}.`)
    .update(rawBody)
    .digest();

  return signatures.split(" ").some((entry) => {
    const [version, value] = entry.split(",");
    if (version !== "v1" || !value) return false;
    const received = Buffer.from(value, "base64");
    return (
      received.length === expected.length &&
      crypto.timingSafeEqual(received, expected)
    );
  });
}

const app = express();

// express.raw keeps req.body as a Buffer of the exact bytes Embers sent.
app.post("/embers", express.raw({ type: "application/json" }), (req, res) => {
  if (!verifyEmbersWebhook(process.env.EMBERS_WEBHOOK_SECRET, req.headers, req.body)) {
    return res.status(401).send("invalid signature");
  }
  const event = JSON.parse(req.body.toString("utf8"));
  res.sendStatus(204);
  // Process `event` here, after acknowledging.
});

app.listen(3000);

import base64
import hashlib
import hmac
import json
import os
import time

from flask import Flask, request

TOLERANCE_SECONDS = 300


def verify_embers_webhook(secret: str, headers, raw_body: bytes) -> bool:
    event_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not (event_id and timestamp and signatures):
        return False

    try:
        age = abs(int(time.time()) - int(timestamp))
    except ValueError:
        return False
    if age > TOLERANCE_SECONDS:
        return False

    key = base64.b64decode(secret.removeprefix("whsec_"))
    signed_content = f"{event_id}.{timestamp}.".encode("utf-8") + raw_body
    expected = base64.b64encode(
        hmac.new(key, signed_content, hashlib.sha256).digest()
    )

    for entry in signatures.split(" "):
        version, _, value = entry.partition(",")
        if version == "v1" and hmac.compare_digest(value.encode("utf-8"), expected):
            return True
    return False


app = Flask(__name__)


@app.post("/embers")
def embers_webhook():
    # get_data() returns the raw bytes; request.headers is case-insensitive.
    raw_body = request.get_data()
    if not verify_embers_webhook(
        os.environ["EMBERS_WEBHOOK_SECRET"], request.headers, raw_body
    ):
        return "invalid signature", 401
    event = json.loads(raw_body)
    # Hand `event` to a background job, then acknowledge.
    return "", 204