API Documentation

POST https://your-server.com/hooks
Content-Type: application/json
X-Webhook-ID: 42
X-Webhook-Type: order.created
X-Webhook-Timestamp: 1712345678
X-Webhook-Signature: sha256=3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b3c4d5e6f7a8b9c0d1e2f3a4b

{
  "id": 42,
  "type": "order.created",
  "data": {
    "order_id": "ord_123",
    "amount": 4900
  },
  "timestamp": 1712345678
}

// Express webhook receiver with HMAC-SHA256 signature verification
import crypto from "crypto";
import express from "express";

const app = express();

// IMPORTANT: use raw body parser so you can compute the HMAC over the exact bytes sent
app.post("/hooks", express.raw({ type: "application/json" }), (req, res) => {
  const signature = req.headers["x-webhook-signature"] as string;
  const secret = process.env.WEBHOOK_SECRET!; // whsec_9f8e7d6c...

  if (!signature?.startsWith("sha256=")) {
    return res.status(401).json({ error: "Missing signature" });
  }

  const expected = "sha256=" + crypto
    .createHmac("sha256", secret)
    .update(req.body) // raw Buffer
    .digest("hex");

  const isValid = crypto.timingSafeEqual(
    Buffer.from(expected),
    Buffer.from(signature)
  );

  if (!isValid) {
    return res.status(401).json({ error: "Invalid signature" });
  }

  const event = JSON.parse(req.body.toString());
  console.log("Received event:", event.type, event.id);

  // Acknowledge immediately; process asynchronously
  res.status(200).json({ received: true });

  // Handle event types
  if (event.type === "order.created") {
    // ... process order
  }
});

app.listen(3000);

import hmac
import hashlib
import os
from flask import Flask, request, jsonify, abort

app = Flask(__name__)
WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"]  # whsec_9f8e7d6c...

@app.route("/hooks", methods=["POST"])
def webhook():
    signature = request.headers.get("X-Webhook-Signature", "")
    if not signature.startswith("sha256="):
        abort(401)

    raw_body = request.get_data()
    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET.encode(),
        raw_body,
        hashlib.sha256
    ).hexdigest()

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

    event = request.get_json()
    print(f"Received {event['type']} (id={event['id']})")

    # Acknowledge immediately
    return jsonify(received=True), 200

if __name__ == "__main__":
    app.run(port=3000)