Webhooks

import crypto from "node:crypto";
// Express: app.post("/webhook", express.raw({ type: "application/json" }), handler)
function verified(req, secret) {
  const [t, v1] = (req.get("X-PesaBridge-Signature-V2") || "").split(",").map((p) => p.split("=")[1]);
  if (!t || Math.abs(Date.now() / 1000 - Number(t)) > 300) return false;
  const want = crypto.createHmac("sha256", secret).update(`${t}.${req.body}`).digest("hex");
  return v1?.length === want.length && crypto.timingSafeEqual(Buffer.from(v1), Buffer.from(want));
}

import hmac, hashlib, time

def verified(raw_body: bytes, headers, secret: str) -> bool:
    parts = dict(p.split("=", 1) for p in headers.get("X-PesaBridge-Signature-V2", "").split(",") if "=" in p)
    t, v1 = parts.get("t"), parts.get("v1", "")
    if not t or abs(time.time() - int(t)) > 300:
        return False
    want = hmac.new(secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(v1, want)

<?php
$raw = file_get_contents("php://input");
parse_str(str_replace(",", "&", $_SERVER["HTTP_X_PESABRIDGE_SIGNATURE_V2"] ?? ""), $sig);
$ok = isset($sig["t"], $sig["v1"]) && abs(time() - (int)$sig["t"]) <= 300
   && hash_equals(hash_hmac("sha256", $sig["t"] . "." . $raw, getenv("PB_WHSEC")), $sig["v1"]);
http_response_code($ok ? 200 : 401);

package webhook

import ("crypto/hmac"; "crypto/sha256"; "encoding/hex")

// Verified checks the V1 signature: hex HMAC-SHA256 of the raw body.
func Verified(raw []byte, sig, secret string) bool {
    m := hmac.New(sha256.New, []byte(secret))
    m.Write(raw)
    return hmac.Equal([]byte(hex.EncodeToString(m.Sum(nil))), []byte(sig))
}