Test Stripe webhooks locally ngrok, localhost, webhook replay

// Minimal receiver: verify stripe-signature (t=..., v1=...) over raw body
// npm i express
import crypto from "node:crypto";
import express from "express";

const app = express();

app.post(
  "/webhook",
  express.raw({ type: "*/*", limit: "2mb" }),
  (req, res) => {
    const secret = process.env.WEBHOOK_SECRET ?? "replace_me";
    const sigHeader = req.header("stripe-signature") ?? "";
    const rawBody = Buffer.isBuffer(req.body) ? req.body : Buffer.from(req.body ?? "");
    const payload = rawBody.toString("utf8");

    if (!sigHeader) return res.status(401).send("missing signature header");

    // Parse: t=123,v1=sig,v0=sig...
    const items = sigHeader.split(",").map((p) => p.trim().split("="));
    const t = items.find((i) => i[0] === "t")?.[1];
    const v1 = items.find((i) => i[0] === "v1")?.[1];
    if (!t || !v1) return res.status(401).send("malformed signature header");

    // Timestamp max-age (replay protection)
    const tolerance = 300;
    const now = Math.floor(Date.now() / 1000);
    const ts = Number.parseInt(t, 10);
    if (!Number.isFinite(ts)) return res.status(401).send("bad timestamp");
    if (Math.abs(now - ts) > tolerance) return res.status(401).send("stale timestamp");

    // Expected: HMAC-SHA256(secret, \`\${t}.\${payload}\`) (hex)
    const toSign = \`\${t}.\${payload}\`;
    const expected = crypto.createHmac("sha256", secret).update(toSign, "utf8").digest("hex");

    const a = Buffer.from(v1);
    const b = Buffer.from(expected);
    if (a.length !== b.length) return res.status(401).send("bad signature");
    if (!crypto.timingSafeEqual(a, b)) return res.status(401).send("bad signature");

    res.sendStatus(202);
  }
);

app.listen(3000, () => console.log("listening on :3000"));

# Minimal receiver: verify stripe-signature (t=..., v1=...) over raw body
# pip install fastapi uvicorn
import hashlib
import hmac
import os
import time
from fastapi import FastAPI, Request, Response

app = FastAPI()

def parse_sig_header(value: str) -> dict:
    parts = {}
    for part in value.split(","):
        part = part.strip()
        if "=" not in part:
            continue
        k, v = part.split("=", 1)
        parts[k] = v
    return parts

@app.post("/webhook")
async def webhook(request: Request) -> Response:
    secret = os.getenv("WEBHOOK_SECRET", "replace_me").encode("utf-8")
    sig_header = request.headers.get("stripe-signature", "")
    raw_body = await request.body()
    payload = raw_body.decode("utf-8")

    if not sig_header:
        return Response("missing signature header", status_code=401)

    parts = parse_sig_header(sig_header)
    t = parts.get("t")
    v1 = parts.get("v1")
    if not t or not v1:
        return Response("malformed signature header", status_code=401)

    # Timestamp max-age (replay protection)
    tolerance = 300
    now = int(time.time())
    try:
        ts = int(t)
    except Exception:
        return Response("bad timestamp", status_code=401)
    if abs(now - ts) > tolerance:
        return Response("stale timestamp", status_code=401)

    # Expected: HMAC-SHA256(secret, f"{t}.{payload}") (hex)
    to_sign = f"{t}.{payload}".encode("utf-8")
    expected = hmac.new(secret, to_sign, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(expected, v1):
        return Response("bad signature", status_code=401)

    return Response(status_code=202)

# Run: uvicorn app:app --reload --port 3000

// Minimal local consumer loop (REST pull):
// 1) GET .../next
// 2) parse X-Hooque-Meta (ackUrl/nackUrl/rejectUrl)
// 3) POST ackUrl on success, otherwise nack/reject
//
// Node 18+ (fetch built-in)
const QUEUE_NEXT_URL =
  process.env.HOOQUE_QUEUE_NEXT_URL ??
  "https://app.hooque.io/queues/cons_dev_webhooks/next";
const TOKEN = process.env.HOOQUE_TOKEN ?? "hq_tok_replace_me";

const headers = { Authorization: `Bearer ${TOKEN}` };

async function processLocally(payload) {
  // TODO: call your local code paths here
  console.log("received event:", payload?.type ?? payload?.id ?? "event");
}

while (true) {
  const resp = await fetch(QUEUE_NEXT_URL, { headers });

  if (resp.status === 204) break; // queue is empty
  if (!resp.ok) throw new Error(`Hooque next() failed: ${resp.status}`);

  const payload = await resp.json();
  const meta = JSON.parse(resp.headers.get("X-Hooque-Meta") ?? "{}");

  try {
    await processLocally(payload);
    await fetch(meta.ackUrl, { method: "POST", headers });
  } catch (err) {
    const reason = String(err);
    const isPermanent = err?.name === "ValidationError";
    const url = isPermanent ? meta.rejectUrl : meta.nackUrl;

    await fetch(url, {
      method: "POST",
      headers: { ...headers, "Content-Type": "application/json" },
      body: JSON.stringify({ reason }),
    });
  }
}

# Minimal local consumer loop (REST pull):
# 1) GET .../next
# 2) parse X-Hooque-Meta (ackUrl/nackUrl/rejectUrl)
# 3) POST ackUrl on success, otherwise nack/reject
import json
import os
import requests

QUEUE_NEXT_URL = os.getenv(
    "HOOQUE_QUEUE_NEXT_URL",
    "https://app.hooque.io/queues/cons_dev_webhooks/next",
)
TOKEN = os.getenv("HOOQUE_TOKEN", "hq_tok_replace_me")
headers = {"Authorization": f"Bearer {TOKEN}"}

def process_locally(payload: dict) -> None:
    print("received event:", payload.get("type") or payload.get("id") or "event")
    # TODO: call your local code paths here

while True:
    resp = requests.get(QUEUE_NEXT_URL, headers=headers, timeout=30)

    if resp.status_code == 204:
        break
    if resp.status_code >= 400:
        raise RuntimeError(f"Hooque next() failed: {resp.status_code} {resp.text}")

    payload = resp.json()
    meta = json.loads(resp.headers.get("X-Hooque-Meta", "{}"))

    try:
        process_locally(payload)
        requests.post(meta["ackUrl"], headers=headers, timeout=30)
    except Exception as err:
        reason = str(err)
        is_permanent = err.__class__.__name__ == "ValidationError"
        url = meta["rejectUrl"] if is_permanent else meta["nackUrl"]
        requests.post(
            url,
            headers={**headers, "Content-Type": "application/json"},
            json={"reason": reason},
            timeout=30,
        )