Webhooks

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

const app = express();
const SECRET = process.env.MEGA_WEBHOOK_SECRET;
const TOLERANCE_SECONDS = 5 * 60;

// Capture the RAW body — signature is over the exact bytes we received.
app.post(
  "/hooks/mega-leads",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const signature = req.get("X-Mega-Signature") || "";
    const timestamp = req.get("X-Mega-Timestamp") || "";
    const rawBody = req.body.toString("utf8");

    // Replay guard: reject stale timestamps.
    const age = Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp));
    if (!timestamp || Number.isNaN(age) || age > TOLERANCE_SECONDS) {
      return res.status(400).send("stale or missing timestamp");
    }

    const expected =
      "sha256=" +
      crypto
        .createHmac("sha256", SECRET)
        .update(`${timestamp}.${rawBody}`)
        .digest("hex");

    const ok =
      signature.length === expected.length &&
      crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected));
    if (!ok) return res.status(401).send("bad signature");

    const event = JSON.parse(rawBody);
    // TODO: de-dupe on req.get("X-Mega-Delivery"), then process event.lead
    res.status(200).send("ok");
  }
);

import hashlib, hmac, time
from flask import Flask, request, abort

app = Flask(__name__)
SECRET = b"mega_whsec_...."   # from your secret manager
TOLERANCE_SECONDS = 5 * 60

@app.post("/hooks/mega-leads")
def mega_leads():
    signature = request.headers.get("X-Mega-Signature", "")
    timestamp = request.headers.get("X-Mega-Timestamp", "")
    raw_body = request.get_data()  # exact bytes

    try:
        age = abs(int(time.time()) - int(timestamp))
    except ValueError:
        abort(400)
    if age > TOLERANCE_SECONDS:
        abort(400)  # replay guard

    expected = "sha256=" + hmac.new(
        SECRET, f"{timestamp}.".encode() + raw_body, hashlib.sha256
    ).hexdigest()

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

    event = request.get_json()
    # TODO: de-dupe on request.headers["X-Mega-Delivery"], then process event["lead"]
    return "ok", 200