Validação de assinatura

signature = "sha256=" + HMAC_SHA256(key=webhookSecret, data=rawBody).hexdigest()

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

export function verifyWebhook(
  signatureHeader: string,
  rawBody: Buffer,
  secret: string,
): boolean {
  // Header format: "sha256=<hex>"
  const hex = signatureHeader.startsWith('sha256=')
    ? signatureHeader.slice(7)
    : signatureHeader;

  const expected = createHmac('sha256', secret).update(rawBody).digest('hex');

  const sigBuf = Buffer.from(hex, 'hex');
  const expBuf = Buffer.from(expected, 'hex');

  if (sigBuf.length !== expBuf.length) return false;
  return timingSafeEqual(sigBuf, expBuf);
}

import express from 'express';

app.post(
  '/webhooks/astronpay',
  express.raw({ type: 'application/json' }),
  (req, res) => {
    const signature = req.header('X-Astronpay-Signature') ?? '';
    if (!verifyWebhook(signature, req.body, process.env.ASTRONPAY_WEBHOOK_SECRET!)) {
      return res.status(401).send('invalid signature');
    }
    const event = JSON.parse(req.body.toString('utf8'));
    // ... processar de forma assíncrona
    res.sendStatus(200);
  },
);

import hmac
import hashlib
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
SECRET = b"meu-webhook-secret"

@app.post("/webhooks/astronpay")
async def handler(request: Request):
    raw = await request.body()
    sig_header = request.headers.get("x-astronpay-signature", "")
    hex_received = sig_header.removeprefix("sha256=")

    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(hex_received, expected):
        raise HTTPException(status_code=401, detail="invalid signature")

    event = await request.json()
    # ... processar
    return {"ok": True}