Verifying Webhooks

use base64::{engine::general_purpose::STANDARD, Engine};
use hmac::{Hmac, Mac};
use serde_json::Value;
use sha2::Sha256;
use subtle::ConstantTimeEq;

type HmacSha256 = Hmac<Sha256>;

fn verify_webhook(body: &Value, secret: &str, signature_header: &str) -> bool {
    // serde_jcs canonicalizes per RFC 8785 (cargo add serde_jcs)
    let canonicalized = match serde_jcs::to_string(body) {
        Ok(s) => s,
        Err(_) => return false,
    };
    let key = match STANDARD.decode(secret) {
        Ok(k) => k,
        Err(_) => return false,
    };
    let mut mac = HmacSha256::new_from_slice(&key).expect("HMAC accepts any key length");
    mac.update(canonicalized.as_bytes());
    let digest = hex::encode(mac.finalize().into_bytes());
    let expected = format!("sha256={digest}");

    expected.as_bytes().ct_eq(signature_header.as_bytes()).into()
}

import { createHmac, timingSafeEqual } from "crypto";
import canonicalize from "canonicalize"; // npm install canonicalize

function verifyWebhook(body, secret, signatureHeader) {
  const canonicalized = canonicalize(body);
  const key = Buffer.from(secret, "base64");
  const hmac = createHmac("sha256", key).update(canonicalized).digest("hex");
  const expected = `sha256=${hmac}`;

  if (expected.length !== signatureHeader.length) return false;
  return timingSafeEqual(Buffer.from(expected), Buffer.from(signatureHeader));
}

// Usage in an Express handler:
app.post("/webhook", (req, res) => {
  const signature = req.headers["x-signature"];
  if (!verifyWebhook(req.body, process.env.WEBHOOK_SECRET, signature)) {
    return res.status(401).send("Invalid signature");
  }
  // Process the event...
  res.sendStatus(200);
});

import hmac, hashlib, base64
from canonicaljson import encode_canonical_json  # pip install canonicaljson

def verify_webhook(body: dict, secret: str, signature_header: str) -> bool:
    canonicalized = encode_canonical_json(body)
    key = base64.b64decode(secret)
    digest = hmac.new(key, canonicalized, hashlib.sha256).hexdigest()
    expected = f"sha256={digest}"
    return hmac.compare_digest(expected, signature_header)