Verificar assinatura

x-oneos-signature: t=1789482001,v1=5f0c3b…e19a2e

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

const SECRET = process.env.ONEOS_WEBHOOK_SECRET // whsec_...
const TOLERANCE_SECONDS = 300

function isValidSignature(rawBody, header) {
  const parts = Object.fromEntries(header.split(',').map((part) => part.split('=')))
  const timestamp = Number(parts.t)
  if (!Number.isInteger(timestamp) || !parts.v1) return false
  if (Math.abs(Date.now() / 1000 - timestamp) > TOLERANCE_SECONDS) return false

  const expected = createHmac('sha256', SECRET).update(`${timestamp}.${rawBody}`).digest()
  const received = Buffer.from(parts.v1, 'hex')
  return expected.length === received.length && timingSafeEqual(expected, received)
}

const app = express()

// corpo BRUTO: o HMAC é sobre os bytes exatos que chegaram
app.post('/webhooks/oneos', express.raw({ type: 'application/json' }), (req, res) => {
  const rawBody = req.body.toString('utf8')
  if (!isValidSignature(rawBody, req.get('x-oneos-signature') ?? '')) {
    return res.sendStatus(400)
  }

  const event = JSON.parse(rawBody)
  res.sendStatus(200) // responda rápido: o limite é 10 s

  enqueue(event) // processe fora do request, deduplicando por event.id
})

import hashlib
import hmac
import json
import os
import time

from flask import Flask, abort, request

SECRET = os.environ["ONEOS_WEBHOOK_SECRET"].encode()  # whsec_...
TOLERANCE_SECONDS = 300

app = Flask(__name__)


def is_valid_signature(raw_body: bytes, header: str) -> bool:
    parts = dict(part.split("=", 1) for part in header.split(",") if "=" in part)
    try:
        timestamp = int(parts["t"])
    except (KeyError, ValueError):
        return False
    if abs(time.time() - timestamp) > TOLERANCE_SECONDS:
        return False
    expected = hmac.new(SECRET, f"{timestamp}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, parts.get("v1", ""))


@app.post("/webhooks/oneos")
def oneos_webhook():
    raw_body = request.get_data()  # bytes exatos, antes de qualquer parse
    if not is_valid_signature(raw_body, request.headers.get("x-oneos-signature", "")):
        abort(400)

    event = json.loads(raw_body)
    enqueue(event)  # deduplique por event["id"]
    return "", 200

<?php
const TOLERANCE_SECONDS = 300;

function oneos_signature_is_valid(string $rawBody, string $header, string $secret): bool
{
    $parts = [];
    foreach (explode(',', $header) as $pair) {
        [$key, $value] = array_pad(explode('=', $pair, 2), 2, '');
        $parts[trim($key)] = trim($value);
    }
    if (!isset($parts['t'], $parts['v1']) || !ctype_digit($parts['t'])) {
        return false;
    }
    if (abs(time() - (int) $parts['t']) > TOLERANCE_SECONDS) {
        return false;
    }
    $expected = hash_hmac('sha256', $parts['t'] . '.' . $rawBody, $secret);
    return hash_equals($expected, $parts['v1']);
}

$rawBody = file_get_contents('php://input'); // corpo bruto
$header = $_SERVER['HTTP_X_ONEOS_SIGNATURE'] ?? '';

if (!oneos_signature_is_valid($rawBody, $header, getenv('ONEOS_WEBHOOK_SECRET'))) {
    http_response_code(400);
    exit;
}

$event = json_decode($rawBody, true);
http_response_code(200);
// enfileire $event e deduplique por $event['id']