Webhooks guide: subscribe and verify events

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

const TOLERANCE_SECONDS = 5 * 60;

// The HMAC key is the UTF-8 bytes of lowercase_hex(sha256(rawSecret)).
// Derive it once at startup, not per request.
function deriveKey(rawSecret) {
  return crypto.createHash("sha256").update(rawSecret, "utf8").digest("hex");
}

function verifySignature(rawBody, signatureHeader, key) {
  const parts = Object.fromEntries(
    signatureHeader.split(",").map((kv) => {
      const i = kv.indexOf("=");
      return i === -1 ? [kv, ""] : [kv.slice(0, i), kv.slice(i + 1)];
    })
  );
  const t = Number(parts.t);
  const v1 = parts.v1;
  if (!Number.isInteger(t) || !v1) return false;

  // Reject deliveries older (or newer) than the tolerance window.
  if (Math.abs(Date.now() / 1000 - t) > TOLERANCE_SECONDS) return false;

  const expected = crypto
    .createHmac("sha256", key)
    .update(`${t}.${rawBody}`, "utf8")
    .digest("hex");

  // Constant-time comparison; lengths must match first.
  return (
    expected.length === v1.length &&
    crypto.timingSafeEqual(Buffer.from(expected, "utf8"), Buffer.from(v1, "utf8"))
  );
}

// Express endpoint — express.raw keeps the body bytes untouched.
const express = require("express");
const app = express();
const KEY = deriveKey(process.env.SSS_WEBHOOK_SECRET);

app.post(
  "/webhooks/signsealship",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const header = req.header("SignSealShip-Signature") ?? "";
    const rawBody = req.body.toString("utf8");

    if (!verifySignature(rawBody, header, KEY)) {
      return res.status(400).send("invalid signature");
    }

    const event = JSON.parse(rawBody);
    switch (event.type) {
      case "passport.sealed":
        // event.data.verifyUrl, event.data.sealedSha256, ...
        break;
      case "room.order_attached":
        // event.data.roomCode, event.data.orderCode, ...
        break;
      case "room.passport_sealed":
        // event.data.roomCode, event.data.passportVerifyCode, ...
        break;
    }

    res.sendStatus(200);
  }
);

app.listen(3000);

import hashlib
import hmac
import json
import os
import time

TOLERANCE_SECONDS = 5 * 60


def derive_key(raw_secret: str) -> bytes:
    """The HMAC key is the UTF-8 bytes of lowercase_hex(sha256(raw_secret))."""
    return hashlib.sha256(raw_secret.encode("utf-8")).hexdigest().encode("utf-8")


def verify_signature(raw_body: bytes, signature_header: str, key: bytes) -> bool:
    parts = dict(
        kv.split("=", 1) for kv in signature_header.split(",") if "=" in kv
    )
    try:
        t = int(parts["t"])
        v1 = parts["v1"]
    except (KeyError, ValueError):
        return False

    # Reject deliveries older (or newer) than the tolerance window.
    if abs(time.time() - t) > TOLERANCE_SECONDS:
        return False

    signed = f"{t}.".encode("utf-8") + raw_body
    expected = hmac.new(key, signed, hashlib.sha256).hexdigest()

    # Constant-time comparison.
    return hmac.compare_digest(expected, v1)


# Flask endpoint - request.get_data() returns the raw body bytes untouched.
from flask import Flask, request

app = Flask(__name__)
KEY = derive_key(os.environ["SSS_WEBHOOK_SECRET"])


@app.post("/webhooks/signsealship")
def signsealship_webhook():
    header = request.headers.get("SignSealShip-Signature", "")

    if not verify_signature(request.get_data(), header, KEY):
        return "invalid signature", 400

    event = json.loads(request.get_data())
    if event["type"] == "passport.sealed":
        pass  # event["data"]["verifyUrl"], event["data"]["sealedSha256"], ...
    elif event["type"] == "room.order_attached":
        pass  # event["data"]["roomCode"], event["data"]["orderCode"], ...
    elif event["type"] == "room.passport_sealed":
        pass  # event["data"]["roomCode"], event["data"]["passportVerifyCode"], ...

    return "", 200