Webhooks

const crypto = require("crypto");
const express = require("express");

function verifyLacesSignature(rawBody, header, secret, toleranceSec = 300) {
  let t = null;
  const sigs = [];
  for (const part of String(header || "").split(",")) {
    const i = part.indexOf("=");
    if (i < 0) continue;
    const k = part.slice(0, i).trim();
    const v = part.slice(i + 1).trim();
    if (k === "t" && /^\d{1,12}$/.test(v)) t = Number(v);
    if (k === "v1") sigs.push(v);
  }
  if (t === null || sigs.length === 0) return false;
  if (Math.abs(Date.now() / 1000 - t) > toleranceSec) return false;

  const expected = crypto
    .createHmac("sha256", secret)
    .update(`${t}.${rawBody}`, "utf8")
    .digest("hex");
  const exp = Buffer.from(expected, "utf8");
  return sigs.some((s) => {
    const got = Buffer.from(s, "utf8");
    return got.length === exp.length && crypto.timingSafeEqual(got, exp);
  });
}

const app = express();

// express.raw keeps the body as a Buffer. Do NOT use express.json() here.
app.post("/laces/webhook", express.raw({ type: "application/json" }), (req, res) => {
  const raw = req.body.toString("utf8");
  const ok = verifyLacesSignature(
    raw,
    req.header("X-Laces-Signature"),
    process.env.LACES_WEBHOOK_SECRET,
  );
  if (!ok) return res.status(400).send("bad signature");

  const event = JSON.parse(raw);
  res.status(200).send("ok"); // acknowledge fast...
  handleEventOnce(req.header("X-Laces-Delivery"), event) // ...then do the work
    .catch((err) => logger.error("laces webhook", err));
});

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

def verify_laces_signature(raw_body: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
    t, sigs = None, []
    for part in (header or "").split(","):
        k, _, v = part.partition("=")
        k, v = k.strip(), v.strip()
        if k == "t" and v.isdigit() and len(v) <= 12:
            t = int(v)
        elif k == "v1":
            sigs.append(v)
    if t is None or not sigs:
        return False
    if abs(time.time() - t) > tolerance:
        return False
    signed = f"{t}.".encode() + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(s, expected) for s in sigs)

app = Flask(__name__)

@app.post("/laces/webhook")
def laces_webhook():
    raw = request.get_data()  # raw bytes, read BEFORE request.json
    if not verify_laces_signature(raw, request.headers.get("X-Laces-Signature", ""),
                                  os.environ["LACES_WEBHOOK_SECRET"]):
        return "bad signature", 400
    event = json.loads(raw)
    enqueue_once(request.headers["X-Laces-Delivery"], event)  # do slow work async
    return "ok", 200

<?php
function verify_laces_signature(string $rawBody, string $header, string $secret, int $tolerance = 300): bool {
    $t = null; $sigs = [];
    foreach (explode(',', $header) as $part) {
        $pos = strpos($part, '=');
        if ($pos === false) continue;
        $k = trim(substr($part, 0, $pos));
        $v = trim(substr($part, $pos + 1));
        if ($k === 't' && ctype_digit($v) && strlen($v) <= 12) $t = (int) $v;
        if ($k === 'v1') $sigs[] = $v;
    }
    if ($t === null || !$sigs) return false;
    if (abs(time() - $t) > $tolerance) return false;
    $expected = hash_hmac('sha256', $t . '.' . $rawBody, $secret);
    foreach ($sigs as $s) {
        if (hash_equals($expected, $s)) return true;
    }
    return false;
}

$raw = file_get_contents('php://input');   // the raw body, not $_POST
$ok  = verify_laces_signature($raw, $_SERVER['HTTP_X_LACES_SIGNATURE'] ?? '', getenv('LACES_WEBHOOK_SECRET'));
if (!$ok) { http_response_code(400); exit('bad signature'); }

$event = json_decode($raw, true);
// Record $_SERVER['HTTP_X_LACES_DELIVERY'] and skip it if already processed.
http_response_code(200);
echo 'ok';