Webhooks

import express from 'express';
import { createHmac, timingSafeEqual } from 'node:crypto';

const SECRET = process.env.INVENT_WEBHOOK_SECRET!;
const TOLERANCE_S = 5 * 60;

const app = express();

// raw body — do NOT use express.json() before verifying
app.post(
  '/hooks/invent',
  express.raw({ type: 'application/json' }),
  (req, res) => {
    const id = req.header('x-webhook-id') ?? '';
    const timestamp = req.header('x-webhook-timestamp') ?? '';
    const header = req.header('x-webhook-signature') ?? '';

    const signature = header
      .split(',')
      .find((part) => part.startsWith('v1='))
      ?.slice(3);

    if (!signature) return res.sendStatus(400);

    if (Math.abs(Date.now() / 1000 - Number(timestamp)) > TOLERANCE_S) {
      return res.sendStatus(400);
    }

    const expected = createHmac('sha256', SECRET)
      .update(`${id}.${timestamp}.${req.body.toString('utf8')}`)
      .digest('hex');

    const a = Buffer.from(expected);
    const b = Buffer.from(signature);

    if (a.length !== b.length || !timingSafeEqual(a, b)) {
      return res.sendStatus(401);
    }

    const event = JSON.parse(req.body.toString('utf8'));

    switch (event.type) {
      case 'broadcast.bounced': {
        // handle it
        break;
      }
      default: {
        break;
      }
    }

    // acknowledge immediately, process asynchronously
    res.sendStatus(200);
  },
);

import hashlib
import hmac
import time

from flask import Flask, request

SECRET = os.environ["INVENT_WEBHOOK_SECRET"]
TOLERANCE_S = 5 * 60

app = Flask(__name__)


@app.post("/hooks/invent")
def receive():
    raw = request.get_data()  # raw bytes, not request.json
    event_id = request.headers.get("x-webhook-id", "")
    timestamp = request.headers.get("x-webhook-timestamp", "")
    header = request.headers.get("x-webhook-signature", "")

    signature = next(
        (p[3:] for p in header.split(",") if p.startswith("v1=")), None
    )

    if not signature:
        return "", 400

    if abs(time.time() - int(timestamp or 0)) > TOLERANCE_S:
        return "", 400

    expected = hmac.new(
        SECRET.encode(),
        f"{event_id}.{timestamp}.".encode() + raw,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(expected, signature):
        return "", 401

    event = request.get_json()

    if event["type"] == "broadcast.bounced":
        pass  # handle it

    return "", 200