Webhooks

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

const SECRET = process.env.LANCER_WEBHOOK_SECRET; // whsec_…
const TOLERANCE_SECONDS = 5 * 60;

function verify(rawBody, timestamp, signature) {
  if (!timestamp || !signature) return false;
  const age = Math.abs(Date.now() / 1000 - Number(timestamp));
  if (!Number.isFinite(age) || age > TOLERANCE_SECONDS) return false;
  const expected = 'sha256=' + crypto
    .createHmac('sha256', SECRET)
    .update(`${timestamp}.${rawBody}`)
    .digest('hex');
  const a = Buffer.from(expected);
  const b = Buffer.from(signature);
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

const app = express();

// Keep the raw body: verification must use the exact bytes Lancer sent.
app.post('/lancer/webhooks', express.raw({ type: 'application/json' }), (req, res) => {
  const raw = req.body.toString('utf8');
  if (!verify(raw, req.get('X-Lancer-Timestamp'), req.get('X-Lancer-Signature'))) {
    return res.status(401).send('bad signature');
  }
  const delivery = JSON.parse(raw);
  // De-duplicate on delivery.id, then hand off to a queue and answer fast.
  if (delivery.event === 'payment.succeeded') {
    console.log('paid', delivery.data.amountAgorot / 100, 'ILS');
  }
  res.sendStatus(204);
});

app.listen(3000);

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

SECRET = os.environ["LANCER_WEBHOOK_SECRET"].encode()  # whsec_…
TOLERANCE_SECONDS = 5 * 60

app = Flask(__name__)

def verify(raw: bytes, timestamp: str | None, signature: str | None) -> bool:
    if not timestamp or not signature:
        return False
    try:
        if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
            return False
    except ValueError:
        return False
    mac = hmac.new(SECRET, timestamp.encode() + b"." + raw, hashlib.sha256)
    return hmac.compare_digest("sha256=" + mac.hexdigest(), signature)

@app.post("/lancer/webhooks")
def lancer_webhook():
    raw = request.get_data()  # the exact bytes Lancer sent
    if not verify(raw, request.headers.get("X-Lancer-Timestamp"), request.headers.get("X-Lancer-Signature")):
        abort(401)
    delivery = json.loads(raw)
    # De-duplicate on delivery["id"], then hand off to a queue and answer fast.
    if delivery["event"] == "lead.created":
        print("new lead", delivery["data"]["name"])
    return "", 204