Webhooks

import { createHmac, timingSafeEqual } from "node:crypto";

export function verifyHeyrafikiWebhook(rawBody, signatureHeader, secret) {
  const parts = Object.fromEntries(
    signatureHeader.split(",").map((part) => part.trim().split("=", 2)),
  );
  const timestamp = Number(parts.t);
  const supplied = Buffer.from(parts.v1 ?? "", "hex");
  const expected = Buffer.from(
    createHmac("sha256", secret)
      .update(String(timestamp) + "." + rawBody)
      .digest("hex"),
    "hex",
  );

  const fresh = Math.abs(Date.now() / 1000 - timestamp) <= 300;
  return fresh && supplied.length === expected.length &&
    timingSafeEqual(supplied, expected);
}

import hashlib
import hmac
import time

def verify_heyrafiki_webhook(raw_body: bytes, header: str, secret: str) -> bool:
    parts = dict(part.strip().split("=", 1) for part in header.split(","))
    timestamp = int(parts["t"])
    signed = str(timestamp).encode() + b"." + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    fresh = abs(time.time() - timestamp) <= 300
    return fresh and hmac.compare_digest(parts.get("v1", ""), expected)

package webhooks

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "strconv"
    "strings"
    "time"
)

func Verify(rawBody []byte, header, secret string, now time.Time) bool {
    parts := map[string]string{}
    for _, part := range strings.Split(header, ",") {
        pair := strings.SplitN(strings.TrimSpace(part), "=", 2)
        if len(pair) == 2 {
            parts[pair[0]] = pair[1]
        }
    }

    timestamp, err := strconv.ParseInt(parts["t"], 10, 64)
    supplied, decodeErr := hex.DecodeString(parts["v1"])
    if err != nil || decodeErr != nil ||
        now.Sub(time.Unix(timestamp, 0)).Abs() > 5*time.Minute {
        return false
    }

    mac := hmac.New(sha256.New, []byte(secret))
    mac.Write([]byte(strconv.FormatInt(timestamp, 10) + "."))
    mac.Write(rawBody)
    return hmac.Equal(supplied, mac.Sum(nil))
}