Webhooks in depth

# Change what an endpoint receives
curl -X PATCH https://api.vocatech.com/v1/webhooks/42 \
  -H "Authorization: Bearer $VOCATECH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{ "event_filters": ["call.ended", "call.transcription"] }'

# Pause it: new events stop
curl -X PATCH https://api.vocatech.com/v1/webhooks/42 \
  -H "Authorization: Bearer $VOCATECH_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{ "enabled": false }'

# Deliveries that failed, newest first
curl "https://api.vocatech.com/v1/webhooks/failures?limit=20" \
  -H "Authorization: Bearer $VOCATECH_API_KEY"

# Delete it: its pending retries stop too
curl -X DELETE https://api.vocatech.com/v1/webhooks/42 \
  -H "Authorization: Bearer $VOCATECH_API_KEY"

import { vt } from './vocatech.mjs';

// Change what an endpoint receives
await vt('PATCH', '/v1/webhooks/42', { event_filters: ['call.ended', 'call.transcription'] });

// Pause it: new events stop
await vt('PATCH', '/v1/webhooks/42', { enabled: false });

// Deliveries that failed, newest first
const { failures, meta } = await vt('GET', '/v1/webhooks/failures?limit=20');
console.log(meta.total, failures.map((f) => [f.event_type, f.attempts, f.error_message]));

// Delete it: its pending retries stop too
await vt('DELETE', '/v1/webhooks/42');

from vocatech import vt

# Change what an endpoint receives
vt("PATCH", "/v1/webhooks/42", {"event_filters": ["call.ended", "call.transcription"]})

# Pause it: new events stop
vt("PATCH", "/v1/webhooks/42", {"enabled": False})

# Deliveries that failed, newest first
failed = vt("GET", "/v1/webhooks/failures", params={"limit": 20})
for f in failed["failures"]:
    print(f["event_type"], f["attempts"], f["error_message"])

# Delete it: its pending retries stop too
vt("DELETE", "/v1/webhooks/42")

<?php
require __DIR__ . '/vocatech.php';

// Change what an endpoint receives
vt('PATCH', '/v1/webhooks/42', ['event_filters' => ['call.ended', 'call.transcription']]);

// Pause it: new events stop
vt('PATCH', '/v1/webhooks/42', ['enabled' => false]);

// Deliveries that failed, newest first
$failed = vt('GET', '/v1/webhooks/failures?limit=20');
foreach ($failed['failures'] as $f) {
    echo $f['event_type'], ' ', $f['attempts'], ' ', $f['error_message'], PHP_EOL;
}

// Delete it: its pending retries stop too
vt('DELETE', '/v1/webhooks/42');

// verify.mjs
import crypto from 'node:crypto';

// header:  the X-Vocatech-Signature value
// rawBody: the request body as a Buffer, exactly as it arrived
// secret:  the webhook's secret_key, as the text you were given
export function verifyVocatech(header, rawBody, secret, toleranceSec = 300) {
  const parts = {};
  for (const piece of String(header || '').split(',')) {
    const at = piece.indexOf('=');
    if (at > 0) parts[piece.slice(0, at).trim()] = piece.slice(at + 1).trim();
  }
  if (!/^\d+$/.test(parts.t || '') || !parts.v1) return false;
  if (Math.abs(Date.now() / 1000 - Number(parts.t)) > toleranceSec) return false;

  const expected = crypto
    .createHmac('sha256', secret)
    .update(`t=${parts.t}.`)
    .update(rawBody)
    .digest('hex');

  const a = Buffer.from(expected, 'utf8');
  const b = Buffer.from(parts.v1, 'utf8');
  // timingSafeEqual throws when the lengths differ, so check them first.
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

# verify.py
import hashlib
import hmac
import time


def verify_vocatech(header, raw_body, secret, tolerance=300):
    """header: X-Vocatech-Signature. raw_body: bytes as received. secret: str."""
    parts = {}
    for piece in (header or "").split(","):
        key, sep, value = piece.strip().partition("=")
        if sep:
            parts[key] = value
    t, v1 = parts.get("t", ""), parts.get("v1", "")
    if not t.isdigit() or not v1:
        return False
    if abs(time.time() - int(t)) > tolerance:
        return False

    signed = b"t=" + t.encode() + b"." + raw_body  # bytes, never a decoded str
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected.encode(), v1.encode())

