Webhooks: receber e verificar

PHP

<?php
$body = file_get_contents('php://input');           // o corpo exatamente como chegou
$header = $_SERVER['HTTP_URDUME_SIGNATURE'] ?? '';
$secret = getenv('URDUME_WEBHOOK_SECRET');            // whsec_… (o valor completo)

$parts = [];
foreach (explode(',', $header) as $piece) {
    [$key, $value] = array_pad(explode('=', trim($piece), 2), 2, '');
    $parts[$key][] = $value;
}
$timestamp = (int) ($parts['t'][0] ?? 0);
$expected = hash_hmac('sha256', $timestamp . '.' . $body, $secret);
$valid = abs(time() - $timestamp) <= 300
    && array_filter($parts['v1'] ?? [], static fn ($s) => hash_equals($expected, $s)) !== [];
if (!$valid) {
    http_response_code(400);
    exit;
}

$event = json_decode($body, true);
if ($event['type'] === 'passport_session.completed') {
    $session = $event['data']['object'];
    // ex.: marcar o produto $session['metadata']['produto_id'] como "com passaporte"
}
http_response_code(200);

Node

const crypto = require('crypto');
const express = require('express');

// O corpo tem de chegar em bruto (Buffer) para a assinatura bater certo.
app.post('/webhooks/urdume', express.raw({ type: 'application/json' }), (req, res) => {
  const parts = {};
  for (const piece of (req.get('Urdume-Signature') || '').split(',')) {
    const [key, value] = piece.trim().split('=');
    (parts[key] ||= []).push(value);
  }
  const timestamp = Number(parts.t?.[0] || 0);
  const expected = crypto
    .createHmac('sha256', process.env.URDUME_WEBHOOK_SECRET)
    .update(`${timestamp}.`)
    .update(req.body)
    .digest('hex');
  const valid = Math.abs(Date.now() / 1000 - timestamp) <= 300 &&
    (parts.v1 || []).some((s) => s.length === expected.length &&
      crypto.timingSafeEqual(Buffer.from(s), Buffer.from(expected)));
  if (!valid) return res.sendStatus(400);

  const event = JSON.parse(req.body);
  if (event.type === 'passport_session.completed') {
    // ...
  }
  res.sendStatus(200);
});

Python

import hashlib, hmac, json, os, time
from flask import abort, request

@app.post("/webhooks/urdume")
def urdume_webhook():
    body = request.get_data()  # bytes, exatamente como chegaram
    parts = {}
    for piece in request.headers.get("Urdume-Signature", "").split(","):
        key, _, value = piece.strip().partition("=")
        parts.setdefault(key, []).append(value)
    timestamp = int(parts.get("t", ["0"])[0] or 0)
    expected = hmac.new(
        os.environ["URDUME_WEBHOOK_SECRET"].encode(), f"{timestamp}.".encode() + body, hashlib.sha256
    ).hexdigest()
    valid = abs(time.time() - timestamp) <= 300 and any(
        hmac.compare_digest(expected, s) for s in parts.get("v1", [])
    )
    if not valid:
        abort(400)

    event = json.loads(body)
    if event["type"] == "passport_session.completed":
        session = event["data"]["object"]
        # ...
    return "", 200