Verifying Kalba.ai webhook signatures

Every webhook POST carries:

Kalba-Signature: t=<unix seconds>,v1=<hex hmac_sha256(secret, t + "." + rawBody)>
Content-Type: application/json

secret is the whsec_... value shown once when you created the endpoint (POST /v1/webhook-endpoints). Verify with a tolerance of ±300 seconds and a constant-time comparison. Always verify against the RAW request body bytes — re-serializing the parsed JSON will change key order/whitespace and break the HMAC.

Respond with any 2xx within 10 seconds. Non-2xx (and redirects) are retried on a 1 m → 5 m → 30 m → 2 h → 12 h backoff, then marked exhausted — missed events are always replayable via GET /v1/events?after=<last seen evt_ id>.

Node / TypeScript

import { createHmac, timingSafeEqual } from "crypto"

export function verifyKalbaSignature(
  secret: string,
  signatureHeader: string,
  rawBody: string,
  toleranceSec = 300
): boolean {
  const parts: Record<string, string> = {}
  for (const part of signatureHeader.split(",")) {
    const [k, v] = part.split("=")
    if (k && v) parts[k.trim()] = v.trim()
  }
  const t = parts["t"]
  const v1 = parts["v1"]
  if (!t || !v1 || !Number.isFinite(Number(t))) return false
  if (Math.abs(Date.now() / 1000 - Number(t)) > toleranceSec) return false

  const expected = createHmac("sha256", secret).update(`${t}.${rawBody}`).digest("hex")
  const a = Buffer.from(v1, "utf8")
  const b = Buffer.from(expected, "utf8")
  return a.length === b.length && timingSafeEqual(a, b)
}

// Express example — note express.raw() to keep the raw bytes:
import express from "express"
const app = express()
app.post("/kalba-hook", express.raw({ type: "application/json" }), (req, res) => {
  const ok = verifyKalbaSignature(
    process.env.KALBA_WEBHOOK_SECRET!,
    req.header("Kalba-Signature") ?? "",
    req.body.toString("utf8")
  )
  if (!ok) return res.status(401).send("bad signature")
  const event = JSON.parse(req.body.toString("utf8"))
  // ... handle event.type / event.data (dedupe by event.id) ...
  res.status(200).send("ok")
})

Python

import hashlib
import hmac
import time


def verify_kalba_signature(secret: str, signature_header: str, raw_body: bytes,
                           tolerance_sec: int = 300) -> bool:
    parts = {}
    for part in signature_header.split(","):
        if "=" in part:
            k, v = part.split("=", 1)
            parts[k.strip()] = v.strip()
    t, v1 = parts.get("t"), parts.get("v1")
    if not t or not v1 or not t.isdigit():
        return False
    if abs(time.time() - int(t)) > tolerance_sec:
        return False
    expected = hmac.new(secret.encode(), f"{t}.".encode() + raw_body,
                        hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, v1)


# Flask example — request.get_data() returns the raw bytes:
# from flask import Flask, request
# app = Flask(__name__)
#
# @app.post("/kalba-hook")
# def kalba_hook():
#     if not verify_kalba_signature(KALBA_WEBHOOK_SECRET,
#                                   request.headers.get("Kalba-Signature", ""),
#                                   request.get_data()):
#         return "bad signature", 401
#     event = request.get_json()
#     # ... handle event["type"] / event["data"] (dedupe by event["id"]) ...
#     return "ok", 200

Notes

• No ordering guarantee — order/dedupe by event.id / event.created.
• Test-mode endpoints only ever receive "mode": "test" events; live endpoints only live.
• Unknown event type values must be ignored, not errored (additive versioning).