Sécurité et signature

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

const app = express();
const SECRET = process.env.PULSE_SECRET; // pulse_whsec_...

function verify(rawBody, header, secret, toleranceSec = 300) {
  if (!header) return false;
  const p = Object.fromEntries(header.split(",").map((s) => s.trim().split("=")));
  if (!p.t || !p.v1) return false;
  if (Math.abs(Date.now() / 1000 - Number(p.t)) > toleranceSec) return false; // rejoué ?
  const mac = crypto.createHmac("sha256", secret).update(`${p.t}.${rawBody}`).digest("hex");
  const a = Buffer.from(mac);
  const b = Buffer.from(p.v1);
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

// IMPORTANT : le corps BRUT, pas express.json()
app.post("/webhooks/pulse-sell", express.raw({ type: "application/json" }), (req, res) => {
  const raw = req.body.toString("utf8");
  if (!verify(raw, req.get("x-pulsesell-signature"), SECRET)) return res.sendStatus(400);

  const event = JSON.parse(raw);
  // Dédoublonnez sur event.id : un envoi peut arriver plus d'une fois
  console.log(event.event, event.id);
  res.sendStatus(200);
});

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

app = Flask(__name__)
SECRET = os.environ["PULSE_SECRET"]  # pulse_whsec_...

def verify(raw_body: bytes, header: str, secret: str, tolerance=300) -> bool:
    try:
        parts = dict(p.strip().split("=", 1) for p in header.split(","))
        t, v1 = parts["t"], parts["v1"]
    except (AttributeError, KeyError, ValueError):
        return False
    if abs(time.time() - int(t)) > tolerance:
        return False  # rejoué ?
    mac = hmac.new(secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(mac, v1)

@app.post("/webhooks/pulse-sell")
def hook():
    raw = request.get_data()  # corps brut
    if not verify(raw, request.headers.get("X-PulseSell-Signature", ""), SECRET):
        abort(400)
    event = request.get_json()
    # Dédoublonnez sur event["id"]
    return "", 200

<?php
$secret = getenv('PULSE_SECRET'); // pulse_whsec_...
$raw = file_get_contents('php://input');           // corps brut
$header = $_SERVER['HTTP_X_PULSESELL_SIGNATURE'] ?? '';

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

$fresh = ctype_digit($t) && abs(time() - (int) $t) <= 300;      // rejoué ?
$mac = hash_hmac('sha256', $t . '.' . $raw, $secret);

if (!$fresh || !hash_equals($mac, $v1)) {
    http_response_code(400);
    exit;
}

$event = json_decode($raw, true);
// Dédoublonnez sur $event['id']
http_response_code(200);