Signature Verification

import base64
import hashlib
import hmac
import json
import time

class WebhookVerificationError(Exception):
    pass

def verify_webhook(raw_body: bytes, headers: dict, secret: str) -> dict:
    msg_id        = headers["webhook-id"]
    msg_timestamp = headers["webhook-timestamp"]
    msg_signature = headers["webhook-signature"]

    # Reject stale payloads (older than 5 minutes)
    if abs(time.time() - int(msg_timestamp)) > 300:
        raise WebhookVerificationError("Timestamp too old")

    # Construct the signed content
    to_sign = f"{msg_id}.{msg_timestamp}.{raw_body.decode()}".encode()

    # Decode the signing secret (strip "whsec_" prefix, then base64-decode)
    key = base64.b64decode(secret.removeprefix("whsec_"))

    # Compute expected signature
    expected = base64.b64encode(
        hmac.new(key, to_sign, hashlib.sha256).digest()
    ).decode()

    # Compare against all provided signatures (space-separated, each prefixed "v1,")
    provided = [s.split(",", 1)[1] for s in msg_signature.split() if s.startswith("v1,")]
    if not any(hmac.compare_digest(expected, s) for s in provided):
        raise WebhookVerificationError("Invalid signature")

    return json.loads(raw_body)

const crypto = require("crypto");

class WebhookVerificationError extends Error {}

function verifyWebhook(rawBody, headers, secret) {
  const msgId        = headers["webhook-id"];
  const msgTimestamp = headers["webhook-timestamp"];
  const msgSignature = headers["webhook-signature"];

  // Reject stale payloads (older than 5 minutes)
  if (Math.abs(Date.now() / 1000 - parseInt(msgTimestamp)) > 300) {
    throw new WebhookVerificationError("Timestamp too old");
  }

  // Construct the signed content
  const toSign = `${msgId}.${msgTimestamp}.${rawBody}`;

  // Decode the signing secret (strip "whsec_" prefix, then base64-decode)
  const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");

  // Compute expected signature
  const expected = crypto.createHmac("sha256", key).update(toSign).digest("base64");

  // Compare against all provided signatures (space-separated, each prefixed "v1,")
  const provided = msgSignature.split(" ")
    .filter(s => s.startsWith("v1,"))
    .map(s => s.slice(3));

  if (!provided.some(s => crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(s)))) {
    throw new WebhookVerificationError("Invalid signature");
  }

  return JSON.parse(rawBody);
}

require "base64"
require "json"
require "openssl"

class WebhookVerificationError < StandardError; end

def verify_webhook(raw_body, headers, secret)
  msg_id        = headers["webhook-id"]
  msg_timestamp = headers["webhook-timestamp"]
  msg_signature = headers["webhook-signature"]

  # Reject stale payloads (older than 5 minutes)
  raise WebhookVerificationError, "Timestamp too old" if (Time.now.to_i - msg_timestamp.to_i).abs > 300

  # Construct the signed content
  to_sign = "#{msg_id}.#{msg_timestamp}.#{raw_body}"

  # Decode the signing secret (strip "whsec_" prefix, then base64-decode)
  key = Base64.strict_decode64(secret.delete_prefix("whsec_"))

  # Compute expected signature
  expected = Base64.strict_encode64(OpenSSL::HMAC.digest("sha256", key, to_sign))

  # Compare against all provided signatures (space-separated, each prefixed "v1,")
  provided = msg_signature.split.filter_map { |s| s.delete_prefix("v1,") if s.start_with?("v1,") }
  raise WebhookVerificationError, "Invalid signature" unless provided.any? { 
    |s| s.bytesize == expected.bytesize && OpenSSL.fixed_length_secure_compare(expected, s)
  }

  JSON.parse(raw_body)
end

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/base64"
    "encoding/json"
    "errors"
    "fmt"
    "math"
    "net/http"
    "strconv"
    "strings"
    "time"
)

var ErrWebhookVerification = errors.New("webhook verification failed")

func verifyWebhook(body []byte, headers http.Header, secret string) (map[string]any, error) {
    msgID        := headers.Get("webhook-id")
    msgTimestamp := headers.Get("webhook-timestamp")
    msgSignature := headers.Get("webhook-signature")

    // Reject stale payloads (older than 5 minutes)
    ts, err := strconv.ParseInt(msgTimestamp, 10, 64)
    if err != nil || math.Abs(float64(time.Now().Unix()-ts)) > 300 {
        return nil, fmt.Errorf("%w: stale timestamp", ErrWebhookVerification)
    }

    // Construct the signed content
    toSign := fmt.Sprintf("%s.%s.%s", msgID, msgTimestamp, body)

    // Decode the signing secret (strip "whsec_" prefix, then base64-decode)
    key, err := base64.StdEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_"))
    if err != nil {
        return nil, err
    }

    // Compute expected signature
    mac := hmac.New(sha256.New, key)
    mac.Write([]byte(toSign))
    expected := base64.StdEncoding.EncodeToString(mac.Sum(nil))

    // Compare against all provided signatures (space-separated, each prefixed "v1,")
    for _, sig := range strings.Fields(msgSignature) {
        if strings.HasPrefix(sig, "v1,") {
            if hmac.Equal([]byte(expected), []byte(strings.TrimPrefix(sig, "v1,"))) {
                var payload map[string]any
                return payload, json.Unmarshal(body, &payload)
            }
        }
    }
    return nil, ErrWebhookVerification
}