Webhooks & Reports

{
  "id": "evt_7f3a9c1e2b4d6f8a0c1e2b4d",
  "event": "email.opened",
  "createdAt": "2026-09-26T10:15:00.000Z",
  "data": { "sendEventId": "…", "leadEmail": "[email protected]", "mailboxEmail": "[email protected]", "campaignId": "…", "openedAt": "…" }
}

import crypto from "node:crypto";
import express from "express";
 
const app = express();
const SECRET = process.env.SENDBOX_WEBHOOK_SECRET;
 
app.post("/sendbox-webhook", express.raw({ type: "application/json" }), (req, res) => {
  const timestamp = req.get("X-Sendbox-Timestamp");
  const header = req.get("X-Sendbox-Signature") ?? "";
  if (!timestamp || Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return res.sendStatus(400);
 
  const expected = crypto.createHmac("sha256", SECRET).update(`${timestamp}.${req.body}`).digest();
  const valid = header.split(",").some((part) => {
    const given = Buffer.from(part.trim().replace(/^v1=/, ""), "hex");
    return given.length === expected.length && crypto.timingSafeEqual(given, expected);
  });
  if (!valid) return res.sendStatus(401);
 
  const event = JSON.parse(req.body);
  // if (alreadyProcessed(event.id)) return res.sendStatus(200);
  console.log(event.event, event.data);
  res.sendStatus(200);
});

import hmac, hashlib, os, time
from flask import Flask, request, abort
 
app = Flask(__name__)
SECRET = os.environ["SENDBOX_WEBHOOK_SECRET"].encode()
 
@app.post("/sendbox-webhook")
def sendbox_webhook():
    timestamp = request.headers.get("X-Sendbox-Timestamp", "")
    header = request.headers.get("X-Sendbox-Signature", "")
    if not timestamp.isdigit() or abs(time.time() - int(timestamp)) > 300:
        abort(400)
    raw = request.get_data()
    expected = hmac.new(SECRET, timestamp.encode() + b"." + raw, hashlib.sha256).hexdigest()
    if not any(hmac.compare_digest(expected, p.strip().removeprefix("v1=")) for p in header.split(",")):
        abort(401)
    event = request.get_json()
    # skip if event["id"] was already processed
    return "", 200