Verify signatures

import { verifyWebhook, WebhookVerificationError } from "@larynth/sdk";

const secret = process.env.LARYNTH_WEBHOOK_SECRET!;

export function handle(headers: Record<string, string>, rawBody: string) {
  try {
    verifyWebhook(secret, headers, rawBody);
  } catch (error) {
    if (error instanceof WebhookVerificationError) {
      return { status: 400 };
    }
    throw error;
  }
  const event = JSON.parse(rawBody);
  return { status: 200, event };
}

import json
import os

from larynth import WebhookVerificationError, verify

secret = os.environ["LARYNTH_WEBHOOK_SECRET"]


def handle(headers: dict[str, str], raw_body: bytes):
    try:
        verify(secret, headers, raw_body)
    except WebhookVerificationError:
        return 400, None
    return 200, json.loads(raw_body)

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

function expectedSignature(secret: string, id: string, timestamp: string, rawBody: string) {
  const key = Buffer.from(secret.replace(/^whsec_/, ""), "base64");
  const mac = createHmac("sha256", key).update(`${id}.${timestamp}.${rawBody}`).digest("base64");
  return `v1,${mac}`;
}

function isValid(
  secret: string,
  headers: Record<string, string>,
  rawBody: string,
  nowSeconds = Math.floor(Date.now() / 1000),
) {
  const id = headers["webhook-id"];
  const timestamp = headers["webhook-timestamp"];
  if (!id || !timestamp || Math.abs(nowSeconds - Number(timestamp)) > 300) return false;
  const expected = Buffer.from(expectedSignature(secret, id, timestamp, rawBody));
  return headers["webhook-signature"].split(" ").some((candidate) => {
    const got = Buffer.from(candidate);
    return got.length === expected.length && timingSafeEqual(got, expected);
  });
}

const secret = "whsec_AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=";
const body =
  '{"id":"msg_0123456789abcdef","type":"call.ended","timestamp":"2026-11-01T12:00:00Z","data":{"call_id":"c1"}}';
const headers = {
  "webhook-id": "msg_0123456789abcdef",
  "webhook-timestamp": "1793534400",
  "webhook-signature": "v1,49s/QnkINzN3EO848IVtTP/5Tpo+qBtrkQsYjg9+dhs=",
};

console.log(expectedSignature(secret, headers["webhook-id"], headers["webhook-timestamp"], body));
console.log(isValid(secret, headers, body, 1793534400));

import base64
import hashlib
import hmac
import time


def expected_signature(secret: str, msg_id: str, timestamp: str, raw_body: str) -> str:
    key = base64.b64decode(secret.removeprefix("whsec_"))
    signed = f"{msg_id}.{timestamp}.{raw_body}".encode()
    mac = hmac.new(key, signed, hashlib.sha256).digest()
    return "v1," + base64.b64encode(mac).decode()


def is_valid(secret: str, headers: dict[str, str], raw_body: str, now: float | None = None) -> bool:
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    if not msg_id or not timestamp:
        return False
    if abs((time.time() if now is None else now) - int(timestamp)) > 300:
        return False
    expected = expected_signature(secret, msg_id, timestamp, raw_body)
    return any(
        hmac.compare_digest(candidate, expected)
        for candidate in headers["webhook-signature"].split(" ")
    )


secret = "whsec_AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8="
body = '{"id":"msg_0123456789abcdef","type":"call.ended","timestamp":"2026-11-01T12:00:00Z","data":{"call_id":"c1"}}'
headers = {
    "webhook-id": "msg_0123456789abcdef",
    "webhook-timestamp": "1793534400",
    "webhook-signature": "v1,49s/QnkINzN3EO848IVtTP/5Tpo+qBtrkQsYjg9+dhs=",
}

print(expected_signature(secret, headers["webhook-id"], headers["webhook-timestamp"], body))
print(is_valid(secret, headers, body, now=1793534400))