Webhooks

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

const WEBHOOK_SECRET = process.env.ICN_WEBHOOK_SECRET; // whsec_... (use as-is)
const app = express();

// Keep the raw body: the signature covers the exact bytes we sent.
app.post("/webhooks/icryptonow", express.raw({ type: "application/json" }), (req, res) => {
  const ts = req.get("X-Timestamp");
  const raw = req.body.toString("utf8");
  const expected = crypto
    .createHmac("sha256", WEBHOOK_SECRET)
    .update(`${ts}\nPOST\n${req.originalUrl}\n${raw}`)
    .digest("hex");

  const given = Buffer.from(req.get("X-Signature") ?? "", "utf8");
  const fresh = Math.abs(Date.now() - Number(ts)) < 5 * 60 * 1000;
  if (!fresh || given.length !== 64 || !crypto.timingSafeEqual(given, Buffer.from(expected))) {
    return res.sendStatus(401);
  }

  const event = JSON.parse(raw);
  // Deliveries are at-least-once: skip event_ids you have already processed.
  queueForProcessing(event);
  res.sendStatus(200); // reply fast; do the work asynchronously
});

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

WEBHOOK_SECRET = os.environ["ICN_WEBHOOK_SECRET"]  # whsec_... (use as-is)
app = Flask(__name__)

@app.post("/webhooks/icryptonow")
def icryptonow_webhook():
    ts = request.headers.get("X-Timestamp", "")
    raw = request.get_data(as_text=True)  # exact bytes we sent
    path = request.full_path.rstrip("?")  # path + query of your webhook URL
    msg = f"{ts}\nPOST\n{path}\n{raw}".encode()
    expected = hmac.new(WEBHOOK_SECRET.encode(), msg, hashlib.sha256).hexdigest()

    fresh = ts.isdigit() and abs(time.time() * 1000 - int(ts)) < 5 * 60 * 1000
    if not fresh or not hmac.compare_digest(expected, request.headers.get("X-Signature", "")):
        abort(401)

    event = json.loads(raw)
    # Deliveries are at-least-once: skip event_ids you have already processed.
    queue_for_processing(event)
    return "", 200