Webhooks

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

const TOLERANCE_SECONDS = 300;

export function verifyWamiaSignature(secret: string, header: string, rawBody: string): boolean {
  const parts = Object.fromEntries(header.split(',').map((kv) => kv.split('=') as [string, string]));
  const timestamp = Number(parts.t);
  if (!Number.isFinite(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('hex');
  const a = Buffer.from(expected, 'hex');
  const b = Buffer.from(parts.v1, 'hex');
  return a.length === b.length && timingSafeEqual(a, b);
}

// Express: keep the raw body — JSON re-serialisation breaks the signature.
app.post('/wamia/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  const raw = req.body.toString('utf8');
  if (!verifyWamiaSignature(process.env.WAMIA_WEBHOOK_SECRET!, req.header('Wamia-Signature') ?? '', raw)) {
    return res.status(401).end();
  }
  const event = JSON.parse(raw) as { id: string; type: string; occurredAt: string; data: Record<string, unknown> };
  // Deduplicate on event.id, then enqueue; answer 2xx within 10 seconds.
  res.status(202).end();
});

<?php
const WAMIA_TOLERANCE_SECONDS = 300;

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

// WordPress REST route, WooCommerce plugin: register_rest_route('wamia/v1', '/webhook', ...)
$raw = file_get_contents('php://input');
$header = $_SERVER['HTTP_WAMIA_SIGNATURE'] ?? '';
if (!wamia_verify_signature(getenv('WAMIA_WEBHOOK_SECRET'), $header, $raw)) {
    http_response_code(401);
    exit;
}
$event = json_decode($raw, true, 512, JSON_THROW_ON_ERROR);
// Deduplicate on $event['id'], queue the job (Action Scheduler), answer 202 at once.
http_response_code(202);

import hmac, hashlib, time

TOLERANCE_SECONDS = 300

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

# Odoo controller: @http.route('/wamia/webhook', type='http', auth='public', methods=['POST'], csrf=False)
def wamia_webhook(self, **kwargs):
    raw = request.httprequest.get_data()
    header = request.httprequest.headers.get("Wamia-Signature", "")
    if not verify_wamia_signature(secret, header, raw):
        return Response(status=401)
    event = json.loads(raw)
    # Deduplicate on event["id"], enqueue with queue_job, answer 202 at once.
    return Response(status=202)