Webhooks

// Node.js 20+: npm install express
import { createHmac, timingSafeEqual } from 'node:crypto';
import express from 'express';

const TOLERANCE_SEC = 5 * 60;

/** true if the signature is valid and the request is no older than 5 minutes. */
export function verifySlicerSignature(rawBody, header, secret, nowMs = Date.now()) {
  if (typeof header !== 'string') return false;
  const pairs = header.split(',').map((part) => part.trim().split('='));
  const timestamp = pairs.find(([key]) => key === 't')?.[1];
  const signatures = pairs.filter(([key]) => key === 'v1').map(([, value]) => value);
  if (!timestamp || !/^\d+$/.test(timestamp) || signatures.length === 0) return false;
  if (Math.abs(nowMs / 1000 - Number(timestamp)) > TOLERANCE_SEC) return false;

  const expected = createHmac('sha256', secret).update(`${timestamp}.`).update(rawBody).digest();
  return signatures.some((value) => {
    const received = Buffer.from(value, 'hex');
    return received.length === expected.length && timingSafeEqual(received, expected);
  });
}

const app = express();
const seen = new Set(); // in production — a database or Redis

// express.raw: the body arrives as a Buffer, exactly as Slicer signed it
app.post('/slicer/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  if (!verifySlicerSignature(req.body, req.get('Slicer-Signature'), process.env.SLICER_WEBHOOK_SECRET)) {
    return res.status(400).send('invalid signature');
  }
  const event = JSON.parse(req.body.toString('utf8'));
  res.sendStatus(204); // answer fast, process afterwards

  if (seen.has(event.id)) return;
  seen.add(event.id);
  setImmediate(() => {
    if (event.type === 'project.succeeded') console.log(`Video ready: ${event.data.title} — ${event.data.appUrl}`);
  });
});

app.listen(3000);

# pip install flask
import hashlib
import hmac
import json
import os
import time

from flask import Flask, abort, request

TOLERANCE_SEC = 5 * 60


def verify_slicer_signature(raw_body, header, secret, now=None):
    """True if the signature is valid and the request is no older than 5 minutes."""
    if not header:
        return False
    pairs = [part.strip().split("=", 1) for part in header.split(",") if "=" in part]
    timestamp = next((value for key, value in pairs if key == "t"), None)
    signatures = [value for key, value in pairs if key == "v1"]
    if timestamp is None or not timestamp.isdigit() or not signatures:
        return False
    if abs((now or time.time()) - int(timestamp)) > TOLERANCE_SEC:
        return False
    expected = hmac.new(secret.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(expected, signature) for signature in signatures)


app = Flask(__name__)
seen = set()  # in production — a database or Redis


@app.post("/slicer/webhook")
def slicer_webhook():
    raw = request.get_data()  # the raw body, before JSON parsing
    if not verify_slicer_signature(raw, request.headers.get("Slicer-Signature"), os.environ["SLICER_WEBHOOK_SECRET"]):
        abort(400)
    event = json.loads(raw)
    if event["id"] not in seen:
        seen.add(event["id"])
        if event["type"] == "project.succeeded":
            print(f"Video ready: {event['data']['title']} — {event['data']['appUrl']}")
    return "", 204