Webhooks

import hashlib, hmac, os, time
from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["CAPIBO_WEBHOOK_SECRET"].encode()

@app.post("/hooks/capibo")
def capibo_webhook():
    header = dict(p.split("=", 1) for p in request.headers["Capibo-Signature"].split(","))
    signed = f"{header['t']}.".encode() + request.get_data()
    expected = hmac.new(SECRET, signed, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, header["v1"]) or time.time() - int(header["t"]) > 300:
        abort(400)

    event = request.get_json()
    if event["type"] == "video.accepted":
        enqueue_download(event["data"])  # your code
    return "", 204

import crypto from "node:crypto";
import express from "express";

const app = express();
const secret = process.env.CAPIBO_WEBHOOK_SECRET;

app.post("/hooks/capibo", express.raw({ type: "application/json" }), (req, res) => {
  const header = Object.fromEntries(req.get("Capibo-Signature").split(",").map((p) => p.split("=")));
  const expected = crypto.createHmac("sha256", secret).update(`${header.t}.${req.body}`).digest("hex");
  const fresh = Date.now() / 1000 - Number(header.t) < 300;
  const valid = header.v1?.length === expected.length && crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(header.v1));
  if (!fresh || !valid) {
    return res.sendStatus(400);
  }

  const event = JSON.parse(req.body);
  if (event.type === "video.accepted") enqueueDownload(event.data); // your code
  res.sendStatus(204);
});