Vobiz Messaging Webhooks

curl -X POST "https://api.vobiz.ai/api/v1/sms/webhooks" \
  -H "X-Auth-ID: MA_XXXXXXXX" \
  -H "X-Auth-Token: YOUR_AUTH_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://example.com/webhooks/vobiz-messaging",
    "secret": "a-long-random-secret-you-generate",
    "events": ["message.received", "message.delivered", "message.failed"]
  }'

import requests

BASE = "https://api.vobiz.ai/api/v1/sms"
HEADERS = {
    "X-Auth-ID": "MA_XXXXXXXX",
    "X-Auth-Token": "YOUR_AUTH_TOKEN",
    "Content-Type": "application/json",
}

r = requests.post(
    f"{BASE}/webhooks",
    headers=HEADERS,
    json={
        "url": "https://example.com/webhooks/vobiz-messaging",
        "secret": "a-long-random-secret-you-generate",
        "events": ["message.received", "message.delivered", "message.failed"],
    },
)
r.raise_for_status()
print(r.json()["id"])

const BASE = "https://api.vobiz.ai/api/v1/sms";
const headers = {
  "X-Auth-ID": "MA_XXXXXXXX",
  "X-Auth-Token": "YOUR_AUTH_TOKEN",
  "Content-Type": "application/json",
};

const res = await fetch(`${BASE}/webhooks`, {
  method: "POST",
  headers,
  body: JSON.stringify({
    url: "https://example.com/webhooks/vobiz-messaging",
    secret: "a-long-random-secret-you-generate",
    events: ["message.received", "message.delivered", "message.failed"],
  }),
});
if (!res.ok) throw new Error(await res.text());
console.log((await res.json()).id);

import express from "express";
import crypto from "node:crypto";

const SECRET = process.env.VOBIZ_WEBHOOK_SECRET; // the secret you registered
const seen = new Set(); // replace with a shared store in production

function verify(rawBody, signatureHeader) {
  const expected = crypto.createHmac("sha256", SECRET).update(rawBody).digest("hex");
  const a = Buffer.from(expected, "utf8");
  const b = Buffer.from(signatureHeader ?? "", "utf8");
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

const app = express();

app.post(
  "/webhooks/vobiz-messaging",
  express.raw({ type: "application/json" }), // keep the raw bytes
  (req, res) => {
    if (!verify(req.body, req.get("X-Webhook-Signature"))) {
      return res.status(401).end();
    }
    const id = req.get("X-Webhook-ID");
    if (seen.has(id)) return res.status(200).end(); // redelivery
    seen.add(id);

    const envelope = JSON.parse(req.body.toString("utf8"));
    res.status(200).end(); // acknowledge quickly

    switch (req.get("X-Webhook-Event")) {
      case "message.delivered":
      case "message.failed":
      case "message.status":
        console.log(envelope.data.messageId, envelope.data.data.status);
        break;
      case "message.received":
        console.log("inbound from", envelope.data.from);
        break;
    }
  }
);

app.listen(3000);

import hmac
import hashlib
import os

from flask import Flask, request, abort

SECRET = os.environ["VOBIZ_WEBHOOK_SECRET"].encode()  # the secret you registered
seen = set()  # replace with a shared store in production

app = Flask(__name__)


def verify(raw_body: bytes, signature: str | None) -> bool:
    expected = hmac.new(SECRET, raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature or "")


@app.post("/webhooks/vobiz-messaging")
def vobiz_messaging():
    raw = request.get_data()  # raw bytes, before any JSON parsing
    if not verify(raw, request.headers.get("X-Webhook-Signature")):
        abort(401)

    event_id = request.headers.get("X-Webhook-ID")
    if event_id in seen:
        return "", 200  # redelivery
    seen.add(event_id)

    envelope = request.get_json(force=True)
    event = request.headers.get("X-Webhook-Event")
    if event in ("message.delivered", "message.failed", "message.status"):
        print(envelope["data"]["messageId"], envelope["data"]["data"]["status"])
    elif event == "message.received":
        print("inbound from", envelope["data"]["from"])
    return "", 200