Webhooks

import base64
import hashlib
import hmac
import time


def verify(secret: str, headers: dict, raw_body: bytes, tolerance: int = 300) -> bool:
    msg_id = headers["webhook-id"]
    timestamp = headers["webhook-timestamp"]
    signatures = headers["webhook-signature"]
    if abs(time.time() - int(timestamp)) > tolerance:
        return False
    key = base64.b64decode(secret.split("_", 1)[1])
    signed = f"{msg_id}.{timestamp}.".encode() + raw_body
    expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode()
    return any(
        hmac.compare_digest(expected, s[3:])
        for s in signatures.split(" ")
        if s.startswith("v1,")
    )

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

function verify(secret, headers, rawBody, tolerance = 300) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  const signatures = headers["webhook-signature"];
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > tolerance) return false;
  const key = Buffer.from(secret.slice(secret.indexOf("_") + 1), "base64");
  const expected = crypto
    .createHmac("sha256", key)
    .update(`${id}.${timestamp}.`)
    .update(rawBody)
    .digest("base64");
  return signatures.split(" ").some(
    (s) =>
      s.startsWith("v1,") &&
      s.length - 3 === expected.length &&
      crypto.timingSafeEqual(Buffer.from(s.slice(3)), Buffer.from(expected)),
  );
}

package fiveapi

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/base64"
	"strconv"
	"strings"
	"time"
)

// Verify checks the signature of a Five API webhook. secret is the whsec_... value from the console.
func Verify(secret, id, timestamp, signatures string, rawBody []byte) bool {
	ts, err := strconv.ParseInt(timestamp, 10, 64)
	if err != nil {
		return false
	}
	if age := time.Now().Unix() - ts; age > 300 || age < -300 {
		return false
	}
	key, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_"))
	if err != nil {
		return false
	}
	mac := hmac.New(sha256.New, key)
	mac.Write([]byte(id + "." + timestamp + "."))
	mac.Write(rawBody)
	expected := base64.StdEncoding.EncodeToString(mac.Sum(nil))
	for _, s := range strings.Split(signatures, " ") {
		if strings.HasPrefix(s, "v1,") && hmac.Equal([]byte(s[3:]), []byte(expected)) {
			return true
		}
	}
	return false
}

const express = require("express");
const app = express();

// express.raw keeps the exact bytes the signature was computed over
app.post("/fiveapi", express.raw({ type: "application/json" }), (req, res) => {
  if (!verify(process.env.FIVEAPI_WEBHOOK_SECRET, req.headers, req.body)) {
    return res.status(400).send("invalid signature");
  }
  const event = JSON.parse(req.body);
  res.sendStatus(200); // acknowledge first
  if (alreadyProcessed(event.id)) return; // retries reuse the same webhook-id
  setImmediate(() => handleEvent(event)); // then do the work
});

function handleEvent(event) {
  const task = event.data;
  switch (event.type) {
    case "task.completed":
      return saveVideo(task.id, task.output.video.url, task.metadata);
    case "task.failed":
      return markFailed(task.id, task.error); // resubmit when task.error.retryable
    case "task.cancelled":
      return markCancelled(task.id);
  }
}

import json
import os
import threading

from flask import Flask, abort, request

app = Flask(__name__)


@app.post("/fiveapi")
def fiveapi_webhook():
    raw_body = request.get_data()  # the exact bytes, before JSON parsing
    if not verify(os.environ["FIVEAPI_WEBHOOK_SECRET"], request.headers, raw_body):
        abort(400)
    event = json.loads(raw_body)
    if not already_processed(event["id"]):  # retries reuse the same webhook-id
        threading.Thread(target=handle_event, args=(event,)).start()
    return "", 200