Webhooks

// Express. This route reads the raw body: don't let express.json() parse it first.
const crypto = require('crypto');
const express = require('express');

const app = express();
const SECRET = process.env.CR3D_WEBHOOK_SECRET;            // whsec_…
const TOLERANCE_SECONDS = 300;

function verifySignature(header, rawBody) {
  const fields = {};
  for (const part of (header || '').split(',')) {
    const i = part.indexOf('=');
    if (i > 0) fields[part.slice(0, i).trim()] = part.slice(i + 1).trim();
  }
  const t = fields.t || '';
  if (!/^\d+$/.test(t) || !fields.v1) return false;
  if (Math.abs(Date.now() / 1000 - Number(t)) > TOLERANCE_SECONDS) return false;

  const expected = crypto.createHmac('sha256', SECRET).update(`${t}.`).update(rawBody).digest();
  const given = Buffer.from(fields.v1, 'hex');
  return given.length === expected.length && crypto.timingSafeEqual(given, expected);
}

// express.raw stops at 100 KB by default, and a platform.finished with a large result is bigger.
app.post('/webhooks/cr3dentials', express.raw({ type: 'application/json', limit: '10mb' }), (req, res) => {
  if (!Buffer.isBuffer(req.body) || !verifySignature(req.get('Zktls-Signature'), req.body)) {
    return res.status(400).send('invalid signature');
  }
  res.status(200).json({ received: true });                 // acknowledge first

  const event = JSON.parse(req.body.toString('utf8'));
  setImmediate(() => handleEvent(event));                   // yours
});

app.listen(3000);

# Flask. request.get_data() returns the raw body bytes. Run it with: flask --app <this file> run --port 3000
import hashlib
import hmac
import json
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["CR3D_WEBHOOK_SECRET"].encode()        # whsec_…
TOLERANCE_SECONDS = 300


def verify_signature(header, raw_body):
    fields = dict(
        part.strip().split("=", 1) for part in (header or "").split(",") if "=" in part
    )
    t, v1 = fields.get("t", ""), fields.get("v1", "")
    if not (t.isascii() and t.isdigit()) or not v1:
        return False
    if abs(time.time() - int(t)) > TOLERANCE_SECONDS:
        return False
    expected = hmac.new(SECRET, t.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected.encode(), v1.encode())


@app.post("/webhooks/cr3dentials")
def cr3dentials_webhook():
    raw_body = request.get_data()
    if not verify_signature(request.headers.get("Zktls-Signature"), raw_body):
        abort(400)
    event = json.loads(raw_body)
    handle_event(event)                                     # yours: keep it quick, or queue it
    return {"received": True}, 200