Webhooks

{
  "id": "evt_8sQ1mB4nZpL2xR7wT0dK",
  "type": "payment.successful",
  "mode": "test",
  "created_at": "2026-09-22T02:10:34.159Z",
  "data": {
    "object": {
      "id": "pay_lfsWb45Pf5wJ1ZGgROzH",
      "object": "payment",
      "mode": "test",
      "status": "successful",
      "amount": 10000,
      "currency": "USD",
      "captured_amount": 10000,
      "refunded_amount": 0,
      "reference": "ORD-1001",
      "route": {
        "provider": { "code": "demo_acquirer_b", "name": "NATIO Demo Acquirer B" },
        "attempts": 2,
        "rule": "Cards → Acquirer A, fallback Acquirer B"
      },
      "failure": null,
      "created_at": "2026-09-22T02:10:33.845Z"
    }
  }
}

POST /webhooks/natio HTTP/1.1
content-type: application/json
user-agent: NATIO-Webhooks/1.0
natio-signature: t=1758507034,v1=9f2c1b...c47a
natio-event-id: evt_8sQ1mB4nZpL2xR7wT0dK
natio-event-type: payment.successful
natio-delivery-id: whd_3pQ8xV2kR9mL1nW6tY4z
natio-delivery-attempt: 1

Natio-Signature: t=<unix seconds>,v1=<hex HMAC-SHA256 of "<t>.<raw body>">

import { createHmac, timingSafeEqual } from "node:crypto";

const TOLERANCE_SECONDS = 300; // 5 minutes

/**
 * Verify a Natio-Signature header against the RAW request body.
 * @param {string} rawBody  the exact bytes NATIO sent, as a string
 * @param {string} header   the value of the Natio-Signature header
 * @param {string} secret   your endpoint signing secret (whsec_...)
 */
export function verifyNatioSignature(rawBody, header, secret) {
  if (!header) return false;

  const parts = Object.fromEntries(
    header.split(",").map((kv) => {
      const i = kv.indexOf("=");
      return [kv.slice(0, i).trim(), kv.slice(i + 1).trim()];
    }),
  );

  const t = Number(parts.t);
  const v1 = parts.v1;
  if (!Number.isFinite(t) || !v1) return false;

  // Reject replays: the timestamp is signed, so it cannot be tampered with.
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - t) > TOLERANCE_SECONDS) return false;

  const expected = createHmac("sha256", secret).update(`${t}.${rawBody}`).digest("hex");

  const a = Buffer.from(expected, "utf8");
  const b = Buffer.from(v1, "utf8");
  return a.length === b.length && timingSafeEqual(a, b);
}

// ---------------------------------------------------------------------------
// Express: express.raw() keeps the body as a Buffer, so nothing re-serialises it.
// ---------------------------------------------------------------------------
import express from "express";

const app = express();
const seen = new Set(); // replace with a durable store keyed on event id

app.post("/webhooks/natio", express.raw({ type: "application/json" }), (req, res) => {
  const rawBody = req.body.toString("utf8");

  if (!verifyNatioSignature(rawBody, req.get("natio-signature"), process.env.NATIO_WEBHOOK_SECRET)) {
    return res.sendStatus(400);
  }

  const event = JSON.parse(rawBody);

  // 1. Acknowledge immediately — do not process inside the request.
  res.sendStatus(200);

  // 2. Deduplicate on the event id: the same event can arrive more than once.
  if (seen.has(event.id)) return;
  seen.add(event.id);

  // 3. Hand off to your queue.
  void enqueue(event);
});

app.listen(3000);

import hashlib
import hmac
import json
import time

TOLERANCE_SECONDS = 300  # 5 minutes


def verify_natio_signature(raw_body: bytes, header: str | None, secret: str) -> bool:
    """Verify a Natio-Signature header against the RAW request body."""
    if not header:
        return False

    parts = {}
    for kv in header.split(","):
        key, _, value = kv.partition("=")
        parts[key.strip()] = value.strip()

    try:
        t = int(parts["t"])
        v1 = parts["v1"]
    except (KeyError, ValueError):
        return False

    # Reject replays.
    if abs(int(time.time()) - t) > TOLERANCE_SECONDS:
        return False

    signed_payload = f"{t}.".encode("utf-8") + raw_body
    expected = hmac.new(secret.encode("utf-8"), signed_payload, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, v1)


# ---------------------------------------------------------------------------
# Flask: request.get_data() returns the raw bytes, before any JSON parsing.
# ---------------------------------------------------------------------------
import os

from flask import Flask, request

app = Flask(__name__)
seen = set()  # replace with a durable store keyed on event id


@app.post("/webhooks/natio")
def natio_webhook():
    raw_body = request.get_data()

    if not verify_natio_signature(raw_body, request.headers.get("Natio-Signature"), os.environ["NATIO_WEBHOOK_SECRET"]):
        return "", 400

    event = json.loads(raw_body)

    if event["id"] not in seen:
        seen.add(event["id"])
        enqueue(event)  # process asynchronously

    return "", 200

curl -X POST https://api.natio.me/v1/webhooks/test \
  -H "Authorization: Bearer natio_sk_test_..." \
  -H "Content-Type: application/json" \
  -d '{ "event_type": "payment.successful" }'