Validando a assinatura dos webhooks

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

const app = express();
const secret = process.env.NYXPAG_WEBHOOK_SECRET;

function verify(rawBody, header, toleranceSeconds = 300) {
  const parts = Object.fromEntries(
    header.split(",").map((kv) => kv.trim().split("=")),
  );
  const { t, v1 } = parts;
  if (!t || !v1) return false;

  if (Math.abs(Date.now() / 1000 - Number(t)) > toleranceSeconds) return false;

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

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

app.post("/webhooks/nyxpag", express.raw({ type: "application/json" }), (req, res) => {
  const rawBody = req.body.toString("utf8");
  if (!verify(rawBody, req.header("X-NyxPag-Signature") ?? "")) {
    return res.status(401).send("invalid signature");
  }

  const event = JSON.parse(rawBody);
  // 1. grave o evento de forma durável  2. responda 2xx  3. processe depois
  res.sendStatus(200);
});

import hashlib
import hmac
import os
import time

from flask import Flask, request, abort

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


def verify(raw_body: bytes, header: str, tolerance: int = 300) -> bool:
    parts = dict(item.strip().split("=", 1) for item in header.split(","))
    t, v1 = parts.get("t"), parts.get("v1")
    if not t or not v1:
        return False
    if abs(time.time() - int(t)) > tolerance:
        return False

    signed = t.encode() + b"." + raw_body
    expected = hmac.new(SECRET, signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, v1)


@app.post("/webhooks/nyxpag")
def nyxpag_webhook():
    if not verify(request.get_data(), request.headers.get("X-NyxPag-Signature", "")):
        abort(401)

    event = request.get_json()
    # grave o evento de forma durável, responda 2xx e processe depois
    return "", 200

<?php
$secret = getenv('NYXPAG_WEBHOOK_SECRET');
$rawBody = file_get_contents('php://input');
$header = $_SERVER['HTTP_X_NYXPAG_SIGNATURE'] ?? '';

$parts = [];
foreach (explode(',', $header) as $item) {
    [$k, $v] = array_pad(explode('=', trim($item), 2), 2, '');
    $parts[$k] = $v;
}

$t = $parts['t'] ?? '';
$v1 = $parts['v1'] ?? '';

if ($t === '' || $v1 === '' || abs(time() - (int) $t) > 300) {
    http_response_code(401);
    exit('invalid signature');
}

$expected = hash_hmac('sha256', $t . '.' . $rawBody, $secret);

if (!hash_equals($expected, $v1)) {
    http_response_code(401);
    exit('invalid signature');
}

$event = json_decode($rawBody, true);
// grave o evento de forma durável, responda 2xx e processe depois
http_response_code(200);