Signature dan retry webhook

signed_payload = event_id + "." + timestamp + "." + raw_body
signature      = hex(HMAC_SHA256(webhook_secret, signed_payload))
header         = "v1=" + signature

raw = request.raw_body
event_id = header("X-AntaraPay-Event-Id")
ts = header("X-AntaraPay-Timestamp")
assert abs(unix_now() - int(ts)) <= 300
message = utf8(event_id + "." + ts + ".") + raw
expected = hmac_sha256(secret, message)
assert constant_time_equal(expected, decode_signature(header_sig))
event = json_parse(raw)
assert event.event_id == event_id
persist_once(event_id, event)
return 204

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

const app = express();
const SECRET = process.env.ANTARAPAY_WEBHOOK_SECRET;

// Raw body is required: do NOT use express.json() on this route.
app.post(
  "/webhooks/antarapay",
  express.raw({ type: "application/json", limit: "256kb" }),
  async (req, res) => {
    const eventId = req.get("X-AntaraPay-Event-Id");
    const ts = req.get("X-AntaraPay-Timestamp");
    const header = req.get("X-AntaraPay-Signature") || "";

    if (!eventId || !/^[0-9]+$/.test(ts || "")) return res.sendStatus(400);
    if (Math.abs(Date.now() / 1000 - Number(ts)) > 300) return res.sendStatus(400);

    const expected = crypto
      .createHmac("sha256", SECRET)
      .update(Buffer.concat([Buffer.from(eventId + "." + ts + "."), req.body]))
      .digest();
    const given = Buffer.from(header.startsWith("v1=") ? header.slice(3) : "", "hex");
    if (given.length !== expected.length || !crypto.timingSafeEqual(given, expected)) {
      return res.sendStatus(401);
    }

    const event = JSON.parse(req.body.toString("utf8"));
    if (event.event_id !== eventId) return res.sendStatus(400);

    // Insert into a durable inbox with a UNIQUE(event_id) constraint.
    // If it already exists, still answer 2xx without reprocessing.
    await saveOnce(eventId, event);
    res.sendStatus(204);
  },
);

import hashlib
import hmac
import json
import os
import time

from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["ANTARAPAY_WEBHOOK_SECRET"].encode()


@app.post("/webhooks/antarapay")
def antarapay_webhook():
    raw = request.get_data()  # bytes, before any JSON parsing
    event_id = request.headers.get("X-AntaraPay-Event-Id", "")
    ts = request.headers.get("X-AntaraPay-Timestamp", "")
    header = request.headers.get("X-AntaraPay-Signature", "")

    if not event_id or not ts.isdigit():
        abort(400)
    if abs(time.time() - int(ts)) > 300:
        abort(400)

    message = (event_id + "." + ts + ".").encode() + raw
    expected = "v1=" + hmac.new(SECRET, message, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, header):
        abort(401)

    event = json.loads(raw)
    if event["event_id"] != event_id:
        abort(400)

    save_once(event_id, event)  # UNIQUE(event_id); 2xx if already stored
    return "", 204