Webhooks.

{
  "event": "message.created",
  "id": "evt_8a1d8e2c-8f2a-4a8b-9c0e-6a1d9e2c3f4b",
  "created_at": "2026-04-17T14:02:31.000Z",
  "organization_id": "5515fbe0-745b-4b01-9ed4-828767c84524",
  "data": {
    "message": {
      "id": "msg_01HYT4...",
      "conversation_id": "cnv_01HYT3...",
      "sender_type": "visitor",
      "sender_id": "vis_01HYT2...",
      "content": "Hi — I can’t reset my password.",
      "created_at": "2026-04-17T14:02:30.842Z"
    }
  }
}

import crypto from 'node:crypto';
import express from 'express';

const app = express();

// Capture the raw body so we can verify the signature against it.
app.post(
  '/webhooks/campfire',
  express.raw({ type: 'application/json' }),
  (req, res) => {
    // Prefer the documented header; fall back to the transition header.
    const signature =
      req.get('X-Campfire-Signature') ??
      (req.get('X-Campfire-Signature-256') ?? '').replace(/^sha256=/, '');
    const expected = crypto
      .createHmac('sha256', process.env.CAMPFIRE_SIGNING_SECRET)
      .update(req.body) // Buffer of the raw body
      .digest('hex');

    const sigBuf = Buffer.from(signature, 'utf8');
    const expBuf = Buffer.from(expected, 'utf8');
    const ok =
      sigBuf.length === expBuf.length &&
      crypto.timingSafeEqual(sigBuf, expBuf);

    if (!ok) return res.status(401).send('invalid signature');

    const event = JSON.parse(req.body.toString('utf8'));
    // ... handle event ...
    res.status(200).send('ok');
  },
);

import hashlib
import hmac
import os

from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["CAMPFIRE_SIGNING_SECRET"].encode()


@app.post("/webhooks/campfire")
def handle_campfire():
    # Prefer the documented header; fall back to the transition header.
    signature = request.headers.get("X-Campfire-Signature", "")
    if not signature:
        signature = request.headers.get("X-Campfire-Signature-256", "")
        signature = signature.removeprefix("sha256=")

    # Use the raw bytes. Do not re-serialize the parsed JSON.
    expected = hmac.new(SECRET, request.get_data(), hashlib.sha256).hexdigest()

    if not hmac.compare_digest(signature, expected):
        abort(401)

    event = request.get_json()
    # ... handle event ...
    return "ok", 200