Verify webhook signatures

import { Webhook } from "svix";

const secret = process.env.WEBHOOK_SIGNING_SECRET; // whsec_...

// `rawBody` must be the raw request body string, not the parsed object.
const wh = new Webhook(secret);
const payload = wh.verify(rawBody, {
  "svix-id": req.headers["svix-id"],
  "svix-timestamp": req.headers["svix-timestamp"],
  "svix-signature": req.headers["svix-signature"],
}); // throws on an invalid signature

from svix.webhooks import Webhook, WebhookVerificationError

secret = os.environ["WEBHOOK_SIGNING_SECRET"]  # whsec_...

wh = Webhook(secret)
try:
    payload = wh.verify(raw_body, dict(request.headers))
except WebhookVerificationError:
    return "", 400

import svix "github.com/svix/svix-webhooks/go"

wh, err := svix.NewWebhook(secret) // whsec_...
if err != nil {
    // handle bad secret
}
if err := wh.Verify(rawBody, req.Header); err != nil {
    http.Error(w, "invalid signature", http.StatusBadRequest)
    return
}

require "svix"

wh = Svix::Webhook.new(secret) # whsec_...
begin
  payload = wh.verify(raw_body, headers)
rescue Svix::WebhookVerificationError
  halt 400
end

use Svix\Webhook;

$wh = new Webhook($secret); // whsec_...
try {
    $payload = $wh->verify($rawBody, $headers);
} catch (\Svix\Exception\WebhookVerificationException $e) {
    http_response_code(400);
    exit;
}

import com.svix.Webhook;

Webhook webhook = new Webhook(secret); // whsec_...
webhook.verify(rawBody, headers); // throws WebhookVerificationException

const crypto = require("crypto");

function verify(rawBody, headers, secret) {
  const id = headers["svix-id"];
  const ts = headers["svix-timestamp"];
  const sigHeader = headers["svix-signature"]; // "v1,aaa v1,bbb"

  if (Math.abs(Date.now() / 1000 - Number(ts)) > 300) return false; // replay guard

  const signedContent = `${id}.${ts}.${rawBody}`;
  const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");
  const expected = crypto.createHmac("sha256", key).update(signedContent).digest("base64");

  return sigHeader.split(" ").some((part) => {
    const sig = part.split(",")[1];
    return (
      sig &&
      sig.length === expected.length &&
      crypto.timingSafeEqual(Buffer.from(sig), Buffer.from(expected))
    );
  });
}

import base64, hashlib, hmac, time

def verify(raw_body: str, headers, secret: str) -> bool:
    svix_id = headers["svix-id"]
    svix_ts = headers["svix-timestamp"]
    sig_header = headers["svix-signature"]  # "v1,aaa v1,bbb"

    if abs(time.time() - int(svix_ts)) > 300:  # replay guard
        return False

    signed_content = f"{svix_id}.{svix_ts}.{raw_body}"
    key = base64.b64decode(secret.removeprefix("whsec_"))
    expected = base64.b64encode(
        hmac.new(key, signed_content.encode(), hashlib.sha256).digest()
    ).decode()

    return any(
        hmac.compare_digest(part.split(",", 1)[1], expected)
        for part in sig_header.split()
        if "," in part
    )

func verify(rawBody, id, ts, sigHeader, secret string) bool {
    n, err := strconv.ParseInt(ts, 10, 64)
    if err != nil || math.Abs(float64(time.Now().Unix()-n)) > 300 {
        return false // replay guard
    }

    signed := id + "." + ts + "." + rawBody
    key, _ := base64.StdEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_"))
    mac := hmac.New(sha256.New, key)
    mac.Write([]byte(signed))
    expected := base64.StdEncoding.EncodeToString(mac.Sum(nil))

    for _, part := range strings.Fields(sigHeader) {
        if sig := strings.SplitN(part, ",", 2); len(sig) == 2 &&
            hmac.Equal([]byte(sig[1]), []byte(expected)) {
            return true
        }
    }
    return false
}