Webhooks

import crypto from "crypto";

export function verifyLumeoSignature(
  rawBody: string,
  signature: string,
  secret: string,
): boolean {
  const expected = crypto
    .createHmac("sha256", secret)
    .update(rawBody, "utf8")
    .digest("hex");
  return crypto.timingSafeEqual(
    Buffer.from(expected),
    Buffer.from(signature),
  );
}

// Next.js App Router
export async function POST(req: Request) {
  const raw = await req.text();
  const sig = req.headers.get("x-lumeo-signature") ?? "";
  if (!verifyLumeoSignature(raw, sig, process.env.LUMEO_WEBHOOK_SECRET!)) {
    return new Response("Forbidden", { status: 403 });
  }
  const event = JSON.parse(raw);
  // handle event.type ...
  return new Response("ok");
}

import hashlib
import hmac

def verify_lumeo_signature(raw_body: bytes, signature: str, secret: str) -> bool:
    expected = hmac.new(
        secret.encode("utf-8"),
        raw_body,
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

# Flask example
from flask import Flask, request, abort
import json

app = Flask(__name__)

@app.route("/webhooks/lumeo", methods=["POST"])
def lumeo_webhook():
    raw = request.get_data()
    sig = request.headers.get("X-Lumeo-Signature", "")
    if not verify_lumeo_signature(raw, sig, os.environ["LUMEO_WEBHOOK_SECRET"]):
        abort(403)
    event = json.loads(raw)
    # handle event["type"] ...
    return "ok", 200

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "net/http"
)

func verifyLumeoSignature(body []byte, signature, secret string) bool {
    mac := hmac.New(sha256.New, []byte(secret))
    mac.Write(body)
    expected := hex.EncodeToString(mac.Sum(nil))
    return hmac.Equal([]byte(expected), []byte(signature))
}