Event subscription notification

package webhook

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

// rawBody is the request body exactly as received (raw bytes).
func IsValidSignature(rawBody []byte, signatureHeader, secret string) bool {
	// Header: "t=<unix timestamp>,v1=<hmac-hex>"
	var t, v1 string
	for _, part := range strings.Split(signatureHeader, ",") {
		kv := strings.SplitN(part, "=", 2)
		if len(kv) != 2 {
			continue
		}
		switch strings.TrimSpace(kv[0]) {
		case "t":
			t = strings.TrimSpace(kv[1])
		case "v1":
			v1 = strings.TrimSpace(kv[1])
		}
	}
	if t == "" || v1 == "" {
		return false
	}

	// Headout signs "<t>.<raw body>".
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(t + "."))
	mac.Write(rawBody)
	computed := hex.EncodeToString(mac.Sum(nil))

	// Constant-time comparison.
	return hmac.Equal([]byte(computed), []byte(v1))
}

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.HexFormat;

// rawBody is the request body exactly as received (raw bytes).
boolean isValidSignature(byte[] rawBody, String signatureHeader, String secret) throws Exception {
    // Header: "t=<unix timestamp>,v1=<hmac-hex>"
    String t = null, v1 = null;
    for (String part : signatureHeader.split(",")) {
        String[] kv = part.split("=", 2);
        if (kv.length != 2) continue;
        if (kv[0].trim().equals("t")) t = kv[1].trim();
        else if (kv[0].trim().equals("v1")) v1 = kv[1].trim();
    }
    if (t == null || v1 == null) return false;

    // Headout signs "<t>.<raw body>".
    Mac mac = Mac.getInstance("HmacSHA256");
    mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
    mac.update(t.getBytes(StandardCharsets.US_ASCII));
    mac.update((byte) '.');
    byte[] computed = mac.doFinal(rawBody);
    String computedHex = HexFormat.of().formatHex(computed);

    // Constant-time comparison. Compares hex strings directly so a malformed
    // (non-hex or wrong-length) v1 fails the comparison instead of throwing.
    return MessageDigest.isEqual(computedHex.getBytes(StandardCharsets.UTF_8), v1.getBytes(StandardCharsets.UTF_8));
}

import javax.crypto.Mac
import javax.crypto.spec.SecretKeySpec
import java.nio.charset.StandardCharsets
import java.security.MessageDigest

// rawBody is the request body exactly as received (raw bytes).
fun isValidSignature(rawBody: ByteArray, signatureHeader: String, secret: String): Boolean {
    // Header: "t=<unix timestamp>,v1=<hmac-hex>"
    val parts = signatureHeader.split(",").mapNotNull {
        val kv = it.split("=", limit = 2)
        if (kv.size == 2) kv[0].trim() to kv[1].trim() else null
    }.toMap()
    val t = parts["t"] ?: return false
    val v1 = parts["v1"] ?: return false

    // Headout signs "<t>.<raw body>".
    val mac = Mac.getInstance("HmacSHA256")
    mac.init(SecretKeySpec(secret.toByteArray(StandardCharsets.UTF_8), "HmacSHA256"))
    mac.update(t.toByteArray(StandardCharsets.US_ASCII))
    mac.update('.'.code.toByte())
    val computed = mac.doFinal(rawBody).joinToString("") { "%02x".format(it) }

    // Constant-time comparison.
    return MessageDigest.isEqual(computed.toByteArray(), v1.toByteArray())
}

const crypto = require("crypto");

function isValidSignature(rawBody, signatureHeader, secret) {
  // Header: "t=<unix timestamp>,v1=<hmac-hex>"
  const parts = Object.fromEntries(
    signatureHeader.split(",").map((p) => p.split("=").map((s) => s.trim()))
  );
  const { t, v1 } = parts;
  if (!t || !v1) return false;

  // rawBody must be the raw string/Buffer as received — NOT JSON.stringify(parsedObject).
  const computed = crypto.createHmac("sha256", secret).update(`${t}.${rawBody}`).digest("hex");

  // Constant-time comparison. timingSafeEqual throws on a length mismatch, so
  // check that first — a malformed v1 should fail the comparison, not throw.
  const computedBuf = Buffer.from(computed);
  const providedBuf = Buffer.from(v1);
  if (computedBuf.length !== providedBuf.length) return false;
  return crypto.timingSafeEqual(computedBuf, providedBuf);
}

import hashlib
import hmac


def is_valid_signature(raw_body: bytes, signature_header: str, secret: str) -> bool:
    # Header: "t=<unix timestamp>,v1=<hmac-hex>"
    parts = {}
    for part in signature_header.split(","):
        key, _, value = part.partition("=")
        parts[key.strip()] = value.strip()
    t, v1 = parts.get("t"), parts.get("v1")
    if not t or not v1:
        return False

    # Headout signs "<t>.<raw body>". raw_body must be the raw bytes as received.
    signed = t.encode() + b"." + raw_body
    computed = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()

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