<?php
// verify.php
// $header: X-Vocatech-Signature. $raw: the body as received. $secret: secret_key.
function vt_verify(string $header, string $raw, string $secret, int $tolerance = 300): bool
{
    $parts = [];
    foreach (explode(',', $header) as $piece) {
        [$key, $value] = array_pad(explode('=', trim($piece), 2), 2, '');
        $parts[$key] = $value;
    }
    $t = $parts['t'] ?? '';
    $v1 = $parts['v1'] ?? '';
    if (!ctype_digit($t) || $v1 === '' || abs(time() - (int) $t) > $tolerance) {
        return false;
    }

    $expected = hash_hmac('sha256', "t={$t}.{$raw}", $secret);
    return hash_equals($expected, $v1); // constant time
}

// Express: npm install express
import express from 'express';
import { verifyVocatech } from './verify.mjs';

const app = express();
const SECRET = process.env.VOCATECH_WEBHOOK_SECRET;
const seen = new Set(); // use your database in production

// express.raw keeps the exact bytes. express.json() would parse them,
// and a re-encoded body no longer matches the signature.
app.post('/webhooks/vocatech', express.raw({ type: 'application/json' }), (req, res) => {
  const signature = req.get('X-Vocatech-Signature');
  if (!Buffer.isBuffer(req.body) || !verifyVocatech(signature, req.body, SECRET)) {
    return res.sendStatus(401);
  }
  res.sendStatus(200); // answer within 5 seconds, then work

  const event = JSON.parse(req.body.toString('utf8'));
  if (seen.has(event.id)) return; // a retry of one you already have
  seen.add(event.id);
  setImmediate(() => handle(event));
});

function handle(event) {
  console.log(event.event_type, event.data);
}

app.listen(3000);

# Flask: pip install flask
import json
import os
import threading

from flask import Flask, abort, request
from verify import verify_vocatech

app = Flask(__name__)
SECRET = os.environ["VOCATECH_WEBHOOK_SECRET"]
seen = set()  # use your database in production


@app.post("/webhooks/vocatech")
def vocatech_webhook():
    raw = request.get_data()  # the exact bytes, before any parsing
    if not verify_vocatech(request.headers.get("X-Vocatech-Signature"), raw, SECRET):
        abort(401)

    event = json.loads(raw)
    if event["id"] not in seen:  # skip a retry of one you already have
        seen.add(event["id"])
        threading.Thread(target=handle, args=(event,)).start()
    return "", 200  # answer within 5 seconds, then work


def handle(event):
    print(event["event_type"], event["data"])

<?php
// Laravel, routes/api.php (API routes skip the CSRF check)
use App\Jobs\HandleVocatechEvent;
use Illuminate\Http\Request;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\Route;

Route::post('/webhooks/vocatech', function (Request $request) {
    $raw = $request->getContent(); // the exact bytes, not $request->all()
    $ok = vt_verify(
        $request->header('X-Vocatech-Signature', ''),
        $raw,
        config('services.vocatech.webhook_secret')
    );
    if (!$ok) {
        abort(401);
    }

    $event = json_decode($raw, true);
    // Cache::add is false when the id is already there: a retry you already have.
    if (Cache::add('vocatech-event:' . $event['id'], true, now()->addDays(8))) {
        HandleVocatechEvent::dispatch($event); // your queued job does the work
    }
    return response()->noContent(); // 204, well within 5 seconds
});