Outbound Webhooks

Node.js

import { createPublicKey, verify } from "node:crypto";

// 1. Cache the JWKS (e.g. for 5 minutes).
let jwksCache: { fetchedAt: number; keys: Map<string, ReturnType<typeof createPublicKey>> } | null = null;
async function getKey(kid: string) {
  if (!jwksCache || Date.now() - jwksCache.fetchedAt > 5 * 60_000) {
    const res = await fetch("https://sunrift-hub.com/api/public/jwks.json");
    const jwks = await res.json();
    const keys = new Map();
    for (const jwk of jwks.keys) keys.set(jwk.kid, createPublicKey({ key: jwk, format: "jwk" }));
    jwksCache = { fetchedAt: Date.now(), keys };
  }
  return jwksCache.keys.get(kid);
}

// 2. Verify each incoming delivery against the raw body.
export async function verifyWebhook(req: { headers: Record<string,string>; rawBody: Buffer }) {
  const sig = req.headers["x-hub-signature"];
  const kid = req.headers["x-hub-signature-kid"];
  const ts  = req.headers["x-hub-signature-timestamp"];
  if (!sig || !kid || !ts) return false;
  if (Math.abs(Date.now() / 1000 - Number(ts)) > 300) return false; // 5 min window
  const key = await getKey(kid);
  if (!key) return false;
  const sigBuf = Buffer.from(sig.replace(/-/g, "+").replace(/_/g, "/"), "base64");
  const msg = Buffer.from(`${ts}.${req.rawBody.toString("utf8")}`);
  return verify(null, msg, key, sigBuf);
}

Python

import time, base64, httpx
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.exceptions import InvalidSignature

_jwks_cache: dict | None = None
def get_key(kid: str):
    global _jwks_cache
    if not _jwks_cache or time.time() - _jwks_cache["t"] > 300:
        jwks = httpx.get("https://sunrift-hub.com/api/public/jwks.json").json()
        keys = {}
        for jwk in jwks["keys"]:
            x = base64.urlsafe_b64decode(jwk["x"] + "==")
            keys[jwk["kid"]] = Ed25519PublicKey.from_public_bytes(x)
        _jwks_cache = {"t": time.time(), "keys": keys}
    return _jwks_cache["keys"].get(kid)

def verify_webhook(raw_body: bytes, headers: dict) -> bool:
    sig, kid, ts = headers.get("x-hub-signature"), headers.get("x-hub-signature-kid"), headers.get("x-hub-signature-timestamp")
    if not (sig and kid and ts) or abs(time.time() - int(ts)) > 300:
        return False
    key = get_key(kid)
    if not key:
        return False
    sig_bytes = base64.urlsafe_b64decode(sig + "==")
    try:
        key.verify(sig_bytes, f"{ts}.{raw_body.decode()}".encode())
        return True
    except InvalidSignature:
        return False