Outbound webhooks

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

const app = express();
const SECRET = process.env.SENDERZ_WEBHOOK_SECRET;

function isValid(rawBody, header, secret) {
  if (typeof header !== "string") return false;
  const expected =
    "sha256=" +
    crypto.createHmac("sha256", secret).update(rawBody).digest("hex");
  const a = Buffer.from(header, "utf8");
  const b = Buffer.from(expected, "utf8");
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

app.post(
  "/hooks/senderz",
  express.raw({ type: "application/json" }),
  (req, res) => {
    if (!isValid(req.body, req.get("X-Sender-Signature"), SECRET)) {
      return res.sendStatus(401);
    }

    const { event, data, deliveryId } = JSON.parse(req.body.toString("utf8"));

    // Acknowledge first, then process out of band.
    res.sendStatus(200);
    void handle(event, data, deliveryId);
  },
);

import hashlib
import hmac
import json
import os

from flask import Flask, request

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


def is_valid(raw_body: bytes, header: str | None) -> bool:
    if not header:
        return False
    digest = hmac.new(SECRET, raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(header, f"sha256={digest}")


@app.post("/hooks/senderz")
def receive():
    raw = request.get_data()
    if not is_valid(raw, request.headers.get("X-Sender-Signature")):
        return "", 401

    payload = json.loads(raw)
    enqueue(payload["event"], payload["data"], payload["deliveryId"])
    return "", 200

<?php
$secret = getenv('SENDERZ_WEBHOOK_SECRET');
$raw = file_get_contents('php://input');
$header = $_SERVER['HTTP_X_SENDER_SIGNATURE'] ?? '';

$expected = 'sha256=' . hash_hmac('sha256', $raw, $secret);

if (!hash_equals($expected, $header)) {
    http_response_code(401);
    exit;
}

$payload = json_decode($raw, true);
http_response_code(200);
// Queue $payload['event'], $payload['data'], $payload['deliveryId'].