Verify and route webhook events safely

import express from "express";

const app = express();

app.post(
  "/blendfi-webhooks",
  express.raw({ type: "application/json" }),
  (req, res) => {
    res.sendStatus(200);
  },
);

app.listen(3000);

from flask import Flask, request

app = Flask(__name__)

@app.post("/blendfi-webhooks")
def handle():
    return "", 200

if __name__ == "__main__":
    app.run(port=3000)

import crypto from "node:crypto";

const SECRET = process.env.BLENDFI_WEBHOOK_SECRET;

function verify(req) {
  const sigHdr = req.header("x-blendfi-signature");
  if (!sigHdr) throw new Error("missing webhook headers");

  // Parse "t=<unix>,v1=<hex>" into its components.
  const parts = Object.fromEntries(
    sigHdr.split(",").map((p) => p.split("=", 2)),
  );
  const ts = parts.t;
  const provided = parts.v1;
  if (!ts || !provided) throw new Error("malformed signature header");

  // Replay window, 300s per BlendFi convention
  const skew = Math.abs(Date.now() / 1000 - Number(ts));
  if (skew > 300) throw new Error("timestamp outside 300s window");

  // Reconstruct signed payload from RAW body bytes
  const signedPayload = `${ts}.${req.body.toString("utf8")}`;
  const expected = crypto
    .createHmac("sha256", SECRET)
    .update(signedPayload)
    .digest("hex");

  const a = Buffer.from(provided);
  const b = Buffer.from(expected);
  if (!(a.length === b.length && crypto.timingSafeEqual(a, b))) {
    throw new Error("signature mismatch");
  }
}

import hmac, hashlib, time, os

SECRET = os.environ["BLENDFI_WEBHOOK_SECRET"].encode()

def verify(request):
    sig_hdr = request.headers.get("X-Blendfi-Signature", "")
    if not sig_hdr:
        raise ValueError("missing webhook headers")

    # Parse "t=<unix>,v1=<hex>" into its components.
    parts = dict(p.split("=", 1) for p in sig_hdr.split(",") if "=" in p)
    ts = parts.get("t")
    provided = parts.get("v1")
    if not ts or not provided:
        raise ValueError("malformed signature header")

    # Replay window
    if abs(time.time() - int(ts)) > 300:
        raise ValueError("timestamp outside 300s window")

    raw = request.get_data()  # raw bytes; do not use request.json
    signed_payload = f"{ts}.".encode() + raw
    expected = hmac.new(SECRET, signed_payload, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(provided, expected):
        raise ValueError("signature mismatch")

import Redis from "ioredis";
const redis = new Redis(process.env.REDIS_URL);

async function alreadyProcessed(eventId) {
  // SET NX with TTL = "claim if not already claimed"
  const claimed = await redis.set(`webhook:${eventId}`, "1", "EX", 600, "NX");
  return claimed === null; // null means key existed → already processed
}

import redis as redislib
redis = redislib.Redis.from_url(os.environ["REDIS_URL"])

def already_processed(event_id: str) -> bool:
    # SET NX with TTL = "claim if not already claimed"
    claimed = redis.set(f"webhook:{event_id}", "1", ex=600, nx=True)
    return claimed is None

import { Queue } from "bullmq";
const queue = new Queue("blendfi-webhooks", { connection: { url: process.env.REDIS_URL } });

app.post(
  "/blendfi-webhooks",
  express.raw({ type: "application/json" }),
  async (req, res) => {
    try {
      verify(req);
    } catch (e) {
      return res.status(401).send(e.message);
    }

    const id = req.header("x-blendfi-event-id");
    if (await alreadyProcessed(id)) {
      return res.sendStatus(200); // dedup; no-op replay
    }

    res.sendStatus(200); // ack first
    await queue.add("event", JSON.parse(req.body.toString("utf8")));
  },
);

from rq import Queue
from redis import Redis as RedisSync

queue = Queue("blendfi-webhooks", connection=RedisSync.from_url(os.environ["REDIS_URL"]))

@app.post("/blendfi-webhooks")
def handle():
    try:
        verify(request)
    except ValueError as e:
        return str(e), 401

    event_id = request.headers["X-Blendfi-Event-Id"]
    if already_processed(event_id):
        return "", 200

    queue.enqueue("worker.process_event", request.json)
    return "", 200

import { Worker } from "bullmq";

const handlers = {
  "conversion.created": (data) => trackPending(data.id),
  "conversion.completed": (data) => markOrderDelivered(data.id),
  "conversion.failed": (data) => alertOps(data.id, data.failure_reason),
  "conversion.standby": (data) => triageStandby(data.id, data.standby_reason),
  "conversion.abandoned": (data) => escalateToOps(data.id),
  "user.kyc_approved": (data) => unlockUser(data.id),
  "user.kyc_rejected": (data) => notifyUser(data.id, data.latest_submission.rejection_reason),
};

new Worker(
  "blendfi-webhooks",
  async (job) => {
    const event = job.data;
    const handler = handlers[event.type];
    if (handler) await handler(event.data.object);
    // Unknown types are silently OK, additive event additions in the future.
  },
  { connection: { url: process.env.REDIS_URL } },
);

def process_event(event: dict):
    handlers = {
        "conversion.created": lambda d: track_pending(d["id"]),
        "conversion.completed": lambda d: mark_order_delivered(d["id"]),
        "conversion.failed": lambda d: alert_ops(d["id"], d["failure_reason"]),
        "conversion.standby": lambda d: triage_standby(d["id"], d["standby_reason"]),
        "conversion.abandoned": lambda d: escalate_to_ops(d["id"]),
        "user.kyc_approved": lambda d: unlock_user(d["id"]),
        "user.kyc_rejected": lambda d: notify_user(
            d["id"], d["latest_submission"]["rejection_reason"]
        ),
    }
    handler = handlers.get(event["type"])
    if handler:
        handler(event["data"]["object"])